server: count-triggered WAL merge holds the store write lock across the generation read #257

Problem

Sweepable::flush for both RolloutStore and GenericStore (crates/lance-context-server/src/sweeper.rs) seals under the read lock, then takes the write lock and calls maybe_merge_own_shard() / maybe_merge_wal():

let mut guard = self.write().await;
guard.maybe_merge_wal().await   // prepare (reads up to MERGE_MAX_GENERATIONS from object storage) + commit

merge_own_shard_if_ready does the whole merge inside that call — including read_flushed_generations, which opens up to ROLLOUT_MERGE_MAX_GENERATIONS generation datasets from object storage. That read takes seconds to tens of seconds on ABFS/S3, and the store's exclusive lock is held for all of it. Every add/get/list on the store blocks, and the next 30s flush tick blocks behind it too.

The cleanup path (Sweepable::merge_wal) already uses the correct prepare/commit split after #256 — read lock for prepare_cleanup_merge, write lock only for commit_prepared_merge. flush() was not updated in #256 and the rollout impl has had this shape since #238.

Observed in production after #256 landed: with one hot generic store the 30s flush sweeper logged flush sweeper timed out 52 times in 15 minutes and request handlers hit Too many concurrent writers … Attempted 5 times. Once the backlog drained the warnings went to zero, confirming lock hold time scales with the merge read.

The same pattern exists in POST /api/v1/generic/{name}/merge-wal (routes/generic.rs), which does store.write().await; guard.cleanup_wal().

Proposed fix

Apply the existing prepare/commit split to the count-triggered path:

async fn flush(&self) -> Result<(), String> {
    { self.read().await.flush().await? }                                  // seal
    let prepared = { self.read().await.prepare_count_merge().await? };    // expensive read, shared lock
    if let Some((ms, m, p)) = prepared {
        self.write().await.commit_prepared_merge(&ms, &m, p).await?;      // short, exclusive
    }
    Ok(())
}

with a new StorageBase::prepare_count_merge() that encapsulates the threshold check (returns None when merge_after_generations == 0, otherwise prepare_merge_if_ready(merge_after_generations)), mirroring the existing prepare_cleanup_merge(). Thin passthroughs on RolloutStore and GenericStore. Same change for the merge-wal route handlers.

A secondary bug in the same function: if the seal succeeds and the merge fails, flush() returns the merge error, which flush_pass logs as flush sweeper failed and counts under rollout_wal_flush_total{result="failed"}. The flush did not fail. Merge errors should be logged/counted under the cleanup metric.

Superseded by the unified maintainer design in the follow-up issue if that is adopted; this is the minimal fix.