diff --git a/src/gc.rs b/src/gc.rs index cd434de..fdfa0f1 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -65,11 +65,10 @@ //! report data (`Ok(SweepReport)` — ADR-012 §1), nothing deleted. //! //! The `#![allow(dead_code)]` below covers the companion-seam scope: -//! the window family has its first in-tree caller in wave 3 (store -//! construction runs `ensure_gc_schema` on the SQL-backed kv engine), -//! while `window_stage`/`window_take` wait for the sweep's durable -//! form and the fleet family waits for postgres's wave. It shrinks as -//! those land. +//! the window family has its in-tree callers (store construction runs +//! `ensure_gc_schema` on the SQL-backed kv engine; the sweep's durable +//! form drives `window_stage`/`window_take`), while the fleet family +//! waits for postgres's wave. It shrinks as that lands. #![allow(dead_code)] diff --git a/src/store.rs b/src/store.rs index e26860b..bb2e3a0 100644 --- a/src/store.rs +++ b/src/store.rs @@ -204,25 +204,26 @@ //! `EngineWindowState` seam) at construction, dispatched on the //! kv-engine enum — a mem arm is structurally nothing, the in-process //! engines execute the ADR-005 window protocol inline, so there is no -//! schema to ensure — and the SQL-backed sweep form (wave 3's -//! sweep-durable-window task) stages its candidates on the kv engine -//! the constructor built. The host branch keys off the engine -//! representation — one more kv-engine-enum match arm — so wave 5's +//! schema to ensure — and the sweep's SQL-backed form (**the durable +//! form**, ADR-008 §2 staged re-arbitrated delete; ADR-012 §6.7 the +//! executor is the next sweep) stages its candidates on the kv engine +//! the constructor built. The form branch keys off the engine +//! representation ([`Store::window_host`] — one kv-engine-enum match +//! arm, never a flag): the durable form runs where the host is +//! SQL-backed, the inline form verbatim on mem/local, so wave 5's //! pg-lo dormancy (the dual-SQL instance's large-engine window clauses //! are dormant when the kv engine already hosts) slots in without //! reshaping. Both sqlite constructors carry the delete-window delay //! (ADR-008 §2: default 5 min, constructor-tunable per deployment — //! [`DEFAULT_DELETE_WINDOW_DELAY`]; `0` is a legal degenerate tuning, -//! execute-on-next-sweep) as the cadence knob the durable sweep form -//! consumes; the mem/local constructors' inline form has no delay in -//! effect. +//! execute-on-next-sweep) as the cadence knob the durable form's +//! staging consumes (`visible_after = mark_end + window`); the +//! mem/local constructors' inline form has no delay in effect. //! //! The `#![allow(dead_code)]` below marks the work still ahead, not -//! the put path (which is driven below): the GC home's fleet clauses -//! and the durable window's cadence knob (until the sweep's durable -//! form consumes it) hold the real scope. The kv engine set is no -//! longer empty in default builds — sqlite landed in wave 3 — and the -//! empty-set `allow`s came off with it. +//! the GC protocol (which is driven below — the durable window's +//! cadence knob is consumed by the durable form's staging). The fleet +//! clauses hold the remaining scope until the fleet engine's wave. #![allow(dead_code)] @@ -238,9 +239,7 @@ use crate::facade::{ Batch, Blob, EntryMeta, GcAbortCause, LivenessSource, Pin, ProtectCallback, Registration, SweepReport, }; -#[cfg(feature = "sqlite")] -use crate::gc::EngineWindowState; -use crate::gc::{Arbitration, GcHome}; +use crate::gc::{Arbitration, EngineWindowState, GcHome, WindowCandidate}; use crate::hashing::{Algorithm, BlobHasher}; use crate::keys::Key; use crate::local::{LocalBackend, LocalCursor, LocalStaging}; @@ -1372,32 +1371,53 @@ impl Store { } /// **`sweep()`** — the explicit mark-and-sweep with the delete - /// window (ADR-005 §Sweep), in its non-SQL inline form (ADR-012 §5: - /// `local`/`mem` stage candidates in-process and execute the window - /// protocol inline in the same sweep). Sweeps are never errors: + /// window (ADR-005 §Sweep), in two forms under one signature: the + /// **durable form** runs where the kv engine is SQL-backed and hosts + /// the window (ADR-012 §5's one-window-host rule — sqlite in today's + /// set; the branch keys off the engine enum, never a flag), and the + /// **inline form** (ADR-012 §5: `local`/`mem` stage candidates + /// in-process and execute the window protocol inline in the same + /// sweep) runs verbatim elsewhere. Sweeps are never errors: /// every abort is report data in `SweepReport.aborted` - /// (`Ok(SweepReport)`), and a completed sweep's counts are exact: - /// `candidates_staged == deleted + cancelled_by_arbitration` - /// (`deleted`/`cancelled` are both 0 on any abort). + /// (`Ok(SweepReport)`), and a completed sweep's counts are exact + /// relative to its form's schedule — the inline form reports + /// `candidates_staged == deleted + cancelled_by_arbitration` over its + /// same-sweep staged set; the durable form reports + /// `candidates_staged` as the candidates staged *this* sweep and + /// `deleted`/`cancelled` as the *overdue* set executed this sweep + /// (staged and executed are different sets there — the delay + /// separates them, ADR-012 §6.7; `deleted`/`cancelled` are both 0 on + /// any abort). /// /// The protocol, as run here: /// /// 1. **Single-flight**: the in-process sweep lease (ADR-013 §2's /// single-node analog); a contended second sweeper aborts with - /// [`GcAbortCause::SweeperLock`], nothing deleted. + /// [`GcAbortCause::SweeperLock`], nothing deleted, nothing + /// staged or executed. /// 2. **Zero registered liveness sources** aborts with - /// [`GcAbortCause::NoLivenessSources`] before anything is marked - /// — the safe default (nothing deleted). + /// [`GcAbortCause::NoLivenessSources`] before anything is marked — + /// the safe default (nothing deleted). /// 3. **Protect**: each registered [`ProtectCallback`] is consulted /// pre-sweep; its keys union into the live set; a callback that /// fails to answer aborts with [`GcAbortCause::ProtectFailed`] — /// a flaky protection source skips the sweep rather than risk - /// deletion (iroh's `ProtectOutcome::Abort` conclusion). - /// 4. **Mark**: the whole pool enumerated (both tiers' complete - /// `list()`, ADR-003 — a malformed pool list is fatal - /// `Err(KeyInvalid)`); the live set computed from all three - /// sources (registered sources, pins, protect callbacks); the - /// candidate-dead set staged (sorted, in-process). + /// deletion (iroh's `ProtectOutcome::Abort` conclusion). On the + /// durable form this consult's coverage extends to the executor + /// phase, so an abort here precedes any overdue execution too. + /// 4. **Durable form (SQL-backed kv engine)**, per sweep under the + /// lease: **execute overdue** first (window_take re-arbitrates + /// each overdue candidate at execution time — pin-before-publish + /// during the window delay cancels the candidate — and deletes in + /// batches), then **mark** (unchanged), then **stage** the fresh + /// candidate-dead set at `visible_after = mark_end + window` (the + /// constructor-tunable delay, ADR-008 §2). The executor of the + /// staged rows is the next sweep (ADR-008 §2; ADR-012 §6.7). + /// **Inline form**: the mark phase enumerates the whole pool + /// (both tiers' complete `list()`, ADR-003 — a malformed pool + /// list is fatal `Err(KeyInvalid)`); the live set computed from + /// all three sources (registered sources, pins, protect + /// callbacks); the candidate-dead set staged (sorted, in-process). /// 5. **Delete window**: per candidate, **arbitration at delete /// time** (not once upfront) under the arbitration lock — pinned /// or re-observed-live cancels the candidate (counted in @@ -1409,7 +1429,9 @@ impl Store { /// is never deleted by a sweep started before it committed — the /// in-flight delete markers plus the blocked pin mint are the /// mechanism ([`crate::gc`]'s module doc), and the concurrency - /// tests below assert it. + /// tests below assert it (the durable form's race holds *across the + /// window delay* — a put landing during the delay pins in-process + /// and the execution-time re-arbitration cancels the candidate). pub async fn sweep(&self) -> Result { let started = SystemTime::now(); @@ -1438,10 +1460,16 @@ impl Store { result } - /// The mark → arbitrate-at-delete-time → batched commit body, - /// under the sweep lease. Operational failures (engine I/O, a - /// malformed pool list) are the `Err` half of the return shape - /// (ADR-012 §1); every deliberate no-op is report data instead. + /// The sweep body, under the sweep lease: the protect consult + /// (whose abort shape is [`GcAbortCause::ProtectFailed`] — and whose + /// coverage extends to the durable form's execution phase), then + /// the form branch on the delete window's host — the kv-engine + /// enum itself (ADR-012 §5's one-window-host rule), never a flag: + /// the **durable form** on a SQL-backed kv engine, the **inline + /// form** verbatim on `local`/`mem`. Operational failures (engine + /// I/O, a malformed pool or window list) are the `Err` half of the + /// return shape (ADR-012 §1); every deliberate no-op is report data + /// instead. async fn sweep_window(&self, started: SystemTime) -> Result { let mut live: BTreeSet = BTreeSet::new(); @@ -1462,6 +1490,39 @@ impl Store { } } + match &self.kv { + #[cfg(feature = "mem")] + KvEngine::Mem(_) => self.sweep_window_inline(live, started).await, + #[cfg(feature = "sqlite")] + KvEngine::Sqlite(engine) => { + // The durable form (ADR-008 §2; ADR-012 §6.7): the + // overdue staged rows execute first — each re-arbitrated + // at execution time — then this sweep's own mark and + // stage at `visible_after = mark_end + window`. + let (deleted, cancelled_by_arbitration) = + self.execute_overdue_window(engine).await?; + let candidates_staged = self.mark_and_stage_window(engine, live).await?; + Ok(Self::sweep_report( + candidates_staged, + deleted, + cancelled_by_arbitration, + None, + started, + )) + } + } + } + + /// The inline form (ADR-012 §5): the mark → arbitrate-at-delete- + /// time → batched commit body, unchanged — the non-SQL engines + /// stage candidates in-process and execute the window protocol + /// inline in the same sweep. Preserved verbatim across the durable + /// form's arrival. + async fn sweep_window_inline( + &self, + mut live: BTreeSet, + started: SystemTime, + ) -> Result { let mut pool = typed_keys(self.kv().list().await?)?; if let Some(large) = self.large() { pool.extend(typed_keys(large.list().await?)?); @@ -1503,6 +1564,78 @@ impl Store { )) } + /// **The durable form's executor step** (ADR-012 §6.7): take the + /// overdue staged candidates (`visible_after <= now`) off the + /// window and execute each through the same + /// [`Store::arbitrated_delete`] path — the re-arbitration happens + /// at execution time, so a put landing (pin-before-publish) during + /// the window delay cancels its candidate here — in ~100-key + /// batches. The consult posture on this path is the sweep's protect + /// consult (recorded named design point): it already ran for this + /// sweep, above, and its coverage extends to the executor phase — + /// so these arbitrations carry `consult_protect = false`, the same + /// posture as the inline form's delete-time arbitration. A vanished + /// candidate (deleted during the delay by a direct delete) settles + /// into the cancelled count. + async fn execute_overdue_window( + &self, + host: &W, + ) -> Result<(u64, u64), Error> { + let overdue: Vec = typed_keys(host.window_take(SystemTime::now()).await?)?; + let mut deleted = 0u64; + let mut cancelled_by_arbitration = 0u64; + for chunk in overdue.chunks(SWEEP_DELETE_BATCH) { + for key in chunk { + match self.arbitrated_delete(key, false).await? { + crate::gc::DeleteOutcome::Deleted(true) => deleted += 1, + _ => cancelled_by_arbitration += 1, + } + } + } + Ok((deleted, cancelled_by_arbitration)) + } + + /// **The durable form's mark + stage step**: the mark phase exactly + /// the inline form's (pool enumerated, live set from pins/registered + /// sources plus the protect consult's keys, candidate-dead set + /// computed), then the candidate-dead set staged into the window at + /// `visible_after = mark_end + window` — the constructor-tunable + /// delay (ADR-008 §2; the store carries the value its constructor + /// took), so execution waits at earliest for the next sweep past + /// the deadline. Returns the staged-candidate count. + async fn mark_and_stage_window( + &self, + host: &W, + mut live: BTreeSet, + ) -> Result { + let mut pool = typed_keys(self.kv().list().await?)?; + if let Some(large) = self.large() { + pool.extend(typed_keys(large.list().await?)?); + } + live.extend(self.gc.pinned_keys()); + for source in self.gc.registered_sources() { + live.extend(source.live_keys()); + } + + let enumerated: BTreeSet = pool.into_iter().collect(); + let mut candidates: Vec = enumerated.difference(&live).cloned().collect(); + candidates.sort(); + + let mark_end = SystemTime::now(); + let visible_after = mark_end.checked_add(self.window_delay).ok_or_else(|| { + Error::Io("the delete-window delay overflows the system clock".to_owned()) + })?; + let staged: Vec = candidates + .iter() + .map(|key| WindowCandidate { + key: key.as_bytes().to_vec(), + visible_after, + }) + .collect(); + host.window_stage(&staged).await?; + Ok(candidates.len() as u64) + } + /// A finished report constructor — every sweep outcome is a /// completed record (ADR-012 §1: the report is data, not an error /// carrier). @@ -4011,3 +4144,501 @@ mod sqlite_construction_tests { drop(pin); } } + +/// The sweep's durable form (feature `sqlite`): the staged +/// re-arbitrated delete (ADR-008 §2; ADR-012 §5/§6.7) — the first +/// durable delete window, single-node — beside the inline form. The +/// two-sweep exact-count shapes, the race across the delay, the +/// vanished-candidate form, the abort shapes, and the delay knob's +/// staging semantics. +#[cfg(all(test, feature = "sqlite"))] +mod sqlite_window_tests { + use super::*; + use crate::backend::Backend; + use std::fs; + use std::path::{Path, PathBuf}; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + + /// A fresh temp site per test (the construction tests' helper + /// shape); Drop removes the file tree and the WAL sidecars. + struct TempSite { + database: PathBuf, + root: PathBuf, + } + + impl TempSite { + fn path(&self) -> &Path { + &self.root + } + + fn drop_sidecar(&self, suffix: &str) { + let base = self.database.to_string_lossy().into_owned(); + let _ = fs::remove_file(format!("{base}{suffix}")); + } + } + + impl Drop for TempSite { + fn drop(&mut self) { + self.drop_sidecar("-wal"); + self.drop_sidecar("-shm"); + let _ = fs::remove_file(&self.database); + let _ = fs::remove_dir_all(&self.root); + } + } + + fn temp_site(tag: &str) -> TempSite { + static SEQ: AtomicU64 = AtomicU64::new(0); + let seq = SEQ.fetch_add(1, Ordering::Relaxed); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let base = std::env::temp_dir().join(format!( + "alkblobs-store-window-{tag}-{}-{nanos:x}-{seq}", + std::process::id() + )); + fs::create_dir_all(&base).expect("temp site created"); + TempSite { + database: base.join("pool.db"), + root: base.join("local-root"), + } + } + + fn key_for(content: &[u8]) -> Key { + Key::derive(content, content.len() as u64) + } + + fn sqlite_engine(store: &Store) -> &SqliteBackend { + match store.kv() { + #[cfg(feature = "mem")] + KvEngine::Mem(_) => panic!("these tests construct sqlite stores"), + #[cfg(feature = "sqlite")] + KvEngine::Sqlite(engine) => engine, + } + } + + async fn plant_kv(store: &Store, key: &Key, value: &[u8]) { + sqlite_engine(store) + .put(&key.as_bytes(), value) + .await + .expect("kv plant"); + } + + async fn plant_large(store: &Store, key: &Key, value: &[u8]) { + match store.large() { + Some(LargeEngine::Local(engine)) => engine + .put(&key.as_bytes(), value) + .await + .expect("large plant"), + None => panic!("planting into a mode with no large tier"), + } + } + + /// A set-backed liveness source — the live-shared seam, shared + /// with the wave-2 tests' fixture shape. + #[derive(Default)] + struct SetSource { + live: parking_lot::Mutex>, + } + + impl SetSource { + fn add(&self, key: Key) { + self.live.lock().insert(key); + } + } + + impl LivenessSource for SetSource { + fn live_keys(&self) -> Vec { + self.live.lock().iter().cloned().collect() + } + + fn contains(&self, key: &Key) -> bool { + self.live.lock().contains(key) + } + } + + /// A set-backed protect callback; `None` models the + /// failed-to-answer case. + #[derive(Default)] + struct SetProtect { + set: parking_lot::Mutex>>, + } + + impl SetProtect { + fn set_keys(&self, keys: Vec) { + *self.set.lock() = Some(keys); + } + + fn fail(&self) { + *self.set.lock() = None; + } + } + + impl ProtectCallback for SetProtect { + fn protect(&self) -> Option> { + self.set.lock().clone() + } + } + + /// A protect callback that blocks mid-consult — the contention + /// window for the single-flight assert on a sqlite store. + struct LatchProtect { + entered: std::sync::mpsc::Sender<()>, + release: parking_lot::Mutex>>, + } + + impl ProtectCallback for LatchProtect { + fn protect(&self) -> Option> { + self.entered.send(()).ok(); + let receiver = self.release.lock().take(); + if let Some(receiver) = receiver { + let _ = receiver.recv(); + } + Some(Vec::new()) + } + } + + /// A non-destructive read of the window is not on the seam — but a + /// far-horizon `window_take` drains it entirely, which is exactly + /// the observation the assertions below need (and leaves the + /// window clean for the next step of the test). + async fn window_drain(store: &Store) -> Vec { + let far = SystemTime::now() + .checked_add(Duration::from_secs(365 * 24 * 3600)) + .unwrap(); + let raws = sqlite_engine(store).window_take(far).await.unwrap(); + typed_keys(raws).expect("the window holds typed keys") + } + + /// The two-sweep exact-count shape at the degenerate delay (the + /// test tuning, ADR-008 §2): sweep 1 stages N — nothing deleted, + /// the pool intact; sweep 2 executes exactly N — + /// `deleted + cancelled == N` against the staged set, staged-this- + /// sweep and executed-this-sweep different sets, and the window + /// left empty. + #[tokio::test] + async fn durable_two_sweep_exact_counts_at_delay_zero() { + let site = temp_site("two-sweep"); + let store = Store::dual_tier_sqlite_local(&site.database, site.path(), 64, Duration::ZERO) + .await + .expect("dual-tier sqlite+local constructs"); + let source = Arc::new(SetSource::default()); + store.register_liveness_source(source.clone()); + + let live = key_for(b"kept by the registered source"); + plant_kv(&store, &live, b"kept by the registered source").await; + source.add(live.clone()); + + let dead_kv = key_for(b"dead kv entry in the window"); + plant_kv(&store, &dead_kv, b"dead kv entry in the window").await; + let dead_large = key_for(b"dead large entry in the window"); + plant_large(&store, &dead_large, b"dead large entry in the window").await; + + let first = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!( + first.candidates_staged, 2, + "both tiers' dead entries staged" + ); + assert_eq!( + first.deleted, 0, + "staging is not executing — the delay separates" + ); + assert_eq!(first.cancelled_by_arbitration, 0); + + assert!( + store.has(&dead_kv.as_bytes()).await.unwrap(), + "sweep 1 staged, did not execute: the kv entry is intact" + ); + assert!( + store.has(&dead_large.as_bytes()).await.unwrap(), + "sweep 1 staged, did not execute: the large entry is intact" + ); + assert!(store.has(&live.as_bytes()).await.unwrap()); + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, None); + assert_eq!(second.candidates_staged, 0, "nothing newly dead"); + assert_eq!( + second.deleted + second.cancelled_by_arbitration, + 2, + "sweep 2 executed exactly the staged set" + ); + assert_eq!(second.deleted, 2); + assert_eq!(second.cancelled_by_arbitration, 0); + + assert!(!store.has(&dead_kv.as_bytes()).await.unwrap()); + assert!(!store.has(&dead_large.as_bytes()).await.unwrap()); + assert!(store.has(&live.as_bytes()).await.unwrap()); + assert!( + window_drain(&store).await.is_empty(), + "the executed rows left the window" + ); + } + + /// The delay knob is the staging's cadence (ADR-008 §2): at a + /// future-facing delay the staged rows are not overdue at + /// `window_take(now)` — no execution, no deletion — and a repeated + /// sweep re-stages the still-dead set without duplicating rows. + #[tokio::test] + async fn durable_stage_at_a_future_delay_waits_past_the_window() { + let site = temp_site("future-delay"); + let store = Store::kv_only_sqlite(&site.database, 64, Duration::from_secs(600)) + .await + .expect("kv-only sqlite constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let dead_a = key_for(b"dead under a future-facing delay a"); + plant_kv(&store, &dead_a, b"dead under a future-facing delay a").await; + let dead_b = key_for(b"dead under a future-facing delay b"); + plant_kv(&store, &dead_b, b"dead under a future-facing delay b").await; + + let first = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!(first.candidates_staged, 2); + assert_eq!(first.deleted, 0, "nothing is overdue at staging time"); + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.candidates_staged, 2, "still dead, re-staged"); + assert_eq!( + second.deleted, 0, + "the window delay pushed execution out again" + ); + + let drained = window_drain(&store).await; + let mut expected = vec![dead_a.clone(), dead_b.clone()]; + expected.sort(); + assert_eq!(drained, expected); + assert!(store.has(&dead_a.as_bytes()).await.unwrap()); + assert!(store.has(&dead_b.as_bytes()).await.unwrap()); + } + + /// The race across the delay (the window form of ADR-005's + /// test-asserted property): a put landing during the window (the + /// pin minted between the staging sweep and the executor sweep, + /// pin-before-publish) survives — the execution-time re-arbitration + /// cancels its candidate — while the unpinned twin deletes. The + /// cancelled candidate re-stages (and executes) only once its + /// protection drops. + #[tokio::test] + async fn durable_race_across_the_delay_put_during_window_survives() { + let site = temp_site("race-across-delay"); + let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) + .await + .expect("kv-only sqlite constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let pinned = key_for(b"the twin a put lands under during the window"); + plant_kv( + &store, + &pinned, + b"the twin a put lands under during the window", + ) + .await; + let unpinned = key_for(b"the unpinned twin"); + plant_kv(&store, &unpinned, b"the unpinned twin").await; + + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(staged.candidates_staged, 2); + assert_eq!(staged.deleted, 0); + + { + let _pin = store.gc().acquire(&pinned).await; + assert_eq!(store.gc().pin_count(&pinned), 1, "pin-before-publish"); + + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(executed.aborted, None); + assert_eq!( + executed.deleted + executed.cancelled_by_arbitration, + 2, + "execution-time re-arbitration ran over the taken set" + ); + assert_eq!(executed.deleted, 1, "the unpinned twin deleted"); + assert_eq!( + executed.cancelled_by_arbitration, 1, + "the pinned candidate cancelled" + ); + assert!( + store.has(&pinned.as_bytes()).await.unwrap(), + "a put landing during the window survives the next sweep's execution" + ); + assert!(!store.has(&unpinned.as_bytes()).await.unwrap()); + } + + // The pin dropped: the still-present candidate re-stages, and + // then (and only then) executes — its window row was consumed + // by the cancellation, not by a delete. + let restaged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + restaged.candidates_staged, 1, + "unpinned, the candidate re-stages" + ); + assert_eq!(restaged.deleted, 0); + + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(executed.deleted, 1, "the re-staged candidate executes"); + assert!(!store.has(&pinned.as_bytes()).await.unwrap()); + } + + /// The vanished-candidate form: a staged candidate deleted during + /// the delay by a direct delete settles into + /// `cancelled_by_arbitration` at execution time — the vanished- + /// candidate accounting carries over (nothing to delete, counts + /// stay exact). + #[tokio::test] + async fn durable_vanished_candidate_settles_into_cancelled() { + let site = temp_site("vanished"); + let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) + .await + .expect("kv-only sqlite constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + + let victim = key_for(b"deleted during the delay"); + plant_kv(&store, &victim, b"deleted during the delay").await; + + let staged = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(staged.candidates_staged, 1); + + store + .delete(&victim.as_bytes()) + .await + .expect("the direct delete lands on the unprotected entry"); + + let executed = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(executed.aborted, None); + assert_eq!(executed.candidates_staged, 0); + assert_eq!(executed.deleted, 0); + assert_eq!( + executed.cancelled_by_arbitration, 1, + "the vanished candidate settles into the cancelled count" + ); + assert!(window_drain(&store).await.is_empty()); + } + + /// The abort shapes on a sqlite store delete nothing, execute + /// nothing overdue, and stage nothing: zero sources + /// (`NoLivenessSources`), a failing protect callback + /// (`ProtectFailed`), and lease contention (`SweeperLock`). + #[tokio::test] + async fn durable_aborts_delete_nothing_and_execute_nothing() { + let site = temp_site("aborts-sources"); + let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) + .await + .expect("kv-only sqlite constructs"); + let dead = key_for(b"dead while no sources are registered"); + plant_kv(&store, &dead, b"dead while no sources are registered").await; + + let aborted = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(aborted.aborted, Some(GcAbortCause::NoLivenessSources)); + assert_eq!(aborted.candidates_staged, 0); + assert_eq!(aborted.deleted, 0); + assert!(store.has(&dead.as_bytes()).await.unwrap()); + assert!(window_drain(&store).await.is_empty(), "nothing staged"); + } + + /// `ProtectFailed` aborts before the executor step on the durable + /// form: an overdue row staged by an earlier sweep is not taken + /// while the protection consult fails — abort precedes execution. + #[tokio::test] + async fn durable_protect_failed_aborts_before_executing_the_overdue_set() { + let site = temp_site("aborts-protect"); + let store = Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) + .await + .expect("kv-only sqlite constructs"); + store.register_liveness_source(Arc::new(SetSource::default())); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + + let staged_dead = key_for(b"staged dead while the consult answers"); + plant_kv( + &store, + &staged_dead, + b"staged dead while the consult answers", + ) + .await; + protect.set_keys(Vec::new()); + let first = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(first.aborted, None); + assert_eq!(first.candidates_staged, 1); + + protect.fail(); + let later_dead = key_for(b"dead while the consult fails"); + plant_kv(&store, &later_dead, b"dead while the consult fails").await; + + let aborted = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(aborted.aborted, Some(GcAbortCause::ProtectFailed)); + assert_eq!(aborted.candidates_staged, 0); + assert_eq!(aborted.deleted, 0); + assert_eq!(aborted.cancelled_by_arbitration, 0); + assert!( + store.has(&staged_dead.as_bytes()).await.unwrap(), + "the overdue candidate was not executed past the abort" + ); + assert!( + store.has(&later_dead.as_bytes()).await.unwrap(), + "and nothing was deleted" + ); + assert_eq!( + window_drain(&store).await, + vec![staged_dead], + "the overdue row still waits in the window" + ); + } + + /// Contention on the single-node lease aborts the second sweeper + /// as report data on the durable form too — nothing executed, + /// nothing staged, nothing deleted (`Ok(SweepReport)`, ADR-012 §1). + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn durable_lease_contention_aborts_the_second_sweeper_as_data() { + let site = temp_site("aborts-lease"); + let store = Arc::new( + Store::kv_only_sqlite(&site.database, 64, Duration::ZERO) + .await + .expect("kv-only sqlite constructs"), + ); + store.register_liveness_source(Arc::new(SetSource::default())); + + let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + store.register_protect_callback(Arc::new(LatchProtect { + entered: entered_tx, + release: parking_lot::Mutex::new(Some(release_rx)), + })); + + let dead = key_for(b"dead under the contended sweeps"); + plant_kv(&store, &dead, b"dead under the contended sweeps").await; + + let first = { + let store = Arc::clone(&store); + tokio::spawn(async move { store.sweep().await }) + }; + + while entered_rx + .recv_timeout(std::time::Duration::from_millis(500)) + .is_err() + { + tokio::task::yield_now().await; + } + + let second = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(second.aborted, Some(GcAbortCause::SweeperLock)); + assert_eq!(second.deleted, 0); + assert_eq!(second.candidates_staged, 0, "nothing staged"); + assert!(store.has(&dead.as_bytes()).await.unwrap()); + assert!(window_drain(&store).await.is_empty(), "nothing executed"); + + release_tx.send(()).ok(); + let first = first + .await + .expect("first sweep task joined") + .expect("first sweep returns a report"); + assert_eq!(first.aborted, None); + assert_eq!(first.candidates_staged, 1, "the lease holder stages"); + assert_eq!(first.deleted, 0, "and executes nothing at staging time"); + assert!( + store.has(&dead.as_bytes()).await.unwrap(), + "the lease holder staged, did not execute" + ); + } +} diff --git a/tasks/store-sweep-durable-window.md b/tasks/store-sweep-durable-window.md index dc3cc73..06e296d 100644 --- a/tasks/store-sweep-durable-window.md +++ b/tasks/store-sweep-durable-window.md @@ -1,7 +1,7 @@ --- id: store-sweep-durable-window name: the sweep's durable-window form — staged candidates, next-sweep execution -status: pending +status: completed depends_on: [store-sqlite-construction] scope: moderate risk: medium @@ -94,6 +94,71 @@ window, single-node: > To be filled by implementation agent +The durable form's shape as it landed: + +- **The branch is the kv-engine enum match itself** (`sweep_window`'s + `match &self.kv`): `KvEngine::Sqlite(engine)` runs the durable form, + `KvEngine::Mem(_)` runs the inline form. The two durable helpers are + generic over `W: EngineWindowState` (the seam has `Backend` + associated types, so a `&dyn` handout is not object-safe — the + concrete `&SqliteBackend` passes directly). Wave 5's pg-lo slots one + more match arm in. +- **The inline body was extracted verbatim** into + `sweep_window_inline` (the wave-2/all-features mem+local suites run + green unchanged); the protect consult stays in `sweep_window` above + the branch, shared by both forms. +- **The protect-consult posture on overdue execution is recorded as a + doc point** on `execute_overdue_window`: the consult already ran for + the sweep (its abort, `ProtectFailed`, precedes execution) and its + coverage extends to the executor phase, so the overdue + re-arbitrations carry `consult_protect = false` — the inline form's + same posture. +- **Malformed window rows are fatal** (`Error::KeyInvalid` via the + shared `typed_keys` boundary) — the pool-list discipline extended to + the window read. +- **`visible_after` overflow is `Err(Error::Io)`** (checked clock add); + no panics on the library path. +- **Report shape**: `candidates_staged` = staged this sweep; + `deleted`/`cancelled_by_arbitration` = the overdue set executed this + sweep (a vanished candidate settles into the cancelled count). The + `sweep()` doc states the exact-count invariants relative to each + form's schedule. + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +The sweep's durable SQL-backed form landed beside the inline form in +`src/store.rs` (the first durable delete window, single-node; ADR-008 +§2 staged re-arbitrated delete, ADR-012 §5/§6.7 the-executor-is-the- +next-sweep): + +- `sweep_window` branches on the delete window's host — the kv-engine + enum (`KvEngine::Sqlite` → durable; `KvEngine::Mem` → inline, + extracted verbatim). Under the lease: the protect consult (abort + `ProtectFailed` precedes any execution), then for the durable form + **execute overdue** — `window_take(now)` → execution-time + re-arbitration through `arbitrated_delete(key, false)` in + ~100-key batches (vanishes count as cancelled) — then **mark** + (unchanged), then **stage** at `visible_after = mark_end + window` + (the constructor-tunable delay; `Duration::ZERO` = execute-on-next- + sweep). Report counts exact relative to the schedule. +- Abort shapes unchanged and unit-tested on sqlite: + `NoLivenessSources`/`ProtectFailed`/`SweeperLock` abort before + anything is executed or staged; the `ProtectFailed` test proves the + overdue set is left waiting in the window (abort precedes + `window_take`). +- Unit tests (`sqlite_window_tests`): two-sweep exact counts at delay 0 + (sweep 1 stages both tiers' dead entries without executing; sweep 2 + executes exactly the staged set), future-delay staging (no execution + before the window closes; re-stage is idempotent), the race across + the delay (a put landing during the window survives via the + execution-time re-arbitration; the unpinned twin deletes; the + cancelled candidate re-stages and executes once unpinned), the + vanished-candidate form, and the three abort shapes. + +Verification: `cargo test` (69 lib + 5 integration, default), 124 +tests `--all-features`, 101 tests `--no-default-features --features +mem`, `cargo clippy --all-targets -- -D warnings` (default and +`--all-features`), and `cargo fmt --check` all green; the mem/local +wave-2 suites pass unchanged (the inline form untouched). \ No newline at end of file