fix(mem-wal): retry a failed L0 flush instead of committing past it

[hamersaw]

Fixes #8293.

The bug

A failed MemTableFlushHandler::flush_memtable was logged and dropped, on the stated assumption that "the worst case for a transient flush failure is replay from the WAL on next open" (TaskDispatcher::run). That assumption broke as soon as the next generation committed: MemTableFlusher::update_manifest unconditionally stamped the committing generation's coordinates, advancing replay_after_wal_entry_position past WAL entries no SSTable held. The failed generation's rows then survived only as a retained FrozenMemTable and died with the process.

Measured on the regression test with the fix reverted: after generation 2's flush fails and generation 3 commits, the manifest reads current_generation=4, cursor=3, and reopening replays 0 rows. Generation 2's row is gone.

The fix

Generations must reach L0 in order, because the manifest commit that records the SSTable and advances the cursor is the shard's single atomic "this reached L0" agreement. flush_generation_with_retry retries a failed generation. The dispatcher awaits handle() inline on a channel of its own, so the loop head-of-line blocks the flush queue: generation N+1 cannot start, let alone commit, while N is still failing. Ordering becomes structural rather than something a later commit has to remember to check.

While retrying, the cost is memory and never data. Frozen memtables queue up behind the retry until backpressure stalls writes, and every row stays WAL-durable and readable throughout. A process that dies mid-retry leaves the uncommitted generations as WAL entries past the cursor, which the next open replays.

Three things break the loop:

• A budget, then a poison. See below.
• Shutdown, via a CancellationToken clone. Cancellation is only polled between messages, so without it abort() would join a dispatcher that never returns.
• Errors a retry cannot clear — Fenced, InvalidInput, PrerequisiteFailed. All deterministic; retrying would spin on a condition waiting cannot change and bury a real bug in an endless log. This list is an optimization, not a safety property: a deterministic error not named here costs the full budget before poisoning, which is slower but still terminates.

Two supporting changes fell out of the restructure:

• The WAL-completion await moves out of the retry loop. It consumes a once-cell and cannot be re-awaited, and a failure there means the rows are not durable — nothing an L0 retry can fix.
• Terminal bookkeeping (completion signal, backpressure drain, frozen retirement) now runs once per generation rather than once per attempt. The completion cell is a WatchableOnceCell, so signalling per attempt would publish a failure the generation later recovered from.

The budget, and the poison on exhaustion

Head-of-line blocking is why the budget has to be finite. The loop holds the only flush task, and close() awaits the flush watchers before shutdown_all() cancels the token the loop selects on — so a loop whose only exit was that cancellation would deadlock close() on a permanently-down object store, with nothing left to wake the watchers.

Exhausting the budget instead poisons the writer, the same fail-stop the WAL appender already uses for its own persistence failures (WalRetryConfig → Error::writer_poisoned). Waiters and later writes fail fast with the typed reason, and reopening replays the WAL entries the failed generation still covers, so the fail-stop costs no data.

ShardWriterConfig gains max_l0_flush_retries (default 8, roughly 12s of backoff), l0_flush_retry_base_delay (50ms), and l0_flush_retry_max_delay (30s), mirroring the wal_persist_retry_* pair. The L0 budget is larger than the WAL's 3 on purpose: a stalled L0 flush risks no data, since the rows are already WAL-durable, so it is worth waiting out an object-store blip rather than forcing a reopen.

Three consequences worth calling out:

• A generation queued behind one that poisoned returns immediately rather than burning its own budget against a store already known to be dead.
• close() now reports a poisoned writer even when no watcher is left to carry the failure. The handler drains the watcher on its way out, so close() used to find an empty watcher list and return Ok(()) for rows that never reached L0. That predates this PR but is the same silent-success class, so it is fixed here.
• Flush watchers keep the fence reason. DurabilityResult carries only a message, and callers key their reopen-versus-retry decision on fence_reason() rather than on the text, so the poison latch is consulted when restoring the error.

The manifest backstop

ShardManifestStore::check_generation_contiguous refuses, at the commit, any generation that is not the manifest's current_generation.

Under correct operation this never fires; the retry loop is what enforces ordering. It is here because the consequence of the invariant breaking is silent data loss rather than a visible error, and because a process-local queue cannot speak for a commit path added later. commit_update's closure became fallible so the check runs against the manifest actually written, not one a conflict retry left stale.

Tests

• test_flush_retries_until_it_commits_holding_later_generations — breaks storage, seals generations 2 and 3 behind the failure, asserts the manifest does not budge while 2 retries, then heals storage and asserts both land in order ([1,2,3]) with nothing left for replay. Fails with retry disabled (current_generation stays at 2).
• test_flush_poisons_writer_when_retries_are_exhausted — a permanently-broken store exhausts the budget; the drain, a later put, a scan, and close() all surface PersistenceFailure, and reopening replays the row that never reached L0. close() runs under a timeout, since the bug it covers is a hang rather than a wrong value.
• test_flush_retry_gives_up_when_fenced_by_successor — a successor claiming the epoch terminates the loop instead of spinning against a shard the writer no longer owns.
• test_flusher_refuses_generation_gap — the manifest backstop.
• FailControls::fail_sstable_puts extends the existing WAL fault-injection store to generation writes.

cargo test -p lance --lib mem_wal → 578 passed. cargo fmt --all, cargo clippy -p lance --tests --benches -- -D warnings, and RUSTDOCFLAGS="-D warnings" cargo doc all clean.

Known cost

Each retry attempt starts a fresh {hash}_gen_{n} directory, so nothing carries over. In the common case that is cheap — write_data_file is the first IO, before any index build, so an unavailable store aborts the attempt before index work. A late failure (index write, bloom filter, PK sidecar, or the commit itself) does redo everything, and leaves the abandoned directory behind; the budget caps how many such directories one failing generation can produce, but does not clean them up.

Retrying a stable prepared artifact in place would fix both, and would also let the commit distinguish an ambiguous manifest write (the PUT landed, the ack did not) from a real failure — today that case reports the generation as failed even though it committed, which is a spurious error rather than data loss. Both need the lost-ack case thought through, so they are left as follow-ups.

🤖 Generated with Claude Code

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The manifest contiguity guard is the right invariant, but the whole-generation retry policy needs a bounded, fail-stop lifecycle before this is safe to operate. Keep the queue-head ordering and backstop, then use bounded cancellation-aware retries followed by a latched writer error and reopen/WAL replay; retries must also reuse or clean uncommitted generation artifacts and reconcile ambiguous manifest commits.

Please mark this PR with the breaking-change label.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The retry budget fixes the original unbounded loop, but the fail-stop transition is still incomplete: a poisoned L0 generation can be forgotten by durability acknowledgements, and a writer already waiting under backpressure can remain stuck after poison. Make the terminal state observable by every drain/fence and recheck it before any post-wait write mutation.

Please mark this PR with the breaking-change label.

[lance-gatekeeper]

This restores the poison only while a watcher is still present. flush_memtable signals the failure and then removes that watcher, so a later wait_for_flush_drain() observes an empty queue and returns Ok(()) although the generation never reached L0. SealFence::wait() bypasses this helper altogether and flattens the same poison into Error::IO with no FenceReason, despite the documented reopen-versus-retry contract. Preserve a typed terminal outcome that every captured fence and every drain can consult, and check it before any success return.

Reproducer

Against this head, I extended test_flush_poisons_writer_when_retries_are_exhausted in two ways and ran:

cargo test -p lance --lib test_flush_poisons_writer_when_retries_are_exhausted

For the fence path:

let fence = writer.force_seal_active().await.unwrap();
let err = fence.wait().await.unwrap_err();
assert_eq!(
    err.fence_reason(),
    Some(FenceReason::PersistenceFailure)
);

The assertion failed with left: None. After the test's first expected failed drain, I also added:

tokio::time::sleep(Duration::from_millis(10)).await;
writer
    .wait_for_flush_drain()
    .await
    .expect_err("a second drain must still report the poisoned writer");

That failed because the second drain returned Ok(()).

[lance-gatekeeper]

Latching poison here does not cover a put already inside maybe_apply_backpressure: that put passed check_poisoned() before waiting, the loop discards the failed watcher's outcome, and put_memtable_no_wait never rechecks before inserting. If the remaining active bytes stay over the threshold after the failed generation is removed, the put sleeps forever with no watcher left; if they fall below it, the put can mutate after the fail-stop boundary. Propagate the failed watcher/poison through backpressure and recheck poison immediately after the wait, before acquiring the state lock and inserting.

Reproducer

In test_flush_poisons_writer_when_retries_are_exhausted, I set:

l0_flush_retry_base_delay: Duration::from_millis(20),
max_unflushed_memtable_bytes: 1,

Then, after sealing the failing generation, I raced a put already entering backpressure with the drain:

let (_put_result, _drain_result) = tokio::join!(
    writer.put(vec![create_test_batch(&schema, 1, 1)]),
    writer.wait_for_flush_drain(),
);

Running timeout 5 cargo test -p lance --lib test_flush_poisons_writer_when_retries_are_exhausted exited 124 after printing running 1 test; the put remained parked after the L0 retry exhausted and poisoned the writer.