perf: claim rows with ordered per-queue scans; add id to _pending_idx

[lesnik512]

Closes #217. For 0.15.0: a fetch now reads about limit rows however large the backlog is, and _pending_idx becomes (queue, next_attempt_at, id). Upgrading needs a concurrent index rebuild, and the docs give the migration.

The problem

The claim statement filtered with queue = ANY(:queues) AND (acquired_token IS NULL OR <expired lease>), then ran ORDER BY next_attempt_at, id LIMIT n FOR UPDATE SKIP LOCKED. Postgres can't serve that OR or the = ANY with an ordered index scan, and it doesn't push the row limit through LockRows. So every fetch read and sorted every ready row on the queue. Draining a backlog got slower roughly with the square of its size, and it was worst exactly when the consumer was catching up.

The rewrite

The new shape is the one measured in the issue:

• The queue array is unnest-ed.
• Each queue gets two CROSS JOIN LATERAL branches, unleased rows and expired leases. Each branch is an ordered LIMIT n FOR UPDATE SKIP LOCKED scan.
• The branches are combined with UNION ALL, and an outer ORDER BY next_attempt_at, id LIMIT n picks the final rows.
• The UPDATE ... RETURNING is unchanged.

It's a SQLAlchemy Core statement built once per client, like the DLQ CTE from #226, with the queues, limit, lease TTL and new token as bind parameters. The SQL text is the same for any number of queues, so asyncpg's prepared-statement cache keeps working.

_pending_idx gains id as its last column. Rows published in one transaction share next_attempt_at, and without id a fetch reads the whole tie group.

Measured: EXPLAIN ANALYZE, limit 10, each run against its own index shape, best of three. "Rows read" is the largest number of rows any plan node read from the outbox table.

ready rows	PG 17 old	PG 17 new	PG 13 old	PG 13 new	rows read old → new
10k, tied	1.39 ms	0.08 ms	1.48 ms	0.09 ms	10,000 → 10
100k, tied	15.2 ms	0.08 ms	17.1 ms	0.09 ms	100,000 → 10
1M, tied	200 ms	0.14 ms	171 ms	0.12 ms	1,000,000 → 10
1M, spread	227 ms	0.07 ms	369 ms	0.09 ms	1,000,000 → 10
1M, spread, 5% expired leases	247 ms	9.9 ms	395 ms	24.8 ms	1,000,000 → 50,000

The last row is the expired-lease branch. It still sorts every expired lease on its queue. An index for that branch stays deferred (#220) until it's measured with a realistic number of hung handlers.

The rewrite briefly locks up to 2 × limit rows per queue while claiming only limit. Other fetchers skip those rows until the claim commits, which affects fairness, not correctness.

Drift, no grace period

As decided, validate_schema() reports the old index shape as drift. Alembic's index comparison already catches the changed column list, so no new probe was needed. The error reads missing index over ['queue', 'next_attempt_at', 'id'] (name='outbox_pending_idx', ...).

Upgrade

docs/operations/alembic.md has a new "Upgrading to 0.15" section with the migration:

• build the new index CONCURRENTLY under a temporary name, drop the old one CONCURRENTLY, then rename;
• everything inside autocommit_block(), plus a matching downgrade.

The section also covers:

• Order: run it before deploying 0.15. The old code works on the new index.
• Failed builds: a failed concurrent build leaves an INVALID index that has to be dropped before retrying.
• Autogenerate: don't apply the autogenerated migration, which rebuilds the index without CONCURRENTLY.

I ran the documented upgrade() and downgrade() through Alembic against real Postgres. The old shape fails validate_schema(), passes after upgrade(), and downgrade() restores the old index.

Tests

Two tests failed before the change:

• test_fetch_reads_only_the_rows_it_claims_from_a_backlog seeds 20,000 tied ready rows, captures the statement fetch executes, and runs it under EXPLAIN ANALYZE inside a rolled-back transaction. It asserts that no plan node reads more than 2 × limit rows from the outbox. The old statement read 19,990. This checks rows read, not node names or timings, so it doesn't depend on which plan Postgres picks, and it also catches an index without id.
• test_validate_schema_reports_the_pre_0_15_pending_index_shape checks the drift error for the old index.

Two new contract tests pin the outer ORDER BY ... LIMIT in both clients: the oldest rows across several queues, and the oldest across pending rows and expired leases. They passed before and after.

The invariant test on the statement's shape now checks the new form:

• each branch keeps its partial-index predicate;
• the queues are bound as one unnest-ed array;
• there are two LATERAL joins.

Docs

• alembic.md: the initial migration, index descriptions, drift table, and the upgrade section.
• how-it-works.md: the query illustration and why it has this shape.
• ADR 0003, "fetch query shape".
• just adr-check passes.

Verification

• 684 tests pass in docker at 100% coverage on PostgreSQL 17, and the full suite also passes on PostgreSQL 13.
• The full suite also passes on the floor dependency set (SQLAlchemy 2.0.0, Python 3.12). The first floor run found every real fetch failing with function unnest(unknown) is not unique: SQLAlchemy 2.0.0 sends the array bind without a type, the same problem fix: flush buffered deletes before the lease expires and match them on (id, token) #223 hit. The queue array now has an explicit cast.
• just bench-check passes, and no gated counter moved.
• ruff is clean, ty shows the same 6 diagnostics as main, and the docs build.

Release note (0.15.0, needs a migration)

The fetch query now claims rows with ordered per-queue index scans, so a fetch costs the same with ten ready rows or a million (about 200 ms down to 0.1 ms per fetch at a million ready rows). <table>_pending_idx gains id as its last column, and validate_schema() reports the old shape as drift. Rebuild the index concurrently before deploying, using the migration in the Alembic guide's "Upgrading to 0.15" section.

[github-actions]

Benchmark gate

✅ gate passed

scenario	msg/s	delete/msg	WALrec/msg	WALB/msg	fpi	upd	del	dead_tup
consumer/w1/b10	1317	1.000	6.69	913	242	5000	5000	10000
consumer/w1/b100	1405	1.000	7.09	1143	243	5000	5000	10000
consumer/w2/b10	1625	1.000	6.70	916	242	5000	5000	10000
consumer/w2/b100	1973	1.000	7.09	1143	243	5000	5000	10000
consumer/w4/b10	1673	1.000	6.72	942	242	5000	5000	10000
consumer/w4/b100	2037	1.000	7.09	1143	243	5000	5000	10000
consumer/w1/b100/tfbs100	8014	0.010	6.12	1157	265	5000	5000	10000
consumer/w1/b100/interval1-10	1491	1.000	7.09	1145	244	5000	5000	10000
producer/w1/b100	4224	0.000	3.04	597	1	0	0	0
producer/w1/b100/ppt1	900	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).