TestOutboxBroker: fake only the storage, publish through the real producer #236

This was generated by AI during triage.

Problem Statement

A developer who tests an app with TestOutboxBroker gets a publish path that doesn't match production. In production every publish goes one way: broker.publish, publisher.publish or an OutboxResponse, then the publish middlewares, then OutboxProducer. The producer builds the envelope, inserts the row on the caller's session, sends the NOTIFY and emits the published event. The test broker swaps out that whole publishing side instead of only the storage underneath it:

• it patches broker.publish and broker.publish_batch with mocks;
• it replaces the producer with FakeOutboxProducer;
• it replaces the client with FakeOutboxClient.

The test module therefore keeps its own copies of the envelope encoding, the published event, the Postgres column checks and the dispatch, and every gap between a copy and production is a bug that users find themselves:

• publish middlewares and their metrics never run in tests (Route TestOutboxBroker publish through _basic_publish #219);
• the caller's session is thrown away, so a rolled-back publish is still delivered (TestOutboxBroker dispatches publishes from rolled-back transactions #218);
• the published event in tests had different tags until fix: align the fake client with the real one and return fetched rows oldest first #233 copied the production ones a second time.

Each production change to publishing needs a matching hand edit in the test module, so these gaps keep coming back.

Solution

TestOutboxBroker fakes only the storage. Publishing runs the same code in tests and in production: the real publish middlewares and the real OutboxProducer. The producer writes rows through the client. OutboxClient writes them with SQL, as today. FakeOutboxClient keeps them in memory and records which rows can now be dispatched. The test broker wraps the real producer in a thin delegating producer: once the real publish returns, it dispatches the recorded rows in sync mode, or wakes the fetch loop in loop mode. The fake publish patches, the fake producer and the copied encoding, metric and validation code are deleted. A contract test keeps the two clients' insert behaviour in step, the same way it already does for fetch and the terminal writes.

This fixes #219 outright and leaves #218 as a small change confined to the fake client's insert, to be decided separately (see Out of Scope).

User Stories

1. As a developer testing with TestOutboxBroker, I want broker.publish to run my publish middlewares, so that middleware behaviour I rely on in production is exercised in tests.
2. As a developer, I want publisher.publish under the test broker to run the same producer as production, so that my tests can't pass on code that production rejects.
3. As a developer, I want a returned OutboxResponse in tests to be published through the real producer, so that chained publishes behave the same as in production.
4. As a developer using the Prometheus or OpenTelemetry publish middleware, I want its publish metrics recorded in tests, so that I can assert on them without a database.
5. As a developer using a PrometheusRecorder or OpenTelemetryRecorder, I want the published event in tests to come from the production producer, so that its tags (status, exception_type, count, size_bytes, duration_seconds) never drift from production.
6. As a developer, I want a body that can't be encoded to fail in tests exactly as it fails in production, with the same exception and the same published{status="error"} event.
7. As a developer, I want custom serializers (pydantic models, dataclasses) to encode identically in tests and production, so that a handler's input is the same in both.
8. As a developer, I want publish_batch in tests to insert the whole batch before any handler runs, so that a handler never sees a partly inserted batch, the same as production's single INSERT.
9. As a developer, I want a batch with one invalid row to store nothing in tests, so that atomicity matches production.
10. As a developer, I want publishing a row with a timer_id that's already live in the queue to return None in tests, so that timer dedup matches production.
11. As a developer, I want a delivered timer row's timer_id to be reusable in tests, as the glossary defines it.
12. As a developer, I want queue names and timer_ids over 255 characters, and headers that can't be stored as JSON, rejected in tests as Postgres rejects them.
13. As a developer in sync mode, I want the handler to have finished when publish returns, so that I can assert straight after publishing, as with other FastStream test brokers.
14. As a developer in sync mode, I want a timer (activate_in / activate_at) to be dispatched immediately, so that tests don't sleep.
15. As a developer using run_loops=True, I want a published row to wake the fetch loop, so that loop-mode tests don't wait for a poll interval.
16. As a developer using run_loops=True, I want timers to stay gated until next_attempt_at, so that scheduling tests stay meaningful.
17. As a developer, I want feed() on the test broker to keep working for rows I insert directly, so that existing tests that seed rows keep passing.
18. As a developer, I want fake_client.rows and fake_client.dlq_rows to keep showing stored rows, so that existing assertions keep working.
19. As a developer, I want publishing to a queue with no subscriber to leave the row in the fake store, as today.
20. As a developer, I want broker.publish in tests to require a session, as production does, with a mock AsyncSession accepted, so that a call that fails in production also fails in tests.
21. As a developer upgrading, I want a ready-made fake_session() from the testing module, so that the upgrade is passing session=fake_session() instead of writing my own mock.
22. As a developer upgrading, I want the release notes to tell me that session=None no longer works under the test broker and to show fake_session() instead.
23. As a developer, I want the published event to be emitted before the handler runs in sync mode, and a handler failure never to be recorded as a failed publish, so that publish metrics mean the same as in production.
24. As a developer with a broker built without an engine, I want broker.publish to keep working, so that a publish-only service doesn't need one.
25. As a developer, I want cancel_timer and fetch_unprocessed under the test broker to keep behaving as today.
26. As a maintainer, I want one producer implementation, so that a change to encoding, metrics or validation is made once.
27. As a maintainer, I want the row insert behind the client interface next to fetch, the terminal writes and cancel_timer, so that all SQL on the outbox table lives in one place.
28. As a maintainer, I want the contract suite to cover inserts, so that OutboxClient and FakeOutboxClient can't drift on insert semantics.
29. As a maintainer, I want the publish benchmark counters to stay the same, so that moving the insert costs nothing in production.
30. As a maintainer, I want the test module noticeably smaller, so that there is less copied logic to keep in step.
31. As a maintainer, I want TestOutboxBroker dispatches publishes from rolled-back transactions #218 reduced to a change inside the fake client's insert, so that it can be judged on its own.

Implementation Decisions

• The insert moves into the client.
  • AbstractOutboxClient gains single-row and batch insert operations that take the caller's session.
  • The single-row insert returns the new row id, or None on a timer_id conflict.
  • The batch insert returns nothing, matching publish_batch today.
  • The NOTIFY becomes part of the insert, with the same rules as today: only when a row landed and isn't future-dated, and deduplicated per queue within a transaction.
• The insert operations are public on the client interface, with contract docstrings that point callers to broker.publish: calling them directly skips publish middlewares and the published event.
• OutboxClient gets the SQL unchanged. It takes over the producer's prebuilt INSERT statements, including the variant that sends pg_notify in RETURNING (perf: send the publish NOTIFY in the INSERT's RETURNING #235), the batch insert followed by its own SELECT pg_notify, and the per-transaction NOTIFY memo. The statements themselves don't change.
• OutboxClient is always built, and its engine is optional. Today it only exists when the broker gets an engine, while publishing needs only the caller's session. Inserts work without an engine; fetch, ping and schema validation still need one, and the broker's "not connected" errors key off the missing engine instead of a missing client.
• OutboxProducer becomes the only producer. It builds the envelope, calls the client's insert and emits the published event. It looks the client up from the broker config at publish time, so the test broker's client swap is all it takes to redirect inserts.
• FakeOutboxClient gets the same insert operations.
  • It validates rows the way Postgres would: column lengths, JSON-storable headers and live timer_id dedup.
  • It allocates ids.
  • It records each newly stored row as dispatchable, or the whole batch once all of it is stored. It never runs a handler itself.
  • feed() keeps working and goes through the same storage.
• TestOutboxBroker stops patching publishing. It removes the broker.publish and publish_batch patches and keeps the client swap. It replaces the producer with a thin delegating producer that calls the real OutboxProducer and, after it returns, takes the recorded rows:
  • sync mode: find the subscriber for each row's queue, claim the row, and dispatch it, ignoring next_attempt_at;
  • loop mode: wake the matching subscriber's fetch loop.
  • Dispatching after the real producer returns keeps published ahead of the handler's events, keeps handler time out of the published duration, and keeps a re-raised configuration error from being recorded as a failed publish. Running the handler inside the client's insert would break all three.
• Deleted: FakeOutboxProducer, the fake publish functions, the test-module copy of the published event, and the Postgres-column copy except what the fake insert needs.
• The session is validated as in production: broker.publish under the test broker requires an AsyncSession. A mock built with spec=AsyncSession passes, as it does for publisher.publish today. This is a breaking change for test code (about 96 call sites in this repo's own tests) and goes in the 0.16.0 upgrade note.
• fake_session(): a module-level function in the testing module returning a no-op mock AsyncSession whose in_transaction() is true, so it passes both the session check and the OutboxResponse open-transaction guard. Module-level rather than on the test broker, because async with TestOutboxBroker(broker) as br yields the broker, not the test broker.
• Unchanged behaviour:
  • OutboxResponse's open-transaction guard;
  • the request-reply refusal;
  • validate_schema raising under the test broker;
  • the first-matching-subscriber rule for overlapping subscribers.
• Docs and agent notes:
  • The testing guide's sections on sessions and the fake broker are rewritten: one producer, a session is required, and publish middlewares now run in tests.
  • The architecture note in the agent guide (the client and fake client implement the same rules twice) is updated so that inserts are part of the contract.
• Delivery: two PRs. The first moves the insert into the client with no user-visible change; the second switches the test broker and carries the breaking change.
• Release: 0.16.0. This closes Route TestOutboxBroker publish through _basic_publish #219. TestOutboxBroker dispatches publishes from rolled-back transactions #218 waits for a separate decision.

Testing Decisions

• What makes a good test here: tests go through the public surface only: TestOutboxBroker, broker.publish, publisher.publish, a returned OutboxResponse, publish_batch, handler side effects, the recorder events, and fake_client.rows. No test should depend on the internal hand-off between the fake client and the test broker.
• Two seams, both existing:
  • The client contract suite, which runs every scenario against OutboxClient on real Postgres and against FakeOutboxClient. New insert scenarios:
    • the returned id;
    • None on a live timer_id conflict;
    • timer_id reuse after delivery;
    • batch atomicity when one row is invalid;
    • column-length and header validation, where each adapter declares the exception it raises (Postgres and the fake can't raise the same type);
    • timers stored with their next_attempt_at.
  • The test-broker behaviour tests, written failing first:
    • a publish middleware's publish hook runs for broker.publish and publisher.publish (Route TestOutboxBroker publish through _basic_publish #219); OutboxResponse already runs it and gets a regression test;
    • the native Prometheus and OpenTelemetry publish metrics are recorded under the test broker;
    • broker.publish without a session raises under the test broker as in production.
• Engine-less publish: a broker without an engine publishes on a real session (integration test).
• Must keep passing: the existing sync-mode, loop-mode, batch, timer, DLQ, metrics and OutboxResponse tests under the test broker. Tests that relied on the removed patches, such as session=None calls, change to pass a mock session.
• Production paths: the existing integration tests for publish, NOTIFY dedup (savepoints, autobegin, batch after single and single after batch, timer conflicts) and the single-statement publish keep passing unchanged against real Postgres. The publish benchmark gate must pass with unchanged counters.
• Prior art:
  • the contract-suite harness pattern of one seed, fetch and terminal helper per adapter;
  • the existing middleware tests under the test broker for the Prometheus and OpenTelemetry middleware;
  • the statement-counting integration tests for NOTIFY dedup.

Out of Scope

• Transaction awareness in the fake (TestOutboxBroker dispatches publishes from rolled-back transactions #218): delivering on commit and discarding on rollback for real sessions. After this refactor it is one change inside the fake client's insert: buffer rows per transaction, deliver on commit, discard on rollback or close, and handle savepoints. Running the handler before commit() returns needs SQLAlchemy's semi-private await_only (renamed await_ in 2.1, old name kept), so whether to do it, or to document real-broker testing and close TestOutboxBroker dispatches publishes from rolled-back transactions #218, is decided separately on TestOutboxBroker dispatches publishes from rolled-back transactions #218.
• Production changes: no change to the fetch, terminal-write or DLQ paths, and none to SQL semantics. The insert SQL moves but doesn't change.
• Multiple subscribers on one queue in tests: dispatch stays first-match.

Further Notes

• Origin: this came out of triaging TestOutboxBroker dispatches publishes from rolled-back transactions #218. Its research is on that issue:
  • commit hooks under every join_transaction_mode;
  • await_only on SQLAlchemy 2.0.0 and 2.1.5;
  • prior art from Django's captureOnCommitCallbacks and Rails 5's after_commit in tests.
• Effect on the duplication: "the client and the fake implement the same rules twice" is a deliberate duplication that the contract suite polices. This refactor narrows that duplication to storage, the only layer that has to differ, and removes the duplication of the producer, which the contract suite never covered.