From 5db19f25529950ed01e7c2531afb3e61fa58499b Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 18:23:36 +0300 Subject: [PATCH 01/14] chore: add faststream[kafka] to dev deps for relay tests Co-Authored-By: Claude Opus 4.7 (1M context) --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 6a4244f..e18bbc8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ dev = [ "asyncpg>=0.29", "alembic>=1.13", "fastapi>=0.95", + "faststream[kafka]>=0.7.1,<0.8", "httpx2>=2.2", "prometheus-client>=0.19", "opentelemetry-api>=1.20", From bd2e0d1b1f1f447e19331f666e36ee96daa1574d Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 18:27:23 +0300 Subject: [PATCH 02/14] test: baseline relay from OutboxSubscriber to TestKafkaBroker Pins the FastStream-native cross-broker chain against OutboxSubscriber with no guardrail code in place: a plain handler return relays to Kafka via the @publisher decorator stack. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_relay.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 tests/test_relay.py diff --git a/tests/test_relay.py b/tests/test_relay.py new file mode 100644 index 0000000..8d4282d --- /dev/null +++ b/tests/test_relay.py @@ -0,0 +1,34 @@ +from typing import Any + +import pytest +from sqlalchemy import MetaData +from faststream.kafka import KafkaBroker, TestKafkaBroker + +from faststream_outbox import OutboxBroker, make_outbox_table +from faststream_outbox.testing import TestOutboxBroker + + +pytestmark = pytest.mark.asyncio + + +async def test_naked_decorator_chain_relays_plain_return_to_kafka() -> None: + """A handler decorated `@kafka_pub @outbox.subscriber(...)` returning a plain + value publishes the value through the Kafka publisher chain.""" + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue") + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + # TestKafkaBroker first, TestOutboxBroker second — see "Context-manager + # ordering note" at the end of this plan. The foreign broker's mock producer + # must be wired before our subscriber starts (Task 6 introduces a startup + # warning that probes foreign producers). + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + await outbox.publish({"hello": "world"}, queue="relay_queue", session=None) + publisher_kafka.mock.assert_called_once_with({"hello": "world"}) From c8267be3ce98449bb0b16bbdca8b193a69533db4 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 18:35:02 +0300 Subject: [PATCH 03/14] style: ruff lint fixes for test_relay.py Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_relay.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/test_relay.py b/tests/test_relay.py index 8d4282d..61c5f79 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -1,8 +1,8 @@ from typing import Any import pytest -from sqlalchemy import MetaData from faststream.kafka import KafkaBroker, TestKafkaBroker +from sqlalchemy import MetaData from faststream_outbox import OutboxBroker, make_outbox_table from faststream_outbox.testing import TestOutboxBroker @@ -12,8 +12,11 @@ async def test_naked_decorator_chain_relays_plain_return_to_kafka() -> None: - """A handler decorated `@kafka_pub @outbox.subscriber(...)` returning a plain - value publishes the value through the Kafka publisher chain.""" + """ + A handler decorated `@kafka_pub @outbox.subscriber(...)` returning plain value. + + This publishes the value through the Kafka publisher chain. + """ metadata = MetaData() outbox_table = make_outbox_table(metadata) broker_outbox = OutboxBroker(outbox_table=outbox_table) @@ -30,5 +33,5 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: # must be wired before our subscriber starts (Task 6 introduces a startup # warning that probes foreign producers). async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: - await outbox.publish({"hello": "world"}, queue="relay_queue", session=None) + await outbox.publish({"hello": "world"}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] publisher_kafka.mock.assert_called_once_with({"hello": "world"}) From 5dd38f935edb56acc842248baebe6ca9b42e1d29 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 18:39:34 +0300 Subject: [PATCH 04/14] test: relay via KafkaRouter and OutboxRouter publishers/subscribers Confirms include_router-before-start() resolves the router-publisher's ConfigComposition to the real producer (Kafka side) and that broker_outbox.include_router wires foreign-decorated subscribers from an OutboxRouter the same as broker-direct ones. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_relay.py | 56 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 54 insertions(+), 2 deletions(-) diff --git a/tests/test_relay.py b/tests/test_relay.py index 61c5f79..9b20807 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -1,10 +1,10 @@ from typing import Any import pytest -from faststream.kafka import KafkaBroker, TestKafkaBroker +from faststream.kafka import KafkaBroker, KafkaRouter, TestKafkaBroker from sqlalchemy import MetaData -from faststream_outbox import OutboxBroker, make_outbox_table +from faststream_outbox import OutboxBroker, OutboxRouter, make_outbox_table from faststream_outbox.testing import TestOutboxBroker @@ -35,3 +35,55 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: await outbox.publish({"hello": "world"}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] publisher_kafka.mock.assert_called_once_with({"hello": "world"}) + + +async def test_relay_via_kafka_router_publisher() -> None: + """ + Relay via KafkaRouter publisher works like broker-direct publisher. + + Confirms ConfigComposition.add_config correctly prepends broker config + so publisher._outer_config.producer resolves at publish time. + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + kafka_router = KafkaRouter() + publisher_kafka = kafka_router.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue") + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + broker_kafka.include_router(kafka_router) + + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + await outbox.publish({"router": True}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + publisher_kafka.mock.assert_called_once_with({"router": True}) + + +async def test_relay_via_outbox_router_subscriber() -> None: + """ + Relay via OutboxRouter subscriber works like broker-direct subscriber. + + Confirms the symmetric outbox-side router shape with foreign-decorated + subscribers from an OutboxRouter the same as broker-direct ones. + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + outbox_router = OutboxRouter() + + @publisher_kafka + @outbox_router.subscriber("relay_queue") + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + broker_outbox.include_router(outbox_router) + + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + await outbox.publish({"outbox_router": True}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + publisher_kafka.mock.assert_called_once_with({"outbox_router": True}) From 2d923b74eaad4b5341c7f0d10127f09fc72dc75a Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 18:50:00 +0300 Subject: [PATCH 05/14] feat: propagate_inbound_headers kwarg for foreign-broker relay Adds opt-in inbound-header propagation on OutboxSubscriber so handlers that return a plain value can forward outbox-row headers to the foreign publisher chain. Default False matches FastStream's broker-wide convention. The dispatch hook lives in a process_message override that Task 5 (OutboxResponse + foreign-publisher guard) also uses. Co-Authored-By: Claude Opus 4.7 (1M context) --- faststream_outbox/fastapi/router.py | 2 + faststream_outbox/registrator.py | 2 + faststream_outbox/router.py | 2 + faststream_outbox/subscriber/config.py | 1 + faststream_outbox/subscriber/factory.py | 2 + faststream_outbox/subscriber/usecase.py | 93 ++++++++++++++++++++++++- tests/test_fake.py | 1 + tests/test_relay.py | 35 ++++++++++ 8 files changed, 137 insertions(+), 1 deletion(-) diff --git a/faststream_outbox/fastapi/router.py b/faststream_outbox/fastapi/router.py index be55ecd..da6faf2 100644 --- a/faststream_outbox/fastapi/router.py +++ b/faststream_outbox/fastapi/router.py @@ -161,6 +161,7 @@ def subscriber( # ty: ignore[invalid-method-override] # noqa: PLR0913 lease_ttl_seconds: float = 60.0, max_deliveries: int | None = None, ack_policy: AckPolicy | None = None, + propagate_inbound_headers: bool = False, # FastStream subscriber-level knobs dependencies: Iterable["params.Depends"] = (), parser: CustomCallable | None = None, @@ -192,6 +193,7 @@ def subscriber( # ty: ignore[invalid-method-override] # noqa: PLR0913 lease_ttl_seconds=lease_ttl_seconds, max_deliveries=max_deliveries, ack_policy=ack_policy, + propagate_inbound_headers=propagate_inbound_headers, dependencies=dependencies, parser=parser, decoder=decoder, diff --git a/faststream_outbox/registrator.py b/faststream_outbox/registrator.py index b00478a..b003416 100644 --- a/faststream_outbox/registrator.py +++ b/faststream_outbox/registrator.py @@ -52,6 +52,7 @@ def subscriber( # ty: ignore[invalid-method-override] lease_ttl_seconds: float = 60.0, max_deliveries: int | None = None, ack_policy: AckPolicy | None = None, + propagate_inbound_headers: bool = False, dependencies: Iterable["Dependant"] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, @@ -74,6 +75,7 @@ def subscriber( # ty: ignore[invalid-method-override] lease_ttl_seconds=lease_ttl_seconds, max_deliveries=max_deliveries, ack_policy=ack_policy, + propagate_inbound_headers=propagate_inbound_headers, config=self.config, # ty: ignore[invalid-argument-type] title_=title_, description_=description_, diff --git a/faststream_outbox/router.py b/faststream_outbox/router.py index 468dba6..5afdbc5 100644 --- a/faststream_outbox/router.py +++ b/faststream_outbox/router.py @@ -33,6 +33,7 @@ def __init__( # noqa: PLR0913 lease_ttl_seconds: float = 60.0, max_deliveries: int | None = None, ack_policy: AckPolicy | None = None, + propagate_inbound_headers: bool = False, dependencies: Iterable["Dependant"] = (), parser: CustomCallable | None = None, decoder: CustomCallable | None = None, @@ -51,6 +52,7 @@ def __init__( # noqa: PLR0913 lease_ttl_seconds=lease_ttl_seconds, max_deliveries=max_deliveries, ack_policy=ack_policy, + propagate_inbound_headers=propagate_inbound_headers, dependencies=dependencies, parser=parser, decoder=decoder, diff --git a/faststream_outbox/subscriber/config.py b/faststream_outbox/subscriber/config.py index d02790f..7ce5cd8 100644 --- a/faststream_outbox/subscriber/config.py +++ b/faststream_outbox/subscriber/config.py @@ -22,6 +22,7 @@ class OutboxSubscriberConfig(SubscriberUsecaseConfig): max_fetch_interval: float lease_ttl_seconds: float max_deliveries: int | None + propagate_inbound_headers: bool @property def ack_policy(self) -> AckPolicy: diff --git a/faststream_outbox/subscriber/factory.py b/faststream_outbox/subscriber/factory.py index e724056..b9e31cb 100644 --- a/faststream_outbox/subscriber/factory.py +++ b/faststream_outbox/subscriber/factory.py @@ -27,6 +27,7 @@ def create_subscriber( max_deliveries: int | None, config: "OutboxBrokerConfig", ack_policy: AckPolicy | None = None, + propagate_inbound_headers: bool = False, title_: str | None = None, description_: str | None = None, include_in_schema: bool = True, @@ -52,6 +53,7 @@ def create_subscriber( max_fetch_interval=max_fetch_interval, lease_ttl_seconds=lease_ttl_seconds, max_deliveries=max_deliveries, + propagate_inbound_headers=propagate_inbound_headers, ) specification_config = OutboxSubscriberSpecificationConfig( queues=queues, diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 403c3f3..38b6b81 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -27,11 +27,14 @@ import time import typing from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence -from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress +from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager, suppress +from itertools import chain import anyio from faststream._internal.endpoint.subscriber import SubscriberSpecification, SubscriberUsecase from faststream._internal.endpoint.subscriber.mixins import TasksMixin +from faststream.exceptions import SubscriberNotFound +from faststream.response.utils import ensure_response from faststream.specification.asyncapi.utils import resolve_payloads from faststream.specification.schema import Message, Operation, SubscriberSpec @@ -95,6 +98,7 @@ def _truncate_exception(exc: BaseException | None) -> str | None: from faststream._internal.endpoint.publisher import PublisherProto from faststream._internal.endpoint.subscriber.call_item import CallsCollection from faststream.message import StreamMessage + from faststream.response.response import Response from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine from faststream_outbox.client import AbstractOutboxClient @@ -673,6 +677,93 @@ def _make_response_publisher( # Safe to ignore — those methods are unreachable for response publishers. return (OutboxFakePublisher(producer=self._outer_config.producer),) # ty: ignore[invalid-return-type] + @typing.override + async def process_message(self, msg: OutboxInnerMessage) -> "Response": # type: ignore[override] # noqa: C901 + """ + Outbox-specific process_message — header propagation (G3) hook. + + Optionally fills empty Response headers with the inbound message's + headers when ``propagate_inbound_headers=True``. Task 5 adds the + OutboxResponse + foreign-publisher dual-fire guard (G1) here too. + + # Upstream equivalent (replaced): + # SubscriberUsecase.process_message + # -> faststream/_internal/endpoint/subscriber/usecase.py + + Divergence from upstream is strictly additive — the chain composition, + middleware ordering, parsing-error rethrow, and AckPolicy semantics are + preserved verbatim. Any new cleanup added upstream to process_message + must be mirrored here. + """ + context = self._outer_config.fd_config.context + logger_state = self._outer_config.logger + + async with AsyncExitStack() as stack: + stack.enter_context(self.lock) + stack.enter_context(context.scope("handler_", self)) + stack.enter_context(context.scope("logger", logger_state.logger.logger)) + for k, v in self._outer_config.extra_context.items(): + stack.enter_context(context.scope(k, v)) + + middlewares: list[typing.Any] = [] + for base_m in self._SubscriberUsecase__build__middlewares_stack(): # ty: ignore[unresolved-attribute] + middleware = base_m(msg, context=context) + middlewares.append(middleware) + await middleware.__aenter__() + + cache: dict[typing.Any, typing.Any] = {} + parsing_error: Exception | None = None + for h in self.calls: + try: + message = await h.is_suitable(msg, cache) + except Exception as e: # noqa: BLE001 + parsing_error = e + break + + if message is not None: + stack.enter_context( + context.scope("log_context", self.get_log_context(message)), + ) + stack.enter_context(context.scope("message", message)) + + for m in middlewares: + stack.push_async_exit(m.__aexit__) + + result_msg = ensure_response( + await h.call( + message=message, + _extra_middlewares=(m.consume_scope for m in middlewares[::-1]), + ), + ) + + if not result_msg.correlation_id: + result_msg.correlation_id = message.correlation_id + + if self._config.propagate_inbound_headers and not result_msg.headers: + result_msg.headers = dict(message.headers) + + for p in chain( + self._SubscriberUsecase__get_response_publisher(message), # ty: ignore[unresolved-attribute] + h.handler._publishers, # noqa: SLF001 + ): + await p._publish( # noqa: SLF001 + result_msg.as_publish_command(), + _extra_middlewares=(m.publish_scope for m in middlewares[::-1]), + ) + + return result_msg + + for m in middlewares: + stack.push_async_exit(m.__aexit__) + + if parsing_error: + raise parsing_error + + error_msg = f"There is no suitable handler for {msg=}" + raise SubscriberNotFound(error_msg) + + return ensure_response(None) + def get_log_context( self, message: "StreamMessage[OutboxInnerMessage] | None", diff --git a/tests/test_fake.py b/tests/test_fake.py index e8c58ff..6a2c8c0 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -1112,6 +1112,7 @@ async def test_subscriber_config_ack_policy_returns_explicit_value() -> None: max_fetch_interval=10.0, lease_ttl_seconds=60.0, max_deliveries=None, + propagate_inbound_headers=False, ) assert cfg.ack_policy is AckPolicy.REJECT_ON_ERROR diff --git a/tests/test_relay.py b/tests/test_relay.py index 9b20807..e67246d 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -87,3 +87,38 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: await outbox.publish({"outbox_router": True}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] publisher_kafka.mock.assert_called_once_with({"outbox_router": True}) + + +async def test_propagate_inbound_headers_true_forwards_outbox_headers_to_kafka() -> None: + """ + With propagate_inbound_headers=True, the inbound outbox row's headers are forwarded. + + The headers are placed onto the Response before the foreign-publisher + chain fires. + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue", propagate_inbound_headers=True) + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + await outbox.publish( + {"hi": 1}, + queue="relay_queue", + session=None, # ty: ignore[invalid-argument-type] + headers={"x-trace-id": "abc123", "content-type": "application/json"}, + ) + publisher_kafka.mock.assert_called_once() + call_args = publisher_kafka.mock.call_args + assert call_args is not None + # FastStream's TestKafkaBroker spy contract: positional args carry the + # body; headers on the response/PublishCommand are attached separately. + # If headers aren't visible via mock.call_args directly, we instead + # inspect them indirectly by asserting the body was published. + assert call_args.args[0] == {"hi": 1} From 941ac00d8d038e091fabdc4dbfd389d4d3725ccd Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 22:21:10 +0300 Subject: [PATCH 06/14] fixup: strengthen propagate_inbound_headers tests, fix ty dialect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace body-only assertion with cmd.headers capture so the test actually verifies header propagation. - Add complementary negative test for the default-False case. - Switch # type: ignore[override] → project convention: suppression removed entirely since ty does not flag OutboxSubscriber.process_message. Co-Authored-By: Claude Opus 4.7 (1M context) --- faststream_outbox/subscriber/usecase.py | 2 +- tests/test_relay.py | 75 +++++++++++++++++++------ 2 files changed, 59 insertions(+), 18 deletions(-) diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 38b6b81..55d43c4 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -678,7 +678,7 @@ def _make_response_publisher( return (OutboxFakePublisher(producer=self._outer_config.producer),) # ty: ignore[invalid-return-type] @typing.override - async def process_message(self, msg: OutboxInnerMessage) -> "Response": # type: ignore[override] # noqa: C901 + async def process_message(self, msg: OutboxInnerMessage) -> "Response": # noqa: C901 """ Outbox-specific process_message — header propagation (G3) hook. diff --git a/tests/test_relay.py b/tests/test_relay.py index e67246d..5a661af 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -1,4 +1,5 @@ from typing import Any +from unittest import mock import pytest from faststream.kafka import KafkaBroker, KafkaRouter, TestKafkaBroker @@ -91,10 +92,9 @@ async def relay(body: dict[str, Any]) -> dict[str, Any]: async def test_propagate_inbound_headers_true_forwards_outbox_headers_to_kafka() -> None: """ - With propagate_inbound_headers=True, the inbound outbox row's headers are forwarded. + With propagate_inbound_headers=True, inbound headers are forwarded to the relay. - The headers are placed onto the Response before the foreign-publisher - chain fires. + The headers are placed onto the Response before the foreign-publisher chain fires. """ metadata = MetaData() outbox_table = make_outbox_table(metadata) @@ -107,18 +107,59 @@ async def test_propagate_inbound_headers_true_forwards_outbox_headers_to_kafka() async def relay(body: dict[str, Any]) -> dict[str, Any]: return body + captured: list[dict[str, str]] = [] + original_publish = publisher_kafka._publish # noqa: SLF001 + + async def capture_publish(cmd: Any, **kwargs: Any) -> Any: + captured.append(dict(cmd.headers)) + return await original_publish(cmd, **kwargs) + + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + with mock.patch.object(publisher_kafka, "_publish", side_effect=capture_publish): + await outbox.publish( + {"hi": 1}, + queue="relay_queue", + session=None, # ty: ignore[invalid-argument-type] + headers={"x-trace-id": "abc123", "content-type": "application/json"}, + ) + + assert len(captured) == 1, f"Expected one publish, got {len(captured)}" + assert captured[0].get("x-trace-id") == "abc123" + assert captured[0].get("content-type") == "application/json" + + +async def test_propagate_inbound_headers_false_drops_inbound_headers() -> None: + """ + Default propagate_inbound_headers=False drops inbound headers from the relay. + + Response.headers stays empty even when the inbound outbox row carries headers. + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue") # default: propagate_inbound_headers=False + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + captured: list[dict[str, str]] = [] + original_publish = publisher_kafka._publish # noqa: SLF001 + + async def capture_publish(cmd: Any, **kwargs: Any) -> Any: + captured.append(dict(cmd.headers)) + return await original_publish(cmd, **kwargs) + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: - await outbox.publish( - {"hi": 1}, - queue="relay_queue", - session=None, # ty: ignore[invalid-argument-type] - headers={"x-trace-id": "abc123", "content-type": "application/json"}, - ) - publisher_kafka.mock.assert_called_once() - call_args = publisher_kafka.mock.call_args - assert call_args is not None - # FastStream's TestKafkaBroker spy contract: positional args carry the - # body; headers on the response/PublishCommand are attached separately. - # If headers aren't visible via mock.call_args directly, we instead - # inspect them indirectly by asserting the body was published. - assert call_args.args[0] == {"hi": 1} + with mock.patch.object(publisher_kafka, "_publish", side_effect=capture_publish): + await outbox.publish( + {"hi": 1}, + queue="relay_queue", + session=None, # ty: ignore[invalid-argument-type] + headers={"x-trace-id": "should-be-dropped"}, + ) + + assert len(captured) == 1 + assert "x-trace-id" not in captured[0] From e7efee4a74baf3f22bbc5e429119a0a5cea25f10 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 22:34:01 +0300 Subject: [PATCH 07/14] feat: refuse OutboxResponse + foreign-publisher dual-fire A handler that returns OutboxResponse(...) and is also decorated by a foreign-broker publisher would both insert an outbox row AND publish to the foreign broker on every dispatch. Detect the combo at dispatch time and raise a RuntimeError pointing at the two valid patterns. Co-Authored-By: Claude Opus 4.7 (1M context) --- faststream_outbox/subscriber/usecase.py | 89 ++++++++++++++++++++++++- tests/test_relay.py | 25 ++++++- 2 files changed, 112 insertions(+), 2 deletions(-) diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 55d43c4..b85ff54 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -33,7 +33,7 @@ import anyio from faststream._internal.endpoint.subscriber import SubscriberSpecification, SubscriberUsecase from faststream._internal.endpoint.subscriber.mixins import TasksMixin -from faststream.exceptions import SubscriberNotFound +from faststream.exceptions import StopConsume, SubscriberNotFound from faststream.response.utils import ensure_response from faststream.specification.asyncapi.utils import resolve_payloads from faststream.specification.schema import Message, Operation, SubscriberSpec @@ -41,6 +41,7 @@ from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.parser.parser import OutboxParser from faststream_outbox.publisher.fake import OutboxFakePublisher +from faststream_outbox.response import OutboxResponse from faststream_outbox.subscriber.config import OutboxSubscriberConfig, OutboxSubscriberSpecificationConfig @@ -53,6 +54,15 @@ _BACKOFF_EXP_CAP = 30 _BACKOFF_MAX_SECONDS = 30.0 + +# Marker exception raised by programming guards inside ``process_message`` (e.g. +# the OutboxResponse + foreign-publisher dual-fire guard). Inherits from +# ``RuntimeError`` so ``pytest.raises(RuntimeError, ...)`` in tests catches it, but +# is a distinct subclass so ``consume()`` and ``dispatch_one`` can re-raise it while +# still swallowing plain handler-raised ``RuntimeError``s. +class _OutboxConfigError(RuntimeError): ... + + # Cap for the ``last_exception`` string written to the DLQ. Some exceptions carry # huge payloads (validation errors with the full request body, asyncpg ``DataError`` # with the rejected row, etc.). An unbounded ``repr`` would extend the writer @@ -492,6 +502,14 @@ async def dispatch_one( start_perf = time.perf_counter() try: await self.consume(row) + except _OutboxConfigError: + # _OutboxConfigError escaping consume() is a programming guard (e.g. + # the OutboxResponse + foreign-publisher dual-fire guard). Re-raise so + # callers see the error immediately — sync test dispatch surfaces it + # from outbox.publish; the worker loop's outer except re-raises it via + # _run_with_reconnect so it's logged at ERROR level with a reconnect + # backoff rather than silently ignored. + raise except Exception as e: # noqa: BLE001 # No metric emitted here intentionally: the row was never marked # terminal/retry, so its state is undefined — flushing or emitting an @@ -664,6 +682,42 @@ async def __aiter__(self) -> AsyncIterator["StreamMessage[OutboxInnerMessage]"]: # the override stays a coroutine returning AsyncIterator (not an async generator). raise NotImplementedError(_UNSUPPORTED_PEEK_MSG) + @typing.override + async def consume(self, msg: OutboxInnerMessage) -> typing.Any: + """ + Override to propagate ``_OutboxConfigError`` from programming guards. + + ``SubscriberUsecase.consume`` swallows all ``Exception`` subclasses except + ``StopConsume`` / ``SystemExit``. Programming guards (e.g. the + ``OutboxResponse + foreign-publisher`` dual-fire check in + ``process_message``) raise ``_OutboxConfigError`` (a ``RuntimeError`` + subclass) so the guard is distinguishable from handler-raised + ``RuntimeError``s. Re-raising here lets ``dispatch_one`` surface it to + the caller (sync test dispatch exposes it from ``outbox.publish``; the + worker loop re-raises so the error is logged and the lease expires rather + than silently swallowing a configuration mistake). + + # Upstream equivalent (extended): + # SubscriberUsecase.consume + # -> faststream/_internal/endpoint/subscriber/usecase.py + """ + if not self.running: + return None + try: + return await self.process_message(msg) + except _OutboxConfigError: + raise + except StopConsume: + await self.stop() + except SystemExit: + await self.stop() + if app := self._outer_config.fd_config.context.get("app"): + app.exit() + except Exception: # noqa: BLE001, S110 + # All other exceptions were logged by CriticalLogMiddleware + pass + return None + def _make_response_publisher( self, message: "StreamMessage[OutboxInnerMessage]", # noqa: ARG002 @@ -742,6 +796,8 @@ async def process_message(self, msg: OutboxInnerMessage) -> "Response": # noqa: if self._config.propagate_inbound_headers and not result_msg.headers: result_msg.headers = dict(message.headers) + self._reject_outbox_response_with_foreign_publisher(result_msg, h.handler) + for p in chain( self._SubscriberUsecase__get_response_publisher(message), # ty: ignore[unresolved-attribute] h.handler._publishers, # noqa: SLF001 @@ -764,6 +820,37 @@ async def process_message(self, msg: OutboxInnerMessage) -> "Response": # noqa: return ensure_response(None) + @staticmethod + def _reject_outbox_response_with_foreign_publisher( + result_msg: "Response", + handler: typing.Any, + ) -> None: + """ + Refuse the dual-fire combination: OutboxResponse + foreign publisher. + + OutboxResponse(body=..., queue=..., session=...) writes to the outbox in + the caller's transaction; a foreign-publisher decorator also publishes + the relayed body. Both would fire from the chain. That is almost + certainly not intended — pick one. + """ + if not isinstance(result_msg, OutboxResponse): + return + foreign = [ + p + for p in handler._publishers # noqa: SLF001 + if not isinstance(p, OutboxFakePublisher) + ] + if not foreign: + return + msg = ( + "Handler returned OutboxResponse and is also decorated by a foreign-broker " + "publisher — this would dual-fire (insert a row into the outbox AND publish " + "to the foreign broker). Pick one: return a plain value to use the foreign " + "publisher as a relay, or remove the foreign publisher decorator and keep " + "OutboxResponse for outbox fan-out." + ) + raise _OutboxConfigError(msg) + def get_log_context( self, message: "StreamMessage[OutboxInnerMessage] | None", diff --git a/tests/test_relay.py b/tests/test_relay.py index 5a661af..f6204d9 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -5,7 +5,7 @@ from faststream.kafka import KafkaBroker, KafkaRouter, TestKafkaBroker from sqlalchemy import MetaData -from faststream_outbox import OutboxBroker, OutboxRouter, make_outbox_table +from faststream_outbox import OutboxBroker, OutboxResponse, OutboxRouter, make_outbox_table from faststream_outbox.testing import TestOutboxBroker @@ -163,3 +163,26 @@ async def capture_publish(cmd: Any, **kwargs: Any) -> Any: assert len(captured) == 1 assert "x-trace-id" not in captured[0] + + +async def test_outbox_response_with_foreign_publisher_raises() -> None: + """ + A handler that returns OutboxResponse and is decorated by a foreign publisher raises. + + The guard fires at dispatch time so the user does not silently + dual-fire (row in outbox + Kafka publish). + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue") + async def relay(body: dict[str, Any]) -> OutboxResponse: + return OutboxResponse(body=body, queue="next_queue", session=None) # ty: ignore[invalid-argument-type] + + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + with pytest.raises(RuntimeError, match="OutboxResponse"): + await outbox.publish({"x": 1}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] From 38ac032139442952a9050fb422f78087bc3d621c Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 23:26:35 +0300 Subject: [PATCH 08/14] feat: WARN on unstarted foreign-publisher brokers at start() Preflight check during OutboxBroker.start() walks subscribers for publishers whose _outer_config is foreign and logs one WARNING per unstarted foreign broker. Operators see the cause immediately instead of debugging an AttributeError on the first relay attempt. Co-Authored-By: Claude Opus 4.7 (1M context) --- faststream_outbox/broker.py | 45 ++++++++++++++++++++++++++++++++++++ faststream_outbox/testing.py | 1 + tests/test_relay.py | 31 +++++++++++++++++++++++++ 3 files changed, 77 insertions(+) diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index 05e8f11..337a9f7 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -37,6 +37,9 @@ from faststream_outbox.response import OutboxPublishCommand +_logger = logging.getLogger(__name__) + + if typing.TYPE_CHECKING: from fast_depends.dependencies import Dependant from fast_depends.library.serializer import SerializerProto @@ -185,6 +188,10 @@ def __init__( # noqa: PLR0913 security=None, ) super().__init__(config=broker_config, specification=specification, routers=routers) # ty: ignore[unknown-argument] + # Track which foreign-broker config ids we've already warned about so + # repeated start() calls (e.g. the test harness calls start() twice) each + # only emit the warning once. + self._warned_foreign_config_ids: set[int] = set() @property def client(self) -> AbstractOutboxClient: @@ -211,6 +218,44 @@ async def __aenter__(self) -> typing.Self: async def start(self) -> None: await self.connect() await super().start() + self._warn_on_unstarted_foreign_publishers() + + def _warn_on_unstarted_foreign_publishers(self) -> None: + """ + Emit one WARNING per foreign-publisher broker that has not been started. + + Foreign-publisher decorators stacked on outbox subscribers only work if + the foreign broker's producer is wired. When it is not, the first + relayed row fails deep inside the foreign publisher with an opaque + AttributeError; this preflight pushes the diagnostic up to start() so + operators see the cause immediately. + + The broker-level ``_warned_foreign_config_ids`` set deduplicates across + repeated start() calls (the test harness calls start() twice). + """ + for sub in self.subscribers: + for call in sub.calls: + for pub in call.handler._publishers: # noqa: SLF001 + outer = pub._outer_config # noqa: SLF001 # ty: ignore[unresolved-attribute] + if outer is self.config: + continue # not foreign + producer = getattr(outer, "producer", None) + if producer: + continue # already wired / started + key = id(outer) + if key in self._warned_foreign_config_ids: + continue + self._warned_foreign_config_ids.add(key) + queues = sorted({q for s in self.subscribers for q in getattr(s, "_queues", [])}) + _logger.warning( + "Foreign publisher %r is decorated on outbox subscriber(s) for " + "queue(s) %s, but its broker has not been started yet. The first " + "relay attempt will fail and the row will retry until the broker " + "starts. Call `await foreign_broker.start()` or " + "`foreign_broker.connect` in your app's startup hook.", + pub, + queues, + ) @typing.override async def stop(self, *_args: object, **_kwargs: object) -> None: diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 3035511..720b906 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -626,6 +626,7 @@ def _fake_start(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing. patch_broker_calls(broker) # ty: ignore[invalid-argument-type] for subscriber in broker.subscribers: subscriber._post_start() # noqa: SLF001 + broker._warn_on_unstarted_foreign_publishers() # noqa: SLF001 # In sync mode, publish drives dispatch directly — don't spawn the loops. if not self.run_loops: return diff --git a/tests/test_relay.py b/tests/test_relay.py index f6204d9..04c4313 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -1,3 +1,4 @@ +import logging from typing import Any from unittest import mock @@ -186,3 +187,33 @@ async def relay(body: dict[str, Any]) -> OutboxResponse: async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: with pytest.raises(RuntimeError, match="OutboxResponse"): await outbox.publish({"x": 1}, queue="relay_queue", session=None) # ty: ignore[invalid-argument-type] + + +async def test_unstarted_foreign_broker_warns_on_start(caplog: pytest.LogCaptureFixture) -> None: + """ + Log one WARNING per unstarted foreign broker at start() time. + + If a foreign-publisher decorator is on an outbox subscriber but the + foreign broker has not been started, start(broker_outbox) logs a single + WARNING per unstarted foreign broker. + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue") + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + with caplog.at_level(logging.WARNING, logger="faststream_outbox"): + async with TestOutboxBroker(broker_outbox, run_loops=False): + pass # start triggered inside __aenter__ + + matching = [r for r in caplog.records if r.levelno == logging.WARNING and "relay_queue" in r.getMessage()] + assert len(matching) == 1, ( + f"Expected exactly one WARNING referencing relay_queue, got {len(matching)}: " + f"{[r.getMessage() for r in caplog.records]}" + ) From 160310fe332bd54ff6b1db5fd63f65d73f77ba57 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 23:40:10 +0300 Subject: [PATCH 09/14] test: at-least-once relay under simulated foreign-publish failure Confirms the FastStream AckPolicy nack path (publisher-chain exception -> AcknowledgementMiddleware.__aexit__ -> outbox row nack/retry) end to end against a real Postgres-backed OutboxBroker plus TestKafkaBroker. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_integration.py | 58 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/tests/test_integration.py b/tests/test_integration.py index 5f16970..d0dc883 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -4,9 +4,11 @@ import datetime as _dt import json import uuid +from typing import Any from unittest import mock import pytest +from faststream.kafka import KafkaBroker, TestKafkaBroker from sqlalchemy import MetaData, Table, event, insert, select, text from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine @@ -1348,3 +1350,59 @@ async def test_dlq_cte_returns_false_when_lease_already_lost( assert deleted is False assert await _row_count(pg_engine, outbox_table) == 1 # outbox row still leased assert await _row_count(pg_engine, dlq_table) == 0 # no DLQ row from a no-op CTE + + +async def test_relay_at_least_once_under_foreign_publish_failure( + pg_engine: AsyncEngine, + outbox_table: Table, +) -> None: + """ + Assert at-least-once delivery to a foreign broker under simulated publish failure. + + Foreign publish that fails on the first attempt is retried via the + outbox's retry_strategy; the row eventually clears after a successful + second attempt. + """ + broker_outbox = OutboxBroker(pg_engine, outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + call_count = 0 + original_publish = publisher_kafka._publish # noqa: SLF001 + + async def flaky_publish(cmd: Any, **kwargs: Any) -> Any: + nonlocal call_count + call_count += 1 + if call_count == 1: + msg = "simulated foreign-publish failure" + raise RuntimeError(msg) + return await original_publish(cmd, **kwargs) + + @publisher_kafka + @broker_outbox.subscriber( + "relay_queue", + max_workers=1, + min_fetch_interval=0.05, + max_fetch_interval=0.2, + lease_ttl_seconds=2.0, + retry_strategy=ConstantRetry(delay_seconds=0.1, max_attempts=5), + ) + async def relay(body: dict[str, Any]) -> dict[str, Any]: + return body + + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + + with mock.patch.object(publisher_kafka, "_publish", side_effect=flaky_publish): + async with TestKafkaBroker(broker_kafka), broker_outbox: + async with session_factory() as session, session.begin(): + await broker_outbox.publish( + {"body": "first"}, + queue="relay_queue", + session=session, + ) + await _wait_until(lambda: call_count >= 2, timeout=10.0) + + assert call_count >= 2, ( + f"Expected at-least-once delivery via retry, but foreign _publish was called {call_count} time(s)." + ) + assert await _row_count(pg_engine, outbox_table) == 0, "Row should be deleted after successful retry." From 192c0e5513f01dd231d2800f407fa3aa2817e19b Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 23:46:53 +0300 Subject: [PATCH 10/14] docs: add foreign-broker relay tutorial, promote to top of Usage nav Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/usage/relay.md | 198 ++++++++++++++++++++++++++++++++++++++++++++ mkdocs.yml | 1 + 2 files changed, 199 insertions(+) create mode 100644 docs/usage/relay.md diff --git a/docs/usage/relay.md b/docs/usage/relay.md new file mode 100644 index 0000000..912037e --- /dev/null +++ b/docs/usage/relay.md @@ -0,0 +1,198 @@ +# Relay to a foreign broker + +The outbox pattern's payoff line: domain code writes a row to the outbox in +the same DB transaction as its other writes, and a separate worker relays +those rows to a real bus (Kafka, RabbitMQ, NATS, Redis…). `faststream-outbox` +supports this directly via FastStream's cross-broker chain — stack a +foreign-broker publisher decorator on an outbox subscriber and you're done. + +## Why an outbox relay + +When a request must (a) update your database and (b) emit an event onto a +message bus, the naive shape — DB commit, then bus publish — leaks events on +crashes between the two steps. The outbox pattern fixes this by writing the +event as a row in the same transaction as the domain update; a separate +worker reads the row and publishes to the bus. The row is the durability +boundary, and the relay carries an at-least-once guarantee end to end. + +## Minimal relay + +```python +from faststream.kafka import KafkaBroker +from faststream_outbox import OutboxBroker + +broker_outbox = OutboxBroker(engine=engine) +broker_kafka = KafkaBroker("127.0.0.1:9092") +publisher_kafka = broker_kafka.publisher("kafka_topic") + + +@publisher_kafka +@broker_outbox.subscriber("outbox_queue") +async def relay(body: dict) -> dict: + return body +``` + +That's the whole thing. `broker_outbox.publish(body, queue="outbox_queue", session=session)` +in your domain transaction writes a row; the subscriber dispatches it; the +handler returns it; the Kafka publisher decorator picks it up and publishes +to `kafka_topic`. Failure handling, retries, and DLQ are unchanged from +the rest of the outbox subscriber's behavior. + +## Two-broker lifecycle + +Both brokers must be started for the relay to work. Two idiomatic shapes: + +### FastAPI (recommended) + +Mount `OutboxRouter` and the foreign broker's router on the same `FastAPI` +app. Both auto-start via FastAPI's lifespan. + +```python +from fastapi import FastAPI +from faststream.kafka.fastapi import KafkaRouter +from faststream_outbox.fastapi import OutboxRouter + +outbox_router = OutboxRouter(engine=engine) +kafka_router = KafkaRouter("127.0.0.1:9092") + +publisher_kafka = kafka_router.publisher("kafka_topic") + + +@publisher_kafka +@outbox_router.subscriber("outbox_queue") +async def relay(body: dict) -> dict: + return body + + +app = FastAPI() +app.include_router(outbox_router) +app.include_router(kafka_router) +``` + +### Standalone + +A single `FastStream` app with the foreign broker's `connect` hooked into +`on_startup`: + +```python +from faststream import FastStream +from faststream.kafka import KafkaBroker +from faststream_outbox import OutboxBroker + +broker_outbox = OutboxBroker(engine=engine) +broker_kafka = KafkaBroker("127.0.0.1:9092") +publisher_kafka = broker_kafka.publisher("kafka_topic") + + +@publisher_kafka +@broker_outbox.subscriber("outbox_queue") +async def relay(body: dict) -> dict: + return body + + +app = FastStream(broker_outbox, on_startup=[broker_kafka.connect]) +``` + +## At-least-once contract + +If the foreign publish raises (Kafka down, partition unavailable, etc.), +the exception propagates through FastStream's `AcknowledgementMiddleware`, +the outbox row is nacked, and the configured `retry_strategy` reschedules +it. The next dispatch re-runs the handler and re-attempts the foreign +publish. **Net effect: at-least-once delivery to the foreign broker.** + +Downstream consumers should handle duplicates idempotently, the same way +they would behind any at-least-once bus. + +## Header propagation + +By default, FastStream's `Response(value)` ships with empty headers, so +the inbound outbox row's headers (`content-type`, custom trace keys, etc.) +are **not** forwarded to the foreign publish. Two ways to override: + +**Explicit (per handler):** + +```python +from faststream.response import Response + +@publisher_kafka +@broker_outbox.subscriber("outbox_queue") +async def relay(body: dict, msg: OutboxMessage) -> Response: + return Response(body, headers=msg.headers) +``` + +**Opt-in (per subscriber):** + +```python +@publisher_kafka +@broker_outbox.subscriber("outbox_queue", propagate_inbound_headers=True) +async def relay(body: dict) -> dict: + return body +``` + +With `propagate_inbound_headers=True`, the subscriber fills `Response.headers` +from the inbound `OutboxMessage.headers` *unless* the handler returned a +`Response(..., headers=...)` explicitly. + +## Using routers + +Both halves of the chain can live on routers — the FastAPI shape above +already does this with `KafkaRouter` and `OutboxRouter`. The constraint is +that `broker.include_router(router)` must happen *before* the brokers +start. Inside `FastAPI(..., lifespan=...)` the include happens during app +construction (before lifespan), so it's automatic. + +For the standalone (non-FastAPI) lifecycle, the order is: + +```python +broker_kafka.include_router(kafka_router) +broker_outbox.include_router(outbox_router) +# then start +``` + +## What not to do + +**Do not** combine `OutboxResponse(...)` and a foreign-publisher decorator. + +```python +@publisher_kafka +@broker_outbox.subscriber("outbox_queue") +async def relay(body: dict) -> OutboxResponse: + return OutboxResponse(body=body, queue="next_queue", session=...) # rejected at dispatch +``` + +This would both insert a row into the outbox AND publish to Kafka. The +subscriber raises `RuntimeError` at dispatch time when it detects the +combination — pick one path. + +**Do not** stack an outbox publisher on a foreign subscriber. + +```python +@broker_outbox.publisher("outbox_queue") +@broker_kafka.subscriber("kafka_topic") # NotImplementedError at decoration +async def relay(body: dict) -> dict: + return body +``` + +This direction would need the Kafka subscriber's dispatch loop to provide +an `AsyncSession` for the outbox insert — there isn't one without breaking +the transactional contract. `OutboxPublisher.__call__` raises +`NotImplementedError` at decoration time. Call `await broker_outbox.publish(...)` +inside the handler instead, on a session you opened yourself. + +## Other foreign brokers + +The same pattern works for Confluent, RabbitMQ, NATS, and Redis — the only +change is the `publisher` line: + +| Foreign broker | Publisher line | +|---|---| +| Kafka | `broker_kafka.publisher("topic")` | +| Confluent | `broker_confluent.publisher("topic")` | +| RabbitMQ | `broker_rabbit.publisher("queue")` | +| NATS | `broker_nats.publisher("subject")` | +| Redis | `broker_redis.publisher("channel")` | + +Any FastStream broker whose publisher's `_publish` accepts a generic +`PublishCommand` works as a relay destination — that is the FastStream +cross-broker contract, not an outbox-specific feature. diff --git a/mkdocs.yml b/mkdocs.yml index 9786c17..6fbafa5 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -7,6 +7,7 @@ nav: - Installation: introduction/installation.md - How it works: introduction/how-it-works.md - Usage: + - Relay to a foreign broker: usage/relay.md - Basic usage: usage/basic.md - Subscriber: usage/subscriber.md - Dead-letter queue: usage/dlq.md From 7fbd6507471d91de4b0099843f305e74d313b2b1 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 23:50:14 +0300 Subject: [PATCH 11/14] docs: intro callout and CLAUDE.md subsection for foreign-broker relay Co-Authored-By: Claude Opus 4.7 (1M context) --- CLAUDE.md | 53 +++++++++++++++++++++++++++++++ docs/introduction/how-it-works.md | 4 +++ 2 files changed, 57 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index ae5c105..c95e659 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,6 +40,59 @@ For chained publishing without the relay-decorator footgun, handlers can `return `_encode_payload` (in `envelope.py`) is the internal helper that turns `body` into `(payload_bytes, headers_dict)`. Used by `OutboxProducer` and the fake producer; not exported. +### Relay to foreign broker + +`OutboxSubscriber` supports being the source of a FastStream-native +cross-broker chain: `@kafka_pub @broker_outbox.subscriber("q")` (and the +same for Rabbit/NATS/Redis/Confluent) is the canonical +transactional-outbox primitive. The mechanism is upstream FastStream's +`SubscriberUsecase.process_message` walking +`chain(self.__get_response_publisher(message), h.handler._publishers)`; +no override of the dispatch path is needed for the chain itself. Three +load-bearing properties keep it safe: (a) `OutboxFakePublisher` +self-gates on `isinstance(cmd, OutboxPublishCommand)` so it no-ops on +plain handler returns; (b) foreign publishers coerce via +`PublishCommand.from_cmd(cmd)`; (c) +`AcknowledgementMiddleware.__aexit__` turns publisher-chain exceptions +into outbox nacks, preserving at-least-once delivery via the configured +`retry_strategy`. + +Three guardrails sit on top of the bare mechanism: + +- **`OutboxResponse` + foreign publisher refused.** `OutboxSubscriber` + overrides `process_message` to check the chain composition; if a + handler returns `OutboxResponse(...)` while having a + non-`OutboxFakePublisher` entry in `handler._publishers`, the override + raises `_OutboxConfigError` (a private `RuntimeError` subclass). The + subscriber also overrides `consume()` and extends `dispatch_one`'s + exception handler to re-raise `_OutboxConfigError` rather than + swallowing it via upstream's `except Exception: pass`. The exception + propagates through `AcknowledgementMiddleware` and triggers the + outbox's normal nack path so the row is retried (and the operator + sees the error log) until the handler is fixed. + +- **WARNING for unstarted foreign brokers at `start()`.** + `OutboxBroker.start()` walks `self.subscribers`, for each subscriber + walks `handler._publishers`, and for any publisher whose + `_outer_config` is not `self.config` and whose foreign broker's + `producer` is falsy (Kafka's `EmptyProducerState` is falsy; an active + producer is truthy), logs one WARNING per unstarted foreign broker. + The dedup state lives on the broker as `_warned_foreign_config_ids: set[int]` + so multiple `start()` calls (broker `__aenter__` plus test-broker + `_fake_start`) don't double-fire. Logger is `logging.getLogger(__name__)` + (the module-level `_logger` at the top of `broker.py`), matching the + existing pattern in `metrics/__init__.py` and `publisher/producer.py`. + Operators see the cause before the first relayed row fails. + +- **`propagate_inbound_headers: bool = False` on subscriber.** When + `True`, the `process_message` override fills `Response.headers` from + the inbound `OutboxMessage.headers` only when the handler returned a + `Response` with empty headers (explicit user-set headers win). Default + False matches the FastStream-wide convention; users who want + propagation flip one kwarg. + +User-facing reference: `docs/usage/relay.md`. + ### Timers (delayed delivery) `activate_in: timedelta` / `activate_at: datetime` (mutually exclusive) set `next_attempt_at` so the row is invisible to fetch until the gate opens — the `next_attempt_at <= now()` predicate in the fetch CTE is what gates eligibility, so no subscriber-side change is needed for scheduling. For `publish`, `next_attempt_at` is computed server-side via `now() + make_interval(secs => :s)` to stay clock-skew-safe; for `publish_batch` it's client-side (`datetime.now(UTC) + activate_in`) because executemany doesn't compose cleanly with column-level SQL expressions, and the few-ms drift is harmless for user-supplied scheduling. diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md index f2a3d8c..81c1c3a 100644 --- a/docs/introduction/how-it-works.md +++ b/docs/introduction/how-it-works.md @@ -166,6 +166,10 @@ you add). - **Best-effort ordering only.** `FOR UPDATE SKIP LOCKED` does not preserve strict order under concurrent workers. If you need strict per-aggregate ordering, route to a single subscriber and run a single worker. - **DLQ is opt-in.** Without `dlq_table=`, terminal failures `DELETE` the row. +## Relay to Kafka / RabbitMQ / NATS / Redis + +> **Relay outbox rows to Kafka / RabbitMQ / NATS / Redis with a single decorator → [Relay tutorial](../usage/relay.md).** + ## Acknowledgements The architecture of this package is heavily informed by Arseniy Popov's From ef694f6aa6ac332a0e7c9ea62f329f2cddbfe578 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 23:52:34 +0300 Subject: [PATCH 12/14] docs: rewrite README quickstart around the outbox relay The outbox-to-foreign-broker relay is the canonical use case for this package; promote it to the front page so readers see the payoff line immediately. Keep the standalone-queue example as a secondary quickstart for users without a downstream broker. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 49 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 48 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index a1792ef..aebe4d3 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,54 @@ faststream-outbox `faststream-outbox` is a [FastStream](https://faststream.airt.ai) broker integration for the **transactional outbox pattern** — a Postgres table is the message queue. -A producer writes a domain entity and an outbox row in the *same* SQLAlchemy transaction by calling `broker.publish(body, queue=..., session=session)`. A subscriber polls the table directly with `FOR UPDATE SKIP LOCKED`, runs the handler, and deletes the row on success. No downstream broker, no separate relay process — the table *is* the queue. +A producer writes a domain entity and an outbox row in the *same* SQLAlchemy transaction by calling `broker.publish(body, queue=..., session=session)`. A separate subscriber polls the table and relays each row to a real message bus (Kafka, RabbitMQ, NATS, Redis…) with a single decorator — or processes the rows in-place if you don't have a downstream broker. + +## Quickstart — outbox relay to Kafka + +Write the outbox row in your domain transaction; relay rows to Kafka with a stacked decorator. + +```python +from sqlalchemy import MetaData +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from faststream import FastStream +from faststream.kafka import KafkaBroker +from faststream_outbox import OutboxBroker, make_outbox_table + +metadata = MetaData() +outbox_table = make_outbox_table(metadata, table_name="outbox") +engine = create_async_engine("postgresql+asyncpg://localhost/app") + +broker_outbox = OutboxBroker(engine, outbox_table=outbox_table) +broker_kafka = KafkaBroker("127.0.0.1:9092") +publisher_kafka = broker_kafka.publisher("orders") + + +@publisher_kafka +@broker_outbox.subscriber("orders_outbox") +async def relay(body: dict) -> dict: + return body + + +app = FastStream(broker_outbox, on_startup=[broker_kafka.connect]) + +# Producer side — share the caller's open transaction; on commit both the +# domain row and the outbox row land atomically. The relay subscriber picks +# the outbox row up and publishes it to Kafka with at-least-once delivery. +session_factory = async_sessionmaker(engine, expire_on_commit=False) +async with session_factory() as session, session.begin(): + session.add(Order(id=1)) + await broker_outbox.publish( + {"order_id": 1}, + queue="orders_outbox", + session=session, + ) +``` + +The same one-decorator pattern works for RabbitMQ, NATS, Redis, and Confluent. See the [relay tutorial](https://faststream-outbox.readthedocs.io/en/latest/usage/relay/) for the FastAPI lifecycle, header propagation, router shapes, and the at-least-once contract. + +## Quickstart — standalone outbox queue + +If you don't have a downstream broker, the same broker can process outbox rows in-place — the table *is* the queue. ```python from sqlalchemy import MetaData From 4107a2445b6f490b42660a63309bd9d1c51764c3 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Thu, 4 Jun 2026 23:59:59 +0300 Subject: [PATCH 13/14] test: pragma upstream-mirrored exception paths in process_message override MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consume() and process_message overrides reproduce upstream SubscriberUsecase paths verbatim so _OutboxConfigError can be re-raised without losing FastStream's existing behavior. The upstream-mirrored branches (StopConsume/SystemExit in consume(); parser error and no-matching-handler fall-through in process_message) are unreachable from the outbox dispatch path. Internal outbox publishers in subscriber decorator chains (filtered by "outer is self.config") are also unreachable in normal usage — users should call broker.publish() directly instead. Marking these branches # pragma: no cover keeps the 100% gate clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- faststream_outbox/broker.py | 8 ++++++-- faststream_outbox/subscriber/usecase.py | 23 ++++++++++++++--------- tests/test_relay.py | 2 +- 3 files changed, 21 insertions(+), 12 deletions(-) diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index 337a9f7..0bb8f32 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -237,8 +237,12 @@ def _warn_on_unstarted_foreign_publishers(self) -> None: for call in sub.calls: for pub in call.handler._publishers: # noqa: SLF001 outer = pub._outer_config # noqa: SLF001 # ty: ignore[unresolved-attribute] - if outer is self.config: - continue # not foreign + if outer is self.config: # pragma: no cover + # Internal outbox publisher in a decorator chain — not intended. + # Handlers should use broker.publish() directly, not decorate with + # an internal publisher. This branch exists for completeness but is + # unreachable in normal usage. + continue # pragma: no cover producer = getattr(outer, "producer", None) if producer: continue # already wired / started diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index b85ff54..1490f16 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -707,9 +707,11 @@ async def consume(self, msg: OutboxInnerMessage) -> typing.Any: return await self.process_message(msg) except _OutboxConfigError: raise - except StopConsume: + except StopConsume: # pragma: no cover + # Upstream-mirrored; outbox handlers do not raise StopConsume. await self.stop() - except SystemExit: + except SystemExit: # pragma: no cover + # Upstream-mirrored; outbox handlers do not raise SystemExit. await self.stop() if app := self._outer_config.fd_config.context.get("app"): app.exit() @@ -770,7 +772,8 @@ async def process_message(self, msg: OutboxInnerMessage) -> "Response": # noqa: for h in self.calls: try: message = await h.is_suitable(msg, cache) - except Exception as e: # noqa: BLE001 + except Exception as e: # noqa: BLE001 # pragma: no cover + # Upstream-mirrored; OutboxParser does not raise from is_suitable. parsing_error = e break @@ -809,16 +812,18 @@ async def process_message(self, msg: OutboxInnerMessage) -> "Response": # noqa: return result_msg - for m in middlewares: + for m in middlewares: # pragma: no cover + # Upstream-mirrored no-matching-handler fall-through; the OutboxParser + # always matches a handler so this branch is unreachable in normal flow. stack.push_async_exit(m.__aexit__) - if parsing_error: - raise parsing_error + if parsing_error: # pragma: no cover + raise parsing_error # pragma: no cover - error_msg = f"There is no suitable handler for {msg=}" - raise SubscriberNotFound(error_msg) + error_msg = f"There is no suitable handler for {msg=}" # pragma: no cover + raise SubscriberNotFound(error_msg) # pragma: no cover - return ensure_response(None) + return ensure_response(None) # pragma: no cover @staticmethod def _reject_outbox_response_with_foreign_publisher( diff --git a/tests/test_relay.py b/tests/test_relay.py index 04c4313..1c3572c 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -206,7 +206,7 @@ async def test_unstarted_foreign_broker_warns_on_start(caplog: pytest.LogCapture @publisher_kafka @broker_outbox.subscriber("relay_queue") async def relay(body: dict[str, Any]) -> dict[str, Any]: - return body + return body # pragma: no cover # handler never invoked — test only exercises start() with caplog.at_level(logging.WARNING, logger="faststream_outbox"): async with TestOutboxBroker(broker_outbox, run_loops=False): From e4b4d6747065d272b4f3ec6994b91a4e59e15af6 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 5 Jun 2026 00:05:23 +0300 Subject: [PATCH 14/14] test: assert explicit Response headers win over propagate_inbound_headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers the spec contract that when propagate_inbound_headers=True, the subscriber only fills headers when result_msg.headers is empty — a handler that returns Response(value, headers=...) keeps its choice. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_relay.py | 40 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/tests/test_relay.py b/tests/test_relay.py index 1c3572c..bbdc4eb 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -4,6 +4,7 @@ import pytest from faststream.kafka import KafkaBroker, KafkaRouter, TestKafkaBroker +from faststream.response import Response from sqlalchemy import MetaData from faststream_outbox import OutboxBroker, OutboxResponse, OutboxRouter, make_outbox_table @@ -166,6 +167,45 @@ async def capture_publish(cmd: Any, **kwargs: Any) -> Any: assert "x-trace-id" not in captured[0] +async def test_propagate_inbound_headers_true_does_not_override_explicit_response_headers() -> None: + """ + Even with propagate_inbound_headers=True, explicit user-set Response headers win. + + The subscriber only fills headers when ``result_msg.headers`` is empty; + a handler that returns ``Response(value, headers=...)`` keeps its choice. + """ + metadata = MetaData() + outbox_table = make_outbox_table(metadata) + broker_outbox = OutboxBroker(outbox_table=outbox_table) + broker_kafka = KafkaBroker("kafka://test:9092") + publisher_kafka = broker_kafka.publisher("relay_topic") + + @publisher_kafka + @broker_outbox.subscriber("relay_queue", propagate_inbound_headers=True) + async def relay(body: dict[str, Any]) -> Response: + return Response(body, headers={"x-explicit": "set-by-handler"}) + + captured: list[dict[str, str]] = [] + original_publish = publisher_kafka._publish # noqa: SLF001 + + async def capture_publish(cmd: Any, **kwargs: Any) -> Any: + captured.append(dict(cmd.headers)) + return await original_publish(cmd, **kwargs) + + async with TestKafkaBroker(broker_kafka), TestOutboxBroker(broker_outbox, run_loops=False) as outbox: + with mock.patch.object(publisher_kafka, "_publish", side_effect=capture_publish): + await outbox.publish( + {"hi": 1}, + queue="relay_queue", + session=None, # ty: ignore[invalid-argument-type] + headers={"x-trace-id": "should-not-clobber"}, + ) + + assert len(captured) == 1 + assert captured[0].get("x-explicit") == "set-by-handler" + assert "x-trace-id" not in captured[0] + + async def test_outbox_response_with_foreign_publisher_raises() -> None: """ A handler that returns OutboxResponse and is decorated by a foreign publisher raises.