Skip to content

Commit 8112f73

Browse files
authored
fix: reject subscriber settings that delete data or never consume (#227)
1 parent ece6614 commit 8112f73

7 files changed

Lines changed: 174 additions & 67 deletions

File tree

‎docs/usage/router.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,8 @@ async def handle_order(order_id: int) -> None:
2222
## No `prefix`
2323

2424
Unlike some FastStream routers, `OutboxRouter` does not accept a
25-
`prefix` argument. Queues are routed by their literal name, so producers
25+
`prefix` argument, and `broker.include_router(router, prefix=...)` raises
26+
`ValueError`. Queues are routed by their literal name, so producers
2627
and consumers must agree on the exact string. If you want namespacing
2728
(e.g., one Postgres instance shared across services), put it in the queue
2829
name itself:

‎docs/usage/subscriber.md‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -90,7 +90,7 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`:
9090
| `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 |
9191
| `max_fetch_interval` | `10.0` s | Ceiling for the adaptive idle backoff (with jitter) |
9292
| `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. |
93-
| `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. |
93+
| `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. |
9494
| `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). |
9595
| `ack_policy` | `AckPolicy.NACK_ON_ERROR` | See [Ack policy](#ack-policy) |
9696
| `retry_strategy` | `ExponentialRetry(...)` | See [Retry strategies](#retry-strategies) |
@@ -108,11 +108,14 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`:
108108
async def handle_urgent(body: dict) -> None: ...
109109
```
110110

111-
Subscriber options are validated when the subscriber is created: likely-wrong
112-
combinations (`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries`
113-
without retry, `min_fetch_interval > max_fetch_interval`, etc.) warn or
114-
raise. Every construction path (`@broker.subscriber`, `@router.subscriber`,
115-
`OutboxRoute`) is checked.
111+
Subscriber options are validated when the subscriber is created. Values that
112+
would delete data or never consume raise: a `max_deliveries` below 1, a NaN or
113+
infinite interval, a `bool` passed as a count, and a queue name that is empty,
114+
longer than 255 characters, or listed twice. Likely-wrong combinations
115+
(`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries` with `NoRetry()`,
116+
`min_fetch_interval > max_fetch_interval`, etc.) warn or raise. Every
117+
construction path (`@broker.subscriber`, `@router.subscriber`, `OutboxRoute`)
118+
is checked.
116119

117120
The table above lists the outbox-specific knobs. The standard FastStream
118121
subscriber kwargs pass through unchanged too: `dependencies`, `parser`,

‎faststream_outbox/registrator.py‎

Lines changed: 26 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -10,34 +10,17 @@
1010
from faststream_outbox.message import OutboxInnerMessage
1111
from faststream_outbox.publisher.factory import create_publisher
1212
from faststream_outbox.publisher.usecase import OutboxPublisher
13-
from faststream_outbox.retry import ExponentialRetry
1413
from faststream_outbox.subscriber.factory import create_subscriber
1514

1615

1716
if TYPE_CHECKING:
1817
from fast_depends.dependencies import Dependant
18+
from faststream._internal.types import BrokerMiddleware
1919

2020
from faststream_outbox.retry import RetryStrategyProto
2121
from faststream_outbox.subscriber.usecase import OutboxSubscriber
2222

2323

24-
def _default_retry_strategy() -> "RetryStrategyProto":
25-
"""Fallback retry policy when the user passes nothing.
26-
27-
An outbox is a reliability primitive; defaulting to "delete on first error" turns
28-
every transient handler failure into silent data loss. Defaulting to a bounded
29-
exponential retry keeps the contract intuitive — users who actually want
30-
delete-on-error opt in explicitly with ``NoRetry()``.
31-
"""
32-
return ExponentialRetry(
33-
initial_delay_seconds=1.0,
34-
multiplier=2.0,
35-
max_delay_seconds=300.0,
36-
max_attempts=10,
37-
jitter_factor=0.2,
38-
)
39-
40-
4124
class OutboxRegistrator(Registrator[OutboxInnerMessage, "OutboxBrokerConfig"]): # ty: ignore[unresolved-reference]
4225
@override
4326
def subscriber( # ty: ignore[invalid-method-override]
@@ -65,11 +48,10 @@ def subscriber( # ty: ignore[invalid-method-override]
6548
if not queue_list:
6649
msg = "subscriber() requires at least one queue name"
6750
raise ValueError(msg)
68-
resolved_retry_strategy = retry_strategy if retry_strategy is not None else _default_retry_strategy()
6951
subscriber = create_subscriber(
7052
queues=queue_list,
7153
max_workers=max_workers,
72-
retry_strategy=resolved_retry_strategy,
54+
retry_strategy=retry_strategy,
7355
fetch_batch_size=fetch_batch_size,
7456
min_fetch_interval=min_fetch_interval,
7557
max_fetch_interval=max_fetch_interval,
@@ -99,6 +81,30 @@ def subscriber( # ty: ignore[invalid-method-override]
9981
dependencies_=dependencies,
10082
)
10183

84+
@override
85+
def include_router(
86+
self,
87+
router: "Registrator[OutboxInnerMessage, Any]",
88+
*,
89+
prefix: str = "",
90+
dependencies: Sequence["Dependant"] = (),
91+
middlewares: Sequence["BrokerMiddleware[OutboxInnerMessage]"] = (),
92+
include_in_schema: bool | None = None,
93+
) -> None:
94+
if prefix:
95+
msg = (
96+
f"include_router(prefix={prefix!r}) is not supported: outbox queues are matched by their "
97+
"literal name, so a prefixed subscriber would never receive a row. Put the namespace "
98+
"in the queue names instead."
99+
)
100+
raise ValueError(msg)
101+
super().include_router(
102+
router,
103+
dependencies=dependencies,
104+
middlewares=middlewares,
105+
include_in_schema=include_in_schema,
106+
)
107+
102108
@override
103109
def publisher( # ty: ignore[invalid-method-override]
104110
self,

‎faststream_outbox/retry.py‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -176,3 +176,20 @@ def _delay_seconds(self, *, attempts_count: int) -> float:
176176
# Absolute backstop so an unbounded config can't emit a delay Postgres'
177177
# make_interval() can't represent.
178178
return min(delay, _MAX_DELAY_SECONDS)
179+
180+
181+
def _default_retry_strategy() -> "RetryStrategyProto":
182+
"""Fallback retry policy when the user passes nothing.
183+
184+
An outbox is a reliability primitive; defaulting to "delete on first error" turns
185+
every transient handler failure into silent data loss. Defaulting to a bounded
186+
exponential retry keeps the contract intuitive — users who actually want
187+
delete-on-error opt in explicitly with ``NoRetry()``.
188+
"""
189+
return ExponentialRetry(
190+
initial_delay_seconds=1.0,
191+
multiplier=2.0,
192+
max_delay_seconds=300.0,
193+
max_attempts=10,
194+
jitter_factor=0.2,
195+
)

‎faststream_outbox/subscriber/config.py‎

Lines changed: 44 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import math
12
import sys
23
import typing
34
import warnings
@@ -9,7 +10,8 @@
910
from faststream._internal.constants import EMPTY
1011
from faststream.middlewares import AckPolicy
1112

12-
from faststream_outbox.retry import NoRetry
13+
from faststream_outbox.response import _MAX_QUEUE_LENGTH
14+
from faststream_outbox.retry import NoRetry, _default_retry_strategy
1315

1416

1517
if typing.TYPE_CHECKING:
@@ -45,6 +47,23 @@ def _subscriber_warn(message: str) -> None:
4547
warnings.warn(message, UserWarning, stacklevel=stacklevel)
4648

4749

50+
def _validate_queues(queues: list[str]) -> None:
51+
"""Reject queue names that no published row can ever match, with the same rules as publish."""
52+
for queue in queues:
53+
if not isinstance(queue, str):
54+
msg = f"queue must be a str, got {type(queue).__name__}"
55+
raise TypeError(msg)
56+
if not queue:
57+
msg = "queue must be a non-empty string"
58+
raise ValueError(msg)
59+
if len(queue) > _MAX_QUEUE_LENGTH:
60+
msg = f"queue must be at most {_MAX_QUEUE_LENGTH} characters (got {len(queue)})"
61+
raise ValueError(msg)
62+
if duplicates := sorted({q for q in queues if queues.count(q) > 1}):
63+
msg = f"duplicate queue names in one subscriber: {duplicates}"
64+
raise ValueError(msg)
65+
66+
4867
@dataclass(kw_only=True)
4968
class OutboxSubscriberConfig(SubscriberUsecaseConfig):
5069
_outer_config: "OutboxBrokerConfig" # ty: ignore[invalid-mutable-override]
@@ -70,6 +89,8 @@ def __post_init__(self) -> None:
7089
if parent_post_init is not None:
7190
parent_post_init()
7291
self._validate()
92+
if self.retry_strategy is None:
93+
self.retry_strategy = _default_retry_strategy()
7394

7495
def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob checks
7596
"""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
84105
# did (e.g. the NACK+NoRetry warning fires only on an explicit NACK_ON_ERROR, not
85106
# on the default that resolves to NACK_ON_ERROR via the ``ack_policy`` property).
86107
ack_policy = None if self._ack_policy is EMPTY else self._ack_policy
108+
_validate_queues(self.queues)
109+
self._validate_numbers()
87110
if self.max_workers <= 0:
88111
msg = f"max_workers must be >= 1, got {self.max_workers}"
89112
raise ValueError(msg)
@@ -132,11 +155,10 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob
132155
"ack_policy=NACK_ON_ERROR with retry_strategy=NoRetry() has the same effect "
133156
"as REJECT_ON_ERROR (one attempt, then delete). Pick one for clarity.",
134157
)
135-
if self.max_deliveries is not None and (self.retry_strategy is None or is_no_retry):
158+
if self.max_deliveries is not None and is_no_retry:
136159
_subscriber_warn(
137-
"max_deliveries is set but no retry_strategy is configured (or NoRetry was "
138-
"passed); the delivery cap is unreachable on the happy path since the row "
139-
"is deleted after the first attempt.",
160+
"max_deliveries is set but retry_strategy=NoRetry(); the delivery cap is "
161+
"unreachable on the happy path since the row is deleted after the first attempt.",
140162
)
141163
if self.lease_ttl_seconds <= self.max_fetch_interval:
142164
_subscriber_warn(
@@ -147,6 +169,23 @@ def _validate(self) -> None: # noqa: C901 # flat sequence of independent knob
147169
f"2 * max_fetch_interval + P99(handler).",
148170
)
149171

172+
def _validate_numbers(self) -> None:
173+
"""Reject bools passed as counts, NaN or infinite intervals, and a ``max_deliveries`` below 1."""
174+
for name in ("max_workers", "fetch_batch_size", "terminal_flush_batch_size", "max_deliveries"):
175+
value = getattr(self, name)
176+
if value is None and name == "max_deliveries":
177+
continue
178+
if isinstance(value, bool) or not isinstance(value, int):
179+
msg = f"{name} must be an int, got {value!r}"
180+
raise TypeError(msg)
181+
for name in ("min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds"):
182+
if not math.isfinite(getattr(self, name)):
183+
msg = f"{name} must be a finite number, got {getattr(self, name)}"
184+
raise ValueError(msg)
185+
if self.max_deliveries is not None and self.max_deliveries < 1:
186+
msg = f"max_deliveries must be >= 1 or None (unlimited), got {self.max_deliveries}"
187+
raise ValueError(msg)
188+
150189
@property
151190
def ack_policy(self) -> AckPolicy:
152191
if self._ack_policy is EMPTY:

‎tests/test_fake.py‎

Lines changed: 7 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@
33
import logging
44
import typing
55
import uuid
6-
import warnings as _warnings
76
from collections.abc import Awaitable, Callable, Mapping
87
from types import TracebackType
98
from unittest.mock import AsyncMock, MagicMock, patch
@@ -483,18 +482,7 @@ async def test_outbox_route_accepts_ack_policy() -> None:
483482
async def handle(body: str) -> None:
484483
del body
485484

486-
# REJECT_ON_ERROR + default ExponentialRetry triggers a misconfig warning at
487-
# registration; pair with NoRetry() to keep the test focused on ack_policy.
488-
router = OutboxRouter(
489-
handlers=(
490-
OutboxRoute(
491-
handle,
492-
queues="orders",
493-
ack_policy=AckPolicy.REJECT_ON_ERROR,
494-
retry_strategy=NoRetry(),
495-
),
496-
),
497-
)
485+
router = OutboxRouter(handlers=(OutboxRoute(handle, queues="orders", ack_policy=AckPolicy.REJECT_ON_ERROR),))
498486

499487
broker = _make_broker()
500488
broker.include_router(router)
@@ -1356,17 +1344,11 @@ async def test_fake_broker_reject_on_error_deletes_row_on_first_failure() -> Non
13561344
broker = _make_broker()
13571345
attempts: list[str] = []
13581346

1359-
with _warnings.catch_warnings():
1360-
# REJECT_ON_ERROR + the default exponential retry triggers the misconfig warning;
1361-
# this test asserts only the runtime semantics. Coverage for the warning itself
1362-
# lives in tests/test_unit.py::test_subscriber_warns_on_reject_with_retry_strategy.
1363-
_warnings.simplefilter("ignore", UserWarning)
1364-
1365-
@broker.subscriber("orders", ack_policy=AckPolicy.REJECT_ON_ERROR)
1366-
async def handle(body: str) -> None:
1367-
attempts.append(body)
1368-
msg = "boom"
1369-
raise RuntimeError(msg)
1347+
@broker.subscriber("orders", ack_policy=AckPolicy.REJECT_ON_ERROR)
1348+
async def handle(body: str) -> None:
1349+
attempts.append(body)
1350+
msg = "boom"
1351+
raise RuntimeError(msg)
13701352

13711353
test_broker = TestOutboxBroker(broker)
13721354
async with test_broker:
@@ -1914,15 +1896,7 @@ async def test_fake_dlq_drops_last_exception_when_renderer_returns_none() -> Non
19141896

19151897
async def test_fake_dlq_captures_rejected_failure() -> None:
19161898
"""``REJECT_ON_ERROR`` ack policy + raising handler → DLQ row with reason ``rejected``."""
1917-
with _warnings.catch_warnings():
1918-
# REJECT_ON_ERROR + a retry strategy triggers a misconfig warning at registration
1919-
# (the retry strategy is ignored). Pass NoRetry() to match the policy and ignore
1920-
# the warning — the runtime semantics we care about (single attempt → DLQ) hold.
1921-
_warnings.simplefilter("ignore", UserWarning)
1922-
broker, _received, raised = _make_broker_with_dlq(
1923-
ack_policy=AckPolicy.REJECT_ON_ERROR,
1924-
retry_strategy=NoRetry(),
1925-
)
1899+
broker, _received, raised = _make_broker_with_dlq(ack_policy=AckPolicy.REJECT_ON_ERROR)
19261900
raised.append(ValueError("poison message"))
19271901
test_broker = TestOutboxBroker(broker)
19281902

‎tests/test_unit.py‎

Lines changed: 69 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,8 +53,8 @@
5353
from faststream_outbox.publisher.fake import OutboxFakePublisher
5454
from faststream_outbox.publisher.producer import OutboxProducer
5555
from faststream_outbox.publisher.specification import OutboxPublisherSpecification
56-
from faststream_outbox.registrator import _default_retry_strategy
5756
from faststream_outbox.response import OutboxPublishCommand
57+
from faststream_outbox.retry import _default_retry_strategy
5858
from faststream_outbox.router import OutboxRoute
5959
from faststream_outbox.schema_validation import (
6060
_AUTOGEN_BLIND_HINT,
@@ -3289,7 +3289,7 @@ def test_subscriber_warns_on_nack_with_no_retry() -> None:
32893289

32903290
def test_subscriber_warns_on_max_deliveries_with_no_retry() -> None:
32913291
broker = _make_broker()
3292-
with pytest.warns(UserWarning, match="max_deliveries is set but no retry_strategy"):
3292+
with pytest.warns(UserWarning, match=r"max_deliveries is set but retry_strategy=NoRetry\(\)"):
32933293
_register_subscriber(broker, max_deliveries=5, retry_strategy=NoRetry())
32943294

32953295

@@ -3317,6 +3317,73 @@ def test_subscriber_reject_on_error_with_no_retry_is_silent() -> None:
33173317
_register_subscriber(broker, ack_policy=AckPolicy.REJECT_ON_ERROR, retry_strategy=NoRetry())
33183318

33193319

3320+
@pytest.mark.parametrize("max_deliveries", [0, -1])
3321+
def test_subscriber_rejects_max_deliveries_below_one(max_deliveries: int) -> None:
3322+
"""INVARIANT: ``max_deliveries`` below 1 is refused at registration.
3323+
3324+
Every row has ``deliveries_count >= 1`` by the time it is checked, so a cap of 0 or less marks
3325+
every row over the limit and deletes the queue (or sends it to the DLQ) without the handler ever
3326+
running. Someone reading 0 as "unlimited" would lose the whole queue.
3327+
"""
3328+
with pytest.raises(ValueError, match="max_deliveries must be >= 1"):
3329+
_register_subscriber(_make_broker(), max_deliveries=max_deliveries)
3330+
3331+
3332+
@pytest.mark.parametrize("field", ["min_fetch_interval", "max_fetch_interval", "lease_ttl_seconds"])
3333+
@pytest.mark.parametrize("value", [math.nan, math.inf])
3334+
def test_subscriber_rejects_non_finite_intervals(field: str, value: float) -> None:
3335+
with pytest.raises(ValueError, match=f"{field} must be a finite number"):
3336+
_register_subscriber(_make_broker(), **{field: value})
3337+
3338+
3339+
@pytest.mark.parametrize("field", ["max_workers", "fetch_batch_size", "terminal_flush_batch_size", "max_deliveries"])
3340+
def test_subscriber_rejects_bool_for_integer_settings(field: str) -> None:
3341+
with pytest.raises(TypeError, match=f"{field} must be an int"):
3342+
_register_subscriber(_make_broker(), **{field: True})
3343+
3344+
3345+
@pytest.mark.parametrize(
3346+
("queues", "match"),
3347+
[
3348+
("", "queue must be a non-empty string"),
3349+
("q" * 256, "queue must be at most 255 characters"),
3350+
(["orders", "orders"], "duplicate queue"),
3351+
],
3352+
)
3353+
def test_subscriber_rejects_queues_that_can_never_match(queues: str | list[str], match: str) -> None:
3354+
with pytest.raises(ValueError, match=match):
3355+
_make_broker().subscriber(queues)
3356+
3357+
3358+
def test_subscriber_rejects_a_non_string_queue() -> None:
3359+
with pytest.raises(TypeError, match="queue must be a str"):
3360+
_make_broker().subscriber([1]) # ty: ignore[invalid-argument-type]
3361+
3362+
3363+
def test_include_router_rejects_a_prefix() -> None:
3364+
broker = _make_broker()
3365+
router = OutboxRouter()
3366+
3367+
@router.subscriber("orders")
3368+
async def handle(body: dict) -> None: ...
3369+
3370+
with pytest.raises(ValueError, match="prefix"):
3371+
broker.include_router(router, prefix="billing.")
3372+
3373+
3374+
def test_subscriber_reject_on_error_without_retry_strategy_is_silent() -> None:
3375+
with warnings.catch_warnings():
3376+
warnings.simplefilter("error", UserWarning)
3377+
_register_subscriber(_make_broker(), ack_policy=AckPolicy.REJECT_ON_ERROR)
3378+
3379+
3380+
def test_subscriber_max_deliveries_with_the_default_retry_strategy_is_silent() -> None:
3381+
with warnings.catch_warnings():
3382+
warnings.simplefilter("error", UserWarning)
3383+
sub = _make_broker().subscriber("orders", max_deliveries=5)
3384+
assert isinstance(sub._config.retry_strategy, ExponentialRetry) # noqa: SLF001
3385+
3386+
33203387
# --- MetricsRecorder seam ------------------------------------------------------------
33213388

33223389

0 commit comments

Comments
 (0)