From 72349e3d3edb876954c8b3200d433d3d5d3178cc Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 13:59:46 +0300 Subject: [PATCH] fix: reject subscriber settings that delete data or never consume --- docs/usage/router.md | 3 +- docs/usage/subscriber.md | 15 +++--- faststream_outbox/registrator.py | 46 +++++++++-------- faststream_outbox/retry.py | 17 ++++++ faststream_outbox/subscriber/config.py | 49 ++++++++++++++++-- tests/test_fake.py | 40 +++------------ tests/test_unit.py | 71 +++++++++++++++++++++++++- 7 files changed, 174 insertions(+), 67 deletions(-) diff --git a/docs/usage/router.md b/docs/usage/router.md index e25e14b..344ca08 100644 --- a/docs/usage/router.md +++ b/docs/usage/router.md @@ -22,7 +22,8 @@ async def handle_order(order_id: int) -> None: ## No `prefix` Unlike some FastStream routers, `OutboxRouter` does not accept a -`prefix` argument. Queues are routed by their literal name, so producers +`prefix` argument, and `broker.include_router(router, prefix=...)` raises +`ValueError`. Queues are routed by their literal name, so producers and consumers must agree on the exact string. If you want namespacing (e.g., one Postgres instance shared across services), put it in the queue name itself: diff --git a/docs/usage/subscriber.md b/docs/usage/subscriber.md index 5b6d475..caf53ab 100644 --- a/docs/usage/subscriber.md +++ b/docs/usage/subscriber.md @@ -90,7 +90,7 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`: | `min_fetch_interval` | `1.0` s | Base for the adaptive idle backoff (jittered ±50%, so an actual wait can land below it); no sleep at all while fetches keep returning rows | | `max_fetch_interval` | `10.0` s | Ceiling for the adaptive idle backoff (with jitter) | | `lease_ttl_seconds` | `60.0` s | How long a claim is valid before another fetch may reclaim it. Must exceed your handler's P99 with margin. | -| `max_deliveries` | `None` (unbounded) | Total claims (including lease-expiry re-claims) after which the row is dropped without invoking the handler. Defends against handlers that consistently wedge. | +| `max_deliveries` | `None` (unbounded) | Total claims (including lease-expiry re-claims) after which the row is dropped without invoking the handler. Defends against handlers that consistently wedge. Must be at least `1`; use `None`, not `0`, for no cap. | | `terminal_flush_batch_size` | `1` (off) | Coalesce completed terminal `DELETE`s into one `DELETE … RETURNING` per N rows. `1` is one round-trip per message (unchanged). Higher trades a wider crash-redelivery window for far fewer round-trips. See [Batching terminal deletes](#batching-terminal-deletes). | | `ack_policy` | `AckPolicy.NACK_ON_ERROR` | See [Ack policy](#ack-policy) | | `retry_strategy` | `ExponentialRetry(...)` | See [Retry strategies](#retry-strategies) | @@ -108,11 +108,14 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`: async def handle_urgent(body: dict) -> None: ... ``` -Subscriber options are validated when the subscriber is created: likely-wrong -combinations (`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries` -without retry, `min_fetch_interval > max_fetch_interval`, etc.) warn or -raise. Every construction path (`@broker.subscriber`, `@router.subscriber`, -`OutboxRoute`) is checked. +Subscriber options are validated when the subscriber is created. Values that +would delete data or never consume raise: a `max_deliveries` below 1, a NaN or +infinite interval, a `bool` passed as a count, and a queue name that is empty, +longer than 255 characters, or listed twice. Likely-wrong combinations +(`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries` with `NoRetry()`, +`min_fetch_interval > max_fetch_interval`, etc.) warn or raise. Every +construction path (`@broker.subscriber`, `@router.subscriber`, `OutboxRoute`) +is checked. The table above lists the outbox-specific knobs. The standard FastStream subscriber kwargs pass through unchanged too: `dependencies`, `parser`, diff --git a/faststream_outbox/registrator.py b/faststream_outbox/registrator.py index ccac147..28acd54 100644 --- a/faststream_outbox/registrator.py +++ b/faststream_outbox/registrator.py @@ -10,34 +10,17 @@ from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.publisher.factory import create_publisher from faststream_outbox.publisher.usecase import OutboxPublisher -from faststream_outbox.retry import ExponentialRetry from faststream_outbox.subscriber.factory import create_subscriber if TYPE_CHECKING: from fast_depends.dependencies import Dependant + from faststream._internal.types import BrokerMiddleware from faststream_outbox.retry import RetryStrategyProto from faststream_outbox.subscriber.usecase import OutboxSubscriber -def _default_retry_strategy() -> "RetryStrategyProto": - """Fallback retry policy when the user passes nothing. - - An outbox is a reliability primitive; defaulting to "delete on first error" turns - every transient handler failure into silent data loss. Defaulting to a bounded - exponential retry keeps the contract intuitive — users who actually want - delete-on-error opt in explicitly with ``NoRetry()``. - """ - return ExponentialRetry( - initial_delay_seconds=1.0, - multiplier=2.0, - max_delay_seconds=300.0, - max_attempts=10, - jitter_factor=0.2, - ) - - class OutboxRegistrator(Registrator[OutboxInnerMessage, "OutboxBrokerConfig"]): # ty: ignore[unresolved-reference] @override def subscriber( # ty: ignore[invalid-method-override] @@ -65,11 +48,10 @@ def subscriber( # ty: ignore[invalid-method-override] if not queue_list: msg = "subscriber() requires at least one queue name" raise ValueError(msg) - resolved_retry_strategy = retry_strategy if retry_strategy is not None else _default_retry_strategy() subscriber = create_subscriber( queues=queue_list, max_workers=max_workers, - retry_strategy=resolved_retry_strategy, + retry_strategy=retry_strategy, fetch_batch_size=fetch_batch_size, min_fetch_interval=min_fetch_interval, max_fetch_interval=max_fetch_interval, @@ -99,6 +81,30 @@ def subscriber( # ty: ignore[invalid-method-override] dependencies_=dependencies, ) + @override + def include_router( + self, + router: "Registrator[OutboxInnerMessage, Any]", + *, + prefix: str = "", + dependencies: Sequence["Dependant"] = (), + middlewares: Sequence["BrokerMiddleware[OutboxInnerMessage]"] = (), + include_in_schema: bool | None = None, + ) -> None: + if prefix: + msg = ( + f"include_router(prefix={prefix!r}) is not supported: outbox queues are matched by their " + "literal name, so a prefixed subscriber would never receive a row. Put the namespace " + "in the queue names instead." + ) + raise ValueError(msg) + super().include_router( + router, + dependencies=dependencies, + middlewares=middlewares, + include_in_schema=include_in_schema, + ) + @override def publisher( # ty: ignore[invalid-method-override] self, diff --git a/faststream_outbox/retry.py b/faststream_outbox/retry.py index 6fab447..d866847 100644 --- a/faststream_outbox/retry.py +++ b/faststream_outbox/retry.py @@ -176,3 +176,20 @@ def _delay_seconds(self, *, attempts_count: int) -> float: # Absolute backstop so an unbounded config can't emit a delay Postgres' # make_interval() can't represent. return min(delay, _MAX_DELAY_SECONDS) + + +def _default_retry_strategy() -> "RetryStrategyProto": + """Fallback retry policy when the user passes nothing. + + An outbox is a reliability primitive; defaulting to "delete on first error" turns + every transient handler failure into silent data loss. Defaulting to a bounded + exponential retry keeps the contract intuitive — users who actually want + delete-on-error opt in explicitly with ``NoRetry()``. + """ + return ExponentialRetry( + initial_delay_seconds=1.0, + multiplier=2.0, + max_delay_seconds=300.0, + max_attempts=10, + jitter_factor=0.2, + ) diff --git a/faststream_outbox/subscriber/config.py b/faststream_outbox/subscriber/config.py index 6c84634..a3c3161 100644 --- a/faststream_outbox/subscriber/config.py +++ b/faststream_outbox/subscriber/config.py @@ -1,3 +1,4 @@ +import math import sys import typing import warnings @@ -9,7 +10,8 @@ from faststream._internal.constants import EMPTY from faststream.middlewares import AckPolicy -from faststream_outbox.retry import NoRetry +from faststream_outbox.response import _MAX_QUEUE_LENGTH +from faststream_outbox.retry import NoRetry, _default_retry_strategy if typing.TYPE_CHECKING: @@ -45,6 +47,23 @@ def _subscriber_warn(message: str) -> None: warnings.warn(message, UserWarning, stacklevel=stacklevel) +def _validate_queues(queues: list[str]) -> None: + """Reject queue names that no published row can ever match, with the same rules as publish.""" + for queue in queues: + if not isinstance(queue, str): + msg = f"queue must be a str, got {type(queue).__name__}" + raise TypeError(msg) + if not queue: + msg = "queue must be a non-empty string" + raise ValueError(msg) + if len(queue) > _MAX_QUEUE_LENGTH: + msg = f"queue must be at most {_MAX_QUEUE_LENGTH} characters (got {len(queue)})" + raise ValueError(msg) + if duplicates := sorted({q for q in queues if queues.count(q) > 1}): + msg = f"duplicate queue names in one subscriber: {duplicates}" + raise ValueError(msg) + + @dataclass(kw_only=True) class OutboxSubscriberConfig(SubscriberUsecaseConfig): _outer_config: "OutboxBrokerConfig" # ty: ignore[invalid-mutable-override] @@ -70,6 +89,8 @@ def __post_init__(self) -> None: if parent_post_init is not None: parent_post_init() self._validate() + if self.retry_strategy is None: + self.retry_strategy = _default_retry_strategy() def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob checks """Reject impossible knob values, warn on combos that silently misbehave. @@ -84,6 +105,8 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob # did (e.g. the NACK+NoRetry warning fires only on an explicit NACK_ON_ERROR, not # on the default that resolves to NACK_ON_ERROR via the ``ack_policy`` property). ack_policy = None if self._ack_policy is EMPTY else self._ack_policy + _validate_queues(self.queues) + self._validate_numbers() if self.max_workers <= 0: msg = f"max_workers must be >= 1, got {self.max_workers}" raise ValueError(msg) @@ -132,11 +155,10 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob "ack_policy=NACK_ON_ERROR with retry_strategy=NoRetry() has the same effect " "as REJECT_ON_ERROR (one attempt, then delete). Pick one for clarity.", ) - if self.max_deliveries is not None and (self.retry_strategy is None or is_no_retry): + if self.max_deliveries is not None and is_no_retry: _subscriber_warn( - "max_deliveries is set but no retry_strategy is configured (or NoRetry was " - "passed); the delivery cap is unreachable on the happy path since the row " - "is deleted after the first attempt.", + "max_deliveries is set but retry_strategy=NoRetry(); the delivery cap is " + "unreachable on the happy path since the row is deleted after the first attempt.", ) if self.lease_ttl_seconds <= self.max_fetch_interval: _subscriber_warn( @@ -147,6 +169,23 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob f"2 * max_fetch_interval + P99(handler).", ) + def _validate_numbers(self) -> None: + """Reject bools passed as counts, NaN or infinite intervals, and a ``max_deliveries`` below 1.""" + for name in ("max_workers", "fetch_batch_size", "terminal_flush_batch_size", "max_deliveries"): + value = getattr(self, name) + if value is None and name == "max_deliveries": + continue + if isinstance(value, bool) or not isinstance(value, int): + msg = f"{name} must be an int, got {value!r}" + raise TypeError(msg) + for name in ("min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds"): + if not math.isfinite(getattr(self, name)): + msg = f"{name} must be a finite number, got {getattr(self, name)}" + raise ValueError(msg) + if self.max_deliveries is not None and self.max_deliveries < 1: + msg = f"max_deliveries must be >= 1 or None (unlimited), got {self.max_deliveries}" + raise ValueError(msg) + @property def ack_policy(self) -> AckPolicy: if self._ack_policy is EMPTY: diff --git a/tests/test_fake.py b/tests/test_fake.py index c377e50..abf2168 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -3,7 +3,6 @@ import logging import typing import uuid -import warnings as _warnings from collections.abc import Awaitable, Callable, Mapping from types import TracebackType from unittest.mock import AsyncMock, MagicMock, patch @@ -483,18 +482,7 @@ async def test_outbox_route_accepts_ack_policy() -> None: async def handle(body: str) -> None: del body - # REJECT_ON_ERROR + default ExponentialRetry triggers a misconfig warning at - # registration; pair with NoRetry() to keep the test focused on ack_policy. - router = OutboxRouter( - handlers=( - OutboxRoute( - handle, - queues="orders", - ack_policy=AckPolicy.REJECT_ON_ERROR, - retry_strategy=NoRetry(), - ), - ), - ) + router = OutboxRouter(handlers=(OutboxRoute(handle, queues="orders", ack_policy=AckPolicy.REJECT_ON_ERROR),)) broker = _make_broker() broker.include_router(router) @@ -1356,17 +1344,11 @@ async def test_fake_broker_reject_on_error_deletes_row_on_first_failure() -> Non broker = _make_broker() attempts: list[str] = [] - with _warnings.catch_warnings(): - # REJECT_ON_ERROR + the default exponential retry triggers the misconfig warning; - # this test asserts only the runtime semantics. Coverage for the warning itself - # lives in tests/test_unit.py::test_subscriber_warns_on_reject_with_retry_strategy. - _warnings.simplefilter("ignore", UserWarning) - - @broker.subscriber("orders", ack_policy=AckPolicy.REJECT_ON_ERROR) - async def handle(body: str) -> None: - attempts.append(body) - msg = "boom" - raise RuntimeError(msg) + @broker.subscriber("orders", ack_policy=AckPolicy.REJECT_ON_ERROR) + async def handle(body: str) -> None: + attempts.append(body) + msg = "boom" + raise RuntimeError(msg) test_broker = TestOutboxBroker(broker) async with test_broker: @@ -1914,15 +1896,7 @@ async def test_fake_dlq_drops_last_exception_when_renderer_returns_none() -> Non async def test_fake_dlq_captures_rejected_failure() -> None: """``REJECT_ON_ERROR`` ack policy + raising handler → DLQ row with reason ``rejected``.""" - with _warnings.catch_warnings(): - # REJECT_ON_ERROR + a retry strategy triggers a misconfig warning at registration - # (the retry strategy is ignored). Pass NoRetry() to match the policy and ignore - # the warning — the runtime semantics we care about (single attempt → DLQ) hold. - _warnings.simplefilter("ignore", UserWarning) - broker, _received, raised = _make_broker_with_dlq( - ack_policy=AckPolicy.REJECT_ON_ERROR, - retry_strategy=NoRetry(), - ) + broker, _received, raised = _make_broker_with_dlq(ack_policy=AckPolicy.REJECT_ON_ERROR) raised.append(ValueError("poison message")) test_broker = TestOutboxBroker(broker) diff --git a/tests/test_unit.py b/tests/test_unit.py index f593441..c4fb41d 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -53,8 +53,8 @@ from faststream_outbox.publisher.fake import OutboxFakePublisher from faststream_outbox.publisher.producer import OutboxProducer from faststream_outbox.publisher.specification import OutboxPublisherSpecification -from faststream_outbox.registrator import _default_retry_strategy from faststream_outbox.response import OutboxPublishCommand +from faststream_outbox.retry import _default_retry_strategy from faststream_outbox.router import OutboxRoute from faststream_outbox.schema_validation import ( _AUTOGEN_BLIND_HINT, @@ -3289,7 +3289,7 @@ def test_subscriber_warns_on_nack_with_no_retry() -> None: def test_subscriber_warns_on_max_deliveries_with_no_retry() -> None: broker = _make_broker() - with pytest.warns(UserWarning, match="max_deliveries is set but no retry_strategy"): + with pytest.warns(UserWarning, match=r"max_deliveries is set but retry_strategy=NoRetry\(\)"): _register_subscriber(broker, max_deliveries=5, retry_strategy=NoRetry()) @@ -3317,6 +3317,73 @@ def test_subscriber_reject_on_error_with_no_retry_is_silent() -> None: _register_subscriber(broker, ack_policy=AckPolicy.REJECT_ON_ERROR, retry_strategy=NoRetry()) +@pytest.mark.parametrize("max_deliveries", [0, -1]) +def test_subscriber_rejects_max_deliveries_below_one(max_deliveries: int) -> None: + """INVARIANT: ``max_deliveries`` below 1 is refused at registration. + + Every row has ``deliveries_count >= 1`` by the time it is checked, so a cap of 0 or less marks + every row over the limit and deletes the queue (or sends it to the DLQ) without the handler ever + running. Someone reading 0 as "unlimited" would lose the whole queue. + """ + with pytest.raises(ValueError, match="max_deliveries must be >= 1"): + _register_subscriber(_make_broker(), max_deliveries=max_deliveries) + + +@pytest.mark.parametrize("field", ["min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds"]) +@pytest.mark.parametrize("value", [math.nan, math.inf]) +def test_subscriber_rejects_non_finite_intervals(field: str, value: float) -> None: + with pytest.raises(ValueError, match=f"{field} must be a finite number"): + _register_subscriber(_make_broker(), **{field: value}) + + +@pytest.mark.parametrize("field", ["max_workers", "fetch_batch_size", "terminal_flush_batch_size", "max_deliveries"]) +def test_subscriber_rejects_bool_for_integer_settings(field: str) -> None: + with pytest.raises(TypeError, match=f"{field} must be an int"): + _register_subscriber(_make_broker(), **{field: True}) + + +@pytest.mark.parametrize( + ("queues", "match"), + [ + ("", "queue must be a non-empty string"), + ("q" * 256, "queue must be at most 255 characters"), + (["orders", "orders"], "duplicate queue"), + ], +) +def test_subscriber_rejects_queues_that_can_never_match(queues: str | list[str], match: str) -> None: + with pytest.raises(ValueError, match=match): + _make_broker().subscriber(queues) + + +def test_subscriber_rejects_a_non_string_queue() -> None: + with pytest.raises(TypeError, match="queue must be a str"): + _make_broker().subscriber([1]) # ty: ignore[invalid-argument-type] + + +def test_include_router_rejects_a_prefix() -> None: + broker = _make_broker() + router = OutboxRouter() + + @router.subscriber("orders") + async def handle(body: dict) -> None: ... + + with pytest.raises(ValueError, match="prefix"): + broker.include_router(router, prefix="billing.") + + +def test_subscriber_reject_on_error_without_retry_strategy_is_silent() -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) + _register_subscriber(_make_broker(), ack_policy=AckPolicy.REJECT_ON_ERROR) + + +def test_subscriber_max_deliveries_with_the_default_retry_strategy_is_silent() -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) + sub = _make_broker().subscriber("orders", max_deliveries=5) + assert isinstance(sub._config.retry_strategy, ExponentialRetry) # noqa: SLF001 + + # --- MetricsRecorder seam ------------------------------------------------------------