refactor: DLQ CTE in Core, prebuilt publish INSERTs, drop dead code

[lesnik512]

Cleanup items R1, R2 and R4 from the review. No tracking issue and no behaviour change.

R1: DLQ CTE in SQLAlchemy Core

_build_dlq_cte_stmt built the DELETE-then-archive statement as an f-string. That needed format_table for quoting and schema qualification, plus a noqa: S608. It's now insert(dlq).from_select(...) over a delete(...).returning(...).cte("deleted"). SQLAlchemy hoists the CTE to the top of the statement, which Postgres requires for a data-modifying CTE, and the SQL comes out the same as before. Because the row id, the lease token and the failure context are all bind parameters, the client builds the statement once in __init__ and the compiled cache can reuse it. The existing schema-qualification, adversarial-name and DLQ column tests still pass. They now compile the stored statement.

R2: prebuilt publish INSERTs

_do_publish built a new insert() or pg_insert().on_conflict_do_nothing() on every publish. The producer now builds four statements once: plain or timer upsert, each with or without the server-side activate_in expression. The row values go in as execute parameters. activate_at is a plain parameter and needs no variant of its own.

Median per-publish time over 1000 publishes in one transaction, against the docker Postgres (two runs each):

publish	main	this PR
plain	131-133 µs	74-82 µs
activate_in	227-234 µs	77-79 µs
timer_id	229-272 µs	178-179 µs

Four test_broker_publish_* tests now read the values from the execute parameters instead of from the compiled statement.

R4: dead code

• max(0.0, lease_ttl_seconds) in OutboxClient.fetch and FakeOutboxClient.fetch: lease_ttl_seconds is validated as > 0 at config time. The max(0.0, delay_seconds) guards stay, because a user's retry strategy can return any delay.
• connect / disconnect on OutboxProducer and FakeOutboxProducer: nothing calls them. ProducerProto doesn't declare them, and the broker sets producer.serializer directly. The three tests that existed only to cover them are removed.

Two R4 items stay because they aren't dead. OutboxProducer._parser / _decoder are members of ProducerProto. The subscriber's _queues property is read through getattr by broker.py and registrator.py.

Verification

• 659 tests pass in docker at 100% coverage, including the DLQ and scheduling integration tests against real Postgres.
• The full suite also passes on SQLAlchemy 2.0.0.
• just bench-check passes.
• 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	1109	1.000	6.67	909	242	5000	5000	10000
consumer/w1/b100	1381	1.000	7.10	1144	243	5000	5000	10000
consumer/w2/b10	1310	1.000	6.68	919	242	5000	5000	10000
consumer/w2/b100	1823	1.000	7.10	1144	243	5000	5000	10000
consumer/w4/b10	1386	1.000	6.68	944	242	5000	5000	10000
consumer/w4/b100	1877	1.000	7.14	1179	263	5000	5000	10000
consumer/w1/b100/tfbs100	5911	0.010	6.07	1119	244	5000	5000	10000
consumer/w1/b100/interval1-10	1420	1.000	7.10	1144	243	5000	5000	10000
producer/w1/b100	3031	0.000	3.04	585	1	0	0	0
producer/w1/b100/ppt1	762	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).