feat!: TestOutboxBroker publishes through the real producer

[lesnik512]

Closes #236 and #219. Follows #237. TestOutboxBroker now fakes only the storage, so publishing in tests runs the production producer and publish middlewares.

Why

The test broker patched broker.publish and publish_batch, swapped in FakeOutboxProducer, and kept its own copies of the encoding, the published event and the dispatch. Each copy could drift from production:

• publish middlewares never ran for broker.publish in tests (Route TestOutboxBroker publish through _basic_publish #219);
• the published tags had to be copied a second time in fix: align the fake client with the real one and return fetched rows oldest first #233.

The change

• Removed:
  • the broker.publish / publish_batch patches;
  • FakeOutboxProducer;
  • the TestOutboxBroker.fake_producer attribute;
  • _fake_publish_* and _emit_published;
  • validate_activate_args, which only those fakes used.
• Dispatch: the test broker wraps the real producer in a thin _DispatchingProducer. It calls the real publish / publish_batch, then takes the rows the fake client recorded and dispatches them: sync mode runs the handlers, and loop mode wakes the fetch loops.
• Why after the producer: TestOutboxBroker: fake only the storage, publish through the real producer #236 first had the fake client's insert run the handlers. Dispatching after the producer returns instead keeps three things right:
  • published stays ahead of the handler's events;
  • handler time stays out of the published duration;
  • a re-raised _OutboxConfigError isn't recorded as a failed publish for a row that was stored.
• Leftover rows: rows recorded but never dispatched (for example from a direct fake_client.insert) are dropped when the test broker exits.
• faststream_outbox.testing.fake_session(): returns a mock AsyncSession that passes the session check and reports an open transaction, so OutboxResponse accepts it too.

Breaking change (0.16.0 upgrade note)

broker.publish and publish_batch need a session under TestOutboxBroker, as they do in production. A test that called broker.publish(body, queue="q") or passed session=None now gets TypeError. Pass a mock instead:

from faststream_outbox.testing import fake_session

await broker.publish(body, queue="q", session=fake_session())

Publish middlewares now run in tests. The Prometheus and OpenTelemetry middlewares record publish metrics under TestOutboxBroker, and so does your own publish_scope. FakeOutboxProducer and TestOutboxBroker.fake_producer are gone.

A naive activate_at now raises the production message (OutboxPublishCommand requires activate_at to be timezone-aware).

Tests

• Written first and failing:
  • a publish middleware's publish_scope runs for broker.publish. The publisher.publish and OutboxResponse paths already ran it, and are now covered in the same test.
  • broker.publish and publish_batch without a session raise TypeError under the test broker.
• Flipped: the Prometheus and OpenTelemetry tests that asserted publish metrics don't fire under the test broker now assert they do, for both the broker and publisher paths.
• New:
  • published is emitted before the handler runs;
  • a config error re-raised by sync dispatch leaves a published{status="success"};
  • rows not dispatched in one test-broker context don't run in the next one. That test failed before the drop on exit.
• Migrated: about 96 test call sites now pass session=fake_session(). The two tests that pinned FakeOutboxProducer check the swapped producer slot instead. The fake-producer copies of the B8 and request-refusal tests are deleted; the real producer already has both.

Docs

• testing.md: sessions and fake_session(), one producer for both publish paths, and publish middlewares running in tests. It also notes that the fake store ignores transactions, so a rolled-back row is still delivered (the behaviour TestOutboxBroker dispatches publishes from rolled-back transactions #218 is about).
• observability.md: the test-broker note.
• dlq.md: a code sample.
• AGENTS.md: the architecture note.

Verification

• 741 tests pass in docker at 100% coverage.
• ruff and ty are clean, and mkdocs build --strict passes.

[github-actions]

Benchmark gate

✅ gate passed

scenario	msg/s	delete/msg	WALrec/msg	WALB/msg	fpi	upd	del	dead_tup
consumer/w1/b10	1491	1.000	6.69	913	242	5000	5000	10000
consumer/w1/b100	1619	1.000	7.09	1143	243	5000	5000	10000
consumer/w2/b10	1425	1.000	6.69	914	242	5000	5000	10000
consumer/w2/b100	2032	1.000	7.09	1143	243	5000	5000	10000
consumer/w4/b10	2064	1.000	6.74	944	242	5000	5000	10000
consumer/w4/b100	2247	1.000	7.09	1142	243	5000	5000	10000
consumer/w1/b100/tfbs100	9970	0.010	6.06	1117	243	5000	5000	10000
consumer/w1/b100/interval1-10	909	1.000	7.15	1186	265	5000	5000	10000
producer/w1/b100	3711	0.000	3.04	597	1	0	0	0
producer/w1/b100/ppt1	882	0.000	4.04	637	1	0	0	0

Gated (fails the build): delete_calls + tuple counters (upd/del/ins) + the producer's insert_calls, exact; select_calls and other_calls within +2; wal_records within a 10% band; fetch_calls within 2x. msg/s, WAL bytes and total calls are informational (timing/FPI noise).