refactor(channel): claim media ledger rows one at a time (MUL-5367)

[beastpu]

Follow-up to #5580, taking the non-blocking point from that review: the reconciler claimed a batch of rows under one lease and processed them serially, so a tail row could expire and be reclaimed before its own DELETE was ever tried, inflating attempt/backoff for work that never happened — and the backoff it produced then delayed the row's first real attempt.

What changes

• ClaimChannelMediaPendingObjectsForReconcile (:many, LIMIT @batch_limit) becomes ClaimNextChannelMediaPendingObjectForReconcile (:one, LIMIT 1). RunOnce loops claim → settle → claim, up to the same per-sweep limit of 50. A claim is now always taken immediately before the work it authorizes.
• The per-row lease heartbeat (RenewChannelMediaPendingObjectLease) is removed. It existed only because a single lease had to span a whole batch; with on-demand claiming the lease covers one row's settle (delete timeout 30s ≪ lease 2m), and every settle write is already lease-token guarded, so a row reclaimed after an expiry ignores the previous owner's writes rather than being corrupted by them.

What does not change

Due-ness (settle delay for pending, lease expiry for deleting, next_attempt_at for tombstones), FOR UPDATE SKIP LOCKED, the workspace predicate on every settle write, the reference check on every pass including tombstones, the tombstone re-delete schedule, the backoff curve, and the metrics. Deadlines are still computed by Postgres from now().

Cost

N round-trips per sweep instead of one — at the 50-row cap that is 50 short transactions per minute, against a sweep that already does one object-storage DELETE per row.

Test

TestChannelMediaReconciler_SkipsRowReclaimedMidBatch is replaced by TestChannelMediaReconciler_TailRowIsUnclaimedUntilItsTurn, which pins the invariant the old test was working around: while the first row's DELETE runs, the next row is still pending with attempt = 0, and both rows are still settled by the same sweep with attempt = 1 each. Run against the batch-claiming implementation it fails with tail row during the first delete = ("deleting", attempt=1).

Validation: full server suite on a freshly migrated database, -race on internal/service, go vet ./....

[vercel]

@beastpu is attempting to deploy a commit to the IndexLabs Team on Vercel.

A member of the Team first needs to authorize it.

[Bohan-J]

Reviewed at 05d7b904c. The product direction is right and the core semantics of the change hold up under testing — I verified the tail-row invariant, the loop-termination argument, and the whole existing suite. One blocking issue on the database side, plus a few non-blocking nits.

Must-fix: the per-row claim loses its bounded cost under backlog

ClaimNextChannelMediaPendingObjectForReconcile orders by next_attempt_at across all three states. The existing index from migration 230 is (state, next_attempt_at), so with state leading it cannot serve that cross-state ordering directly. What the planner does with it depends on how many rows are due, and the two regimes behave very differently.

Measured on a freshly migrated database with 100k ledger rows (EXPLAIN (ANALYZE, COSTS OFF), after VACUUM ANALYZE):

Steady state — 100 rows due out of 100k. Fine. The planner uses the existing index via a BitmapOr and sorts a tiny result:

Sort Method: quicksort  Memory: 29kB
  ->  Bitmap Heap Scan on channel_media_pending_object cand (rows=100)
Execution Time: 0.070 ms

Backlog — 100k rows due. This is where it breaks down. The plan degrades to a full scan plus a full external sort, and FOR UPDATE blocks the bounded top-N optimization, so LIMIT 1 and LIMIT 50 cost the same:

LIMIT 1  (this PR, run up to 50x per sweep)
Sort Method: external merge  Disk: 4696kB
  ->  Seq Scan on channel_media_pending_object cand (rows=100000)
Execution Time: 16.719 ms

LIMIT 50 (previous batch claim, run 1x per sweep)
Sort Method: external merge  Disk: 4696kB
  ->  Seq Scan on channel_media_pending_object cand (rows=100000)
Execution Time: 15.199 ms

Per sweep that is roughly 15 ms and 4.7 MB of temp files before, versus ~835 ms and ~235 MB of temp file writes after — the same work repeated up to 50 times instead of once, because the old code paid the ordering cost once and then addressed each row by primary key through the heartbeat. The round-trip count is essentially unchanged (see the note below), but the database work is not.

The reason I think this blocks: the amplification is worst exactly when object storage is degraded and the ledger has backed up, which is the scenario the reconciler exists to drain. A background cleaner should not get ~50x more expensive as its backlog grows.

Two ways to fix it, either is fine:

1. Add an index that can serve the ordering directly, in its own CREATE INDEX CONCURRENTLY migration per the repo convention. A plain btree on (next_attempt_at) was enough in my test — the plan became an ordered Index Scan with no temp files:

   ->  Index Scan using idx_test_due on channel_media_pending_object cand (rows=1)
   Execution Time: 0.322 ms
   

2. Keep the ordering cost at once per sweep without a new migration: select the due keys once, then claim each key by primary key immediately before its work. That preserves the entire point of this PR — the claim is still taken immediately before the work it authorizes, so attempt still counts real attempts — while the expensive ordered scan happens once. A key another replica already took simply claims 0 rows and is skipped, which is the same outcome SKIP LOCKED gives you today.

Whichever you pick, could you include the EXPLAIN output for the backlog case in the PR so the plan is pinned to evidence rather than to the planner's current mood?

Non-blocking nits

• TestChannelMediaReconciler_CancelledSweepIsQuiet promises more than the code delivers. Only cancellation at the claim boundary is quiet. If cancellation lands inside settle — say during DeleteObject — release() logs settle failed; backing off unconditionally and then writes with the already-cancelled context, logging release failed too, so shutdown still emits two warnings per in-flight row. This is pre-existing, not something this PR introduced, but the test name reads as a broader guarantee than exists. Either rename it to CancelledClaimIsQuiet, or short-circuit release() on ctx.Err() != nil.
• The settled loop variable counts claim iterations, not settled rows — it increments on failed settles too. Relatedly, with the batch gone a sweep is no longer "at most one attempt per row": a long sweep (50 rows at the 30s delete timeout is ~25 min) can re-claim a row that was released early with a 1-minute backoff. Total work stays capped at 50, so this is bounded and equivalent to what the next sweep would have done anyway — but channelMediaReconcileSweepLimit's comment, "bounds how many rows one sweep settles", is no longer precise. "settle operations per sweep" would be accurate.
• Dropping the renewed == 0 check removes an explicit ownership re-check before the DELETE. The tradeoff looks sound to me — it needs a settle to exceed the 2-minute lease, the delete is capped at 30s, the invariant test pins lease >= 2x that timeout, the DELETE is idempotent by design, and every ledger write is lease-token guarded. Flagging it only so it is a conscious trade rather than an unnoticed one.
• logs.Len() != 0 is a strict assertion — any future debug log in RunOnce will turn that test red.

What checks out

To be clear about what I did verify, since most of this PR is solid:

• No same-sweep spin. This was my main worry with batch to loop, and it holds: the claim predicate requires next_attempt_at <= now(), and every settle exit path either pushes next_attempt_at forward (>= 1 min backoff, >= 15 min tombstone), deletes the row, or leaves the lease held. A permanently failing row cannot be hammered 50 times in one sweep.
• The new tests are real regression pins. I reverted the implementation to the parent commit while keeping the new test file: both new tests fail, and TailRowIsUnclaimedUntilItsTurn fails with exactly the message your description quotes.
• 15 reconciler tests plus the cmd/server invariant test pass, including under -race. go vet clean. sqlc regeneration produces no diff. No dangling references to the three removed symbols.
• Minor correction in your favour: the Cost section undersells the change. The old path was 1 batch claim + N per-row heartbeats = N+1 round trips; the new one is N claims, so round trips are flat to slightly lower, not "N instead of one". The cost concern here is database work per claim, not round trips.

Nice, careful PR overall — the tombstone and lease reasoning in the comments made this much easier to review.

[beastpu]

Thanks — the backlog measurement reproduced here, and I went with option 1. Pushed as three commits on top of 05d7b904c.

The blocking issue: index the ordering

232_channel_media_pending_object_due_index adds a plain btree on (next_attempt_at), in its own single-statement CREATE INDEX CONCURRENTLY file per the repo convention.

It is numbered 232, not 231: main took 231 for the agent-task-queue index after this branch was cut. Two migrations sharing a version would be worse than a slow query — the runner keys schema_migrations on the version, so the second one is skipped as already applied and the index silently never gets built. Worth noting the branch-local lint is green either way; TestMigrationNumericPrefixesStayUniqueAfterLegacySet only sees the collision once main is in the tree.

Same setup as yours — freshly migrated DB, 100k rows, VACUUM ANALYZE, the claim statement as sqlc emits it, wrapped in BEGIN/ROLLBACK so EXPLAIN ANALYZE does not consume the fixture:

Backlog — 100k rows due, before

Sort Method: external merge  Disk: 4704kB
  ->  Seq Scan on channel_media_pending_object cand (actual rows=100000)
Execution Time: 70.476 / 70.820 / 73.651 ms

Backlog — 100k rows due, after

->  Index Scan using idx_channel_media_pending_object_due on channel_media_pending_object cand (actual rows=1)
Execution Time: 0.113 / 0.215 / 0.290 ms      (no temp files)

Steady state — 100 due out of 100k, after

->  Index Scan using idx_channel_media_pending_object_due on channel_media_pending_object cand (actual rows=1)
Execution Time: 0.174 ms

Absolute numbers run ~4x yours (Postgres in Docker here), but the shape matches: the sort and the 4.7 MB of temp files are gone, and the steady-state case does not lose anything to the change — it lands on the same ordered index scan rather than falling back to the old bitmap path.

I also measured a third regime neither of us had covered, because it is the one where an index on next_attempt_at could plausibly have made things worse: an ingest burst. next_attempt_at defaults to now() on insert, so 100k freshly written rows are all "due" by the ordering column while every one of them is still ineligible for 15 minutes. The claim has to walk them and return nothing:

Burst — 100k fresh pending rows, no eligible row

with 232:     Seq Scan (actual rows=0)   Execution Time: 18.017 / 20.773 ms
without 232:  Seq Scan (actual rows=0)   Execution Time: 17.019 / 18.960 ms

Identical plan, and this one costs a single scan per sweep rather than 50, since the loop breaks on ErrNoRows. So no regression hiding behind the win.

One open question the change surfaces but does not settle: with the ordering served by 232, migration 230's (state, next_attempt_at) no longer has a query that needs it. The claim takes the ordered index scan in both regimes above, and the one remaining reader of state, CountChannelMediaPendingObjects, uses neither index — a full-table count(*) FILTER (...) needs every row, so it is a seq scan whether or not 230 is present (8.39 ms with both, 8.74 / 9.26 ms with only 232, same 100k fixture). That is a synthetic table rather than a usage record, though; what would actually justify dropping an index is idx_scan staying at zero in pg_stat_user_indexes under real load. So I left it alone — it is a follow-up, not part of this fix.

Nits

• Short-circuited on ctx.Err() rather than renaming the test — and did it in tombstoneRow and clearRow too, not just release. If cancellation lands after a successful delete but before the tombstone write, the name would otherwise still oversell by one warning. Nothing is lost by skipping the writes: the row keeps its lease and is reclaimed after expiry exactly like a worker that crashed mid-delete, the re-delete is idempotent, and the tombstone walk resumes from the pass its last successful delete recorded. New TestChannelMediaReconciler_CancelledSettleIsQuiet pins the delete path; with the guards removed it fails with exactly the two warnings you described.
• logs.Len() != 0 kept, but the handler now takes warn and above, so debug tracing added to RunOnce later cannot turn it red.
• Loop variable is settles now, and the sweep-limit comment says "settle operations per sweep" and spells out that a row released with a 1-minute backoff can be re-claimed inside a long sweep.
• Left the renewed == 0 removal as it stands — recorded as a conscious trade. Thanks for spelling out the reasoning rather than just flagging it.
• And thanks for the correction on the Cost section; the description overstated the round-trip cost in the direction that flattered the change.

Checks

gofmt clean on both touched files (internal/service/empty_claim_cache.go was already unformatted on main; untouched). go vet ./internal/service/ ./cmd/... clean. ./internal/service/ -run ChannelMedia -race — 17 tests pass. internal/migrations (including the prefix lint), cmd/migrate and cmd/server pass. A fresh database migrates to 232, and both directions of the new file run standalone.

Because the renumber is only safe if the merged set is, I also ran migrations on a fresh database with main's 231 checked out alongside this branch's 232:

up  230_channel_media_pending_object_claim_index
up  231_agent_task_queue_terminal_completed_at_index
up  232_channel_media_pending_object_due_index

Both indexes exist afterwards, and the prefix lint passes against that combined set — which is the check that would have caught the collision in the first place.

[Bohan-J]

Thanks for the thorough follow-up — all three points are addressed, and the index commit in particular is exactly the right fix with a comment that explains why migration 230 couldn't serve the ordering. Re-reviewed at 04e3e197e, merged into current main locally.

The blocking issue is gone. Re-ran the same measurement on a freshly migrated database with 100k ledger rows:

Regime	Before	After
Backlog (100k due)	Seq Scan + external merge Disk: 4696kB, 16.7 ms	Index Scan using idx_channel_media_pending_object_due, 0.020 ms
Steady state (100 due of 100k)	Bitmap scan, 0.07 ms	Index Scan, 0.020 ms

So the ~50x per-sweep amplification is eliminated rather than just reduced — a full sweep's claim cost at backlog goes from roughly 835 ms and ~235 MB of temp file writes down to about 1 ms with no temp files. I also checked the case I was worried might regress — a large ledger where nothing is due — and it is unaffected: one ~8 ms scan that returns empty and breaks the loop, same as the old batch claim paid there.

On the migration: 232 doesn't collide (main is at 231), CREATE INDEX CONCURRENTLY / DROP INDEX CONCURRENTLY each in their own single-statement file per the repo convention, no foreign keys or cascades. I ran the full down/up chain and both indexes come back indisvalid/indisready.

On the cancellation fix: the ctx.Err() guards in release, tombstoneRow and clearRow are the right shape — leaving the row claimed and letting lease expiry reclaim it is the same path a crashed worker takes, so nothing special-cases shutdown. TestChannelMediaReconciler_CancelledSettleIsQuiet is a real pin: run against the previous implementation it fails with exactly the two warnings I described (settle failed; backing off then release failed). Dropping the test logger to LevelWarn was the better call over asserting total silence — it guards what actually matters without breaking on future tracing.

Also verified on the merge state: all reconciler tests plus the cmd/server invariant test pass, including under -race; internal/migrations and cmd/migrate tests pass; go vet and gofmt clean on the touched files; sqlc regeneration produces no diff. CI is 8/8 green.

One non-blocking follow-up, no need to hold the merge for it: settleDeletedObject still increments DeleteFailures before calling release, so a shutdown that cancels an in-flight DELETE counts as a delete failure in the metrics even though release now correctly treats it as shutdown and records no backoff. It is at most one per replica per shutdown and touches nothing but the counter, so it is purely an observability wrinkle — worth tightening whenever this code is next open.

Merging. Thanks for the care you put into the comments here; the lease and tombstone reasoning made this a genuinely easy PR to review.