fix(wintermute): bound backfill shutdown by a grace budget and claim resync rows first

[afbase]

Two fixes to rsky-wintermute's backfill runner from tonight's soak test on the appview box.

Problem A: hybrid mode cannot stop within systemd's 60 s

With BACKFILL_MODE=hybrid the daemon was SIGKILLed on every stop (three of three cycles); with the mode off it stopped in 16 s. Thread probes and logs showed three unbounded waits in the stop path:

• fetch workers finished their in-flight fetch+parse before logging fetch worker stopped (up to ~35 s after SIGTERM);
• BackfillManager::run then awaited sink.finish(), which drained every in-flight record (up to BACKFILL_MAX_RECORDS_IN_FLIGHT) through Postgres with no deadline;
• enumerate_pages's transient backoff (sleep_secs(delay), up to the 300 s Retry-After cap) did not observe SHUTDOWN at all.

Claimed rows are already returned to pending on restart (reset_claimed, covered by claimed_rows_are_recovered_after_a_crash), so abandoning in-flight work at shutdown is safe.

Design

• BACKFILL_SHUTDOWN_GRACE_SECS (default 20), read in BackfillConfig::from_env, threaded into RunnerConfig::shutdown_grace, documented in the README env table. It is one budget for the whole stop, not per stage: the coordinator records when it first saw the flag, spends what it needs on the workers, and hands the remainder to the sink drain.
• Shutdown-aware sleeps. New runner::sleep_or_shutdown(dur, flag) polls the flag every 250 ms and returns early. Every wait that can run long goes through it: the listRepos retry backoff, the hubble cooldown, the enumerator's 60 s error backoff, the re-enumeration interval, and the coordinator poll. A backoff cut short by a stop returns from enumerate_pages as an incomplete pass (cursor persisted) rather than a host failure, so the host is not put on a 300 s cooldown for being stopped.
• Awaited vs aborted. On shutdown drain awaits the fetch-worker JoinSet under tokio::time::timeout(remaining grace); workers that finish in time settle normally (their repos complete). Anything still in flight past the budget is aborted (JoinSet::shutdown) with a warn naming the sources. An aborted fetch never reached the sink, so it leaves no receipt and no completion task: the row stays claimed and reset_claimed recovers it on the next start. The enumerator task is aborted as before.
• Bounded sink drain. RecordSink gains finish(&self) / uncommitted() / abandon() (defaults are no-ops; NullSink uses them). PgSink::finish now takes &self (the intake sender lives in a Mutex<Option<_>> and is taken to close the channel), which also removes the Arc::into_inner(sink) dance that silently skipped the drain whenever a completion task still held the runner. backfiller::finish_sink awaits finish unbounded until a stop is requested, then under timeout(grace); on timeout it logs at warn how many repos/records are abandoned and that they will be re-fetched from their claimed state on restart, then abandon() aborts the writers. run_and_finish ties the runner and the sink drain together and is used by the manager and the backfill CLI.
• No double completion. A repo's committed receipt is a oneshot that resolves exactly once: Ok (complete), Err (fail with cooldown), or dropped when a writer is aborted (fail(..., "sink dropped") returns it to pending). Fetch tasks aborted before ingest leave the row claimed. ingest also rolls its in-flight counters back if the send is rejected by a closed channel.
• The runner's shutdown flag is now RunnerConfig::shutdown: &'static AtomicBool (default crate::SHUTDOWN), so tests inject their own flag instead of racing each other on the global.

Problem B: resync requests queued behind millions of enumerated repos

RepoStateStore::claim_for_source ordered by cooldown_until only, so live sync 1.1 resync requests (request_resync, last_error = 'resync: ...') competed equally with ordinary pending rows. In the soak, three resync rows sat pending/claimed for 24 minutes with zero completions while 7.8M enumerated repos were pending.

Design

• New repo.priority INTEGER NOT NULL DEFAULT 0 column. request_resync sets 1; complete (and mark_dry_run) clear it to 0 alongside last_error = NULL. A transient failure keeps it, so the retry is still prioritised.
• A partial index repo_priority_claimable ON repo (source, state, cooldown_until) WHERE priority > 0 keeps the priority probe a lookup in a tiny index. claim_for_source claims from it first (ORDER BY cooldown_until, rowid), then fills the remainder from the existing repo_claimable index with priority = 0, all in one transaction. Net order: priority desc, cooldown_until, rowid. EXPLAIN QUERY PLAN confirms both selects and has_claimable stay on their indexes.
• Idempotent in-place upgrade in RepoStateStore::init: PRAGMA table_info(repo) is checked, the column is added with ALTER TABLE ... ADD COLUMN if missing, and waiting resync rows (last_error LIKE 'resync:%', state pending/claimed) are promoted to priority 1 so the box's existing state.sqlite fixes its three stuck rows on the first start. The partial index is created after the migration so an old file does not hit an unknown column.
• README: one sentence in the sync 1.1 section on resync requests being fetched ahead of enumerated repos; state-file table mentions the priority.

Tests

New: shutdown-aware sleep returns promptly when the flag flips (before, at once, and midway); a 300 s Retry-After backoff ends at shutdown without cooling the host; a fetch that finishes within the grace budget is awaited and completes; a fetch still in flight past the budget is aborted with the row left claimed and reset_claimed recovering it, then a hung sink is abandoned at once because the shared budget is spent; drain stops within the budget on shutdown; finish_sink abandons a never-resolving sink after the grace budget (and counts the budget from a stop that arrives midway); run_and_finish returns within the budget against a sink whose finish never resolves; a resync row is claimed before older pending rows for the same source (and a priority row under cooldown still waits); priority clears on completion; a pre-priority state file is upgraded in place with waiting resyncs promoted.

DATABASE_URL=postgresql://postgres:postgres@localhost:55435/bsky_test cargo test -p rsky-wintermute
test result: ok. 261 passed; 0 failed; 2 ignored; 0 measured; 0 filtered out; finished in 16.79s

cargo fmt --all -- --check clean; cargo clippy -p rsky-wintermute --all-targets reports zero diagnostics for the crate. The backfiller tests were run five more times with no flakes.

Not verified here: the stop time on the box itself under real Postgres load; the design bounds the stop at BACKFILL_SHUTDOWN_GRACE_SECS plus the ~250 ms flag-poll granularity, so the default 20 s stays well inside systemd's 60 s.

Commits are unsigned (commit.gpgsign=false).

🤖 Generated with Claude Code

[afbase]

Follow-up: the stop path still hung after the sink was abandoned

gdb on the box showed the backfill thread still inside Runtime::block_on::<BackfillManager::run::{closure}> 33 s after sink did not drain ... abandoning was logged, all runtime workers parked; in one of two cycles it had exited by +35 s, so the hang was data-dependent.

The blocker, found by reading the code: finish_sink ran tokio::time::timeout(grace, &mut finish) on a borrowed pinned future. When the timeout fired only the Timeout wrapper was dropped; the finish future itself stayed alive on the stack until finish_sink returned. PgSink::finish takes the tokio writers mutex on its first poll and then parks on handle.await for a writer that is mid-batch — so the timed-out future kept holding that guard, and PgSink::abandon()'s self.writers.lock().await waited on it forever with nothing left to wake it. When every writer was idle at the stop, finish completed at once and nothing hung: that is the cycle that exited at +35 s.

Changes

1. Fix: in finish_sink the finish future is owned by an inner block and dropped before abandon runs (the select! picks either completion or "stopped, now bounded"). The StuckSink test double now holds a tokio mutex in finish that abandon also needs, the same shape as PgSink, so the existing abandon test is the regression test.
2. Outer bound: run_bounded(runner, sink, slack) races run_and_finish against "SHUTDOWN observed + BACKFILL_SHUTDOWN_GRACE_SECS + slack" with tokio::select!; BackfillManager::run uses it inside block_on with a fixed SHUTDOWN_SLACK of 5 s. Losing the race logs at warn which phase was still pending (runner / sink drain / sink abandon). The runtime is then torn down with rt.shutdown_timeout(2 s) instead of the implicit drop, so a spawn_blocking task (sqlite, CAR parse) cannot hold the thread either.
3. abandon(): the writers lock is taken under a 1 s timeout (warns and returns if the drain still holds it), each aborted writer's handle.await is under a 1 s timeout, and writers that did not settle are counted and logged at warn.
4. No unbounded sqlite on the stop path: log_progress skips sample_state_metrics (SELECT state, COUNT(*) FROM repo GROUP BY state, a full scan of the 13M-row table) when shutting down. It ran at the end of drain and is the likely source of the ~2 s between the "aborting" and "abandoning" log lines. The remaining state-store calls on the stop path are single-row updates.
5. Test: the_whole_stop_path_is_bounded_even_if_abandon_never_resolves — an idle runner with the flag set and a sink whose finish and abandon never resolve returns from run_bounded within grace + slack (300 ms + 200 ms in the test), and an unstopped run is not bounded at all.

DATABASE_URL=postgresql://postgres:postgres@localhost:55435/bsky_test cargo test -p rsky-wintermute
test result: ok. 262 passed; 0 failed; 2 ignored; 0 measured; 0 filtered out; finished in 9.26s

cargo fmt --all -- --check clean; cargo clippy -p rsky-wintermute --all-targets zero diagnostics for the crate.

🤖 Generated with Claude Code