diff --git a/docs/agents.md b/docs/agents.md index 2fa2d61..32aca9a 100644 --- a/docs/agents.md +++ b/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` | diff --git a/docs/api_reference.md b/docs/api_reference.md index 45eaed0..001b034 100644 --- a/docs/api_reference.md +++ b/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. diff --git a/docs/architecture.md b/docs/architecture.md index 29873a8..10f7f3f 100644 --- a/docs/architecture.md +++ b/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`. diff --git a/docs/guide/configuration.md b/docs/guide/configuration.md index 3c4ea8d..4b62178 100644 --- a/docs/guide/configuration.md +++ b/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` diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index ce7eeb8..ea6298b 100644 --- a/docs/troubleshooting.md +++ b/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 diff --git a/docs/user_guide.md b/docs/user_guide.md index 3af4e3a..c2e00c9 100644 --- a/docs/user_guide.md +++ b/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: diff --git a/omni_box/__init__.py b/omni_box/__init__.py index d8131e2..5b752d8 100644 --- a/omni_box/__init__.py +++ b/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", diff --git a/omni_box/application/factories.py b/omni_box/application/factories.py index ad23faf..ec0a53d 100644 --- a/omni_box/application/factories.py +++ b/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: diff --git a/omni_box/core/__init__.py b/omni_box/core/__init__.py index 9299123..cd0f25c 100644 --- a/omni_box/core/__init__.py +++ b/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", ] diff --git a/omni_box/core/constants.py b/omni_box/core/constants.py index c59bba6..e4da075 100644 --- a/omni_box/core/constants.py +++ b/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 diff --git a/omni_box/core/exceptions.py b/omni_box/core/exceptions.py index 9123bdb..0a514d1 100644 --- a/omni_box/core/exceptions.py +++ b/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.""" diff --git a/omni_box/core/pipeline/steps/__init__.py b/omni_box/core/pipeline/steps/__init__.py index 62f179a..f1dbc02 100644 --- a/omni_box/core/pipeline/steps/__init__.py +++ b/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", ] diff --git a/omni_box/core/pipeline/steps/handler.py b/omni_box/core/pipeline/steps/handler.py index f4ffec9..ae08dd2 100644 --- a/omni_box/core/pipeline/steps/handler.py +++ b/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, + ) diff --git a/omni_box/core/pipeline/steps/publisher.py b/omni_box/core/pipeline/steps/publisher.py new file mode 100644 index 0000000..565ce3b --- /dev/null +++ b/omni_box/core/pipeline/steps/publisher.py @@ -0,0 +1,63 @@ +"""Publisher step for the outbox pipeline.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import structlog + +from ...services.results import EventHandlerStatus +from ..step import StepResult +from .handler import HandlerExecutionStep + +if TYPE_CHECKING: + from ...models.entities import BaseEvent + from ..context import ProcessingContext + +logger = structlog.get_logger(__name__) + +# ``ProcessingContext.extra`` key holding the broker error that ended this batch's +# publishing. Per batch, not per step: one processor may run several batches at once. +_DEFERRED_BROKER_ERROR = "publisher_broker_unreachable" + + +class PublisherExecutionStep[T: BaseEvent](HandlerExecutionStep[T]): + """Step that publishes each event through a broker publisher. + + Everything in an outbox batch goes to the same broker, so a broker that + does not answer is a property of the cycle, not of the row. A + ``TransientError`` or a timeout out of the publisher is recorded without + spending an attempt, and the publisher is not called again in this batch: + the remaining events are marked the same way, their schedule untouched, + and come back in the next cycle. Any other exception is about the event + and counts, as it does in ``HandlerExecutionStep``. + """ + + async def execute( + self, + event: T, + context: ProcessingContext[T], + ) -> StepResult: + """Publish the event, unless this batch already found the broker unreachable.""" + deferred = context.extra.get(_DEFERRED_BROKER_ERROR) + if isinstance(deferred, str): + context.mark_failed(event.id, deferred, count_as_attempt=False, status=EventHandlerStatus.RETRY) + return StepResult.next() + return await super().execute(event, context) + + def _timed_out(self, event: T, context: ProcessingContext[T]) -> None: + """A publish the broker did not acknowledge in time is the broker's problem, not the row's.""" + self._retry_later(event, context, f"Publish timed out after {self._timeout}s") + + def _retry_later(self, event: T, context: ProcessingContext[T], error: str) -> None: + super()._retry_later(event, context, error) + context.extra[_DEFERRED_BROKER_ERROR] = error + logger.warning( + "Broker unreachable, deferring the rest of the batch", + event_id=str(event.id), + worker_id=context.worker_id, + error=error, + ) + + +__all__ = ["PublisherExecutionStep"] diff --git a/omni_box/infra/brokers/kafka/publisher.py b/omni_box/infra/brokers/kafka/publisher.py index fcefc8f..763bfd7 100644 --- a/omni_box/infra/brokers/kafka/publisher.py +++ b/omni_box/infra/brokers/kafka/publisher.py @@ -8,8 +8,16 @@ import orjson import structlog from aiokafka import AIOKafkaProducer +from aiokafka.errors import ( + KafkaConnectionError, + KafkaTimeoutError, + NodeNotReadyError, + RequestTimedOutError, + UnknownTopicOrPartitionError, +) from ....core.converters.event import EventConverter +from ....core.exceptions import TransientError from ....core.models.entities import OutboxEvent from ....core.protocols import EventPublisher from ....utils.backoff import ErrorClassifier, calculate_backoff_with_jitter @@ -19,6 +27,32 @@ logger = structlog.get_logger(__name__) +# The aiokafka errors that mean the broker is not answering. aiokafka's own +# ``retriable`` flag is wider than this: it also covers a topic that does not +# exist, which is about the row and not about the broker. +_BROKER_UNREACHABLE: tuple[type[BaseException], ...] = ( + KafkaConnectionError, + KafkaTimeoutError, + NodeNotReadyError, + RequestTimedOutError, +) + + +def _describe(exc: BaseException) -> str: + """``TypeName: message``, without repeating a name the exception already prints. + + aiokafka's errors stringify as ``NodeNotReadyError: node 1 is not ready`` + on their own, and one of them carrying no message stringifies as the name + alone; the builtins print the message only. + """ + name = type(exc).__name__ + detail = str(exc) + if not detail or detail == name: + return name + if detail.startswith(f"{name}: "): + return detail + return f"{name}: {detail}" + class KafkaEventPublisher(EventPublisher): """Kafka publisher that converts OutboxEvent and sends via aiokafka. @@ -27,6 +61,13 @@ class KafkaEventPublisher(EventPublisher): Caller is responsible for ``AIOKafkaProducer`` lifecycle (``start``/``stop``). For at-least-once delivery configure the producer with ``enable_idempotence=True`` and ``acks="all"``. + + 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 the outbox pipeline + records without spending an attempt. Everything else, a payload or a + topic the broker rejects included, is raised as is and counts. """ def __init__( @@ -59,18 +100,31 @@ async def publish(self, event: OutboxEvent, repo: EventRepository[OutboxEvent]) headers=encoded_headers, ) except Exception as e: - classification = ErrorClassifier.classify(e) - if classification.is_transient and attempt < self._max_infra_retries: + reason = await self._unreachable_reason(e, event.topic) + if reason is None: + raise + if attempt < self._max_infra_retries: delay = calculate_backoff_with_jitter(attempt) logger.warning( - "Kafka retry", event_id=str(event.id), attempt=attempt + 1, delay=delay, error=str(e) + "Kafka retry", event_id=str(event.id), attempt=attempt + 1, delay=delay, error=reason ) await asyncio.sleep(delay) continue - raise + raise TransientError(reason) from e else: return + async def _unreachable_reason(self, exc: Exception, topic: str) -> str | None: + """Describe ``exc`` when it means the broker is not answering; ``None`` when it is about the event.""" + if ErrorClassifier.classify(exc, additional_transient=_BROKER_UNREACHABLE).is_transient: + return f"Kafka broker unreachable: {_describe(exc)}" + # aiokafka reports a topic it could not fetch metadata for the same way + # whether the topic is missing or the broker is silent; a metadata + # refresh that fails as well tells the two apart. + if isinstance(exc, UnknownTopicOrPartitionError) and not await self._producer.client.force_metadata_update(): + return f"Kafka broker unreachable: no metadata for topic {topic!r} and no answer to a metadata request" + return None + def _build_headers(self, event: OutboxEvent) -> dict[str, str]: headers = dict(event.headers or {}) headers["event_id"] = str(event.id) diff --git a/tests/integration/kafka/test_publisher.py b/tests/integration/kafka/test_publisher.py index afb2374..d43dee9 100644 --- a/tests/integration/kafka/test_publisher.py +++ b/tests/integration/kafka/test_publisher.py @@ -12,6 +12,7 @@ from aiokafka import AIOKafkaConsumer, AIOKafkaProducer from omni_box.core.converters.event import EnvelopeEventConverter, RawEventConverter, SchemaVersionedConverter +from omni_box.core.exceptions import TransientError from omni_box.core.models.entities import OutboxEvent from omni_box.infra.brokers.kafka.publisher import KafkaEventPublisher from tests.helpers import FakeOutboxStore @@ -178,7 +179,7 @@ async def _no_sleep(delay: float) -> None: assert len(sleeps) == 2 -async def test__kafka_publisher__transient_error_over_limit__raises_original_exception( +async def test__kafka_publisher__transient_error_over_limit__raises_transient_error( monkeypatch: pytest.MonkeyPatch, ) -> None: # Arrange @@ -186,13 +187,15 @@ async def _no_sleep(_delay: float) -> None: return monkeypatch.setattr("omni_box.infra.brokers.kafka.publisher.asyncio.sleep", _no_sleep) - broken = _BrokenProducer(fail_times=10, exc=ConnectionError("permanent transient")) + original = ConnectionError("permanent transient") + broken = _BrokenProducer(fail_times=10, exc=original) publisher = KafkaEventPublisher(cast(AIOKafkaProducer, broken), RawEventConverter(), max_infra_retries=1) # Act / Assert - with pytest.raises(ConnectionError): + with pytest.raises(TransientError, match="Kafka broker unreachable: ConnectionError") as exc_info: await publisher.publish(_make_event(topic="any"), repo=cast("FakeOutboxStore", FakeOutboxStore())) assert broken.calls == 2 # initial + 1 retry + assert exc_info.value.__cause__ is original async def test__kafka_publisher__permanent_error__no_retry_and_raises( diff --git a/tests/integration/postgres/test_broker_outage.py b/tests/integration/postgres/test_broker_outage.py new file mode 100644 index 0000000..00ca410 --- /dev/null +++ b/tests/integration/postgres/test_broker_outage.py @@ -0,0 +1,200 @@ +"""A broker outage against the real outbox table: the attempt budget is not spent on it. + +The relay here is the documented one -- ``OutboxPublisher`` over +``PostgresOutboxRepository`` and ``KafkaEventPublisher`` -- and the broker is a +producer that behaves the way aiokafka does while the container is paused: it +neither answers nor refuses. Cycles run back to back rather than on a timer, so +the outage from the report takes milliseconds instead of four minutes. +""" + +from __future__ import annotations + +import asyncio +from uuid import UUID, uuid4 + +import pytest +from aiokafka.errors import MessageSizeTooLargeError, NodeNotReadyError +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from omni_box import OmniBoxDomainService, OutboxEvent, OutboxPublisher +from omni_box.core.converters import EnvelopeEventConverter +from omni_box.core.models.enums import EventStatus +from omni_box.core.services.results import BatchProcessingResult +from omni_box.infra.brokers.kafka import KafkaEventPublisher +from omni_box.infra.storage.postgres import PostgresOutboxRepository +from omni_box.utils import utc_now +from tests.models import ConcreteOutboxEvent + +pytestmark = pytest.mark.integration + +EVENTS = 20 +"""The size of the reporter's backlog.""" + +MAX_ATTEMPTS = 6 +"""The default budget, and the number of cycles that used to burn all of it.""" + +CYCLES = 7 +"""One cycle more than the budget: on 0.2.0 the seventh already found nothing left to publish.""" + +TOPIC = "orders.events" + + +class _StubClient: + """The producer's client, asked for a metadata refresh when a topic is unknown.""" + + def __init__(self, answers: bool) -> None: + self.answers = answers + + async def force_metadata_update(self) -> bool: + return self.answers + + +class _StubProducer: + """An ``AIOKafkaProducer`` that raises what the real one raises, or records the send. + + ``failure`` is what the broker gives back; setting it to ``None`` is the + broker coming back. + """ + + def __init__(self, failure: BaseException | None = None, *, answers_metadata: bool = False) -> None: + self.failure = failure + self.client = _StubClient(answers_metadata) + self.sent: list[bytes | None] = [] + + async def send_and_wait( + self, + topic: str, + value: bytes, + key: bytes | None = None, + headers: list[tuple[str, bytes]] | None = None, + ) -> None: + if self.failure is not None: + raise self.failure + self.sent.append(key) + + +def _broker(producer: _StubProducer) -> KafkaEventPublisher: + """The reporter's wiring, minus the adapter's own retry loop. + + ``max_infra_retries=0`` keeps a cycle to a single probe and to no sleeping; + the retry loop itself is covered by the adapter's unit tests. + """ + return KafkaEventPublisher(producer, EnvelopeEventConverter(), max_infra_retries=0) + + +async def _create_events(session_factory: async_sessionmaker[AsyncSession]) -> list[UUID]: + """Fill the outbox the way a service would, in one transaction.""" + domain = OmniBoxDomainService(max_attempts=MAX_ATTEMPTS) + ids: list[UUID] = [] + async with session_factory() as session, session.begin(): + repo = PostgresOutboxRepository(session, model_class=ConcreteOutboxEvent) + for i in range(EVENTS): + event = domain.create_outbox_event( + aggregate_type="order", + aggregate_id=uuid4(), + event_type="order.created", + topic=TOPIC, + partition_key=f"order-{i}", + payload={"n": i}, + ) + await repo.create(event) + ids.append(event.id) + return ids + + +async def _relay_cycle( + session_factory: async_sessionmaker[AsyncSession], + producer: _StubProducer, +) -> BatchProcessingResult[OutboxEvent]: + """One tick of a relay loop: fetch, publish, commit.""" + async with session_factory() as session, session.begin(): + publisher = OutboxPublisher( + PostgresOutboxRepository(session, model_class=ConcreteOutboxEvent), + _broker(producer), + publish_timeout=5.0, + ) + return await publisher.publish_batch(worker_id="relay-1", batch_size=100) + + +async def _rows(session_factory: async_sessionmaker[AsyncSession], ids: list[UUID]) -> list[OutboxEvent]: + """Read the rows back as domain entities, in the order they were created.""" + async with session_factory() as session: + repo = PostgresOutboxRepository(session, model_class=ConcreteOutboxEvent) + rows = [await repo.get_by_id(event_id) for event_id in ids] + return [row for row in rows if row is not None] + + +@pytest.fixture +def session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]: + """A session per relay cycle, the way a long-running relay works.""" + return async_sessionmaker(db_engine, expire_on_commit=False, class_=AsyncSession) + + +async def test__relay__broker_unreachable_for_longer_than_the_budget__rows_stay_pending( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + """The reporter's scenario: an outage that outlasts ``max_attempts`` cycles.""" + # Arrange + ids = await _create_events(session_factory) + producer = _StubProducer(NodeNotReadyError("node 1 is not ready")) + created_before = utc_now() + + # Act + per_cycle = [] + for _ in range(CYCLES): + await _relay_cycle(session_factory, producer) + per_cycle.append({(row.status, row.attempts_made) for row in await _rows(session_factory, ids)}) + + # Assert + assert per_cycle == [{(EventStatus.PENDING, 0)}] * CYCLES + rows = await _rows(session_factory, ids) + assert all(row.locked_at is None for row in rows) + assert all(row.last_error == "Kafka broker unreachable: NodeNotReadyError: node 1 is not ready" for row in rows) + # One probe per cycle, not one per row: the rest of the batch is deferred as + # soon as the broker turns out to be unreachable, and keeps its schedule. + assert producer.sent == [] + rescheduled = [row.id for row in rows if row.scheduled_at > created_before] + assert len(rescheduled) == CYCLES + + +async def test__relay__broker_back_after_an_outage__next_cycle_publishes_everything( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + """No ``requeue_failed`` and no operator: the backlog drains on the first cycle after the outage.""" + # Arrange + ids = await _create_events(session_factory) + producer = _StubProducer(NodeNotReadyError("node 1 is not ready")) + for _ in range(CYCLES): + await _relay_cycle(session_factory, producer) + + # Act + producer.failure = None # Kafka is back + await asyncio.sleep(1.5) # the probed rows were put a second ahead + result = await _relay_cycle(session_factory, producer) + + # Assert + assert len(producer.sent) == EVENTS + assert set(result.processed_event_ids) == set(ids) + rows = await _rows(session_factory, ids) + assert [row.status for row in rows] == [EventStatus.COMPLETED] * EVENTS + assert [row.attempts_made for row in rows] == [0] * EVENTS + + +async def test__relay__broker_answers_and_rejects_the_record__every_row_spends_an_attempt( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + """A broker that is there and says no is about the row: the budget is still the right tool.""" + # Arrange + ids = await _create_events(session_factory) + producer = _StubProducer(MessageSizeTooLargeError("The message is 2000000 bytes when serialized")) + + # Act + result = await _relay_cycle(session_factory, producer) + + # Assert + assert {failure.event_id for failure in result.failed_counted} == set(ids) + assert result.failed_noncounted == [] + rows = await _rows(session_factory, ids) + assert [row.status for row in rows] == [EventStatus.PENDING] * EVENTS + assert [row.attempts_made for row in rows] == [1] * EVENTS + assert all("MessageSizeTooLargeError" in (row.last_error or "") for row in rows) diff --git a/tests/unit/application/test_app_factories.py b/tests/unit/application/test_app_factories.py index f77150a..92e73ea 100644 --- a/tests/unit/application/test_app_factories.py +++ b/tests/unit/application/test_app_factories.py @@ -25,6 +25,7 @@ HandlerExecutionStep, MetricsStep, OpenTelemetryStep, + PublisherExecutionStep, SiblingDeduplicationStep, ) from omni_box.core.pipeline.strategies.fetch import FilteredFetchStrategy @@ -206,14 +207,14 @@ def test__create_inbox_processor__handler_step__defaults_to_module_default_timeo # -------- create_outbox_processor -------- -def test__create_outbox_processor__defaults__has_only_handler_step( +def test__create_outbox_processor__defaults__has_only_publisher_step( outbox_repo: MagicMock, publisher: MagicMock ) -> None: # Arrange / Act processor = create_outbox_processor(repo=outbox_repo, publisher=publisher) # Assert - assert [type(s) for s in processor._pipeline._steps] == [HandlerExecutionStep] + assert [type(s) for s in processor._pipeline._steps] == [PublisherExecutionStep] assert processor._job_name == "outbox_processor" assert processor._metrics is None @@ -249,26 +250,26 @@ def test__create_outbox_processor__all_optional_features_enabled__builds_full_pi CircuitBreakerStep, DLQStep, _RecordingStep, - HandlerExecutionStep, + PublisherExecutionStep, _RecordingStep, MetricsStep, ] assert processor._job_name == "custom_outbox" assert processor._metrics is outbox_metrics - handler_step = next(s for s in processor._pipeline._steps if isinstance(s, HandlerExecutionStep)) - assert handler_step._timeout == 2.5 + publisher_step = next(s for s in processor._pipeline._steps if isinstance(s, PublisherExecutionStep)) + assert publisher_step._timeout == 2.5 -def test__create_outbox_processor__handler_step__defaults_to_module_default_timeout( +def test__create_outbox_processor__publisher_step__defaults_to_module_default_timeout( outbox_repo: MagicMock, publisher: MagicMock ) -> None: # Arrange / Act processor = create_outbox_processor(repo=outbox_repo, publisher=publisher) # Assert - handler_step = next(s for s in processor._pipeline._steps if isinstance(s, HandlerExecutionStep)) - assert handler_step._timeout == DEFAULT_PUBLISH_TIMEOUT_SECONDS + publisher_step = next(s for s in processor._pipeline._steps if isinstance(s, PublisherExecutionStep)) + assert publisher_step._timeout == DEFAULT_PUBLISH_TIMEOUT_SECONDS # -------- create_dispatching_processor -------- diff --git a/tests/unit/core/pipeline/steps/test_handler.py b/tests/unit/core/pipeline/steps/test_handler.py index c5d957c..9516e11 100644 --- a/tests/unit/core/pipeline/steps/test_handler.py +++ b/tests/unit/core/pipeline/steps/test_handler.py @@ -6,10 +6,12 @@ import pytest +from omni_box.core.exceptions import TransientError from omni_box.core.models.entities import OutboxEvent from omni_box.core.pipeline.context import ProcessingContext from omni_box.core.pipeline.steps.handler import HandlerExecutionStep from omni_box.core.services.results import EventHandlerResult, EventHandlerStatus +from omni_box.utils import utc_now from tests.helpers import create_fake_event pytestmark = pytest.mark.unit @@ -179,3 +181,27 @@ async def broken_handler(event: OutboxEvent, repo: object) -> None: assert failure.event_id == event.id assert "ValueError" in failure.error assert "kaboom" in failure.error + + +async def test__handler_step__handler_raises_transient_error__marks_failed_noncounted_and_reschedules( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + async def handler(event: OutboxEvent, repo: object) -> None: + raise TransientError("payment gateway unreachable") + + step: HandlerExecutionStep[OutboxEvent] = HandlerExecutionStep(handler) + event = create_fake_event() + before = utc_now() + + # Act + await step.execute(event, context) + + # Assert + assert context.failed_counted == [] + failure = context.failed_noncounted[0] + assert failure.event_id == event.id + assert failure.error == "payment gateway unreachable" + assert failure.next_retry_at is not None + assert failure.next_retry_at > before + assert context.statuses[event.id] == EventHandlerStatus.RETRY diff --git a/tests/unit/core/pipeline/steps/test_publisher_step.py b/tests/unit/core/pipeline/steps/test_publisher_step.py new file mode 100644 index 0000000..638fccb --- /dev/null +++ b/tests/unit/core/pipeline/steps/test_publisher_step.py @@ -0,0 +1,170 @@ +"""Unit tests for ``omni_box.core.pipeline.steps.publisher``.""" + +from __future__ import annotations + +import asyncio +from datetime import timedelta + +import pytest + +from omni_box.core.constants import DEFAULT_TRANSIENT_RETRY_DELAY_SECONDS +from omni_box.core.exceptions import TransientError +from omni_box.core.models.entities import OutboxEvent +from omni_box.core.pipeline.context import ProcessingContext +from omni_box.core.pipeline.steps import HandlerExecutionStep, PublisherExecutionStep +from omni_box.core.services.results import EventHandlerStatus +from omni_box.utils import utc_now +from tests.helpers import create_fake_event + +pytestmark = pytest.mark.unit + + +class _Repo: + pass + + +class _Broker: + """Publisher fake that raises what it is told to for the topics it is given.""" + + def __init__(self, failing: dict[str, BaseException] | None = None) -> None: + self.failing = failing or {} + self.published: list[OutboxEvent] = [] + + async def publish(self, event: OutboxEvent, repo: object) -> None: + if event.topic in self.failing: + raise self.failing[event.topic] + self.published.append(event) + + +@pytest.fixture +def context() -> ProcessingContext[OutboxEvent]: + return ProcessingContext(repo=_Repo(), worker_id="w1") # type: ignore[arg-type] + + +def test__publisher_step__constructed__is_a_handler_execution_step() -> None: + # Arrange / Act + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(_Broker().publish) + + # Assert + assert isinstance(step, HandlerExecutionStep) + + +async def test__publisher_step__publish_succeeds__marks_completed( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + broker = _Broker() + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(broker.publish) + event = create_fake_event() + + # Act + await step.execute(event, context) + + # Assert + assert context.completed_ids == [event.id] + assert broker.published == [event] + + +async def test__publisher_step__transient_error__marks_failed_noncounted_one_tick_ahead( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + broker = _Broker(failing={"topic.down": TransientError("Kafka broker unreachable: NodeNotReadyError")}) + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(broker.publish) + event = create_fake_event(topic="topic.down") + before = utc_now() + + # Act + await step.execute(event, context) + + # Assert + assert context.failed_counted == [] + failure = context.failed_noncounted[0] + assert failure.event_id == event.id + assert failure.error == "Kafka broker unreachable: NodeNotReadyError" + assert failure.next_retry_at is not None + assert before < failure.next_retry_at <= utc_now() + timedelta(seconds=DEFAULT_TRANSIENT_RETRY_DELAY_SECONDS) + assert context.statuses[event.id] == EventHandlerStatus.RETRY + + +async def test__publisher_step__publish_times_out__marks_failed_noncounted( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + async def hanging_publish(event: OutboxEvent, repo: object) -> None: + await asyncio.Future() + + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(hanging_publish, timeout=0.01) + event = create_fake_event() + + # Act + await step.execute(event, context) + + # Assert + assert context.failed_counted == [] + failure = context.failed_noncounted[0] + assert failure.error == "Publish timed out after 0.01s" + assert failure.next_retry_at is not None + + +async def test__publisher_step__transient_error__defers_rest_of_batch_without_calling_publisher( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + broker = _Broker(failing={"topic.down": TransientError("Kafka broker unreachable: RequestTimedOutError")}) + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(broker.publish) + first, second, third = create_fake_event(topic="topic.down"), create_fake_event(), create_fake_event() + + # Act + for event in (first, second, third): + await step.execute(event, context) + + # Assert + assert broker.published == [] + assert context.completed_ids == [] + assert context.failed_counted == [] + assert [f.event_id for f in context.failed_noncounted] == [first.id, second.id, third.id] + probed, *deferred = context.failed_noncounted + assert probed.next_retry_at is not None + for failure in deferred: + assert failure.error == probed.error + assert failure.next_retry_at is None # the schedule of a row we never touched stays as it is + assert context.statuses[failure.event_id] == EventHandlerStatus.RETRY + + +async def test__publisher_step__next_batch__publishes_again_after_a_deferred_one( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + broker = _Broker(failing={"topic.down": TransientError("Kafka broker unreachable")}) + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(broker.publish) + await step.execute(create_fake_event(topic="topic.down"), context) + next_context: ProcessingContext[OutboxEvent] = ProcessingContext(repo=_Repo(), worker_id="w1") # type: ignore[arg-type] + event = create_fake_event() + + # Act + await step.execute(event, next_context) + + # Assert + assert broker.published == [event] + assert next_context.completed_ids == [event.id] + + +async def test__publisher_step__broker_rejects_the_event__counts_and_the_batch_goes_on( + context: ProcessingContext[OutboxEvent], +) -> None: + # Arrange + broker = _Broker(failing={"topic.bad": ValueError("payload rejected")}) + step: PublisherExecutionStep[OutboxEvent] = PublisherExecutionStep(broker.publish) + bad, good = create_fake_event(topic="topic.bad"), create_fake_event() + + # Act + await step.execute(bad, context) + await step.execute(good, context) + + # Assert + assert [f.event_id for f in context.failed_counted] == [bad.id] + assert context.failed_counted[0].error == "ValueError: payload rejected" + assert context.failed_noncounted == [] + assert context.completed_ids == [good.id] + assert broker.published == [good] diff --git a/tests/unit/core/test_exceptions.py b/tests/unit/core/test_exceptions.py index 27f7842..34b3014 100644 --- a/tests/unit/core/test_exceptions.py +++ b/tests/unit/core/test_exceptions.py @@ -12,6 +12,8 @@ EventLockedByAnotherWorkerError, EventNotLockedError, InvalidEventStateError, + OmniBoxError, + TransientError, UnsupportedCapabilityError, ) @@ -126,3 +128,12 @@ def test__unsupported_capability_error__created__includes_repo_and_capability_in assert exc.capability == "BulkOps" assert exc.repo_type == "MyRepo" assert "MyRepo does not support BulkOps" in str(exc) + + +def test__transient_error__raised__is_an_omni_box_error_carrying_the_message() -> None: + # Arrange / Act + exc = TransientError("Kafka broker unreachable: NodeNotReadyError") + + # Assert + assert isinstance(exc, OmniBoxError) + assert str(exc) == "Kafka broker unreachable: NodeNotReadyError" diff --git a/tests/unit/infra/brokers/kafka/test_publisher_pure.py b/tests/unit/infra/brokers/kafka/test_publisher_pure.py index ae7e7ca..2f225d9 100644 --- a/tests/unit/infra/brokers/kafka/test_publisher_pure.py +++ b/tests/unit/infra/brokers/kafka/test_publisher_pure.py @@ -7,8 +7,17 @@ import orjson import pytest +from aiokafka.errors import ( + KafkaConnectionError, + KafkaTimeoutError, + MessageSizeTooLargeError, + NodeNotReadyError, + RequestTimedOutError, + UnknownTopicOrPartitionError, +) from omni_box.core.converters.event import EventConverter +from omni_box.core.exceptions import TransientError from omni_box.core.models.entities import OutboxEvent from omni_box.infra.brokers.kafka.publisher import KafkaEventPublisher @@ -129,7 +138,7 @@ async def test__publish__transient_error__retries_up_to_max_infra_retries(self) assert producer.send_and_wait.call_count == 3 @pytest.mark.asyncio - async def test__publish__transient_error_exceeds_max_retries__raises(self) -> None: + async def test__publish__transient_error_exceeds_max_retries__raises_transient_error(self) -> None: # Arrange event = _make_event() producer = AsyncMock() @@ -143,11 +152,125 @@ async def test__publish__transient_error_exceeds_max_retries__raises(self) -> No with patch("omni_box.infra.brokers.kafka.publisher.asyncio.sleep", new_callable=AsyncMock): # Act / Assert - with pytest.raises(ConnectionError): + with pytest.raises(TransientError, match="Kafka broker unreachable: ConnectionError: conn") as exc_info: await pub.publish(event, repo) # 3 attempts total (attempt 0, 1, 2) — on attempt==max_infra_retries it raises assert producer.send_and_wait.call_count == 3 + assert isinstance(exc_info.value.__cause__, ConnectionError) + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "exc", + [ + KafkaConnectionError("Unable to bootstrap from [('kafka', 9092)]"), + KafkaTimeoutError(), + NodeNotReadyError("node 1 is not ready"), + RequestTimedOutError(), + ], + ids=["KafkaConnectionError", "KafkaTimeoutError", "NodeNotReadyError", "RequestTimedOutError"], + ) + async def test__publish__broker_does_not_answer__retries_then_raises_transient_error(self, exc: Exception) -> None: + # Arrange + event = _make_event() + producer = AsyncMock() + converter = MagicMock(spec=EventConverter) + converter.convert.return_value = {"k": "v"} + pub = KafkaEventPublisher(producer, converter, max_infra_retries=1) + repo = MagicMock() + producer.send_and_wait.side_effect = exc + + with patch("omni_box.infra.brokers.kafka.publisher.asyncio.sleep", new_callable=AsyncMock): + # Act / Assert + with pytest.raises(TransientError, match="Kafka broker unreachable") as exc_info: + await pub.publish(event, repo) + + # Assert + assert producer.send_and_wait.call_count == 2 # initial + 1 retry + assert exc_info.value.__cause__ is exc + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("exc", "expected"), + [ + (NodeNotReadyError("node 1 is not ready"), "NodeNotReadyError: node 1 is not ready"), + (KafkaTimeoutError(), "KafkaTimeoutError"), + (ConnectionError("connection refused"), "ConnectionError: connection refused"), + ], + ids=["aiokafka error with a message", "aiokafka error without one", "builtin"], + ) + async def test__publish__broker_does_not_answer__names_the_error_once_in_the_reason( + self, exc: Exception, expected: str + ) -> None: + # Arrange + event = _make_event() + producer = AsyncMock() + converter = MagicMock(spec=EventConverter) + converter.convert.return_value = {"k": "v"} + pub = KafkaEventPublisher(producer, converter, max_infra_retries=0) + repo = MagicMock() + producer.send_and_wait.side_effect = exc + + # Act / Assert + with pytest.raises(TransientError) as exc_info: + await pub.publish(event, repo) + + # ``last_error`` is read by an operator: aiokafka's own str() already + # carries the class name, and repeating it would be noise. + assert str(exc_info.value) == f"Kafka broker unreachable: {expected}" + + @pytest.mark.asyncio + async def test__publish__unknown_topic_and_no_answer_to_metadata__raises_transient_error(self) -> None: + # Arrange + event = _make_event(topic="orders.events") + producer = AsyncMock() + producer.client.force_metadata_update = AsyncMock(return_value=False) + converter = MagicMock(spec=EventConverter) + converter.convert.return_value = {"k": "v"} + pub = KafkaEventPublisher(producer, converter, max_infra_retries=0) + repo = MagicMock() + producer.send_and_wait.side_effect = UnknownTopicOrPartitionError() + + # Act / Assert + with pytest.raises(TransientError, match=r"no metadata for topic 'orders\.events'"): + await pub.publish(event, repo) + + producer.client.force_metadata_update.assert_awaited_once() + + @pytest.mark.asyncio + async def test__publish__unknown_topic_on_a_broker_that_answers__raises_as_is_without_retry(self) -> None: + # Arrange + event = _make_event(topic="no.such.topic") + producer = AsyncMock() + producer.client.force_metadata_update = AsyncMock(return_value=True) + converter = MagicMock(spec=EventConverter) + converter.convert.return_value = {"k": "v"} + pub = KafkaEventPublisher(producer, converter, max_infra_retries=3) + repo = MagicMock() + producer.send_and_wait.side_effect = UnknownTopicOrPartitionError() + + # Act / Assert + with pytest.raises(UnknownTopicOrPartitionError): + await pub.publish(event, repo) + + assert producer.send_and_wait.call_count == 1 + + @pytest.mark.asyncio + async def test__publish__broker_rejects_the_record__raises_as_is_without_retry(self) -> None: + # Arrange + event = _make_event() + producer = AsyncMock() + converter = MagicMock(spec=EventConverter) + converter.convert.return_value = {"k": "v"} + pub = KafkaEventPublisher(producer, converter, max_infra_retries=3) + repo = MagicMock() + producer.send_and_wait.side_effect = MessageSizeTooLargeError() + + # Act / Assert + with pytest.raises(MessageSizeTooLargeError): + await pub.publish(event, repo) + + assert producer.send_and_wait.call_count == 1 @pytest.mark.asyncio async def test__publish__non_transient_error__raises_immediately(self) -> None: