server: unify the flush and cleanup sweepers into one guarded per-store maintainer #258

Problem

MemWAL maintenance in the server has three entry points with two different lock disciplines:

Consequences beyond #257:

1. Double prepare. Both timers can prepare the same generations of the same shard concurrently. commit_merge correctly detects the loser (re-reads the manifest, returns false if the generations are already drained) so there is no double append, but the loser wasted a full object-store read of up to MERGE_MAX_GENERATIONS generations plus lock time.
2. Flush sweeper still walks kinds serially. fix(server): stop generic-store WAL merges starving behind rollout sweeps #256 parallelised spawn_global_sweeper with tokio::join! but not spawn_flush_sweeper, which still does rollout → datagen → generic. Now that the flush pass does real merge work for generic, a slow rollout walk delays generic's tick — the same starvation shape one level up.
3. Serial walk within a kind. With hundreds of resident rollout stores, one tick's walk can exceed the 30s interval; stores at the end of the list are visited less often.
4. Three places to instrument, three places to get the lock order wrong.

Proposed design

One maintainer per store kind, one 30s ticker, one lock discipline:

async fn tick(store, force_full: bool) -> Result<usize> {
    // 1. seal — read lock, cheap
    store.read().await.flush().await?;
    // 2. decide + prepare — read lock, expensive
    let threshold = if force_full { 1 } else { store.read().await.merge_after_generations() };
    if !force_full && threshold == 0 { return Ok(0); }          // count trigger disabled
    let prepared = store.read().await.prepare_merge_if_ready(threshold).await?;
    // 3. commit — write lock, short
    match prepared {
        Some((ms, m, p)) => store.write().await.commit_prepared_merge(&ms, &m, p).await,
        None => Ok(0),
    }
}

• Every tick: seal, merge if pending ≥ ROLLOUT_MERGE_AFTER_GENERATIONS.
• Every Nth tick (N = ceil(CLEANUP_INTERVAL / FLUSH_INTERVAL), default 10): force_full = true (threshold 1). Preserves today's "time OR count" trigger exactly.
• POST …/merge-wal calls tick(store, force_full = true).
• Per-store merge guard: tokio::sync::Mutex<()> with try_lock; if a merge is in flight for this store, skip the tick. Eliminates double prepare.
• Kinds run with tokio::join!; within a kind, walk with buffer_unordered(K) (K ≈ 4) rather than a serial for, so one slow store does not serialise the rest.
• Keep the per-store timeout(5 × interval).

Corner cases considered

1. Merge longer than one tick. Guard try_lock fails on tick N+1 → skip. Ticker uses MissedTickBehavior::Skip so there is no catch-up burst.
2. add during prepare. Both hold read locks; LSM guarantees merge touches only sealed history. Already the design premise of the prepare/commit split.
3. New generation sealed between prepare's manifest read and commit. commit_merge drains with a relative edit inside a CAS commit_update closure (retain generations ∉ merged), so it composes with the concurrent append. Must not be refactored into an absolute assignment — see the existing doc comment on commit_merge.
4. Epoch fencing. Commit reuses the shard's current epoch; if a new writer claimed the shard meanwhile, commit_update fails cleanly. tick must propagate that error rather than swallow it.
5. Version-pinned or deleted store. prepare_merge_if_ready_inner returns None; commit_merge re-checks both after taking the write lock. Both checks must survive.
6. ROLLOUT_MERGE_AFTER_GENERATIONS = 0. Must mean "count trigger off". Note prepare_merge_if_ready_inner does threshold.max(1), so passing 0 through would merge every tick — handle explicitly as above.
7. Interval config. FLUSH_INTERVAL = 0 → no ticker (also disables the full drain — behaviour change vs today, document it; main.rs already warns when both are off). CLEANUP_INTERVAL = 0 → never force. CLEANUP < FLUSH → N = 1. Non-multiple → round up.
8. Timeout between base-table append and manifest drain. The append is a merge_insert keyed by the id column, so a retry re-merges the same keys idempotently; a timeout here leaks one round of work, not correctness. Log it distinctly.
9. Memory bound. Up to 3 concurrent merges (one per kind), each ≤ ROLLOUT_MERGE_MAX_BYTES. Today's effective bound was already 2. Document 3 × MERGE_MAX_BYTES in the env-var help. Optionally a process-wide Semaphore(1) around prepare; start without and measure.
10. Orphaned shards (owner ordinal no longer exists) are still never merged — out of scope, needs master fan-out for generic stores like rollout has via WORKER_ENDPOINTS.

Tests to add

There are currently no tests asserting sweeper behaviour, which is how #257 shipped.

• Count trigger: store with merge_after_generations = N, N sealed generations → one tick → pending_wal_generations() == 0.
• Forced tick merges a single pending generation.
• Second tick while a merge is in flight skips (guard) and does not double-append (row count unchanged).
• merge_after_generations = 0 never count-merges; forced tick still drains.
• Deleted / version-pinned store: tick is a no-op.

Sequencing

1. A small observability PR first (separate issue): warn + metric when any shard's flushed_generations.len() exceeds a threshold, so read fan-out pages before it OOMs.
2. This refactor, replacing spawn_flush_sweeper, spawn_global_sweeper and the two merge-wal route bodies.
3. Master fan-out for generic stores (feat: add python bindings for context store #10 above).

Open questions where I would like input: keep two tickers feeding one guarded tick() instead of one ticker with an Nth-tick flag (case 7), and whether to add the global prepare semaphore now (case 9).