fix: refuse OutboxResponse on a closed session, write the recorded outcome on config errors

[lesnik512]

Closes #214.

A follow-on row on a closed session

FastStream and FastAPI both tear Depends down inside the handler call, and the response is published after that. With the session dependency the docs showed, the follow-on INSERT ran on a session whose transaction had already committed. SQLAlchemy opened a new transaction for it that nobody committed, so the row and its NOTIFY were rolled back. Meanwhile the inbound row was acked and published recorded success.

OutboxFakePublisher._publish now checks session.in_transaction() and raises _OutboxConfigError when there's no open transaction. That follows the rule from the grilling for config errors: retry, and log at ERROR. The message names both fixes.

_OutboxConfigError moves from subscriber/usecase.py to response.py, so the response publisher can raise it without a circular import. usecase.py still imports the name.

One limitation: a session that is still open but hasn't started a transaction, because the handler made no writes and nothing called begin(), is refused too. The documented pattern opens its transaction up front, so it isn't affected.

Where the session should live

I checked this against real Postgres before writing the docs. Middleware consume_scope runs inside h.call, so it closes before the publish just like Depends. A broker middleware's on_receive / after_processed wrap the whole dispatch, response publish included. FastStream puts the ack middleware outermost, so a session committed in after_processed commits before the ack decision, and a failed commit nacks the inbound row.

docs/usage/publisher.md now shows that middleware, with the handler taking the session through Context(). docs/usage/fastapi.md no longer says OutboxResponse commits with a Depends session; it points FastAPI handlers at router.broker.publish(..., session=session) instead. I haven't checked Context() inside FastAPI-wrapped handlers, so the FastAPI page doesn't suggest the middleware. The troubleshooting page has an entry for the new error.

Config errors write the recorded outcome

A config error comes after the handler has run and unwinds through the ack middleware, which records a nack. The worker used to drop that nack and leave the row leased until expiry. With max_deliveries=None, the handler then re-ran every lease_ttl_seconds forever. dispatch_one now writes the recorded outcome inline and re-raises. The test broker still surfaces the error from publish, and _worker_inner still logs it at ERROR. The write is inline because a buffered row would get a second task_done as the error unwinds. Both the consume() docstring and the comment in _worker_inner now describe this behaviour (B15).

Under AckPolicy.MANUAL the capture middleware records the exception, so assert_state_set nacks and the row is retried rather than rejected. A test pins that.

correlation_id from headers

Upstream fills an empty correlation_id with the inbound id, so OutboxResponse(headers={"correlation_id": X}) always hit the envelope's mismatch check after the handler succeeded, and the inbound row was retried. OutboxResponse now takes the header value when the argument isn't passed.

Tests

Each of these failed before the change:

• An integration test publishes through the live worker with the docs' Depends session. Before, there was no error and the inbound row was acked. Now the row is retried once and an ERROR log names the transaction.
• In sync mode:
  • the no-transaction guard raises, and the inbound row is retried;
  • a config error writes the recorded retry;
  • the correlation_id header reaches the downstream handler.

These pass on both the old and the new code:

• an integration test for the middleware-scoped session, which shows the documented pattern commits the follow-on row;
• the AckPolicy.MANUAL guard.

Performance

The only addition on the hot path is one in_transaction() call per response publish, and that call doesn't touch the database. just bench-check passes.

Verified locally: 662 tests pass in docker at 100% coverage, and the full suite also passes on SQLAlchemy 2.0.0. ruff is clean, ty shows the same 6 diagnostics as main, and the docs build.

[github-actions]

Benchmark gate

✅ gate passed

scenario	msg/s	delete/msg	WALrec/msg	WALB/msg	fpi	upd	del	dead_tup
consumer/w1/b10	1089	1.000	6.68	910	242	5000	5000	10000
consumer/w1/b100	1372	1.000	7.10	1144	243	5000	5000	10000
consumer/w2/b10	1293	1.000	6.68	919	242	5000	5000	10000
consumer/w2/b100	1787	1.000	7.10	1144	243	5000	5000	10000
consumer/w4/b10	1295	1.000	6.69	923	242	5000	5000	10000
consumer/w4/b100	1831	1.000	7.14	1180	263	5000	5000	10000
consumer/w1/b100/tfbs100	5631	0.010	6.07	1119	244	5000	5000	10000
consumer/w1/b100/interval1-10	1356	1.000	7.10	1144	243	5000	5000	10000
producer/w1/b100	2133	0.000	3.04	585	1	0	0	0
producer/w1/b100/ppt1	676	0.000	4.04	626	0	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).