feat: opt-in DLQ on terminal failure (issue #26)

[lesnik512]

Closes #26.

Summary

• Opt-in DLQ. Pass OutboxBroker(..., dlq_table=make_dlq_table(metadata)); default behavior is bit-for-bit identical to today.
• Atomic via single Postgres CTE. OutboxClient.delete_with_lease(..., dlq_payload=...) issues WITH deleted AS (DELETE … RETURNING …) INSERT INTO <dlq> SELECT … FROM deleted so the audit row commits with the DELETE or not at all (DLQ INSERT failure rolls the whole statement back — outbox row stays leased, gets reclaimed/retried). Preserves the writer-connection autocommit fast path (single statement); preserves the lease-token guard.
• Three failure paths, each tagged on the row via OutboxInnerMessage.terminal_failure_reason so _flush_terminal knows whether to build a DLQ payload:
  • allow_delivery False ⇒ "max_deliveries"
  • _nack exhausted (delay None) ⇒ "retry_terminal"
  • _reject ⇒ "rejected" (diverges from issue text — see "Design decisions" below)
  • _ack ⇒ stays None, never touches DLQ
• New dlq_written metric event emitted from _flush_terminal after the CTE commits; tags {queue, subscriber, deliveries_count, failure_reason, exception_type}. Skipped on lease-lost.
• validate_schema() checks the DLQ table when configured (separate _validate_dlq_schema_sync pass alongside the existing outbox one).
• FakeOutboxClient accumulates dlq_rows in-memory so TestOutboxBroker tests can assert audit content without Postgres.

Design decisions (worth flagging for review)

1. _reject writes to DLQ. Issue E3: DLQ / archive on terminal failure (opt-in) #26 lists _reject as opt-out. We took the opposite call after discussion: AckPolicy.REJECT_ON_ERROR is a real failure operators want audited, and manual await msg.reject() still signals "poison, drop with audit." last_exception may be None on manual reject — DLQ column is nullable.
2. dlq_written metric fires in this PR, not deferred. The issue's "wait for E4" note is stale; the metrics-recorder seam already exists, so the emission is a 5-line addition consistent with the other six subscriber events.
3. failure_reason stored as String(32) literals, not int enum or PG ENUM. Human-readable in ad-hoc operator queries; storage delta is irrelevant for an audit table; no Python-side enum-mapping required.
4. CTE composed via text() with quoted identifiers, not SQLAlchemy core composition. The table names come from application-owned Table objects and are passed through dialect.identifier_preparer.quote(); bind parameters carry the dynamic values. # noqa: S608 justified inline. Trade-off: SQLAlchemy core's delete(...).returning(...).cte() + insert().from_select() works but is harder to read for a Postgres-only library.

Test plan

• just lint — ruff format, ruff check, ty check all green
• just test — 383 passed, 100.00% coverage gate satisfied
• Unit tests cover the 4 failure-reason wiring paths + 7 _flush_terminal dispatch behaviors + make_dlq_table factory
• Fake-broker end-to-end exercises all three failure modes + unconfigured no-op + metric fire/no-fire
• Integration tests cover atomic CTE happy path, INSERT-failure rollback, lease-lost no-op, and validate_schema with/without DLQ
• Default path (no dlq_table kwarg) is bit-for-bit identical — existing 240+ tests continue to pass unmodified

Notes for reviewers

• The four FakeOutboxClient.delete_with_lease overrides in test files (test_unit.py × 2, test_fake.py × 2) were updated to (*args, **kwargs) or explicit dlq_payload kwarg to track the abstract signature change.
• One pre-existing test was extended with _warnings.catch_warnings() around the REJECT_ON_ERROR + retry_strategy misconfig warning — matches the existing pattern at tests/test_fake.py:971.
• Docs (README.md, CLAUDE.md) intentionally not touched in this PR — happy to add a follow-up if you'd like the architecture notes in tree.

🤖 Generated with Claude Code