alkblobs/src/facade.rs

//! The store facade's value types (store-api.md §Public surface): the
//! 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`]) 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;

/// The `get` seam's public stream (`(len, stream)` — store-api.md §Read
/// path): a bounded, re-entrant **windowed reader** over a stored entry,
/// uniform across both tiers with the crate-internal read cursor
/// (ADR-010 §1) staying behind it. The kv tier's windows slice
/// materialized content (bounded by tier policy); the large tier's read
/// window-by-window over the committed file — streaming without ever
/// materializing the whole value. `read_at` is a pread-style bounded
/// window (over-reads clamp at the entry's end; a read at EOF is empty,
/// never an error); `read_range` is the same window plus the ADR-006
/// slice digest — plain SHA-256 over exactly the returned bytes, fixed
/// by convention.
///
/// Constructed by the store core only (via `get`); engine cursors never
/// leak onto this surface. A `Blob` stays valid across concurrent pool
/// activity because committed entries are immutable. No verification
/// rides the windows themselves: canonical verification fires at get
/// time on the kv tier's materialized reads (and at put time), per the
/// store core's recorded verification-on-get posture (ADR-006).
pub struct Blob {
    cursor: AnyCursor,
}

impl fmt::Debug for Blob {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Blob")
            .field("len", &self.cursor.len())
            .finish()
    }
}

impl Blob {
    /// Store-core constructor: a `Blob` wraps a cursor-dispatched
    /// engine body; consumers obtain one only through the read seam.
    pub(crate) fn new(cursor: AnyCursor) -> Blob {
        Blob { cursor }
    }

    /// The entry's length in bytes — the metadata-only length the
    /// cursor body carries (the same probe `stat` rides).
    pub fn len(&self) -> u64 {
        self.cursor.len()
    }

    /// Whether the entry holds zero bytes.
    pub fn is_empty(&self) -> bool {
        self.cursor.len() == 0
    }

    /// The bytes at `[offset, offset + len)` — clamped at the entry's
    /// end (pread-style short read); `offset >= len()` (EOF) is empty,
    /// never an error.
    pub async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, crate::Error> {
        self.cursor.read_at(offset, len).await
    }

    /// The bytes at `[offset, offset + len)` plus the ADR-006 slice
    /// digest — SHA-256 over exactly the returned bytes, for the caller
    /// to carry out-of-band.
    pub async fn read_range(
        &self,
        offset: u64,
        len: u64,
    ) -> Result<(Vec<u8>, [u8; 32]), crate::Error> {
        self.cursor.read_range(offset, len).await
    }
}

/// The cheap stat probe's result (store-api.md §Read path): the entry's
/// length and its hash algorithm — the "type" at this layer is the hash
/// algorithm, there is no other (ADR-002). `stat` rides the Backend
/// trait's `size` method (ADR-008): an indexed-column probe on kv
/// engines, a file stat on `local`, a companion-row probe on pg-lo —
/// never a value/content load.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EntryMeta {
    /// The entry's length in bytes.
    pub len: u64,
    /// The key's hash algorithm — ADR-002's two-member policy.
    pub algorithm: Algorithm,
}

/// The RAII put-path pin guard (ADR-005 §Marks 2; store-api.md §Write
/// path).
///
/// **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 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
///   re-put acquires no new pool entry but refcounts the existing one
///   under the same arbitration lock (ADR-005).
/// - Fleet correspondence (ADR-010 §4, ADR-012 §2): in fleet mode the
///   joint tx commits the expiry-carrying pin row and this guard drops
///   at response time (the row's expiry is then the protection); in
///   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 {
    /// 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 fmt::Debug for Pin {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Pin")
            .field("key", &self.key.digest())
            .field("armed", &self.armed)
            .finish()
    }
}

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 (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
///   pin-before-publish exactly as lone puts do).
/// - **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. 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 {
    registry: Arc<RegistryCore>,
    keys: Vec<Key>,
}

impl Batch {
    /// 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). The
    /// contributed guard is defused — its own drop releases nothing;
    /// the batch's one-step release owns the protection from here.
    pub(crate) fn add(&mut self, pin: 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.keys.len()
    }

    /// Whether the group currently protects nothing.
    pub fn is_empty(&self) -> bool {
        self.keys.is_empty()
    }

    /// **Batch commit**: the group's pins drop together in one ordered
    /// step — protection spans the whole batch or none of it, and ends
    /// when the batch ends (ADR-005 §Marks 2; the same release batch
    /// drop drives). A committed batch's members are no longer
    /// batch-protected; durable protection beyond the write group is
    /// the consumer's liveness seam (`Store::register_liveness_source`
    /// or the pin-conversion ordering).
    pub fn commit(mut self) {
        self.release_members();
    }

    /// 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();
    }
}

/// A sweep's abort cause — report data only, never an error (ADR-012 §1:
/// sweeps are never errors; an aborted sweep returns `Ok(SweepReport)`
/// with `aborted: Some(cause)`, and there is no sweep-abort error
/// variant).
///
/// Exactly three variants, ADR-pinned; new causes are new ADRs.
/// `#[non_exhaustive]` makes the ADR-012 §1 ruling — callers match
/// non-exhaustively — structural for downstream crates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum GcAbortCause {
    /// A protection source failed to answer (ADR-005's protect-callback
    /// failure): the sweep aborted, nothing deleted.
    ProtectFailed,
    /// The fleet sweep lease was contended (ADR-008; ADR-013 §2's
    /// exclusive-lease semantics): the second sweeper aborts — nothing
    /// deleted; its next run takes the lock.
    SweeperLock,
    /// The safe default: a sweep invoked with zero registered liveness
    /// sources aborts without deleting (store-api.md §Lifecycle).
    NoLivenessSources,
}

/// A key's pin classification (ADR-010 §2): the engine's only vocabulary
/// for pins — the engine never decides deletion, it only classifies, and
/// the store core receives the answer and applies the decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PinState {
    /// An un-expired pin row exists for the key — the key is protected.
    Protected,
    /// Nothing pins the key.
    Unprotected,
}

/// A sweep's outcome record — always data, never an error carrier
/// (ADR-012 §1). Counts are exact; the test gate asserts sweep outcomes
/// against these fields (store-api.md invariants 4/7).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SweepReport {
    /// Candidates staged into the delete window at the mark phase.
    pub candidates_staged: u64,
    /// Candidates actually deleted at the commit phase — plus, on the
    /// sweep-as-executor path, the overdue staged-window candidates this
    /// sweep re-arbitrated and executed (ADR-012 §6.7); `0` on any abort.
    pub deleted: u64,
    /// Candidates cancelled at delete time by arbitration — the put side
    /// of the sweep-vs-put race won, and protection composed over the
    /// candidate (ADR-005's delete-window protocol).
    pub cancelled_by_arbitration: u64,
    /// `Some(cause)` when the sweep aborted — nothing was deleted;
    /// `None` on a completed sweep.
    pub aborted: Option<GcAbortCause>,
    /// Wall-clock start of the run.
    pub started_at: SystemTime,
    /// Wall-clock finish — an aborted sweep also finishes; the report is
    /// its outcome record either way.
    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::*;
    use crate::hashing::Digest;

    /// The stat value's shape: length plus the key's algorithm — there
    /// is no other type at this layer (store-api.md §Read path).
    #[test]
    fn entry_meta_carries_len_and_algorithm() {
        let key = Key::from(Digest::derive(b"meta", 4));
        let meta = EntryMeta {
            len: 4,
            algorithm: key.algorithm(),
        };
        assert_eq!(meta.len, 4);
        assert_eq!(meta.algorithm, Algorithm::GitBlobSha256);
        assert_eq!(meta.algorithm.digest_len(), 32);
    }

    /// The abort vocabulary's exhaustiveness shape: exactly three
    /// variants, ADR-pinned (ADR-012 §1). The in-crate match below is
    /// exhaustive without a wildcard — adding a variant breaks it,
    /// which is the "new causes are new ADRs" discipline surfacing at
    /// compile time in-crate; `#[non_exhaustive]` enforces the
    /// wildcard discipline on downstream callers.
    #[test]
    fn gc_abort_cause_is_exactly_three_variants() {
        let mut matched: Vec<&'static str> = Vec::new();
        for cause in [
            GcAbortCause::ProtectFailed,
            GcAbortCause::SweeperLock,
            GcAbortCause::NoLivenessSources,
        ] {
            match cause {
                GcAbortCause::ProtectFailed => matched.push("protect"),
                GcAbortCause::SweeperLock => matched.push("lease"),
                GcAbortCause::NoLivenessSources => matched.push("sources"),
            }
        }
        assert_eq!(matched, ["protect", "lease", "sources"]);
    }

    /// Report field semantics: exact counts, `aborted: None` on a
    /// completed run, and the abort shape — nothing deleted, the cause
    /// carried, the report still a finished record.
    #[test]
    fn sweep_report_carries_exact_counts_and_abort_data() {
        let now = SystemTime::now();
        let completed = SweepReport {
            candidates_staged: 3,
            deleted: 2,
            cancelled_by_arbitration: 1,
            aborted: None,
            started_at: now,
            finished_at: now,
        };
        assert!(completed.aborted.is_none());
        assert!(completed.finished_at >= completed.started_at);

        let aborted = SweepReport {
            deleted: 0,
            cancelled_by_arbitration: 0,
            aborted: Some(GcAbortCause::NoLivenessSources),
            ..completed
        };
        assert_eq!(aborted.candidates_staged, 3, "mark-phase count is kept");
        assert_eq!(aborted.deleted, 0, "an aborted sweep deletes nothing");
        assert_eq!(
            aborted.aborted,
            Some(GcAbortCause::NoLivenessSources),
            "the safe default cause is report data"
        );
    }

    /// 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 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(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
    /// is classification only (ADR-010 §2).
    #[test]
    fn pin_state_is_the_two_variant_classification() {
        assert_ne!(PinState::Protected, PinState::Unprotected);
        for state in [PinState::Protected, PinState::Unprotected] {
            match state {
                PinState::Protected | PinState::Unprotected => {}
            }
        }
    }
}