feat: foreign-broker relay from OutboxSubscriber

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
+`<Broker>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.

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

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

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.

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,48 @@ 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:  # 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
+                    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: