feat(gc): the in-process GC protocol — pins, liveness, delete windows, sweep, direct delete

The gc-protocol task (ADR-005; abort-as-data per ADR-012 §1; non-SQL
inline window staging per ADR-012 §5) wired on the Store facade:

- src/gc.rs — GcHome + RegistryCore: one arbitration lock over per-key
  pin refcounts, in-flight delete markers, and the registered source
  lists; async pin minting (acquire) that blocks on in-flight deletes
  (ADR-005's re-put case (a) delete-then-recover without engine I/O
  under the lock); the ordered-step convert primitive (pin release +
  registration in one section); batch one-step release; the
  single-node sweep lease (the in-process analog of ADR-013 §2).
- src/facade.rs — Pin/Batch carry the registry back-reference and own
  their drop release (batch drop decrements every member in one
  ordered step); 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 per ADR-012 §6.4 → GcRefuse/Missing/delete through the
  fall-through), and the registration/unregistration verbs.
- 16 new tests + the facade pin/batch shape test reworked on the wired
  machinery: exact counts across both tiers, abort shapes, dedup
  refcounts, batch drops, live-shared sources, the conversion ordered
  step sequentially and raced, sweep-vs-pinned-put (64 rounds, every
  schedule, byte-identical delete-then-recover), publish-after-mark,
  no-torn reads under a window, single-flight contention, and
  malformed-list sweeps as operational errors.
- tasks/gc-protocol.md — status completed; named design points recorded
  in Notes.

Verification: cargo test (default) 46 passed; --features mem and
--all-features 89 passed each; cargo clippy --all-targets -- -D
warnings (default and --all-features) clean; cargo fmt --check clean;
cargo doc --no-deps at the pre-existing baseline (no new warnings).

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 }
+tokio-postgres = { version = "0.7", optional = true }
+
+[dev-dependencies]
+tokio = { version = "1", default-features = false, features = ["rt-multi-thread"] }

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<RegistryCore>,
     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<RegistryCore>, 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<Pin>,
+    registry: Arc<RegistryCore>,
+    keys: Vec<Key>,
 }
 
 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<RegistryCore>) -> 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<dyn LivenessSource>`); 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<Key>;
+
+    /// 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<Vec<Key>>;
+}
+
+/// 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

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<RegistryCore>,
+}
+
+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<F: FnOnce()>(&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<dyn LivenessSource>) -> 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<dyn ProtectCallback>) -> 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<Arc<dyn LivenessSource>> {
+        self.core.registered_sources()
+    }
+
+    /// Snapshot of the registered protect callbacks — consulted
+    /// pre-sweep, outside the arbitration lock.
+    pub(crate) fn registered_protects(&self) -> Vec<Arc<dyn ProtectCallback>> {
+        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<Key> {
+        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<Arbitration, Error> {
+        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<RegistryState>,
+    /// 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<Key, u64>,
+    /// 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<Key>,
+    /// Registered liveness sources with their registration ids.
+    sources: Vec<(u64, Arc<dyn LivenessSource>)>,
+    /// Registered protect callbacks with their registration ids.
+    protects: Vec<(u64, Arc<dyn ProtectCallback>)>,
+}
+
+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<dyn LivenessSource>) -> 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(&registration, false)
+    }
+
+    pub(crate) fn register_protect(&self, callback: Arc<dyn ProtectCallback>) -> 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(&registration, 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<Arc<dyn LivenessSource>> {
+        self.state
+            .lock()
+            .sources
+            .iter()
+            .map(|(_, s)| Arc::clone(s))
+            .collect()
+    }
+
+    fn registered_protects(&self) -> Vec<Arc<dyn ProtectCallback>> {
+        self.state
+            .lock()
+            .protects
+            .iter()
+            .map(|(_, c)| Arc::clone(c))
+            .collect()
+    }
+
+    pub(crate) fn pinned_keys(&self) -> Vec<Key> {
+        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<Self>, 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<Key>) {
+        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<F: FnOnce()>(&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<Arbitration, Error> {
+        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

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};

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<RegistryCore>` (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<RegistryState>` — 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<dyn
+  LivenessSource>` 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<Vec<Key>>`; `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
+> 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).