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

docs/agents.md

@@ -82,7 +82,10 @@ on. Alternatively, land rows fast with no handler and drain them later with
 
 **State** is three values. `pending` → `completed` on success; `pending` → `pending` with
 `attempts_made + 1` on a counted failure; `pending` → `failed` when `attempts_made` reaches
-`max_attempts`. `failed` is terminal — nothing fetches it again.
+`max_attempts`. `failed` is terminal — nothing fetches it again. A **non-counted** failure
+leaves the row `pending` with `attempts_made` where it was: that is what a broker or a
+downstream that is not there produces, and an outage therefore costs no budget however long
+it lasts.
 
 **Locking** is a `(locked_at, locked_by)` pair on the row plus, on PostgreSQL, the row lock
 the fetching transaction holds. Two workers do not collide because of `SKIP LOCKED`; the
@@ -319,7 +322,8 @@ event, `StepResult.stop()` ends the whole batch.
 
 | Step (`omni_box.core.pipeline.steps`) | What it does |
 |---|---|
-| `HandlerExecutionStep(handler, timeout=30.0)` | awaits the handler, turns the outcome into `mark_completed` / `mark_failed` / `mark_skipped` on the context. Every processor needs one |
+| `HandlerExecutionStep(handler, timeout=30.0)` | awaits the handler, turns the outcome into `mark_completed` / `mark_failed` / `mark_skipped` on the context. Every processor needs one. A `TransientError` out of the handler is recorded without spending an attempt; a timeout counts |
+| `PublisherExecutionStep(publish, timeout=30.0)` | the outbox's handler step, installed by `create_outbox_processor`. A `TransientError` **or a timeout** is recorded without spending an attempt, and it ends this cycle's publishing: the rest of the batch is recorded the same way, unpublished and unrescheduled, so one dead broker costs one probe per cycle |
 | `SiblingDeduplicationStep(enabled=True)` | skips the event when a *completed* row shares its `(message_id, consumer_group)`. Inbox only |
 | `MetricsStep(metrics)` | emits counters and durations at `on_batch_end` |
 | `OpenTelemetryStep(service_name="omni-box")` | one span per event; needs the `opentelemetry` extra, silently inert without it |
@@ -377,7 +381,12 @@ with `isinstance(repo, PostgresInboxRepository)` where a type checker needs to s
 ### Kafka (`omni_box.infra.brokers.kafka`, extra `kafka`)
 
 `KafkaEventPublisher(producer, converter, *, max_infra_retries=3)` — you own the
-`AIOKafkaProducer` lifecycle; set `enable_idempotence=True` and `acks="all"` on it.
+`AIOKafkaProducer` lifecycle; set `enable_idempotence=True` and `acks="all"` on it. A broker
+that does not answer — a connection or node error, a request or client timeout, or a topic it
+cannot fetch metadata for while it ignores a metadata request as well — is retried
+`max_infra_retries` times and then raised as `TransientError`, which costs the row no
+attempt. Everything else, a record the broker rejects or a topic it says it does not have
+included, is raised as it is and counts.
 `KafkaEventConsumer(consumer, *, payload_loader=None, message_id_getter=None,
 event_type_getter=None, source_getter=None, envelope_parser=None)` — you own the
 `AIOKafkaConsumer` and should set `enable_auto_commit=False`. Without a
@@ -457,11 +466,17 @@ protocols, not a third implementation. `EventBatchProcessor` sets
     repositories therefore serialize the identity with `pg_advisory_xact_lock` and look it up
     before inserting: one advisory lock and one `SELECT` per insert, held until *your*
     transaction ends. One more reason to keep those transactions short.
-13. **Nothing classifies errors for you.** `ErrorClassifier` exists in `omni_box.utils` and
-    is used only inside `KafkaEventPublisher`'s own infrastructure retry. In the pipeline,
-    every exception out of a handler or a publisher is a counted failure. To spend no budget
-    on a transient error, return `handler_retry(msg, count_as_attempt=False,
-    next_retry_at=…)` — and `next_retry_at` is mandatory when `count_as_attempt=False`.
+13. **Nothing classifies errors for you; you say so, by raising or returning.** In the
+    pipeline every exception out of a handler or a publisher is a counted failure, with one
+    named exception: `TransientError`, which says the failure belongs to the environment and
+    not to the event. It is recorded without spending an attempt and the row comes back a
+    second later. Raise it from your handler or your broker adapter when the thing you were
+    talking to is not there; `KafkaEventPublisher` raises it once `max_infra_retries` are
+    spent on a broker that does not answer. The equivalent for a handler that returns rather
+    than raises is `handler_retry(msg, count_as_attempt=False, next_retry_at=…)` — and
+    `next_retry_at` is mandatory when `count_as_attempt=False`. `ErrorClassifier` still
+    classifies nothing on your behalf: it lives in `omni_box.utils` and is used only inside
+    `KafkaEventPublisher`'s own infrastructure retry.
 14. **`scheduled_at` is validated against `created_at`**: no more than 60 seconds before it,
     no more than 365 days after. A retry scheduled beyond that is rejected, not clamped.
 15. **`release_stale_locks(stale_timeout_seconds)` must use a timeout comfortably larger
@@ -579,6 +594,7 @@ Everything derives from `OmniBoxError`.
 | `EventAlreadyLockedError` | locking an event that is already locked |
 | `InvalidEventStateError` | a transition from a status that does not allow it — carries `current_status` and `expected_statuses` |
 | `EventConcurrentUpdateError` | an update touched fewer rows than expected: another worker got there first, or the row is gone. Carries `expected`, `actual`, `missing_ids` |
+| `TransientError` | the failure is the environment's, not the event's. Raised by a publisher or a handler; recorded without spending an attempt and retried next cycle |
 | `UnsupportedCapabilityError` | a maintenance call on a repository without `SupportsRetentionPolicies` |
 | `InboxPersistError` | the per-message inbox transaction rolled back; the offset was deliberately not committed. The underlying failure is on `.cause` |
 

docs/api_reference.md

@@ -77,7 +77,8 @@ Fluent builder. Picks `DistributedLockingFetchStrategy` + `BulkCommitStrategy` a
 
 | Step | Purpose | Notes |
 | :--- | :--- | :--- |
-| `HandlerExecutionStep` | Runs the user handler inside the pipeline with a timeout. | Required terminal step in every processor. |
+| `HandlerExecutionStep` | Runs the user handler inside the pipeline with a timeout. | Required terminal step in every processor. A `TransientError` out of the handler is recorded with `count_as_attempt=False` and rescheduled; a timeout counts. |
+| `PublisherExecutionStep` | The outbox's handler step; installed by `create_outbox_processor`. | Subclasses `HandlerExecutionStep`. The publish timeout is transient too, and the first transient failure ends the batch's publishing — the remaining events are recorded the same way, unpublished, with their schedule untouched. |
 | `SiblingDeduplicationStep` | Skips an `InboxEvent` if a sibling row with the same `(message_id, consumer_group)` is already `completed`. | Calls `InboxEventRepository.has_completed_sibling_for_inbox_key`. A no-op on a non-partitioned table — see [storage adapters](storage_adapters.md#inboxeventrepository). |
 | `MetricsStep` | Pushes batch lifecycle counters into an `InboxMetrics` / `OutboxMetrics` sink. | |
 | `OpenTelemetryStep(service_name=...)` | Creates spans for each batch/event. | Requires `opentelemetry` extra. |
@@ -185,6 +186,7 @@ All inherit from `OmniBoxError`.
 
 - Storage: `StorageError`, `StorageConnectionError`, `StorageTimeoutError`, `StorageTransactionError`, `StorageIntegrityError`.
 - Domain / locking: `EventNotLockedError`, `EventLockedByAnotherWorkerError`, `EventAlreadyLockedError`, `InvalidEventStateError`, `EventConcurrentUpdateError`.
+- Transient: `TransientError` — raised by a publisher or a handler to say the failure belongs to the environment and not to the event. The pipeline records it without spending an attempt and retries the row on the next cycle.
 - Misc: `UnsupportedCapabilityError`, `InboxPersistError`.
 
 ## Infrastructure adapters
@@ -201,7 +203,7 @@ All inherit from `OmniBoxError`.
 
 `omni_box.infra.brokers.kafka`:
 
-- `KafkaEventPublisher(producer, converter, *, max_infra_retries=3)` — built on top of `aiokafka.AIOKafkaProducer`. The caller owns the producer lifecycle (`start`/`stop`).
+- `KafkaEventPublisher(producer, converter, *, max_infra_retries=3)` — built on top of `aiokafka.AIOKafkaProducer`. The caller owns the producer lifecycle (`start`/`stop`). A broker that does not answer — connection and node errors, request and client timeouts, and an unknown topic the broker will not confirm with a metadata refresh either — is retried `max_infra_retries` times and then raised as `TransientError`, which costs the row no attempt. Anything else, including a record the broker rejects, is raised as it is and counts.
 - `KafkaEventConsumer` — wraps `aiokafka.AIOKafkaConsumer` and exposes per-record `AckHandle`s. Use `DefaultEnvelopeParser` or provide your own `EnvelopeParser`.
 
 Neither adapter depends on any external "kit" package; only `aiokafka` is required.

docs/architecture.md

@@ -29,7 +29,7 @@ Pure Python, framework-agnostic.
 - Pipeline primitives (`omni_box.core.pipeline`):
   - `EventProcessorBuilder` — fluent builder for `EventBatchProcessor`.
   - `ProcessingPipeline`, `ProcessingContext`, `ProcessingStep`, `StepResult`.
-  - Built-in steps: `HandlerExecutionStep`, `SiblingDeduplicationStep`, `MetricsStep`, `OpenTelemetryStep`, `CircuitBreakerStep`, `DLQStep`.
+  - Built-in steps: `HandlerExecutionStep`, `PublisherExecutionStep`, `SiblingDeduplicationStep`, `MetricsStep`, `OpenTelemetryStep`, `CircuitBreakerStep`, `DLQStep`.
   - Strategies:
     - Fetch: `DistributedLockingFetchStrategy`, `OptimisticLockingFetchStrategy`, `FilteredFetchStrategy`.
     - Commit: `BulkCommitStrategy`, `SingleCommitStrategy`.

docs/guide/configuration.md

@@ -24,7 +24,7 @@ OutboxPublisher(
 )
 ```
 
-- `publish_timeout` — hard timeout for a single `broker.publish` call.
+- `publish_timeout` — hard timeout for a single `broker.publish` call. It is treated as the broker's problem, not the row's: it costs no attempt and ends the cycle's publishing. Keep an eye on how it compares with the adapter's own patience — `AIOKafkaProducer(request_timeout_ms=...)` defaults to 40 s against this timeout's 30 s, so with the defaults the step gives up before aiokafka has said anything, and the log shows a publish timeout where a broker error would be more informative. Set `request_timeout_ms` below `publish_timeout`, or raise `publish_timeout` above it.
 - `concurrency_limit` — wraps `publish_batch` in an `asyncio.Semaphore`.
 
 ## `InboxConsumerRunner`

docs/troubleshooting.md

@@ -54,7 +54,21 @@ Inbox rows are deduplicated by **`(message_id, consumer_group)`** (the unique in
 
 The `DLQStep` only considers counted failures: a transient error never routes a row to DLQ even if it superficially looks like it crossed the threshold. The DLQ move itself is **best-effort** — it runs outside the commit transaction and a failure during `move_to_dlq` is logged and swallowed. Pair the step with an idempotent sink (e.g. Kafka with a unique key) to avoid duplicates on replay.
 
-Inspect failed rows directly:
+### The attempt budget is for the row, not for the outage
+
+An attempt is spent when the failure is about the event: a payload the broker rejects, a topic it does not have, a serialization error. A broker or a downstream that is **not there** is a property of the cycle, and spending the budget on it would turn a long outage into a backlog of terminal `failed` rows that only an operator could bring back.
+
+So a publisher or a handler says so by raising `TransientError`. The pipeline records it without bumping `attempts_made` and reschedules the row a second ahead; `KafkaEventPublisher` raises it once `max_infra_retries` are spent on a broker that does not answer. What you see during an outage is rows sitting in `pending` with `attempts_made` unchanged and `last_error` saying the broker is unreachable — and, because the outbox step stops publishing for the rest of the cycle after the first such failure, one probe per cycle rather than one per row:
+
+```sql
+SELECT status, attempts_made, count(*), max(last_error)
+FROM outbox_events
+GROUP BY 1, 2 ORDER BY 1, 2;
+```
+
+When the broker answers again the next cycle publishes the backlog. `requeue_failed` stays what it always was: the operator's tool for rows that genuinely poisoned themselves, not the way out of an outage.
+
+Inspect genuinely failed rows directly:
 
 ```sql
 SELECT id, event_type, attempts_made, max_attempts, last_error

docs/user_guide.md

@@ -85,6 +85,18 @@ while not shutdown:
 Drop the transaction and the cycle publishes for nothing: the lock and the completion roll
 back with the session, the rows are still `pending`, and the next cycle sends them again.
 
+**When the broker is down.** A publish failure that is about the broker rather than about the
+row — the connection, the node, a request that timed out, the publish timeout itself — does
+not spend the row's attempt budget. `KafkaEventPublisher` raises `TransientError` once its
+own `max_infra_retries` are spent, and the outbox step records the row with
+`attempts_made` untouched and stops publishing for the rest of the cycle: the remaining rows
+are recorded the same way without being sent, since they were going to the same broker. So an
+outage costs one probe per cycle, the rows stay `pending` however long it lasts, and the first
+cycle after the broker answers publishes the backlog. A payload the broker rejects, or a topic
+it says it does not have, is about the row and still spends an attempt — that is what
+`max_attempts` and `failed` are for. Raise `TransientError` from your own publisher or handler
+to get the same treatment.
+
 ## Transactional Inbox
 
 ### Option A — drive consumption with `InboxConsumerRunner`
@@ -229,6 +241,7 @@ from omni_box.core.pipeline.steps import (
     DLQStep,
     HandlerExecutionStep,
     OpenTelemetryStep,
+    PublisherExecutionStep,
     SiblingDeduplicationStep,
 )
 from omni_box.core.pipeline.strategies import (
@@ -252,6 +265,10 @@ processor = builder.build()
 
 The builder auto-picks `DistributedLockingFetchStrategy` + `BulkCommitStrategy` when the repository advertises matching capabilities, so the `with_*` calls above are usually optional.
 
+For an outbox, use `PublisherExecutionStep(broker.publish, timeout=30)` in place of
+`HandlerExecutionStep` — it is what `create_outbox_processor` installs, and it is the step
+that keeps a broker outage off the attempt budget.
+
 ## Outbox payload envelopes
 
 `OutboxPublisher` delegates the body shape to a converter:

omni_box/__init__.py

@@ -34,6 +34,7 @@
     StorageIntegrityError,
     StorageTimeoutError,
     StorageTransactionError,
+    TransientError,
     UnsupportedCapabilityError,
 )
 from .core.converters import EnvelopeEventConverter
@@ -116,6 +117,7 @@
     "StorageIntegrityError",
     "StorageTimeoutError",
     "StorageTransactionError",
+    "TransientError",
     "UnsupportedCapabilityError",
     "__version__",
     "create_dispatching_processor",

omni_box/application/factories.py

@@ -17,6 +17,7 @@
     HandlerExecutionStep,
     MetricsStep,
     OpenTelemetryStep,
+    PublisherExecutionStep,
     SiblingDeduplicationStep,
 )
 from ..core.pipeline.strategies.fetch import FilteredFetchStrategy
@@ -127,7 +128,7 @@ def create_outbox_processor(
         for step in additional_steps_before:
             builder.add_step(step)
 
-    builder.add_step(HandlerExecutionStep(publisher.publish, timeout=publish_timeout))
+    builder.add_step(PublisherExecutionStep(publisher.publish, timeout=publish_timeout))
 
     if additional_steps_after:
         for step in additional_steps_after:

omni_box/core/__init__.py

@@ -12,6 +12,7 @@
     StorageIntegrityError,
     StorageTimeoutError,
     StorageTransactionError,
+    TransientError,
     UnsupportedCapabilityError,
 )
 from .models.entities import BaseEvent, InboxEvent, OutboxEvent
@@ -42,5 +43,6 @@
     "StorageIntegrityError",
     "StorageTimeoutError",
     "StorageTransactionError",
+    "TransientError",
     "UnsupportedCapabilityError",
 ]

omni_box/core/constants.py

@@ -37,6 +37,7 @@
 DEFAULT_PUBLISH_TIMEOUT_SECONDS = 30.0
 DEFAULT_PROCESS_TIMEOUT_SECONDS = 30.0
 DEFAULT_LEASE_TIMEOUT_SECONDS = 300
+DEFAULT_TRANSIENT_RETRY_DELAY_SECONDS = 1.0
 DEFAULT_MARK_RESULTS_MAX_RETRIES = 3
 DEFAULT_MARK_RESULTS_RETRY_DELAY = 1.0
 

omni_box/core/exceptions.py

@@ -102,6 +102,16 @@ def __init__(
         super().__init__(msg)
 
 
+class TransientError(OmniBoxError):
+    """Raised by a publisher or a handler when the failure is the environment's, not the event's.
+
+    The pipeline records it without spending an attempt and reschedules the
+    event, the way ``handler_retry(count_as_attempt=False)`` does for a
+    returned result. ``KafkaEventPublisher`` raises it once its own retries
+    are spent on a broker that does not answer.
+    """
+
+
 class UnsupportedCapabilityError(OmniBoxError):
     """Raised when a required repository capability is not available."""
 

omni_box/core/pipeline/steps/__init__.py

@@ -6,6 +6,7 @@
 from .handler import HandlerExecutionStep
 from .metrics import MetricsStep
 from .otel import OpenTelemetryStep
+from .publisher import PublisherExecutionStep
 
 __all__ = [
     "CircuitBreakerStep",
@@ -14,5 +15,6 @@
     "HandlerExecutionStep",
     "MetricsStep",
     "OpenTelemetryStep",
+    "PublisherExecutionStep",
     "SiblingDeduplicationStep",
 ]

omni_box/core/pipeline/steps/handler.py

@@ -2,9 +2,13 @@
 
 import asyncio
 from collections.abc import Awaitable, Callable
+from datetime import timedelta
 from typing import TYPE_CHECKING
 
-from ...services.results import EventHandlerResult, coerce_handler_outcome
+from ....utils.datetime import utc_now
+from ...constants import DEFAULT_TRANSIENT_RETRY_DELAY_SECONDS
+from ...exceptions import TransientError
+from ...services.results import EventHandlerResult, EventHandlerStatus, coerce_handler_outcome
 from ..step import BaseProcessingStep, StepResult
 
 if TYPE_CHECKING:
@@ -14,7 +18,13 @@
 
 
 class HandlerExecutionStep[T: BaseEvent](BaseProcessingStep[T]):
-    """Step that executes a handler for each event."""
+    """Step that executes a handler for each event.
+
+    A ``TransientError`` out of the handler says the failure belongs to a
+    downstream and not to the event: it is recorded without spending an
+    attempt and rescheduled a moment ahead. Every other exception, the
+    timeout included, counts.
+    """
 
     def __init__(
         self,
@@ -48,14 +58,30 @@ async def execute(
                     next_retry_at=outcome.next_retry_at,
                     status=outcome.status,
                 )
+        except TransientError as e:
+            self._retry_later(event, context, str(e))
         except TimeoutError:
-            context.mark_failed(
-                event.id,
-                f"Handler execution timed out after {self._timeout}s",
-                count_as_attempt=True,
-                status="failed",
-            )
+            self._timed_out(event, context)
         except Exception as e:
             context.mark_failed(event.id, f"{type(e).__name__}: {e}", status="failed")
 
         return StepResult.next()
+
+    def _timed_out(self, event: T, context: ProcessingContext[T]) -> None:
+        """Record the handler timeout. Counts as an attempt: the handler is the event's own work."""
+        context.mark_failed(
+            event.id,
+            f"Handler execution timed out after {self._timeout}s",
+            count_as_attempt=True,
+            status="failed",
+        )
+
+    def _retry_later(self, event: T, context: ProcessingContext[T], error: str) -> None:
+        """Record a failure that is not the event's fault: no attempt spent, retried on the next cycle."""
+        context.mark_failed(
+            event.id,
+            error,
+            count_as_attempt=False,
+            next_retry_at=utc_now() + timedelta(seconds=DEFAULT_TRANSIENT_RETRY_DELAY_SECONDS),
+            status=EventHandlerStatus.RETRY,
+        )