feat(rollout): concurrent writes with periodic flush

[beinan]

Summary

Rollout ingest was throughput-bound under high concurrency with a slow object store. Every add did put → force_seal_active → wait_for_flush_drain behind add(&mut self) and a handler store.write() lock, so a single hot store was capped at ~1/latency and concurrent requests queued on the exclusive lock (the source of the observed p99 = 20–30s — queueing, not slow single writes).

The serialization was ours, not lance's: ShardWriter::{put, force_seal_active, wait_for_flush_drain} are all &self, and put waits for WAL durability outside its internal lock. Verified against lance 7.0 mem_wal source:

• Durability = the WAL entry PUT to object storage (wal.rs atomic_put). That alone means "won't be lost".
• Flush (materializing the memtable into a queryable generation) is a separate, deferrable concern; freeze_memtable swaps in a fresh memtable and hands the old one to a background flusher without holding the state lock.
• Reopen replays un-flushed WAL (replay_memtable_from_wal), so a durable-but-unflushed put is recovered on a fenced-writer reopen — no loss.

Changes

• add is now &self — concurrent appends each do only a durable WAL put and return. The resident writer lives in Mutex<Option<Arc<ShardWriter>>>, locked only to fetch/open/invalidate, never across put, so steady-state appends run concurrently. Fenced writers reopen exactly once via an identity-checked (Arc::ptr_eq) invalidation.
• New flush(&self) seals the active memtable into a queryable generation. New spawn_flush_sweeper flushes every resident store on a fixed interval (ROLLOUT_FLUSH_INTERVAL_SECS, default 30, 0 disables) and runs the count-triggered merge that used to ride the append path.
• Add handler switches store.write() → read().
• MemWAL index init moves to open time so the hot add path stays &self.

Semantic change (accepted)

Read-after-write is now asynchronous, bounded by the flush interval (≤30s). Durability is unchanged — every add waits for its WAL PUT before returning.

Test plan

• New tests: 50 concurrent adds on a shared Arc<RolloutStore>, all durable, none lost; durable-but-not-visible-until-flush; flush-is-noop-without-writes.
• The 27 add-then-read core unit tests and 5 server route tests now flush() before reading (the accepted async-visibility change).
• wal_merge integration test asserts correctness via the deduplicating read path instead of observe().row_count — see WAL merge leaves physical duplicate rows in the base table; count-based reads over-count #180 (pre-existing base-table physical-duplication bug, reproduces on main, unrelated to this change).
• cargo test -p lance-context-core (161 passed) and -p lance-context-server (49 passed).
• cargo fmt --check + cargo clippy --all-targets clean on both crates.
• Manual: high-concurrency ingest against a slow object store; confirm throughput scales past 1/latency and rows appear within the flush interval.

Notes

• The count-triggered merge (ROLLOUT_MERGE_AFTER_GENERATIONS) moved off the append path onto the flush sweeper; the option is unchanged and still used by datagen.
• Follow-up filed as WAL merge leaves physical duplicate rows in the base table; count-based reads over-count #180: WAL merge leaves physical duplicate rows in the base table (reads are correct via id-dedup, but raw count_rows() over-counts).

🤖 Generated with Claude Code