fix(outbox): stop a broker outage from spending the attempt budget

[AlexeyShalaev]

Summary

A relay whose broker does not answer used to spend one attempt on every pending row every
cycle. After max_attempts cycles the whole backlog was failed, which is terminal: when
Kafka came back the relay published nothing, because fetch_and_lock_pending filters on
attempts_made < max_attempts, and the events came back only when somebody ran
requeue_failed on each row. A four-minute outage against a relay ticking every second
turned into a backlog that never drained on its own.

The attempt budget is the right tool for a row that cannot be published — a payload the
broker rejects, a topic that does not exist, a serialization error. It is the wrong tool for
a broker that is not there, which is a property of the cycle and not of any row.

What the fix is

• TransientError (omni_box.core.exceptions, re-exported at the top level) — the
  exception-shaped twin of handler_retry(count_as_attempt=False). A publisher or a handler
  raises it to say the failure is the environment's, not the event's.
• HandlerExecutionStep records it without spending an attempt and puts the row a second
  ahead. Every other exception, the handler timeout included, still counts.
• KafkaEventPublisher raises it once max_infra_retries are spent on a broker that does
  not answer: connection and node errors, request and client timeouts, and
  UnknownTopicOrPartitionError only when a forced metadata refresh fails as well — a
  topic the broker answers about and still does not know keeps counting, so a typo in a topic
  name still ends in failed. aiokafka's own retriable flag is wider than this, which is
  why the check is explicit.
• PublisherExecutionStep (new, installed by create_outbox_processor) goes two steps
  further, because everything in an outbox batch goes to the same broker: the publish timeout
  is transient too — a payload the broker cannot ack in 30 s is not the row's fault — and the
  first transient failure ends this cycle's publishing. The remaining rows are recorded the
  same way, unsent, with their schedule untouched. A dead broker costs one probe per cycle
  instead of a batch of them.

Nothing is added to any signature and no default changes for anyone whose broker is up.

Why aiokafka's errors never reached the retry loop

Measured on 0.2.0 with aiokafka 0.14.0 and the container paused: the first publish of a topic
the producer has never seen ends in UnknownTopicOrPartitionError after request_timeout_ms,
because _wait_on_metadata swallows the failed refresh and reports "no such topic"; once
metadata is cached it is RequestTimedOutError, then NodeNotReadyError. None of those
subclass the builtin ConnectionError or TimeoutError, so ErrorClassifier called them all
permanent and max_infra_retries never ran for aiokafka's own errors at all.

What I rejected

• A classifier hook on the step, or the step consulting ErrorClassifier — core cannot see
  aiokafka's classes, and it would silently change how an inbox handler raising
  ConnectionError is counted.
• Making publish return a handler result — that changes the publisher contract for every
  adapter.
• Treating UnknownTopicOrPartitionError as transient unconditionally — a missing topic would
  then never fail, and would cost a probe every cycle forever.
• Rescheduling the whole batch along with the probed row — one misclassified row would drag
  every row behind it.
• Growing backoff per row. The throttle on a dead broker is the probe itself, once per cycle;
  any row-level delay beyond a tick is pure lag after the broker returns, and growth would
  need state that outlives a cycle. If it is ever wanted, a retry_after on the exception is
  where it goes.

Type of change

• Bug fix
• New feature
• Breaking change
• Documentation update
• Refactoring / internal

Checklist

• Tests added or updated
• make check passes locally (ruff + mypy)
• CHANGELOG.md updated under [Unreleased] — n/a, release-please writes it from the commit
• Documentation updated (if the public API changed), docs/agents.md included

Verification

tests/integration/postgres/test_broker_outage.py is the report against the real outbox
table: 20 rows, max_attempts=6, a KafkaEventPublisher over a producer that raises what a
paused broker raises, and seven relay cycles. On master it fails exactly the way the report
reads — {(PENDING, 1)} after cycle 1, {(FAILED, 6)} at the end, and nothing published when
the broker comes back. A broker that answers and rejects the record still spends an attempt
per row, which the third test pins.

The reporter's own kafka_down_lab.py, unmodified, against this branch:

--- Kafka paused; the relay keeps ticking every cycle ---
  cycle 1 at   5.0 s: {'pending/attempts=0': 20}
  cycle 2 at  10.1 s: {'pending/attempts=0': 20}
  cycle 3 at  15.1 s: {'pending/attempts=0': 20}
  cycle 4 at  20.1 s: {'pending/attempts=0': 20}
  cycle 5 at  25.1 s: {'pending/attempts=0': 20}
  cycle 6 at  30.1 s: {'pending/attempts=0': 20}
  cycle 7 at  35.2 s: {'pending/attempts=0': 20}

--- Kafka is back ---
  next relay cycle: {'completed/attempts=0': 20}; messages on the topic: 20
  after requeue_failed on every failed row and one more cycle: {'completed/attempts=0': 20}; messages on the topic: 20

The outage is also 35 s instead of 240 s, because a cycle is now one probe rather than twenty.

Full gate: make check clean, make test 758 passed / 98.61% coverage, make test-integration
95 passed.

Related issues

Closes #32