Persistence lifecycle: close(), terminal Failed state, wait_for_ops surfacing errors

[pathscale]

Context

Expert review round 2 (2026-07-27) findings 2, 3 and part of the vacuum API feedback. Deliberately deferred to this issue by maintainer decision; the interim mitigations shipped in the fix series are documented below.

Problems

1. Shutdown: PersistenceTask owns a detached engine task. Dropping the task previously either leaked the task (pre-0014: a reopened table raced the old engine on the same files) or aborted it unconditionally (0014: AbortHandle::abort() can cancel mid-batch — the batch sub-futures are not cancellation-safe; a data page can be left half-written while its index events are abandoned). Interim shipped: Drop aborts only when the engine is provably idle (check_wait_triggers), otherwise logs tracing::error and leaves the task running; callers must wait_for_ops() before dropping.
2. Permanent event gaps: BatchOperation::validate strictly defers on event-id gaps (correct — events are positional; force-applying scrambles on-disk nodes). A gap that never fills (only possible if events are discarded after id assignment, i.e. a non-CDC mutation on a persisted index — see Non-CDC index mutations silently burn CDC event ids #167) stalls persistence forever: wait_for_ops() hangs, tracing::error fires repeatedly after 8 attempts. A log line is not an API-visible failure, and the error repeats each attempt.
3. Partial batch application: apply_batch_operation drains all three sub-operations and reports every failed component (data write / primary index / secondary index, each error wrapped with its component name). But the engine loop only tracing::warns and continues to the next batch from potentially inconsistent durable state. Blind retry is not safe either — structural CDC events are not idempotent.
4. VacuumPersistence::apply_move is infallible — it cannot report a closed/failed engine once such states exist.

Required design (from review)

enum PersistenceState { Running, Closing, Failed(Arc<PersistenceError>), Closed }

• Transition atomically to Failed on unrecoverable gap (attempts threshold) or batch-apply error; stop applying later operations; reject new pushes (including apply_move); wake all waiters; preserve root cause.
• wait_for_ops() returns Result (breaking change to the generated API and every test — mechanical but wide) or panics on Failed if a non-breaking variant is preferred.
• pub async fn close(self) -> Result<(), PersistenceError>: stop intake, drain per documented semantics, signal the engine loop to exit, await the JoinHandle, surface terminal errors. Drop stays as the guarded last resort.
• This also fixes the pre-existing hang class where the engine task panics (see the TOC bug history: engine death made wait_for_ops hang until tests added timeouts).

Verification hints

tests/persistence/vacuum.rs and tests/persistence/bulk_load_stall.rs show the timeout-instead-of-hang pattern. Reverting the one-line .with_persistence(..) wiring in the persist impls generator produces a genuine permanent gap on demand (stall repro). The suite is fully deterministic (40/40 clean runs at time of filing), so any flakiness introduced by lifecycle work will be visible immediately.