diff --git a/Cargo.toml b/Cargo.toml index b8395f5..94fe681 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,4 +36,7 @@ hex = "0.4" rusqlite = { version = "0.40", features = ["bundled"], optional = true } sha1 = "0.10" sha2 = "0.10" -tokio-postgres = { version = "0.7", optional = true } \ No newline at end of file +tokio-postgres = { version = "0.7", optional = true } + +[dev-dependencies] +tokio = { version = "1", default-features = false, features = ["rt-multi-thread"] } diff --git a/src/facade.rs b/src/facade.rs index e56ab7d..a1d5889 100644 --- a/src/facade.rs +++ b/src/facade.rs @@ -1,19 +1,37 @@ //! The store facade's value types (store-api.md §Public surface): the -//! shapes put/get/stat/sweep return and the pin lifecycle vocabulary. -//! Engine-blind — nothing here knows which engines a store runs (ADR-007/ -//! 008/010); these types are the contract wave 2's facade behavior -//! implements and every consumer programs against. +//! shapes put/get/stat/sweep return, the pin lifecycle vocabulary, and +//! the registered-liveness vocabulary. Engine-blind — nothing here +//! knows which engines a store runs (ADR-007/008/010); these types are +//! the contract the facade behavior implements and every consumer +//! programs against. //! -//! The pin types ([`Pin`], [`Batch`]) declare the ADR-005 lifecycle -//! contract; the arbitration machinery behind drops and conversions is -//! wave 2's facade logic. The contracts are pinned here so the guard -//! semantics never change shape — wave 2 wires the behavior, not the -//! surface. +//! **The pin types ([`Pin`], [`Batch`]) carry their registry +//! back-reference.** Each guard holds an `Arc` to the store instance's +//! GC registry core ([`crate::gc`]), so a guard's release reaches the +//! per-key refcounts under the arbitration lock wherever the guard +//! goes — the wave-1 shape carried only the key. The guard's drop +//! decrements its refcount; a [`Batch`] holds its members' keys and +//! releases every one of them in **one** ordered step under the same +//! lock on commit or drop (no early single-pin release). The guard +//! types never expose the registry consumer-side: the Arc is private, +//! minting happens on the store core only. +//! +//! **The registered-liveness vocabulary is consumer-owned shared +//! state, never a snapshot.** [`LivenessSource`] and +//! [`ProtectCallback`] are the two consumer-side halves of ADR-005's +//! three liveness sources (the pins are the third); both are +//! registered through the facade and are consulted *live* — at mark +//! time and at delete-time arbitration — never copied at registration +//! (POC #1 finding 7's install-implies-copy failure is the pinning +//! reason). A [`Registration`] is the deregistration token the +//! correction flows of ADR-012 §6.4 need. use std::fmt; +use std::sync::Arc; use std::time::SystemTime; use crate::backend::ReadCursor; +use crate::gc::RegistryCore; use crate::hashing::Algorithm; use crate::keys::Key; use crate::store::AnyCursor; @@ -102,18 +120,27 @@ pub struct EntryMeta { /// The RAII put-path pin guard (ADR-005 §Marks 2; store-api.md §Write /// path). /// -/// **Lifecycle contract (pinned here; the machinery is wave 2):** +/// **Lifecycle contract (ADR-005; the machinery is wired):** /// /// - Acquired **before the entry becomes visible** — pin-before-publish: /// an entry cannot appear in a sweep's `list()` without already /// carrying its pin (or its converted registration, ADR-005's put-side /// invariant). +/// - **Back-reference shape (named design point):** the guard holds an +/// `Arc` to the store's GC registry core (`RegistryCore`, +/// crate-internal) alongside +/// its key, so the drop reaches the per-key refcounts under the +/// arbitration lock the sweep consults — wherever the guard travels. +/// The registry core outlives the `Store` instance when guards still +/// reference it; a late release against a detached registry is +/// harmless (the refcount bookkeeping is self-contained). /// - The guard holds a per-key refcount under the same arbitration lock /// the sweep consults, until **dropped or converted into a registered /// consumer reference** (a registered liveness source or a root tag). -/// - **Conversion ordering (ADR-005, pinned):** the pin's drop and its -/// replacement's registration into a liveness source are one ordered -/// step under the same arbitration lock — a consumer may not release a +/// Conversion goes through the registry's ordered-step primitive +/// (`RegistryCore::convert`, crate-internal): the pin's release and +/// the replacement's registration into a liveness source are one +/// ordered step under the same lock — a consumer may not release a /// pin before its registered source observes the replacement (the /// local analog of the ops-layer pin-token contract). /// - A dedup put's guard protects exactly as a fresh put's does: a @@ -125,30 +152,45 @@ pub struct EntryMeta { /// non-fleet modes the ops layer's token is this guard's in-process /// handle — nothing expires outside fleet mode. pub struct Pin { + registry: Arc, key: Key, + /// False once the guard has been contributed to a [`Batch`] (the + /// release duty moved to the batch's one-step drop) or consumed by + /// the ordered-step conversion. + armed: bool, } impl Pin { - /// Store-core constructor: pins are minted by the put path, never - /// held by consumers before a put exists — the guard's meaning is - /// the put's protection (wave 2 wires the refcount acquisition under - /// the arbitration lock). - #[allow(dead_code)] // wave 2's put path wires the call sites; exercised by the tests - pub(crate) fn new(key: Key) -> Pin { - Pin { key } - } - /// The key whose entry the guard protects. pub fn key(&self) -> &Key { &self.key } + + /// Store-core constructor: pins are minted by the GC registry + /// ([`crate::gc`]) when the refcount is acquired under the + /// arbitration lock — never held by consumers before a put exists. + pub(crate) fn minted(registry: Arc, key: Key) -> Pin { + Pin { + registry, + key, + armed: true, + } + } +} + +impl Drop for Pin { + fn drop(&mut self) { + if self.armed { + self.registry.release_pin(&self.key); + } + } } /// Batch-scoped pin group (ADR-005 §Marks 2; store-api.md): multi-put /// writes — manifest writes are exactly this — group under one /// batch-scoped pin. /// -/// **Lifecycle contract (pinned here; the machinery is wave 2):** +/// **Lifecycle contract (ADR-005; the machinery is wired):** /// /// - Created by the store core's batching put seam; each put within the /// batch contributes its pin (batch-internal puts publish @@ -156,36 +198,60 @@ impl Pin { /// - **A batch's pins drop together on batch commit or batch drop**, in /// one ordered step under the same arbitration lock the sweep /// consults. No early single-pin release: protection spans the whole -/// batch or none of it. +/// batch or none of it. The batch holds its members' keys; the +/// release takes the arbitration lock once and decrements every +/// member refcount inside that one section. /// - Batch drop is the failure path (stream error, caller abandonment): /// every member pin releases together; member liveness never outlives /// the batch. pub struct Batch { - pins: Vec, + registry: Arc, + keys: Vec, } impl Batch { - /// Store-core constructor (wave 2's batching seam wires the - /// arbitration-lock acquisition). - #[allow(dead_code)] // wave 2's batching path wires the call sites; exercised by the tests - pub(crate) fn new() -> Batch { - Batch { pins: Vec::new() } + /// Store-core constructor: the batching put seam creates batches + /// through the GC registry ([`crate::gc`]). + pub(crate) fn created(registry: Arc) -> Batch { + Batch { + registry, + keys: Vec::new(), + } } - /// Contribute a put's pin to the group (store-core call). - #[allow(dead_code)] // wave 2's batching path drives this; exercised by the tests + /// Contribute a put's pin to the group (store-core call). The + /// contributed guard is defused — its own drop releases nothing; + /// the batch's one-step release owns the protection from here. + #[allow(dead_code)] // the batching put seam wires the call sites (store-put-paths); exercised by the tests pub(crate) fn add(&mut self, pin: Pin) { - self.pins.push(pin); + let mut pin = pin; + pin.armed = false; + self.keys.push(pin.key().clone()); } /// How many pins the group currently protects. pub fn len(&self) -> usize { - self.pins.len() + self.keys.len() } /// Whether the group currently protects nothing. pub fn is_empty(&self) -> bool { - self.pins.is_empty() + self.keys.is_empty() + } + + /// The one-step release: every member refcount decremented inside + /// a single arbitration-lock section, sorted into the keys' total + /// order. Driven by [`Drop::drop`] (batch drop) and available for + /// the batching put seam's commit step (which consumes the batch + /// the same way). + pub(crate) fn release_members(&mut self) { + self.registry.release_pins(std::mem::take(&mut self.keys)); + } +} + +impl Drop for Batch { + fn drop(&mut self) { + self.release_members(); } } @@ -248,6 +314,86 @@ pub struct SweepReport { pub finished_at: SystemTime, } +/// A registered liveness source — the live-shared GC seam (ADR-005 +/// §Marks 1; store-api.md §Lifecycle, `register_liveness_source`). +/// +/// **Named design point — the source shape: two sync verbs, live-shared +/// consults, none blocking.** The source is consumer-owned state shared +/// by handle (`Arc`); the store consults the +/// *current* contents every time and never snapshots at registration — +/// POC #1 finding 7's install-implies-copy failure (roots registered +/// after install were invisible to GC) is the pinning reason. The two +/// verbs split by protocol phase: +/// +/// - [`live_keys`](Self::live_keys) — mark-time enumeration, consulted +/// outside the arbitration lock (whole live-set computation); +/// - [`contains`](Self::contains) — delete-time arbitration probe, +/// consulted **under the in-process arbitration lock** (the pin/ +/// liveness lock), so it must answer from an indexed structure and +/// never block or await; the provided default derives it from +/// [`live_keys`](Self::live_keys) (correct but linear — indexed +/// sources override). +/// +/// The verbs are sync (non-async) by the same lock discipline: pin +/// drops are sync RAII, and arbitration consults run under a +/// short-held `parking_lot` lock (store-api.md §Concurrency posture; +/// AGENTS convention 3). A source whose set changes concurrently with +/// a sweep is the consumer's coordination duty — the delete-time +/// re-observation composes the sources' live-shared answers, and the +/// ADR-005 Consequences assign pin/liveness misuse to the consumer +/// contract (store-api.md's invariants document it). +pub trait LivenessSource: Send + Sync + 'static { + /// The keys currently live per this source's own reference state — + /// consulted at mark time, every sweep, never a stale snapshot. + fn live_keys(&self) -> Vec; + + /// Whether the key is live right now — consulted at delete-time + /// arbitration (sweep candidates' cancel check and the direct + /// delete's full protection set). Runs under the arbitration lock: + /// quick, indexed, never blocking. Default: linear over + /// [`live_keys`](Self::live_keys). + fn contains(&self, key: &Key) -> bool { + self.live_keys().contains(key) + } +} + +/// A registered protect callback — ADR-005 §Marks 3's protection +/// source, consulted pre-sweep and at direct-delete arbitration. +/// +/// **Named design point — the callback shape: one sync verb returning +/// the protection set or a failure to answer.** `None` means the +/// protection source could not answer, and every consumer of that +/// answer takes the safe side (iroh's `ProtectOutcome::Abort` +/// conclusion, adopted): a sweep aborts with +/// [`GcAbortCause::ProtectFailed`], nothing deleted (ADR-012 §1: +/// abort is report data); a direct delete refuses conservatively with +/// `Error::GcRefuse` — nothing deleted either way. The verb is sync by +/// the same discipline as [`LivenessSource`] (direct-delete +/// arbitration runs it under the arbitration lock — quick, never +/// blocking); mark-time consults run outside the lock and may be +/// slower. +pub trait ProtectCallback: Send + Sync + 'static { + /// The externally-known-live keys, or `None` when the protection + /// source failed to answer (abort/refuse semantics above). + fn protect(&self) -> Option>; +} + +/// A registration token — the opaque handle the store's +/// `unregister_liveness_source`/`unregister_protect_callback` verbs +/// consume to remove a +/// registered [`LivenessSource`] or [`ProtectCallback`]. ADR-012 §6.4's +/// embedder correction flow ("deregisters/re-registers around the +/// delete") is exactly this: the token makes deregistration expressible +/// without exposing the registry internals. +/// +/// Copyable, but each token deregisters at most once (the second call +/// returns `false`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct Registration { + pub(crate) id: u64, + pub(crate) protect: bool, +} + #[cfg(test)] mod tests { use super::*; @@ -322,21 +468,39 @@ mod tests { ); } - /// Pin/Batch shape invariants: the guard holds its key and the batch - /// groups pins. Wave 2 wires the refcount machinery; the shape these - /// tests pin is what the guard carries and what drops together. - #[test] - fn pin_guard_holds_its_key_and_batch_groups_pins() { - let pin = Pin::new(Key::from(Digest::derive(b"guarded", 7))); + /// Pin/Batch shape invariants: the guard holds its key and the + /// batch groups pins, released in one step on batch drop — the + /// refcounts land under the registry's arbitration lock (GcHome + /// drives the minting; the drop behavior is observable through the + /// refcount probe). + #[tokio::test] + async fn pin_guard_holds_its_key_and_batch_groups_pins() { + let gc = crate::gc::GcHome::default(); + let key = Key::from(Digest::derive(b"guarded", 7)); + let pin = gc.acquire(&key).await; assert_eq!(pin.key().algorithm(), Algorithm::GitBlobSha256); + assert_eq!(gc.pin_count(&key), 1, "acquire minted the refcount"); drop(pin); + assert_eq!(gc.pin_count(&key), 0, "drop released the refcount"); - let mut batch = Batch::new(); + let a = Key::from(Digest::derive(b"a", 1)); + let b = Key::from(Digest::derive(b"b", 1)); + let mut batch = gc.new_batch(); assert!(batch.is_empty(), "fresh batch protects nothing yet"); - batch.add(Pin::new(Key::from(Digest::derive(b"a", 1)))); - batch.add(Pin::new(Key::from(Digest::derive(b"b", 1)))); + batch.add(gc.acquire(&a).await); + batch.add(gc.acquire(&b).await); assert_eq!(batch.len(), 2, "member pins counted"); + assert_eq!( + gc.pin_count(&a) + gc.pin_count(&b), + 2, + "member refcounts held while the batch lives" + ); drop(batch); + assert_eq!( + gc.pin_count(&a) + gc.pin_count(&b), + 0, + "batch drop released every member in one step" + ); } /// The arbitration vocabulary: two states, and the engine's reading diff --git a/src/gc.rs b/src/gc.rs index 8ffe0d2..2a8b88a 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -24,30 +24,487 @@ //! learn liveness (ADR-005). Every clause is contract-tested per its //! applicability row in waves 3–5; `local`/`mem` are contract-excluded. //! -//! The `#![allow(dead_code)]` below marks the seam's scope: its clauses are -//! implemented by the SQL engines of waves 3–5 and are not called yet. It -//! shrinks as those waves land. +//! **The in-process machinery (this module's `GcHome`).** The whole +//! in-process GC state sits behind **one arbitration lock** — a +//! short-held [`parking_lot::Mutex`] over the per-key pin refcounts, +//! the in-flight delete markers, and the registered liveness/protect +//! source lists (ADR-005's "the arbitration lock the sweep consults"; +//! store-api.md §Concurrency posture; AGENTS convention 3). parking_lot +//! does not poison, and these sections hold no `RwLock` other lock +//! kind, so there is no `unwrap_or_else(|e| e.into_inner())` recovery +//! posture to apply (the mem engine's module doc records the same +//! posture). Consumer consults ([`LivenessSource::contains`]/ +//! [`ProtectCallback::protect`], the trait shapes on +//! [`crate::facade`]) run under that lock at delete-time arbitration +//! and outside it at mark time — both documented as quick, non-blocking +//! duties on the traits. Lock order is one-directional everywhere: +//! arbitration lock → consumer locks (the ordered-step conversion +//! primitive is the only nesting, same direction). +//! +//! **Named design points (recorded for the review gate):** +//! +//! - **The sweep-vs-put race is closed with in-flight delete markers,** +//! not by holding the arbitration lock across engine I/O. A sweep's +//! (or direct delete's) arbitration decides *and* inserts the +//! key's delete marker inside one lock section; the engine delete +//! runs outside the lock; the marker removal notifies any put-path +//! pin mint blocked on the key. A re-put whose pin mint lands while +//! the marker is held blocks until the deletion commits, then lands +//! as a fresh pinned put — ADR-005's case (a) delete-then-recover +//! verbatim; a pin mint that lands before the arbitration is case +//! (b), its protection composed at the arbitration itself. Holding +//! the parking_lot lock across the engine delete was rejected: the +//! guard is not `Send`, the sweep future must be, and AGENTS +//! convention 3 forbids blocking I/O under internal locks. +//! - **The single-node sweep is single-flight** — an in-process +//! test-and-set flag is the single-node analog of ADR-013 §2's named +//! exclusive sweep lease (ADR-012 §5's "single-flighter is the +//! in-process lock" posture): one sweeper per arbitration domain, +//! atomic test-and-set, holdership ends at release; a contended +//! second sweeper aborts with [`GcAbortCause::SweeperLock`][crate::GcAbortCause::SweeperLock] as +//! report data (`Ok(SweepReport)` — ADR-012 §1), nothing deleted. +//! +//! The `#![allow(dead_code)]` below covers the companion-seam scope: +//! the engine-state clauses are implemented by the SQL engines of +//! waves 3–5 and are not called yet. It shrinks as those waves land. #![allow(dead_code)] +use std::collections::{BTreeMap, BTreeSet}; +use std::pin::pin; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; use std::time::SystemTime; +use parking_lot::Mutex; +use tokio::sync::Notify; + use crate::backend::Backend; -use crate::facade::PinState; +use crate::facade::{Batch, LivenessSource, Pin, PinState, ProtectCallback, Registration}; +use crate::keys::Key; use crate::Error; /// The store-instance GC state home — held as a field of the public /// facade type ([`Store`](crate::store::Store)). ADR-011 §1's vocabulary /// and ADR-010 §2's ownership rule together: the store instance owns /// all GC state, so in-process GC state carries on the facade -/// instance itself. The wave-2 gc-protocol task fills it — the pin -/// registry (per-key refcounts under the arbitration lock), registered -/// liveness sources, the protect callback, and the in-process delete -/// window (ADR-005's protocol inline, ADR-012 §5's non-SQL staging) — -/// and drives sweep/direct-delete through the facade. Fleet state never -/// lands here: it rides the engine-state companion seam below -/// (ADR-010 §2), hosted on the fleet's kv engine (ADR-012 §4). -pub(crate) struct GcHome; +/// instance itself — the pin registry (per-key refcounts under the +/// arbitration lock), registered liveness sources, the protect +/// callbacks, and the in-process delete window (ADR-005's protocol +/// inline, ADR-012 §5's non-SQL staging), shared with the public pin +/// guards ([`Pin`]/[`Batch`]) through the [`RegistryCore`] `Arc`. +/// Fleet state never lands here: it rides the engine-state companion +/// seam below (ADR-010 §2), hosted on the fleet's kv engine (ADR-012 §4). +pub(crate) struct GcHome { + core: Arc, +} + +impl Default for GcHome { + fn default() -> GcHome { + GcHome { + core: Arc::new(RegistryCore::new()), + } + } +} + +impl GcHome { + /// Mint a pin: acquire the key's refcount under the arbitration + /// lock, blocking (async) while a delete of the key is in flight — + /// ADR-005's re-put case (a): the mint waits until the deletion + /// commits, then the guard protects the fresh put. The put path + /// acquires this **before** publishing (pin-before-publish). + pub(crate) async fn acquire(&self, key: &Key) -> Pin { + self.core.acquire(key).await + } + + /// A fresh batch-scoped pin group (ADR-005 §Marks 2): member pins + /// release together in one ordered step on commit or drop. + pub(crate) fn new_batch(&self) -> Batch { + Batch::created(Arc::clone(&self.core)) + } + + /// The **conversion ordering primitive** (ADR-005, pinned): the + /// pin's release and the replacement closure's registration into a + /// liveness source are one ordered step under the same arbitration + /// lock — a consumer may not release a pin before its registered + /// source observes the replacement. Consumes the pin (its own drop + /// is suppressed; the release happens inside the section, before + /// the closure runs). + pub(crate) fn convert(&self, pin: Pin, register: F) { + self.core.convert(pin, register) + } + + /// The registered liveness source count — the sweep's + /// [`GcAbortCause::NoLivenessSources`] safe default keys off zero. + pub(crate) fn registered_source_count(&self) -> usize { + self.core.registered_source_count() + } + + /// Register a [`LivenessSource`] — the live-shared seam; the token + /// deregisters it. + pub(crate) fn register_source(&self, source: Arc) -> Registration { + self.core.register_source(source) + } + + /// Deregister a previously registered liveness source; `false` when + /// the token is already consumed or was never a source token. + pub(crate) fn unregister_source(&self, registration: Registration) -> bool { + self.core.unregister_source(registration) + } + + /// Register a [`ProtectCallback`]; the token deregisters it. + pub(crate) fn register_protect(&self, callback: Arc) -> Registration { + self.core.register_protect(callback) + } + + /// Deregister a previously registered protect callback; `false` + /// when the token is already consumed or was never a protect token. + pub(crate) fn unregister_protect(&self, registration: Registration) -> bool { + self.core.unregister_protect(registration) + } + + /// Snapshots of the registered liveness sources — cloned under the + /// arbitration lock, consulted outside it (mark-time enumeration + /// may be slow; the delete-time probe path re-reads the live list + /// under the lock instead). + pub(crate) fn registered_sources(&self) -> Vec> { + self.core.registered_sources() + } + + /// Snapshot of the registered protect callbacks — consulted + /// pre-sweep, outside the arbitration lock. + pub(crate) fn registered_protects(&self) -> Vec> { + self.core.registered_protects() + } + + /// The keys currently carrying pins (refcount > 0) — the pin half + /// of the mark-phase live set. + pub(crate) fn pinned_keys(&self) -> Vec { + self.core.pinned_keys() + } + + /// The key's current pin refcount (a testing/put-path observation + /// aid; zero when unpinned). + pub(crate) fn pin_count(&self, key: &Key) -> u64 { + self.core.pin_count(key) + } + + /// The single-node sweep lease: `true` = acquired (this caller + /// sweeps); `false` = contended — the caller aborts with + /// [`GcAbortCause::SweeperLock`] report data. + pub(crate) fn try_begin_sweep(&self) -> bool { + self.core.try_begin_sweep() + } + + /// Release the single-node sweep lease (ADR-013 §2's holdership + /// rule, in-process: holdership ends at release). + pub(crate) fn end_sweep(&self) { + self.core.end_sweep() + } + + /// **Delete-time arbitration** (ADR-005 §Sweep step 2). Decides the + /// key's fate in one arbitration-lock section — pins, in-flight + /// markers (wait and retry), and — per the `consult_protect` flag — + /// the registered liveness/protect sources' delete-time probes: + /// + /// - [`Arbitration::Cancelled`] — protection composed over the + /// candidate (pinned, re-observed live, protect-named/failed); + /// nothing deleted, no marker left. + /// - [`Arbitration::Proceed`] — the key's delete marker is now + /// held; the caller **must** run [`Self::conclude_delete`] on + /// every path after its engine delete (success, failure, or + /// absent) — the marker-release notifies blocked pin mints. + pub(crate) async fn arbitrate( + &self, + key: &Key, + consult_protect: bool, + ) -> Result { + self.core.arbitrate(key, consult_protect).await + } + + /// Release the delete marker taken by a [`Arbitration::Proceed`] + /// arbitration and notify blocked pin mints. Mandatory on every + /// post-arbitration path. + pub(crate) fn conclude_delete(&self, key: &Key) { + self.core.conclude_delete(key) + } +} + +/// The arbitration vocabulary between a delete and a pin mint: the +/// delete is cancelled by protection, or it proceeds holding the key's +/// in-flight delete marker. +pub(crate) enum Arbitration { + /// Protection composed over the key at delete time (ADR-005's + /// "pinned or re-observed live, cancel the candidate"). + Cancelled, + /// The delete marker is held; run the engine delete, then + /// [`GcHome::conclude_delete`]. + Proceed, +} + +/// The outcome of a store-side arbitrated delete: cancelled by +/// protection (nothing deleted), or executed — with whether the entry +/// was still present when the window reached it. +pub(crate) enum DeleteOutcome { + Cancelled, + Deleted(bool), +} + +/// The GC registry core — the shared state behind [`GcHome`], the +/// public pin guards' back-reference ([`Pin`] holds this `Arc`), and +/// **the arbitration lock's owner**. One [`parking_lot::Mutex`] guards +/// the per-key pin refcounts, the in-flight delete markers, and the +/// registered source lists together, so a delete-time decision (pins ∨ +/// markers ∨ sources) and its marker insert are one atomic section +/// against the put path's pin mints. +pub(crate) struct RegistryCore { + /// The arbitration lock's state (see the module doc). + state: Mutex, + /// Wakeup for pin mints blocked on an in-flight delete marker. + delete_done: Notify, + /// The single-node sweep lease (the in-process analog of ADR-013 + /// §2's named exclusive sweep lease). + sweeping: AtomicBool, + /// Registration id allocation. + next_registration: AtomicU64, +} + +struct RegistryState { + /// Per-key pin refcounts (ADR-005 §Marks 2); zero-count keys are + /// removed, so the map is exactly the pinned-key set at mark time. + pins: BTreeMap, + /// In-flight delete markers: keys between an arbitration's Proceed + /// decision and its marker release. Pin mints block on these + /// (ADR-005's re-put case (a)). + deleting: BTreeSet, + /// Registered liveness sources with their registration ids. + sources: Vec<(u64, Arc)>, + /// Registered protect callbacks with their registration ids. + protects: Vec<(u64, Arc)>, +} + +impl RegistryCore { + fn new() -> RegistryCore { + RegistryCore { + state: Mutex::new(RegistryState { + pins: BTreeMap::new(), + deleting: BTreeSet::new(), + sources: Vec::new(), + protects: Vec::new(), + }), + delete_done: Notify::new(), + sweeping: AtomicBool::new(false), + next_registration: AtomicU64::new(1), + } + } + + fn allocate_registration(&self) -> u64 { + self.next_registration.fetch_add(1, Ordering::Relaxed) + } + + pub(crate) fn register_source(&self, source: Arc) -> Registration { + let id = self.allocate_registration(); + self.state.lock().sources.push((id, source)); + Registration { id, protect: false } + } + + pub(crate) fn unregister_source(&self, registration: Registration) -> bool { + self.unregister(®istration, false) + } + + pub(crate) fn register_protect(&self, callback: Arc) -> Registration { + let id = self.allocate_registration(); + self.state.lock().protects.push((id, callback)); + Registration { id, protect: true } + } + + pub(crate) fn unregister_protect(&self, registration: Registration) -> bool { + self.unregister(®istration, true) + } + + fn unregister(&self, registration: &Registration, protect: bool) -> bool { + let mut state = self.state.lock(); + if protect { + let before = state.protects.len(); + state.protects.retain(|(id, _)| *id != registration.id); + state.protects.len() != before + } else { + let before = state.sources.len(); + state.sources.retain(|(id, _)| *id != registration.id); + state.sources.len() != before + } + } + + pub(crate) fn registered_source_count(&self) -> usize { + self.state.lock().sources.len() + } + + fn registered_sources(&self) -> Vec> { + self.state + .lock() + .sources + .iter() + .map(|(_, s)| Arc::clone(s)) + .collect() + } + + fn registered_protects(&self) -> Vec> { + self.state + .lock() + .protects + .iter() + .map(|(_, c)| Arc::clone(c)) + .collect() + } + + pub(crate) fn pinned_keys(&self) -> Vec { + self.state.lock().pins.keys().cloned().collect() + } + + /// The key's current pin refcount (a testing/put-path observation + /// aid; zero when unpinned). + pub(crate) fn pin_count(&self, key: &Key) -> u64 { + self.state.lock().pins.get(key).copied().unwrap_or(0) + } + + pub(crate) async fn acquire(self: &Arc, key: &Key) -> Pin { + loop { + let notified = self.delete_done.notified(); + let mut notified = pin!(notified); + notified.as_mut().enable(); + { + let mut state = self.state.lock(); + if !state.deleting.contains(key) { + *state.pins.entry(key.clone()).or_insert(0) += 1; + return Pin::minted(Arc::clone(self), key.clone()); + } + } + notified.as_mut().await; + } + } + + pub(crate) fn release_pin(&self, key: &Key) { + let mut state = self.state.lock(); + Self::release_in(&mut state, key); + } + + /// The batch's one-step release: every member refcount decremented + /// inside one arbitration-lock section, in the keys' total order + /// (ADR-005: no early single-pin release). + pub(crate) fn release_pins(&self, mut keys: Vec) { + keys.sort(); + let mut state = self.state.lock(); + for key in &keys { + Self::release_in(&mut state, key); + } + } + + fn release_in(state: &mut RegistryState, key: &Key) { + if let Some(count) = state.pins.get_mut(key) { + *count -= 1; + if *count == 0 { + state.pins.remove(key); + } + } + } + + /// The ordered-step conversion (ADR-005's pinned conversion + /// ordering): within one arbitration-lock section, release the + /// pin's refcount and run the registration closure. A concurrent + /// arbitration therefore never observes neither-pin-nor-registered. + /// Consumes the pin with its own drop suppressed (the release ran + /// inside the section). + pub(crate) fn convert(&self, pin: Pin, register: F) { + let pin = std::mem::ManuallyDrop::new(pin); + let key = pin.key().clone(); + let mut state = self.state.lock(); + Self::release_in(&mut state, &key); + register(); + } + + pub(crate) fn try_begin_sweep(&self) -> bool { + self.sweeping + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + } + + pub(crate) fn end_sweep(&self) { + self.sweeping.store(false, Ordering::Release); + } + + pub(crate) async fn arbitrate( + &self, + key: &Key, + consult_protect: bool, + ) -> Result { + loop { + let notified = self.delete_done.notified(); + let mut notified = pin!(notified); + notified.as_mut().enable(); + + enum Plan { + Cancel, + Wait, + Proceed, + } + + let plan = { + let mut state = self.state.lock(); + if state.pins.contains_key(key) { + Plan::Cancel + } else if state.deleting.contains(key) { + Plan::Wait + } else { + let mut protected = false; + if consult_protect { + for (_, callback) in &state.protects { + match callback.protect() { + Some(keys) if keys.iter().any(|live| live == key) => { + protected = true; + break; + } + None => { + // A failed protection source refuses + // conservatively: nothing deleted. + protected = true; + break; + } + Some(_) => {} + } + } + } + if !protected { + for (_, source) in &state.sources { + if source.contains(key) { + protected = true; + break; + } + } + } + if protected { + Plan::Cancel + } else { + state.deleting.insert(key.clone()); + Plan::Proceed + } + } + }; + + match plan { + Plan::Cancel => return Ok(Arbitration::Cancelled), + Plan::Proceed => return Ok(Arbitration::Proceed), + Plan::Wait => notified.as_mut().await, + } + } + } + + pub(crate) fn conclude_delete(&self, key: &Key) { + let removed = self.state.lock().deleting.remove(key); + if removed { + self.delete_done.notify_waiters(); + } + } +} /// A delete-window candidate staged into the durable window: the key and /// the time it becomes visible for deletion — exactly the diff --git a/src/lib.rs b/src/lib.rs index fb1f01a..92eb10e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,7 +10,10 @@ pub mod store; pub use dispatch::DEFAULT_DISPATCH_THRESHOLD; pub use error::Error; -pub use facade::{Batch, EntryMeta, GcAbortCause, Pin, PinState, SweepReport}; +pub use facade::{ + Batch, EntryMeta, GcAbortCause, LivenessSource, Pin, PinState, ProtectCallback, Registration, + SweepReport, +}; pub use hashing::{Algorithm, Digest}; pub use keys::Key; pub use store::{Store, StoreMode}; diff --git a/src/store.rs b/src/store.rs index abfebd4..2362724 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,12 +1,14 @@ //! The `Store` facade type (store-api.md §What this is): the public //! surface consumers program against — constructor modes, the -//! size-threshold dispatch seam, and the read-path ops (`get`/`stat`/ +//! size-threshold dispatch seam, the read-path ops (`get`/`stat`/ //! `read_range`/`has`/`list` with the fall-through general rule, -//! implemented below). The remaining ops land against this type in -//! `store-put-paths` (put, the kv-only rejection behavior) and -//! `gc-protocol` (sweep, direct delete, the pin/liveness machinery). -//! Direct `delete` is *not* a read path — it lands with the GC -//! protocol's arbitration. This module owns the engine representation +//! implemented below), and the GC protocol's ops (`sweep`, direct +//! `delete`, the liveness registration — arbitration-then-delete per +//! ADR-005/012, implemented below). The remaining op: the put seam +//! (`put`, the kv-only rejection behavior, the batching seam) lands +//! in `store-put-paths` against the pin registry minted here. +//! Direct `delete` is *not* a read path — it participates in the +//! delete-window arbitration. This module owns the engine representation //! and the constructor-mode machinery; the pure routing function lives //! in [`crate::dispatch`]. //! @@ -142,10 +144,16 @@ #![allow(dead_code)] +use std::collections::BTreeSet; +use std::sync::Arc; +use std::time::SystemTime; + use crate::backend::{Backend, ReadCursor}; use crate::dispatch::{route, Tier}; -use crate::facade::{Blob, EntryMeta}; -use crate::gc::GcHome; +use crate::facade::{ + Blob, EntryMeta, GcAbortCause, LivenessSource, ProtectCallback, Registration, SweepReport, +}; +use crate::gc::{Arbitration, GcHome}; use crate::keys::Key; use crate::local::{LocalBackend, LocalCursor}; @@ -276,6 +284,14 @@ impl KvEngine { } } + #[allow(unused_variables)] // same empty-set scope as `get` + async fn delete(&self, key: &[u8]) -> Result<(), Error> { + match *self { + #[cfg(feature = "mem")] + KvEngine::Mem(ref engine) => engine.delete(key).await, + } + } + async fn list(&self) -> Result>, Error> { match *self { #[cfg(feature = "mem")] @@ -303,6 +319,12 @@ impl LargeEngine { } } + async fn delete(&self, key: &[u8]) -> Result<(), Error> { + match self { + LargeEngine::Local(engine) => engine.delete(key).await, + } + } + async fn list(&self) -> Result>, Error> { match self { LargeEngine::Local(engine) => engine.list().await, @@ -338,7 +360,7 @@ impl Store { large: None, threshold: crate::dispatch::DEFAULT_DISPATCH_THRESHOLD, mode: MemOnly, - gc: GcHome, + gc: GcHome::default(), } } @@ -355,7 +377,7 @@ impl Store { large: None, threshold, mode: KvOnly, - gc: GcHome, + gc: GcHome::default(), } } @@ -381,7 +403,7 @@ impl Store { large: Some(LargeEngine::Local(LocalBackend::new(root).await?)), threshold, mode: DualTier, - gc: GcHome, + gc: GcHome::default(), }) } @@ -576,6 +598,312 @@ impl Store { } } +/// **The delete-window commit batch size** — ~100/batch, iroh's proven +/// shape (ADR-005 §Sweep; POC #1 finding 3's named shape adopted). +const SWEEP_DELETE_BATCH: usize = 100; + +/// The GC protocol's facade surface (ADR-005; gc-and-namespaces.md; +/// store-api.md §Lifecycle, §Invariants 4/5): the liveness +/// registration verbs, the direct `delete` with its arbitration-then- +/// delete refusal, and the `sweep()` delete-window protocol. This is +/// the architecture's most intricate state machine in its single-node +/// form — the fleet extension (DB-backed pins, sweep lease, SQL +/// arbitration) extends it without changing its invariants. +/// +/// **The state machine, one instance at a time.** All shared GC state +/// rides the crate-internal `GcHome` behind the crate-internal +/// `Store::gc` accessor: one arbitration lock +/// over per-key pin refcounts, in-flight delete markers, and the +/// registered source lists ([`crate::gc`]'s module doc carries the +/// named design points — the marker mechanism that closes the +/// sweep-vs-put race without engine I/O under the lock, and the +/// single-node sweep lease). +/// +/// **Abort-as-data** (ADR-012 §1): every sweep returns +/// `Ok(SweepReport)`; aborts are `GcAbortCause` values in the report, +/// never errors. `Error::GcRefuse` is the direct-delete refusal alone, +/// decided here at the facade — it never fires below the facade (the +/// error module's altitude discipline). The full protection set +/// (ADR-012 §6.4): pinned, re-observed live via a registered source, +/// or named by a protect callback. +impl Store { + /// Register a [`LivenessSource`] — the live-shared GC seam (ADR-005 + /// §Marks 1; `register_liveness_source` verbatim; POC #1 finding + /// 7's finding is the verb's pinning reason). The source is + /// consulted **live** at mark time and at delete-time arbitration — + /// never a snapshot at registration. The returned + /// [`Registration`] token deregisters via + /// [`Store::unregister_liveness_source`] (ADR-012 §6.4's correction + /// flow). Zero registered sources make every sweep abort with + /// [`GcAbortCause::NoLivenessSources`] — the safe default: a + /// consumer without registered liveness sources cannot have its + /// content deleted. + pub fn register_liveness_source(&self, source: Arc) -> Registration { + self.gc.register_source(source) + } + + /// Deregister a liveness source previously registered on this + /// store. `false` when the token was already consumed or never + /// came from [`Store::register_liveness_source`]. + pub fn unregister_liveness_source(&self, registration: Registration) -> bool { + self.gc.unregister_source(registration) + } + + /// Register a [`ProtectCallback`] — ADR-005 §Marks 3's protection + /// source. Consulted pre-sweep (it may add externally-known hashes + /// or fail the sweep into [`GcAbortCause::ProtectFailed`] report + /// data) and at direct-delete arbitration, where a + /// protect-callback-named key is refused (`Error::GcRefuse`, ADR-012 + /// §6.4's full protection set). Registration replaces nothing and + /// composes: each callback's set unions into the protection. + pub fn register_protect_callback(&self, callback: Arc) -> Registration { + self.gc.register_protect(callback) + } + + /// Deregister a protect callback previously registered on this + /// store. `false` when the token was already consumed or never + /// came from [`Store::register_protect_callback`]. + pub fn unregister_protect_callback(&self, registration: Registration) -> bool { + self.gc.unregister_protect(registration) + } + + /// **Direct `delete(key)`** — fall-through consult (kv first, then + /// large; ADR-012 §6.4's "arbitration-then-delete where the entry + /// lives"), then the full protection set arbitration at delete + /// time: a key that is pinned, re-observed live via a registered + /// source, or named by a protect callback is refused with + /// [`Error::GcRefuse`] — nothing deleted. Unprotected keys delete + /// (across both homes when a degenerate pool holds the key twice); + /// absent keys carry [`Error::Missing`]; malformed key bytes + /// [`Error::KeyInvalid`]. A failed protect callback is refused + /// conservatively (`GcRefuse`, nothing deleted) — the direct-path + /// reading of abort-as-data's safe side, recorded in the facade + /// module doc. + /// + /// Outside-of-sweep deletion is unusual by posture — most deletion + /// flows through sweeps (ADR-005); an embedder's correction flow + /// deregisters/re-registers around the delete (ADR-012 §6.4). + pub async fn delete(&self, key: &[u8]) -> Result<(), Error> { + let key = boundary_key(key)?; + let bytes = key.as_bytes(); + + let present = if self.kv().has(&bytes).await? { + true + } else { + match self.large() { + Some(large) => large.has(&bytes).await?, + None => false, + } + }; + if !present { + return Err(Error::Missing); + } + + match self.arbitrated_delete(&key, true).await? { + crate::gc::DeleteOutcome::Cancelled => Err(Error::GcRefuse), + crate::gc::DeleteOutcome::Deleted(true) => Ok(()), + // The entry vanished between the existence probe and the + // window: some other path resolved it after the probe. + crate::gc::DeleteOutcome::Deleted(false) => Err(Error::Missing), + } + } + + /// **`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: + /// 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). + /// + /// 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. + /// 2. **Zero registered liveness sources** aborts with + /// [`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). + /// 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 + /// `cancelled_by_arbitration`); otherwise the candidate deletes + /// (counted in `deleted`), in ~100-key batches. + /// + /// **The test-asserted property** (ADR-005): a visible pool entry is + /// never deleted while liveness protects it, and an in-flight put + /// 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. + pub async fn sweep(&self) -> Result { + let started = SystemTime::now(); + + if self.gc.registered_source_count() == 0 { + return Ok(Self::sweep_report( + 0, + 0, + 0, + Some(GcAbortCause::NoLivenessSources), + started, + )); + } + + if !self.gc.try_begin_sweep() { + return Ok(Self::sweep_report( + 0, + 0, + 0, + Some(GcAbortCause::SweeperLock), + started, + )); + } + + let result = self.sweep_window(started).await; + self.gc.end_sweep(); + 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. + async fn sweep_window(&self, started: SystemTime) -> Result { + let mut live: BTreeSet = BTreeSet::new(); + + for callback in self.gc.registered_protects() { + match callback.protect() { + Some(keys) => { + live.extend(keys); + } + None => { + return Ok(Self::sweep_report( + 0, + 0, + 0, + Some(GcAbortCause::ProtectFailed), + started, + )); + } + } + } + + 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 candidates: Vec = { + let enumerated: BTreeSet = pool.into_iter().collect(); + let mut candidates: Vec = enumerated.difference(&live).cloned().collect(); + candidates.sort(); + candidates + }; + let candidates_staged = candidates.len() as u64; + + let mut deleted = 0u64; + let mut cancelled_by_arbitration = 0u64; + for chunk in candidates.chunks(SWEEP_DELETE_BATCH) { + for key in chunk { + match self.arbitrated_delete(key, false).await? { + // A candidate already resolved elsewhere (deleted + // between mark and its window turn) settles here + // too: the completed sweep's counts stay exact over + // the staged set. + crate::gc::DeleteOutcome::Deleted(true) => deleted += 1, + _ => cancelled_by_arbitration += 1, + } + } + } + + Ok(Self::sweep_report( + candidates_staged, + deleted, + cancelled_by_arbitration, + None, + started, + )) + } + + /// A finished report constructor — every sweep outcome is a + /// completed record (ADR-012 §1: the report is data, not an error + /// carrier). + #[allow(clippy::too_many_arguments)] + fn sweep_report( + candidates_staged: u64, + deleted: u64, + cancelled_by_arbitration: u64, + aborted: Option, + started: SystemTime, + ) -> SweepReport { + SweepReport { + candidates_staged, + deleted, + cancelled_by_arbitration, + aborted, + started_at: started, + finished_at: SystemTime::now(), + } + } + + /// Arbitrate, then delete where the entry lives, with the + /// in-flight delete marker held across the delete and released on + /// every path (the store-side composition of [`GcHome::arbitrate`] + /// + [`GcHome::conclude_delete`]). + async fn arbitrated_delete( + &self, + key: &Key, + consult_protect: bool, + ) -> Result { + match self.gc.arbitrate(key, consult_protect).await? { + Arbitration::Cancelled => Ok(crate::gc::DeleteOutcome::Cancelled), + Arbitration::Proceed => { + let result = self.delete_everywhere(key).await; + self.gc.conclude_delete(key); + Ok(crate::gc::DeleteOutcome::Deleted(result?)) + } + } + } + + /// Delete the key's pool entry wherever it physically lives — the + /// kv home and the large home (normal pools hold exactly one; + /// immutability means both homes of a degenerate duplicate are the + /// same dead content). `false` when the entry was already absent: + /// per-home `Missing` is not an operational error here, it is the + /// presence answer. + async fn delete_everywhere(&self, key: &Key) -> Result { + let bytes = key.as_bytes(); + let mut deleted = match self.kv().delete(&bytes).await { + Ok(()) => true, + Err(Error::Missing) => false, + Err(err) => return Err(err), + }; + if let Some(large) = self.large() { + match large.delete(&bytes).await { + Ok(()) => deleted = true, + Err(Error::Missing) => {} + Err(err) => return Err(err), + } + } + Ok(deleted) + } +} + #[cfg(all(test, feature = "mem"))] mod tests { use super::*; @@ -1144,4 +1472,722 @@ mod tests { ); } } + + // ------------------------------------------------------------------ + // The GC protocol (the gc-protocol work): pin registry, registered + // liveness sources, protect callback, delete windows, direct-delete + // refusal — ADR-005/012/013 on the facade. + // ------------------------------------------------------------------ + + /// A set-backed liveness source — shared-handle semantics, the POC + /// finding-7 shape: the store consults the same set this test + /// mutates, so its answers are live-shared, never snapshots. + #[derive(Default)] + struct SetSource { + live: parking_lot::Mutex>, + } + + impl SetSource { + fn add(&self, key: Key) { + self.live.lock().insert(key); + } + + fn remove(&self, key: &Key) { + self.live.lock().remove(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 (abort/refuse safe side). + #[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 sweep lease's + /// contended-window fixture): it announces its entry on a channel, + /// waits on a second, then answers with an empty protection set. + 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()) + } + } + + /// `sweep()` with zero registered liveness sources is the safe + /// default: `Ok(SweepReport)` carrying `NoLivenessSources`, nothing + /// deleted, the pool intact (store-api.md §Lifecycle). + #[tokio::test] + async fn sweep_with_zero_sources_aborts_as_data_and_deletes_nothing() { + let store = Store::mem_only(); + let key = key_for(b"unprotected content"); + plant_kv(&store, &key.as_bytes(), b"unprotected content").await; + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, Some(GcAbortCause::NoLivenessSources)); + assert_eq!(report.candidates_staged, 0); + assert_eq!(report.deleted, 0); + assert_eq!(report.cancelled_by_arbitration, 0); + assert!( + store.has(&key.as_bytes()).await.unwrap(), + "an aborted sweep deletes nothing" + ); + } + + /// With sources registered, an empty pool sweeps to a zero report + /// — no abort, exact zero counts. + #[tokio::test] + async fn sweep_with_sources_and_an_empty_pool_is_a_zero_report() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!(report.candidates_staged, 0); + assert_eq!(report.deleted, 0); + assert_eq!(report.cancelled_by_arbitration, 0); + } + + /// Exact counts on a mixed pool (store-api.md invariant 4): the + /// registered source's key and a pinned key mark live, both tiers' + /// dead entries stage and delete — `candidates_staged == deleted + + /// cancelled_by_arbitration`. + #[tokio::test] + async fn sweep_counts_are_exact_across_both_tiers() { + let root = temp_root("gc-exact-counts"); + let store = Store::dual_tier_mem_local(root.path(), 1024) + .await + .expect("dual-tier 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.as_bytes(), b"kept by the registered source").await; + source.add(live.clone()); + + let pinned = key_for(b"kept by the pin"); + plant_kv(&store, &pinned.as_bytes(), b"kept by the pin").await; + let pin = store.gc().acquire(&pinned).await; + + let kv_dead = key_for(b"dead kv entry"); + plant_kv(&store, &kv_dead.as_bytes(), b"dead kv entry").await; + let large_dead = key_for(b"dead large entry with bulk"); + plant_large( + &store, + &large_dead.as_bytes(), + b"dead large entry with bulk", + ) + .await; + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!(report.candidates_staged, 2); + assert_eq!(report.deleted, 2, "both tiers' dead entries deleted"); + assert_eq!(report.cancelled_by_arbitration, 0); + assert_eq!( + report.candidates_staged, + report.deleted + report.cancelled_by_arbitration, + "the completed sweep's counts are exact over the staged set" + ); + assert!(store.has(&live.as_bytes()).await.unwrap()); + assert!(store.has(&pinned.as_bytes()).await.unwrap()); + assert!(!store.has(&kv_dead.as_bytes()).await.unwrap()); + assert!(!store.has(&large_dead.as_bytes()).await.unwrap()); + + drop(pin); + } + + /// A dedup re-put's guard refcounts the existing entry: two pins + /// compose, each drop decrements, and only the last drop exposes + /// the entry to a sweep (ADR-005's dedup-refcount ruling). + #[tokio::test] + async fn dedup_reput_guards_refcount_under_the_arbitration_lock() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + let content = b"refcounted content"; + let key = key_for(content); + plant_kv(&store, &key.as_bytes(), content).await; + + let first = store.gc().acquire(&key).await; + let second = store.gc().acquire(&key).await; + assert_eq!(store.gc().pin_count(&key), 2, "dedup re-put refcounts"); + drop(first); + assert_eq!(store.gc().pin_count(&key), 1); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 0, "one refcount still protects"); + assert!(store.has(&key.as_bytes()).await.unwrap()); + + drop(second); + assert_eq!(store.gc().pin_count(&key), 0, "zero removes the entry row"); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 1); + assert_eq!(report.deleted, 1); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + /// Batch pins protect the whole group; batch drop releases every + /// member in one ordered step — no early single-pin release + /// (ADR-005 §Marks 2). + #[tokio::test] + async fn batch_pins_drop_together_in_one_ordered_step() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + let a = key_for(b"batch member a"); + let b = key_for(b"batch member b"); + plant_kv(&store, &a.as_bytes(), b"batch member a").await; + plant_kv(&store, &b.as_bytes(), b"batch member b").await; + + let mut batch = store.gc().new_batch(); + batch.add(store.gc().acquire(&a).await); + batch.add(store.gc().acquire(&b).await); + assert_eq!(batch.len(), 2); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 0, "the batch protects both"); + assert!(store.has(&a.as_bytes()).await.unwrap()); + assert!(store.has(&b.as_bytes()).await.unwrap()); + + drop(batch); + assert_eq!( + store.gc().pin_count(&a) + store.gc().pin_count(&b), + 0, + "batch drop released every member refcount" + ); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.candidates_staged, 2); + assert_eq!(report.deleted, 2); + } + + /// The protect callback adds externally-known hashes (they stay + /// alive) and aborts as report data on failure — nothing deleted, + /// `ProtectFailed` in `Ok(SweepReport)` (ADR-012 §1). + #[tokio::test] + async fn protect_callback_adds_keys_and_aborts_as_data() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + + let named = key_for(b"protected by the callback"); + plant_kv(&store, &named.as_bytes(), b"protected by the callback").await; + let dead = key_for(b"dead while the callback answers"); + plant_kv(&store, &dead.as_bytes(), b"dead while the callback answers").await; + protect.set_keys(vec![named.clone()]); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, None); + assert_eq!(report.candidates_staged, 1, "the named key never staged"); + assert_eq!(report.deleted, 1); + assert!(store.has(&named.as_bytes()).await.unwrap()); + assert!(!store.has(&dead.as_bytes()).await.unwrap()); + + protect.fail(); + let another_dead = key_for(b"dead while the callback fails"); + plant_kv( + &store, + &another_dead.as_bytes(), + b"dead while the callback fails", + ) + .await; + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, Some(GcAbortCause::ProtectFailed)); + assert_eq!( + report.deleted, 0, + "a failed protection source deletes nothing" + ); + assert!(store.has(&another_dead.as_bytes()).await.unwrap()); + assert!(store.has(&named.as_bytes()).await.unwrap()); + } + + /// The live-shared seam: registration never snapshots — content + /// added to the set after registering is seen by the next sweep; + /// content removed exposes the entry again (POC #1 finding 7). + #[tokio::test] + async fn liveness_sources_are_consulted_live_not_snapshotted() { + let store = Store::mem_only(); + let source = Arc::new(SetSource::default()); + store.register_liveness_source(source.clone()); + + let key = key_for(b"live-shared content"); + plant_kv(&store, &key.as_bytes(), b"live-shared content").await; + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "unprotected at mark, deleted"); + + plant_kv(&store, &key.as_bytes(), b"live-shared content").await; + source.add(key.clone()); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the registration after registration is consulted live" + ); + assert!(store.has(&key.as_bytes()).await.unwrap()); + + source.remove(&key); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "deregistration re-exposes the entry"); + assert!(!store.has(&key.as_bytes()).await.unwrap()); + } + + /// Unregistration consumes the token (returning `false` the second + /// time) and the corrected zero-source posture aborts sweeps again + /// — ADR-012 §6.4's correction-flow prerequisite. + #[tokio::test] + async fn unregistration_consumes_the_token_and_restores_the_safe_default() { + let store = Store::mem_only(); + let registration = store.register_liveness_source(Arc::new(SetSource::default())); + + assert!( + store.unregister_liveness_source(registration), + "the token deregisters once" + ); + assert!( + !store.unregister_liveness_source(registration), + "the token does not deregister twice" + ); + + let protect_token = store.register_protect_callback(Arc::new(SetProtect::default())); + assert!( + !store.unregister_liveness_source(protect_token), + "a protect token is not a source token" + ); + assert!(store.unregister_protect_callback(protect_token)); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.aborted, Some(GcAbortCause::NoLivenessSources)); + } + + /// Direct `delete` refuses the full protection set with the typed + /// `Error::GcRefuse` — pinned, registered-source-live, + /// protect-callback-named (ADR-012 §6.4) — and deletes unprotected + /// keys through the fall-through dispatch; absent keys carry + /// `Err(Missing)`; malformed bytes `Err(KeyInvalid)`. + #[tokio::test] + async fn direct_delete_refuses_the_full_protection_set() { + let root = temp_root("gc-direct-delete"); + let store = Store::dual_tier_mem_local(root.path(), 1024) + .await + .expect("dual-tier constructs"); + let source = Arc::new(SetSource::default()); + store.register_liveness_source(source.clone()); + let protect = Arc::new(SetProtect::default()); + store.register_protect_callback(protect.clone()); + + let pinned = key_for(b"pinned refusal"); + plant_kv(&store, &pinned.as_bytes(), b"pinned refusal").await; + let _pin = store.gc().acquire(&pinned).await; + + let source_live = key_for(b"source-live refusal"); + plant_large(&store, &source_live.as_bytes(), b"source-live refusal").await; + source.add(source_live.clone()); + + let protect_named = key_for(b"protect-named refusal"); + plant_kv(&store, &protect_named.as_bytes(), b"protect-named refusal").await; + protect.set_keys(vec![protect_named.clone()]); + + let unprotected_kv = key_for(b"unprotected kv deletes"); + plant_kv( + &store, + &unprotected_kv.as_bytes(), + b"unprotected kv deletes", + ) + .await; + let unprotected_large = key_for(b"unprotected large deletes"); + plant_large( + &store, + &unprotected_large.as_bytes(), + b"unprotected large deletes", + ) + .await; + + assert_eq!( + store.delete(&pinned.as_bytes()).await.unwrap_err(), + Error::GcRefuse, + "pinned keys refuse" + ); + assert!( + store.has(&pinned.as_bytes()).await.unwrap(), + "nothing deleted on refusal" + ); + assert_eq!( + store.delete(&source_live.as_bytes()).await.unwrap_err(), + Error::GcRefuse, + "registered-source-live keys refuse" + ); + assert!(store.has(&source_live.as_bytes()).await.unwrap()); + assert_eq!( + store.delete(&protect_named.as_bytes()).await.unwrap_err(), + Error::GcRefuse, + "protect-callback-named keys refuse" + ); + assert!(store.has(&protect_named.as_bytes()).await.unwrap()); + + store + .delete(&unprotected_kv.as_bytes()) + .await + .expect("unprotected kv keys delete"); + assert!(!store.has(&unprotected_kv.as_bytes()).await.unwrap()); + store + .delete(&unprotected_large.as_bytes()) + .await + .expect("unprotected large keys delete through the fall-through"); + assert!(!store.has(&unprotected_large.as_bytes()).await.unwrap()); + + let absent = key_for(b"never planted"); + assert_eq!( + store.delete(&absent.as_bytes()).await.unwrap_err(), + Error::Missing, + "absent keys carry Missing" + ); + assert_eq!( + store.delete(&[0x03; 33]).await.unwrap_err(), + Error::KeyInvalid, + "malformed key bytes reject at the boundary" + ); + } + + /// Sequential evidence of the conversion ordering primitive: after + /// the ordered step, the pin refcount is released **and** the + /// registration is visible — under one arbitration-lock section + /// (ADR-005's pinned conversion ordering). + #[tokio::test] + async fn pin_conversion_releases_and_registers_in_one_ordered_step() { + let store = Store::mem_only(); + let source = Arc::new(SetSource::default()); + store.register_liveness_source(source.clone()); + + let key = key_for(b"converted protection"); + let pin = store.gc().acquire(&key).await; + assert_eq!(store.gc().pin_count(&key), 1); + + store.gc().convert(pin, || source.add(key.clone())); + + assert_eq!( + store.gc().pin_count(&key), + 0, + "the pin's release is one half of the ordered step" + ); + assert!( + source.contains(&key), + "the registration is the other half — the same section" + ); + + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.candidates_staged, 0, + "the converted registration protects through the source" + ); + } + + /// The ordered step under race: a pin converted into a source is + /// never an unprotected instant — a concurrent direct delete must + /// observe the pin or the registration at every schedule and + /// refuse, never delete (ADR-005's conversion-ordering invariant, + /// raced). + #[tokio::test] + async fn conversion_raced_against_direct_delete_never_deletes() { + let store = Store::mem_only(); + let source = Arc::new(SetSource::default()); + store.register_liveness_source(source.clone()); + + for round in 0..64 { + let content = format!("converted protection round {round}"); + let key = key_for(content.as_bytes()); + let bytes = key.as_bytes(); + plant_kv(&store, &bytes, content.as_bytes()).await; + + let pin = store.gc().acquire(&key).await; + let (result, ()) = tokio::join!(store.delete(&bytes), async { + store.gc().convert(pin, { + let source = Arc::clone(&source); + let key = key.clone(); + move || source.add(key) + }) + },); + assert_eq!( + result, + Err(Error::GcRefuse), + "round {round}: no schedule leaves the key unprotected " + ); + assert!( + store.has(&bytes).await.unwrap(), + "nothing deleted while the protection converted" + ); + + source.remove(&key); + store + .delete(&bytes) + .await + .expect("unprotected now, deletes"); + } + } + + /// The test-asserted property (ADR-005): a sweep racing an + /// in-flight pinned put never deletes the protected entry — the + /// pin-before-publish ordering holds from the sweep side, and the + /// delete-then-recover arm lands byte-identical. Counts stay exact + /// in every schedule, and the surviving pin keeps protecting. + #[tokio::test] + async fn sweep_racing_a_pinned_put_never_deletes_the_protected_entry() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + + for round in 0..64 { + let content = format!("race round {round}"); + let key = key_for(content.as_bytes()); + let bytes = key.as_bytes(); + + plant_kv(&store, &bytes, content.as_bytes()).await; + + let (report, pin) = tokio::join!(store.sweep(), async { + let pin = store.gc().acquire(&key).await; + plant_kv(&store, &bytes, content.as_bytes()).await; + pin + },); + + let report = report.expect("sweeps are never errors"); + assert_eq!(report.aborted, None, "round {round}"); + assert_eq!( + report.candidates_staged, + report.deleted + report.cancelled_by_arbitration, + "round {round}: exact counts in every schedule" + ); + + assert!( + store.has(&bytes).await.unwrap(), + "round {round}: the entry survives (canceled, or delete-then-recovered)" + ); + let (_, blob) = store + .get(&bytes) + .await + .expect("get succeeds") + .expect("entry present"); + assert_eq!( + blob.read_at(0, content.len() as u64).await.unwrap(), + content.as_bytes(), + "round {round}: the recovered re-put is byte-identical" + ); + + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!( + report.deleted, 1, + "round {round}: with the pin dropped the entry is dead — exactly one deletion" + ); + assert!(!store.has(&bytes).await.unwrap()); + } + } + + /// An in-flight put whose publish lands after the sweep's mark is + /// never enumerated by it — the sweep started before the commit + /// and cannot delete what its mark never saw (ADR-005's + /// pin-before-publish invariant, the fresh-put arm). + #[tokio::test] + async fn put_publishing_after_a_sweep_started_is_never_deleted() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + + let content = b"published after the mark"; + let key = key_for(content); + let bytes = key.as_bytes(); + let dead = key_for(b"a dead witness for the concurrent sweep"); + plant_kv( + &store, + &dead.as_bytes(), + b"a dead witness for the concurrent sweep", + ) + .await; + + let (report, pin) = tokio::join!(store.sweep(), async { + let pin = store.gc().acquire(&key).await; + tokio::task::yield_now().await; + tokio::task::yield_now().await; + plant_kv(&store, &bytes, content).await; + pin + },); + + let report = report.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "the dead witness deleted"); + assert_eq!( + report.candidates_staged, 1, + "the in-flight put's entry never listed — nothing staged for it" + ); + assert!( + store.has(&bytes).await.unwrap(), + "the published entry survived" + ); + assert_eq!(store.gc().pin_count(&key), 1, "the pin still protects"); + + drop(pin); + let report = store.sweep().await.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1, "now unprotected, deleted"); + } + + /// `has`/`get` during a delete window observe no torn state: each + /// observation is either the untouched full content or clean + /// absence — existence flips atomically per key, content never + /// partial (the pool's immutable-entry posture). + #[tokio::test] + async fn reads_during_a_delete_window_observe_no_torn_state() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + + let content = b"whole-content visibility under the window"; + let key = key_for(content); + let bytes = key.as_bytes(); + plant_kv(&store, &bytes, content).await; + + let observer = { + let store = &store; + let bytes = bytes.clone(); + let content_len = content.len() as u64; + async move { + let mut seen_true = false; + let mut seen_absence = false; + for _ in 0..200 { + if store.has(&bytes).await.unwrap() { + seen_true = true; + assert!( + !seen_absence, + "no re-appearance: existence flips once per key" + ); + let (_, blob) = store + .get(&bytes) + .await + .expect("get succeeds") + .expect("has said true"); + let whole = blob.read_at(0, content_len).await.unwrap(); + assert_eq!( + whole, content, + "an observed entry presents its exact, whole content" + ); + } else { + seen_absence = true; + } + tokio::task::yield_now().await; + } + (seen_true, seen_absence) + } + }; + + let ((seen_true, seen_absence), report) = tokio::join!(observer, store.sweep()); + let report = report.expect("sweeps are never errors"); + assert_eq!(report.deleted, 1); + assert!(seen_true || seen_absence, "the observer ran alongside"); + assert!( + !store.has(&bytes).await.unwrap(), + "after the window the entry is gone" + ); + } + + /// The single-node sweep is single-flight — the in-process analog + /// of the named exclusive sweep lease (ADR-013 §2): a second + /// sweeper contending behind the first aborts as report data + /// (`SweeperLock`), nothing deleted. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn concurrent_sweeps_single_flight_and_the_second_aborts_as_data() { + let store = Arc::new(Store::mem_only()); + 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 entry under the contended sweeps"); + plant_kv( + &store, + &dead.as_bytes(), + b"dead entry 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), + "the contended sweeper aborts as report data" + ); + assert_eq!(second.deleted, 0); + assert!( + store.has(&dead.as_bytes()).await.unwrap(), + "the contended sweep deleted nothing" + ); + + 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.deleted, 1, "the lease holder sweeps"); + assert!(!store.has(&dead.as_bytes()).await.unwrap()); + } + + /// Sweeps surface operational failures as errors (ADR-012 §1's + /// error half): a malformed pool list is fatal `KeyInvalid`, never + /// a silent skip — through the sweep's own mark enumeration too. + #[tokio::test] + async fn sweep_of_a_malformed_pool_list_is_an_operational_error() { + let store = Store::mem_only(); + store.register_liveness_source(Arc::new(SetSource::default())); + plant_kv(&store, &[0x07; 33], b"stray in the table").await; + + assert_eq!( + store.sweep().await.unwrap_err(), + Error::KeyInvalid, + "a malformed pool list is fatal through the sweep" + ); + } } diff --git a/tasks/gc-protocol.md b/tasks/gc-protocol.md index d43c70f..30302a5 100644 --- a/tasks/gc-protocol.md +++ b/tasks/gc-protocol.md @@ -1,7 +1,7 @@ --- id: gc-protocol name: In-process GC protocol — pins, liveness, delete windows, sweep, direct-delete refusal -status: pending +status: completed depends_on: [store-construction] scope: broad risk: high @@ -82,24 +82,24 @@ extends without changing its invariants. ## Acceptance Criteria -- [ ] Pin registry with per-key refcounts under the arbitration lock; +- [x] Pin registry with per-key refcounts under the arbitration lock; dedup re-puts refcount; `Pin`/`Batch` drops release through it (batch drops in one ordered step) -- [ ] `register_liveness_source` live-shared (consulted at mark and at +- [x] `register_liveness_source` live-shared (consulted at mark and at delete-time arbitration, never snapshotted); source shape decided and recorded -- [ ] Protect callback with abort-as-data (`ProtectFailed`); +- [x] Protect callback with abort-as-data (`ProtectFailed`); `NoLivenessSources` safe default; `sweep()` returns `Ok(SweepReport)` on every abort — never an abort error -- [ ] Delete-window protocol: mark → arbitrate-at-delete-time → +- [x] Delete-window protocol: mark → arbitrate-at-delete-time → batched commit (~100/batch), with exact-count `SweepReport` fields (`candidates_staged`/`deleted`/`cancelled_by_arbitration`) -- [ ] Direct `delete` refuses the full protection set with `GcRefuse` +- [x] Direct `delete` refuses the full protection set with `GcRefuse` and deletes unprotected keys through the fall-through dispatch -- [ ] Concurrency tests: sweep vs in-flight pinned put (the +- [x] Concurrency tests: sweep vs in-flight pinned put (the pin-before-publish property from the sweep side), exact-count outcomes, abort shapes -- [ ] `cargo test` green; `cargo test --features mem` and `cargo test +- [x] `cargo test` green; `cargo test --features mem` and `cargo test --all-features` green; clippy/fmt gates green ## References @@ -115,6 +115,127 @@ extends without changing its invariants. > To be filled by implementation agent +**Named design points decided here** (recorded in `src/gc.rs`'s module +doc and on the traits in `src/facade.rs`): + +- **Guard back-reference shape**: `Pin` holds `Arc` (the + GC home's shared core) alongside its key — so a drop reaches the + per-key refcounts wherever the guard travels, and the registry + outlives the `Store` instance when guards still reference it. No + weak-pointer tag-drop plumbing (iroh's TempTag conclusion without + their mechanism). `Batch` holds its members' keys; contribution + defuses the member guard (`armed` flag), and the batch's release + decrements every member refcount inside **one** arbitration-lock + section, keys in total order — no early single-pin release. +- **Lock granularity**: one `parking_lot::Mutex` — the + pin refcounts, the in-flight delete markers, and the registered + source lists under one lock, so a delete-time decision (pins ∨ + markers ∨ sources) and its marker insert are one atomic section + against pin mints. parking_lot does not poison (no `into_inner` + posture to apply). Consumer consults (`contains`/`protect`) run + under that lock at delete-time arbitration and outside it at mark + time; lock order is one-directional (arbitration lock → consumer + locks; the ordered-step conversion is the only nesting, same + direction). +- **Liveness source shape**: a sync (non-async) trait, + `live_keys()` (mark-time enumeration, consulted outside the lock) + + `contains()` (delete-time probe, consulted under the lock; default + derives it linearly from `live_keys` — indexed sources override). + Sync because pin drops are sync RAII and arbitration consults run + under the short-held lock. Consumers register `Arc` shared handles — the POC finding-7 shape; the store + consults the current contents every time and never snapshots. +- **Protect callback shape**: one sync verb, `protect() → + Option>`; `None` = failed to answer → sweep aborts with + `ProtectFailed` report data, and a direct delete refuses + conservatively (`GcRefuse`, nothing deleted). Callbacks compose + (each registered set unions into the protection). +- **The sweep-vs-put race closes with in-flight delete markers**, not + by holding the arbitration lock across engine I/O: the arbitrate + section marks a Proceed-ing key `deleting`; the engine delete runs + outside the lock; marker release notifies blocked pin mints. A + re-put's mint landing while the marker is held blocks (async, + `tokio::sync::Notify` with the enable-before-check recipe) until the + deletion commits, then lands fresh — ADR-005's case (a) + delete-then-recover verbatim; a mint landing before arbitration is + case (b), protection composed at the arbitration itself. Holding the + lock across the delete was rejected: a `parking_lot` guard is not + `Send`, the sweep future must be, and AGENTS convention 3 forbids + blocking I/O under internal locks. +- **The sweep is single-flight in-process**: an atomic test-and-set + flag is the single-node analog of ADR-013 §2's named exclusive sweep + lease (ADR-012 §5's "single-flighter is the in-process lock" + posture); a contended second sweeper aborts with `SweeperLock` + report data — nothing deleted, the next run takes the lease. +- **Vanished candidates count as arbitration cancels**: a key deleted + between mark and its window turn (e.g. a concurrent direct delete) + settles into `cancelled_by_arbitration`, keeping the completed + sweep's accounting exact: `candidates_staged == deleted + + cancelled_by_arbitration`. +- **Registration vocabulary**: a public opaque `Registration` token + per registration; `unregister_liveness_source` / + `unregister_protect_callback` consume it (the correction-flow + prerequisite of ADR-012 §6.4). The protect-callback registration + verbs are this layer's addition — the direct-delete full-protection + set requires consultable protect state on the store, which store-api + §Lifecycle implies but does not spell out. +- **The ordered-step conversion primitive**: `GcHome::convert(pin, + register)` (crate-internal until a consumer-facing surface exists): + pin release + registration closure inside one arbitration-lock + section; tested sequentially and raced (64-round direct-delete + race: no schedule leaves the key unprotected — every delete + refuses). + +**Scope notes**: the pin minting API (`GcHome::acquire`, async, waits +on in-flight deletes) is wired; the put path's call sites remain for +`store-put-paths`, which now only needs `store.gc().acquire(key)` / +`new_batch()` / `convert` at the publish points. Batch commit semantics +(the seam's terminal step) also rides that task; `Batch::release_members` +is its reuse point. Sweep/`delete`/registration live on `Store` +(`src/store.rs`), with one new `KvEngine`/`LargeEngine` dispatch method +(`delete`). A dev-dependency addition (`tokio`'s `rt-multi-thread`, +test-only) supports the single-flight concurrency test's multi-thread +runtime flavor; the base crate's dependency set is unchanged. + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +The in-process GC protocol (ADR-005; abort-as-data per ADR-012 §1; +non-SQL inline window staging per ADR-012 §5) is wired on `Store`: + +- `src/gc.rs` — `GcHome` + `RegistryCore`: one arbitration lock over + pin refcounts, delete markers, and registration lists; async pin + minting (`acquire`) that blocks on in-flight deletes (the + delete-window mechanism); the ordered-step `convert` primitive; + batch one-step release; source/protect registration bookkeeping; + the single-node sweep lease. +- `src/facade.rs` — `Pin`/`Batch` now carry the registry + back-reference and own their drop-release behavior (the wave-1 + contracts are now wired behavior); new public vocabulary + `LivenessSource`, `ProtectCallback`, `Registration` (re-exported + from `lib.rs`). +- `src/store.rs` — `sweep()` (single-flight → zero-source safe + default → protect consult → mark over both tiers' `list()` union → + arbitrate-at-delete-time → ~100/batch commit → exact-count report + or abort-as-data), direct `delete()` (existence probe → full + protection set arbitration → `GcRefuse`/`Missing`/delete through + the fall-through), and the four registration/unregistration verbs. +- 16 new tests + the facade pin/batch shape test reworked on the wired + machinery: zero-source abort, empty-pool zero report, exact + counts across both tiers, dedup refcounts, batch one-step drops, + protect add + abort-as-data, live-shared (non-snapshotted) sources, + token unregistration, the full direct-delete refusal set (+ + `Missing`/`KeyInvalid`), the ordered-step conversion sequential and + raced (64 rounds, always refused), the sweep-vs-pinned-put race + (64 rounds: entry survives in every schedule, byte-identical + re-put, exact counts, pin keeps protecting after the race), the + publish-after-mark arm, no-torn `has`/`get` reads under a window, + single-flight contention aborting as data, and malformed-list + sweeps as operational errors. + +Verification: `cargo test` (default) 46 passed; `cargo test +--features mem` and `--all-features` 89 passed each; `cargo clippy +--all-targets -- -D warnings` and `--all-features` clean; `cargo fmt +--check` clean; `cargo doc --no-deps` at the pre-existing baseline +(24 warnings carried from earlier waves, none new). \ No newline at end of file