diff --git a/Cargo.lock b/Cargo.lock index 5b1ec6e..5e471b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2300,7 +2300,7 @@ dependencies = [ [[package]] name = "digstore-chain" -version = "0.15.1" +version = "0.16.0" dependencies = [ "aes-gcm", "anyhow", @@ -2339,7 +2339,7 @@ dependencies = [ [[package]] name = "digstore-chunker" -version = "0.15.1" +version = "0.16.0" dependencies = [ "digstore-core", "hex", @@ -2349,7 +2349,7 @@ dependencies = [ [[package]] name = "digstore-cli" -version = "0.15.1" +version = "0.16.0" dependencies = [ "anstream 0.6.21", "anstyle", @@ -2415,7 +2415,7 @@ dependencies = [ [[package]] name = "digstore-core" -version = "0.15.1" +version = "0.16.0" dependencies = [ "aes-gcm-siv", "hex", @@ -2512,7 +2512,7 @@ dependencies = [ [[package]] name = "digstore-remote" -version = "0.15.1" +version = "0.16.0" dependencies = [ "async-trait", "axum", @@ -2568,6 +2568,19 @@ dependencies = [ "toml 0.8.23", ] +[[package]] +name = "digstore-subscription" +version = "0.16.0" +dependencies = [ + "async-trait", + "digstore-core", + "hex", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "directories-next" version = "2.0.0" diff --git a/Cargo.toml b/Cargo.toml index a2dc786..bdcd464 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,11 +1,11 @@ [workspace] resolver = "2" -members = ["crates/digstore-core", "crates/digstore-chunker", "crates/digstore-crypto", "crates/digstore-store", "crates/digstore-guest", "crates/digstore-prover", "crates/digstore-host", "crates/digstore-compiler", "crates/digstore-stage", "crates/digstore-remote", "crates/digstore-cli", "crates/digstore-chain", "crates/dig-resolver"] +members = ["crates/digstore-core", "crates/digstore-chunker", "crates/digstore-crypto", "crates/digstore-store", "crates/digstore-guest", "crates/digstore-prover", "crates/digstore-host", "crates/digstore-compiler", "crates/digstore-stage", "crates/digstore-remote", "crates/digstore-cli", "crates/digstore-chain", "crates/dig-resolver", "crates/digstore-subscription"] exclude = ["crates/digstore-prover/guest", "crates/dig-client-wasm"] [workspace.package] edition = "2021" -version = "0.15.1" +version = "0.16.0" license = "GPL-2.0-only" [workspace.dependencies] diff --git a/crates/digstore-subscription/Cargo.toml b/crates/digstore-subscription/Cargo.toml new file mode 100644 index 0000000..ebf6b1c --- /dev/null +++ b/crates/digstore-subscription/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "digstore-subscription" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "The canonical DIG Network Subscription primitive: a networkless chain-watch state machine that follows one CHIP-0035 store singleton's full lineage and drives per-tip .dig backfill over injected chain/network seams." +repository = "https://github.com/DIG-Network/dig-store" +readme = "README.md" +keywords = ["dig", "chia", "subscription", "chain-watch", "singleton"] +categories = ["data-structures", "asynchronous"] + +[lib] +name = "digstore_subscription" +path = "src/lib.rs" + +[dependencies] +# The canonical `Capsule` (storeId:rootHash) + `Bytes32` identity. A Subscription +# follows a store's lineage of capsules, so it speaks the one ecosystem-wide capsule +# type rather than re-inventing the pair. This is the crate's ONLY runtime dependency: +# the primitive is a networkless leaf so any consumer (node, resolver, SDK) can follow a +# store without dragging in the spend-heavy `digstore-chain` crate. +digstore-core = { path = "../digstore-core" } +async-trait = "0.1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tracing = "0.1" + +[dev-dependencies] +tokio = { version = "1", features = ["rt", "macros"] } +hex = "0.4" diff --git a/crates/digstore-subscription/README.md b/crates/digstore-subscription/README.md new file mode 100644 index 0000000..c23f0dd --- /dev/null +++ b/crates/digstore-subscription/README.md @@ -0,0 +1,7 @@ +# digstore-subscription + +The canonical DIG Network **Subscription** primitive: a networkless chain-watch state +machine that follows one CHIP-0035 store singleton's full lineage and drives per-tip +`.dig` backfill over injected chain/network seams. + +See `SPEC.md` for the normative contract. diff --git a/crates/digstore-subscription/SPEC.md b/crates/digstore-subscription/SPEC.md new file mode 100644 index 0000000..da9da70 --- /dev/null +++ b/crates/digstore-subscription/SPEC.md @@ -0,0 +1,117 @@ +# digstore-subscription — SPEC + +Normative contract for the canonical DIG Network **Subscription** primitive. An independent +reimplementation MUST conform to this document. Cross-references: superproject `SYSTEM.md` +("Subscription — the canonical chain-watch primitive"), CLAUDE.md §5.1 (`.dig` permanence), +§5.3 (client→node read ladder), dig-store `SPEC.md` (`Capsule`, the §21 remote). + +## 1. Concept + +A **Subscription** follows exactly ONE CHIP-0035 store singleton and keeps a local set of +`.dig` files in sync with that singleton's on-chain history. A **capsule** is one immutable +store generation `(store_id, root_hash)`, written `storeId:rootHash`; a store is the ordered +sequence of capsules its singleton has committed, eve → current tip. + +A Subscription MUST: + +1. **Watch** the chain for the store's singleton tip progressing (a new generation / new root). +2. **On tip progression**, find the `.dig` matching the new tip's capsule on the network and + sync it down. +3. **Track the ENTIRE singleton history** and backfill every historical tip's `.dig` where + possible (best-effort; a tip whose `.dig` cannot be found is recorded missing and retried). +4. Be a **distinct, isolated behaviour** — one managed object per store, owning its watch + state, full-history tracking, and per-tip sync status. + +## 2. Architecture — a networkless core over four seams + +The primitive owns POLICY (which capsules to fetch, ordering, reorg handling) and NONE of the +mechanism. Mechanism is four injected traits so the decision core and state machine are pure +and unit-testable with no chain and no network: + +- **`ChainWatch`** — resolve a store's confirmed capsule lineage (eve → tip). Return type + `LineageResult = Result, String>`. This carries the fail-closed contract (§4). +- **`CapsuleFetcher`** — find + sync + verify + land the `.dig` for one capsule. MUST verify the + fetched module against the chain-anchored root before landing it, and be idempotent. It MUST + return `Ok(())` ONLY after the module is verified AND landed such that `HeldCheck` will report + it held — an `Ok(())` that does not actually land the module leaves the capsule missing, so the + next tick re-plans and re-fetches it indefinitely. +- **`HeldCheck`** — whether the `.dig` for a capsule is already held locally. +- **`Persistence`** — load/save the persisted `SubscriptionSet`. A missing/corrupt store loads + as an empty set, never an error. + +A `Lineage` is an ordered, non-empty capsule list all sharing one `store_id`; it is exactly +the `history` a chain walk (dig-store's `sync_datastore_with_history` → `StoreHistory`) +produces. It is modelled chain-free so the crate does not depend on the spend-heavy +`digstore-chain` crate; a consumer adapts `StoreHistory.history` (a `Vec`) straight +into `Lineage::try_new`. + +## 3. State machine + +A `Subscription` tracks, per store: + +- the last-observed confirmed `history` (ordered `Capsule` lineage, eve → tip); +- a per-capsule `TipSync` status keyed by root: `Held`, `Missing`, `Pending`, + `Failed { attempts, last_error }`; +- the `orphaned` capsules — those dropped from the canonical lineage by a reorg (§5). + +`observe_lineage(lineage, held)` folds a freshly-confirmed lineage into the state: it detects +tip progression (new capsules appended) and reorg (§5), and refreshes each capsule's status +from `HeldCheck`. A `Failed`/`Pending` record is preserved across a re-observe (so retry counts +survive) unless the `.dig` is now held. A subscription is bound to exactly ONE store: a lineage +whose `store_id` differs from the subscription's is IGNORED (never folded), so a buggy +`ChainWatch` returning another store's walk cannot repoint the history. + +`record_fetch_result(capsule, result)` sets `Held` on success, else `Failed` with an +incremented attempt count. Attempts MUST accrue across ticks in the real reconcile flow: because +a tick sets `Pending` immediately before recording the result, the `Pending` state carries the +prior failed-attempt count forward, so a capsule that fails on N consecutive ticks records +`Failed { attempts: N }` (never resetting to 1 each tick). + +## 4. Fail-closed invariant (MANDATORY) + +A Subscription MUST NEVER fetch or verify against an unconfirmable root: + +- `ChainWatch` returning `Err(_)` (chain read failed) ⇒ the tick does NOTHING (`Skipped(ChainError)`); +- `ChainWatch` returning `Ok(None)` (no confirmed generation) ⇒ the tick does NOTHING + (`Skipped(NoConfirmedGeneration)`); +- only `Ok(Some(lineage))` — a chain-CONFIRMED walk — drives observe + fetch. + +The decision core `decide` is only ever handed a confirmed `Lineage`, so it structurally cannot +act on an unconfirmable root. + +## 5. Reorg / permanence invariant (MANDATORY — §5.1) + +`.dig` files are permanent, on-chain-anchored artifacts. When a reorg supersedes a root (drops +it from the canonical lineage returned this tick): + +- the superseded capsule is moved to `orphaned` and RETAINED — never deleted; +- its `TipSync` status is preserved (an already-`Held` superseded root stays `Held`); +- the primitive exposes NO eviction/delete API — a held `.dig` can never be evicted by a reorg. + +The canonical lineage each tick is whatever `ChainWatch` returns from the current unspent tip; +new tips append, dropped tips orphan, held content persists. + +## 6. Fetch ordering (deterministic) + +`decide(lineage, held)` emits `SyncAction::Fetch(capsule)` for every capsule not already held: + +1. the current **tip first** (§ clause 2 — a client following the store gets current fastest); +2. then the remaining gaps **oldest → newest** (§ clause 3 backfill); +3. duplicate roots collapse — a store that reverted to a prior root fetches that `.dig` once. + +## 7. Persisted subscription set + +`SubscriptionSet` is an order-preserving, de-duplicated set of lower-case 64-hex store ids. Ids +are normalized (trimmed + lower-cased) on insert so a mixed-case duplicate collapses to one. +`add`/`remove` are idempotent and reject non-64-hex ids. The JSON codec (`encode`/`decode`) is a +schema-versioned document (`{ "version": 1, "stores": [...] }`); `decode` tolerates an empty, +garbage, or legacy bare `{ "stores": [...] }` input as an empty/best-effort set and drops +malformed entries. The document schema is additive-only (a version bump never removes or +repurposes a field), so an old reader ignores unknown fields and a new reader defaults them. + +## 8. Watch cadence + +The host owns the loop and the async runtime; the crate is runtime-agnostic (`reconcile_tick` +for one store, `Subscriptions::reconcile_all` for the set). `watch_interval_from_env` resolves +the poll interval from `DIG_WATCH_INTERVAL` (seconds), defaulting to 30 s and floored at 1 s so +a mis-set value cannot flood the chain source. diff --git a/crates/digstore-subscription/src/decide.rs b/crates/digstore-subscription/src/decide.rs new file mode 100644 index 0000000..fe17035 --- /dev/null +++ b/crates/digstore-subscription/src/decide.rs @@ -0,0 +1,130 @@ +//! The pure decision core — given a confirmed lineage + what is held, decide which +//! capsules to fetch and in what order. No chain, no network, no I/O: fully unit-testable. +//! +//! This generalizes dig-node's `decide_watch` from "gap-fill the ONE confirmed tip" to +//! "gap-fill EVERY historical capsule not held" (#979 clause 3, full-history backfill). +//! +//! Ordering policy (deterministic + tested): +//! 1. **the current tip first** — clause 2: a new tip is the most-wanted `.dig`, fetched +//! ahead of history so a client following the store gets current fastest; +//! 2. **then the remaining gaps, oldest → newest** — clause 3 backfill, in chronological +//! order so history fills forward predictably. +//! +//! FAIL-CLOSED: [`decide`] only ever runs on a chain-CONFIRMED [`Lineage`]. A chain error +//! or a no-generation store never produces a lineage (the [`ChainWatch`](crate::ChainWatch) +//! seam returns `Err`/`Ok(None)`), so this function is never handed an unconfirmable root. + +use digstore_core::Capsule; + +use crate::lineage::Lineage; +use crate::seams::HeldCheck; + +/// One unit of sync work the watcher should perform: fetch the `.dig` for this capsule. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SyncAction { + /// Find + sync + verify + land the `.dig` for this capsule (via [`CapsuleFetcher`](crate::CapsuleFetcher)). + Fetch(Capsule), +} + +/// Decide the ordered fetch worklist for a confirmed lineage: every capsule not already +/// held, tip-first then oldest → newest, with duplicate roots collapsed (a store that +/// reverted to a prior root only needs that `.dig` fetched once). +pub fn decide(lineage: &Lineage, held: &dyn HeldCheck) -> Vec { + let tip = lineage.tip(); + let mut actions = Vec::new(); + let mut queued: Vec = Vec::new(); + + let mut push_if_needed = |capsule: Capsule, actions: &mut Vec| { + // Skip what we already hold, and any root already queued this tick (dedup so a + // repeated root — a revert — is fetched once). + if held.is_held(&capsule) || queued.iter().any(|c| c.root_hash == capsule.root_hash) { + return; + } + queued.push(capsule); + actions.push(SyncAction::Fetch(capsule)); + }; + + // 1. The current tip first (clause 2). + push_if_needed(tip, &mut actions); + // 2. Then historical gaps, oldest → newest (clause 3). The tip is already queued, so + // its (possibly repeated) root is skipped by the dedup. + for &capsule in lineage.capsules() { + push_if_needed(capsule, &mut actions); + } + actions +} + +#[cfg(test)] +mod tests { + use super::*; + use digstore_core::Bytes32; + + fn cap(root: u8) -> Capsule { + Capsule { + store_id: Bytes32([1; 32]), + root_hash: Bytes32([root; 32]), + } + } + + /// A held-check backed by an explicit list of held roots. + struct Held(Vec); + impl HeldCheck for Held { + fn is_held(&self, capsule: &Capsule) -> bool { + self.0.contains(&capsule.root_hash.0[0]) + } + } + + fn lineage(roots: &[u8]) -> Lineage { + Lineage::try_new(roots.iter().map(|&r| cap(r)).collect()).unwrap() + } + + fn fetched_roots(actions: &[SyncAction]) -> Vec { + actions + .iter() + .map(|SyncAction::Fetch(c)| c.root_hash.0[0]) + .collect() + } + + /// **Proves:** a fully-held lineage produces no work. + /// **Catches:** a core that re-fetches held generations. + #[test] + fn nothing_to_do_when_all_held() { + let actions = decide(&lineage(&[1, 2, 3]), &Held(vec![1, 2, 3])); + assert!(actions.is_empty()); + } + + /// **Proves:** the tip is fetched FIRST, then history oldest → newest (clause 2 + 3). + /// **Catches:** a core that backfills history before the current tip. + #[test] + fn tip_first_then_history_oldest_to_newest() { + // Hold nothing; lineage eve=10, 20, 30, tip=40. + let actions = decide(&lineage(&[10, 20, 30, 40]), &Held(vec![])); + assert_eq!( + fetched_roots(&actions), + vec![40, 10, 20, 30], + "tip first, then oldest→newest backfill" + ); + } + + /// **Proves:** only the MISSING historical capsules are queued (backfill skips held). + /// **Catches:** a backfill that ignores the held inventory. + #[test] + fn backfills_only_missing_history() { + // Hold the tip (40) and one mid gen (20); 10 + 30 remain missing. + let actions = decide(&lineage(&[10, 20, 30, 40]), &Held(vec![40, 20])); + assert_eq!(fetched_roots(&actions), vec![10, 30]); + } + + /// **Proves:** a repeated root (a revert to a prior generation) is fetched once. + /// **Catches:** a core that double-fetches the same `.dig`. + #[test] + fn dedupes_repeated_roots() { + // eve=10, 20, then reverted back to 10 as the tip. + let actions = decide(&lineage(&[10, 20, 10]), &Held(vec![])); + assert_eq!( + fetched_roots(&actions), + vec![10, 20], + "root 10 fetched once" + ); + } +} diff --git a/crates/digstore-subscription/src/lib.rs b/crates/digstore-subscription/src/lib.rs new file mode 100644 index 0000000..9cf94c9 --- /dev/null +++ b/crates/digstore-subscription/src/lib.rs @@ -0,0 +1,52 @@ +//! The canonical DIG Network **Subscription** primitive. +//! +//! A [`Subscription`] follows ONE CHIP-0035 store singleton and keeps a local set of `.dig` +//! files in sync with that singleton's on-chain history. It is the reusable engine behind +//! "follow this store and keep it current + complete" — instantiated per store, owning its +//! watch state, its full-history tracking, and its per-tip sync/backfill (#979). +//! +//! Contract (user-normative, #979): +//! 1. **Watch** the chain for the store's singleton tip progressing (a new generation/root). +//! 2. **On tip progression**, find the `.dig` matching the new tip (capsule = `storeId:rootHash`) +//! on the network and sync it down. +//! 3. **Track the ENTIRE singleton history** and backfill every historical tip's `.dig` where +//! possible (best-effort; a tip whose `.dig` can't be found is recorded missing + retried). +//! 4. Be a **distinct, isolated behaviour** — one managed object per store. +//! +//! # Architecture — a networkless core over four seams +//! +//! The primitive owns the POLICY (which capsules to fetch, in what order, how reorgs are +//! handled) and none of the MECHANISM. The mechanism is four injected traits — +//! [`ChainWatch`], [`CapsuleFetcher`], [`HeldCheck`], [`Persistence`] — so the decision core +//! ([`decide`]) and the state machine ([`Subscription`]) are pure and unit-testable with no +//! chain and no network. Consumers (dig-node in Phase 2) provide the real seam impls. +//! +//! # Two load-bearing invariants +//! +//! - **Fail-closed** ([`reconcile_tick`]): a chain error or a store with no confirmed +//! generation does NOTHING — an unconfirmable root is never fetched or verified against. +//! - **Reorg / permanence** (§5.1, [`Subscription::observe_lineage`]): a root superseded by a +//! reorg is retained (moved to [`Subscription::orphaned`]); an already-held `.dig` is never +//! evicted. `.dig` files are permanent, on-chain-anchored artifacts. + +#![forbid(unsafe_code)] + +pub mod decide; +pub mod lineage; +pub mod reconcile; +pub mod seams; +pub mod set; +pub mod subscription; + +pub use decide::{decide, SyncAction}; +pub use lineage::Lineage; +pub use reconcile::{ + reconcile_tick, watch_interval_from_env, SkipReason, SubscriptionDeps, Subscriptions, + TickOutcome, DEFAULT_WATCH_INTERVAL_SECS, MIN_WATCH_INTERVAL_SECS, +}; +pub use seams::{CapsuleFetcher, ChainWatch, HeldCheck, LineageResult, Persistence}; +pub use set::{decode, encode, normalize_store_id, SubscriptionSet, SubscriptionsDoc}; +pub use subscription::{Subscription, TipSync}; + +// The canonical capsule identity a Subscription follows, re-exported for consumers. +pub use digstore_core::{Bytes32, Capsule}; diff --git a/crates/digstore-subscription/src/lineage.rs b/crates/digstore-subscription/src/lineage.rs new file mode 100644 index 0000000..6f81cfa --- /dev/null +++ b/crates/digstore-subscription/src/lineage.rs @@ -0,0 +1,99 @@ +//! The confirmed **lineage** of a store singleton — its ordered capsule history. +//! +//! A [`Lineage`] is the chain-confirmed sequence of capsules `(store_id, root)` a store +//! has ever been, oldest (eve) → newest (the current unspent tip). It is exactly the +//! ordered `history` that a chain walk (dig-store's `sync_datastore_with_history`, which +//! returns `StoreHistory { current, history }`) produces — modelled here as a plain, +//! chain-free value so the Subscription core stays networkless and does not depend on the +//! spend-heavy `digstore-chain` crate. A consumer adapts its `StoreHistory.history` +//! (`Vec`) straight into [`Lineage::try_new`]. +//! +//! A `Lineage` is only ever constructed from a CHAIN-CONFIRMED walk. The fail-closed +//! invariant lives at the [`ChainWatch`](crate::ChainWatch) seam: a chain error or a store +//! with no confirmed generation yields NO lineage (never an empty/partial one), so the +//! decision core is never handed an unconfirmable root to act on. + +use digstore_core::{Bytes32, Capsule}; + +/// The ordered, chain-confirmed capsule history of one store singleton, eve → tip. +/// +/// Invariants (enforced by [`try_new`](Lineage::try_new)): non-empty, and every capsule +/// shares the same `store_id`. The last element is the current tip. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Lineage { + store_id: Bytes32, + capsules: Vec, +} + +impl Lineage { + /// Build a lineage from an ordered (eve → tip) capsule list. + /// + /// Returns `Err` if the list is empty (a confirmed lineage always has at least the eve + /// generation) or if any capsule's `store_id` disagrees with the first — a mixed-store + /// walk is corrupt input and must not drive a sync. + pub fn try_new(capsules: Vec) -> Result { + let store_id = capsules + .first() + .ok_or_else(|| "lineage must have at least the eve capsule".to_string())? + .store_id; + if let Some(bad) = capsules.iter().find(|c| c.store_id != store_id) { + return Err(format!( + "lineage mixes store ids: {} vs {}", + store_id.to_hex(), + bad.store_id.to_hex() + )); + } + Ok(Lineage { store_id, capsules }) + } + + /// The store this lineage belongs to. + pub fn store_id(&self) -> Bytes32 { + self.store_id + } + + /// The full ordered capsule history, eve → tip. + pub fn capsules(&self) -> &[Capsule] { + &self.capsules + } + + /// The current tip capsule (the last, newest generation). + pub fn tip(&self) -> Capsule { + *self + .capsules + .last() + .expect("lineage is non-empty by construction") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cap(store: u8, root: u8) -> Capsule { + Capsule { + store_id: Bytes32([store; 32]), + root_hash: Bytes32([root; 32]), + } + } + + /// **Proves:** a well-formed lineage exposes its store id + ordered capsules + tip. + #[test] + fn builds_and_exposes_tip() { + let l = Lineage::try_new(vec![cap(1, 10), cap(1, 20), cap(1, 30)]).unwrap(); + assert_eq!(l.store_id(), Bytes32([1; 32])); + assert_eq!(l.capsules().len(), 3); + assert_eq!(l.tip(), cap(1, 30), "tip is the newest generation"); + } + + /// **Proves:** an empty lineage is rejected (a confirmed lineage always has the eve gen). + #[test] + fn rejects_empty() { + assert!(Lineage::try_new(vec![]).is_err()); + } + + /// **Proves:** a lineage mixing store ids is rejected as corrupt (never drives a sync). + #[test] + fn rejects_mixed_store_ids() { + assert!(Lineage::try_new(vec![cap(1, 10), cap(2, 20)]).is_err()); + } +} diff --git a/crates/digstore-subscription/src/reconcile.rs b/crates/digstore-subscription/src/reconcile.rs new file mode 100644 index 0000000..0169b22 --- /dev/null +++ b/crates/digstore-subscription/src/reconcile.rs @@ -0,0 +1,445 @@ +//! The async **reconcile tick** — the one place that ties the chain seam → the state +//! machine → the network fetch seam together, best-effort + fail-closed. +//! +//! `reconcile_tick` drives ONE subscription for one tick; [`Subscriptions`] drives a whole +//! set. The host owns the loop/spawn (the crate stays runtime-agnostic — see +//! [`watch_interval_from_env`] for the interval policy the host applies). + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use digstore_core::Bytes32; + +use crate::decide::SyncAction; +use crate::seams::{CapsuleFetcher, ChainWatch, HeldCheck}; +use crate::set::SubscriptionSet; +use crate::subscription::Subscription; + +/// Default interval between chain-watch polls. Modest by design — a new generation confirms +/// on-chain in tens of seconds to minutes, so a ~30 s poll detects it promptly without +/// hammering the chain source. Overridable via `DIG_WATCH_INTERVAL` (seconds). +pub const DEFAULT_WATCH_INTERVAL_SECS: u64 = 30; + +/// The floor on the configured watch interval (seconds) — so a mis-set env var can't turn +/// the loop into a chain-source flood. +pub const MIN_WATCH_INTERVAL_SECS: u64 = 1; + +/// Resolve the watch-poll interval from `DIG_WATCH_INTERVAL` (seconds), floored at +/// [`MIN_WATCH_INTERVAL_SECS`] and defaulting to [`DEFAULT_WATCH_INTERVAL_SECS`]. +pub fn watch_interval_from_env() -> Duration { + let secs = std::env::var("DIG_WATCH_INTERVAL") + .ok() + .and_then(|s| s.trim().parse::().ok()) + .filter(|&n| n > 0) + .unwrap_or(DEFAULT_WATCH_INTERVAL_SECS) + .max(MIN_WATCH_INTERVAL_SECS); + Duration::from_secs(secs) +} + +/// Why a reconcile tick did no fetching for a store — the fail-closed no-ops. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SkipReason { + /// The chain read failed — never fetch against an unconfirmable root. + ChainError, + /// The store has no confirmed on-chain generation yet. + NoConfirmedGeneration, +} + +/// The outcome of reconciling one store this tick. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TickOutcome { + /// No work — the fail-closed no-op (chain error or no confirmed generation). + Skipped(SkipReason), + /// The lineage was confirmed and reconciled: `attempted` fetches, `fetched` succeeded. + Synced { attempted: usize, fetched: usize }, +} + +/// The seams a reconcile tick drives. +#[derive(Clone)] +pub struct SubscriptionDeps { + /// The confirmed-lineage source (fail-closed). + pub chain: Arc, + /// The network `.dig` fetch actuator. + pub fetcher: Arc, + /// The held-module check. + pub held: Arc, +} + +/// Reconcile ONE subscription for one tick: resolve its confirmed lineage, fold it into the +/// subscription's state (tip progression + reorg), then drive the ordered fetch worklist. +/// +/// FAIL-CLOSED: a chain `Err` or a no-generation `Ok(None)` returns a [`TickOutcome::Skipped`] +/// with NO fetch call — an unconfirmable root is never acted on. +pub async fn reconcile_tick(sub: &mut Subscription, deps: &SubscriptionDeps) -> TickOutcome { + let lineage = match deps.chain.lineage(&sub.store_id()).await { + Err(_) => return TickOutcome::Skipped(SkipReason::ChainError), + Ok(None) => return TickOutcome::Skipped(SkipReason::NoConfirmedGeneration), + Ok(Some(lineage)) => lineage, + }; + + sub.observe_lineage(&lineage, deps.held.as_ref()); + + let mut attempted = 0; + let mut fetched = 0; + for SyncAction::Fetch(capsule) in sub.plan(deps.held.as_ref()) { + attempted += 1; + sub.mark_pending(&capsule); + let result = deps.fetcher.fetch(capsule).await; + match &result { + Ok(()) => { + fetched += 1; + tracing::info!(capsule = %capsule.canonical(), "subscription: synced .dig"); + } + Err(e) => tracing::warn!( + capsule = %capsule.canonical(), + error = %e, + "subscription: fetch failed; will retry next tick" + ), + } + sub.record_fetch_result(&capsule, result); + } + TickOutcome::Synced { attempted, fetched } +} + +/// A live set of [`Subscription`]s keyed by store id, driven together each tick. +/// +/// Membership is synced from a [`SubscriptionSet`] (the persisted worklist): a newly +/// subscribed store gets a fresh [`Subscription`]; an unsubscribed store is dropped from the +/// live set (its retained `.dig` files are the consumer's business — this only stops watching). +#[derive(Debug, Default)] +pub struct Subscriptions { + subs: BTreeMap, +} + +impl Subscriptions { + /// An empty live set. + pub fn new() -> Self { + Subscriptions::default() + } + + /// Reconcile membership against the persisted set: add subscriptions for newly-listed + /// stores, drop those no longer listed. Existing subscriptions (and their tracked state) + /// are preserved. + pub fn sync_membership(&mut self, set: &SubscriptionSet) { + let listed: Vec = set.stores().to_vec(); + self.subs.retain(|k, _| listed.contains(k)); + for store_hex in listed { + if let Some(store_id) = parse_store_id(&store_hex) { + self.subs + .entry(store_hex) + .or_insert_with(|| Subscription::new(store_id)); + } + } + } + + /// The live subscriptions (store hex → subscription), for status reporting. + pub fn iter(&self) -> impl Iterator { + self.subs.iter() + } + + /// The subscription for a store id, if watched. + pub fn get(&self, store_hex: &str) -> Option<&Subscription> { + self.subs.get(store_hex) + } + + /// How many stores are watched. + pub fn len(&self) -> usize { + self.subs.len() + } + + /// Whether no stores are watched. + pub fn is_empty(&self) -> bool { + self.subs.is_empty() + } + + /// Reconcile every watched subscription once, returning each store's outcome. + pub async fn reconcile_all(&mut self, deps: &SubscriptionDeps) -> Vec<(Bytes32, TickOutcome)> { + let mut out = Vec::with_capacity(self.subs.len()); + for sub in self.subs.values_mut() { + let outcome = reconcile_tick(sub, deps).await; + out.push((sub.store_id(), outcome)); + } + out + } +} + +/// Parse a 64-hex store id into a [`Bytes32`]. Ids in the set are already sanitized, but a +/// hand-edited file is never trusted. +fn parse_store_id(store_hex: &str) -> Option { + Bytes32::from_hex(store_hex) + .ok() + .filter(|_| store_hex.len() == 64) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::lineage::Lineage; + use crate::seams::LineageResult; + use async_trait::async_trait; + use digstore_core::Capsule; + use std::sync::Mutex; + + fn cap(store: u8, root: u8) -> Capsule { + Capsule { + store_id: Bytes32([store; 32]), + root_hash: Bytes32([root; 32]), + } + } + + /// A chain source scripted per store id. + struct MockChain(BTreeMap); + #[async_trait] + impl ChainWatch for MockChain { + async fn lineage(&self, store_id: &Bytes32) -> LineageResult { + match self.0.get(&store_id.to_hex()) { + Some(Ok(Some(l))) => Ok(Some(l.clone())), + Some(Ok(None)) => Ok(None), + Some(Err(e)) => Err(e.clone()), + None => Ok(None), + } + } + } + + /// A held-check over explicit (store,root) first-byte pairs. + struct Held(Vec<(u8, u8)>); + impl HeldCheck for Held { + fn is_held(&self, c: &Capsule) -> bool { + self.0.contains(&(c.store_id.0[0], c.root_hash.0[0])) + } + } + + /// A fetcher recording every capsule it was asked to pull, with a scripted result. + struct RecordingFetcher { + calls: Mutex>, + fail: bool, + } + #[async_trait] + impl CapsuleFetcher for RecordingFetcher { + async fn fetch(&self, capsule: Capsule) -> Result<(), String> { + self.calls.lock().unwrap().push(capsule); + if self.fail { + Err("pull failed".into()) + } else { + Ok(()) + } + } + } + + fn deps(chain: MockChain, held: Held, fetcher: Arc) -> SubscriptionDeps { + SubscriptionDeps { + chain: Arc::new(chain), + fetcher, + held: Arc::new(held), + } + } + + /// **Proves the fail-closed invariant:** a chain error triggers NO fetch and reports the + /// skip reason. **Catches:** a tick that pulls against an unconfirmable root. + #[tokio::test] + async fn chain_error_fetches_nothing() { + let mut chain = BTreeMap::new(); + chain.insert(Bytes32([1; 32]).to_hex(), Err("coinset 503".into())); + let fetcher = Arc::new(RecordingFetcher { + calls: Mutex::new(vec![]), + fail: false, + }); + let d = deps(MockChain(chain), Held(vec![]), fetcher.clone()); + let mut sub = Subscription::new(Bytes32([1; 32])); + + let outcome = reconcile_tick(&mut sub, &d).await; + assert_eq!(outcome, TickOutcome::Skipped(SkipReason::ChainError)); + assert!( + fetcher.calls.lock().unwrap().is_empty(), + "no fetch on chain error" + ); + } + + /// **Proves the fail-closed invariant:** a store with no confirmed generation fetches + /// nothing. + #[tokio::test] + async fn no_generation_fetches_nothing() { + let mut chain = BTreeMap::new(); + chain.insert(Bytes32([1; 32]).to_hex(), Ok(None)); + let fetcher = Arc::new(RecordingFetcher { + calls: Mutex::new(vec![]), + fail: false, + }); + let d = deps(MockChain(chain), Held(vec![]), fetcher.clone()); + let mut sub = Subscription::new(Bytes32([1; 32])); + + let outcome = reconcile_tick(&mut sub, &d).await; + assert_eq!( + outcome, + TickOutcome::Skipped(SkipReason::NoConfirmedGeneration) + ); + assert!(fetcher.calls.lock().unwrap().is_empty()); + } + + /// **Proves full-history backfill:** a confirmed 3-generation lineage with nothing held + /// fetches ALL three `.dig` (tip first), and reports the summary. + /// **Catches:** a watcher that only pulls the latest tip (the old single-tip behaviour). + #[tokio::test] + async fn confirmed_lineage_backfills_whole_history() { + let lineage = Lineage::try_new(vec![cap(1, 10), cap(1, 20), cap(1, 30)]).unwrap(); + let mut chain = BTreeMap::new(); + chain.insert(Bytes32([1; 32]).to_hex(), Ok(Some(lineage))); + let fetcher = Arc::new(RecordingFetcher { + calls: Mutex::new(vec![]), + fail: false, + }); + let d = deps(MockChain(chain), Held(vec![]), fetcher.clone()); + let mut sub = Subscription::new(Bytes32([1; 32])); + + let outcome = reconcile_tick(&mut sub, &d).await; + assert_eq!( + outcome, + TickOutcome::Synced { + attempted: 3, + fetched: 3 + } + ); + assert_eq!( + *fetcher.calls.lock().unwrap(), + vec![cap(1, 30), cap(1, 10), cap(1, 20)], + "tip first, then oldest→newest backfill" + ); + } + + /// **Proves interruption-retry accrual through the REAL reconcile flow:** across three + /// failing ticks — each of which calls `mark_pending` immediately before + /// `record_fetch_result`, exactly as production does — the capsule's recorded attempt count + /// accrues 1 → 2 → 3 (it does NOT reset to 1 each tick), then a success flips it to `Held`. + /// **Catches:** the masked bug where `mark_pending` set `Pending` before the record, so the + /// prior `Failed` count was lost and `attempts` reset to 1 every tick. + #[tokio::test] + async fn failed_fetch_retry_attempts_accrue_across_real_ticks() { + use crate::subscription::TipSync; + let root = Bytes32([10; 32]); + let lineage = Lineage::try_new(vec![cap(1, 10)]).unwrap(); + let mut chain = BTreeMap::new(); + chain.insert(Bytes32([1; 32]).to_hex(), Ok(Some(lineage))); + let fetcher = Arc::new(RecordingFetcher { + calls: Mutex::new(vec![]), + fail: true, + }); + let d = deps(MockChain(chain), Held(vec![]), fetcher.clone()); + let mut sub = Subscription::new(Bytes32([1; 32])); + + for expected in 1..=3u32 { + let tick = reconcile_tick(&mut sub, &d).await; + assert_eq!( + tick, + TickOutcome::Synced { + attempted: 1, + fetched: 0 + } + ); + match sub.tip_status(&root) { + Some(TipSync::Failed { attempts, .. }) => { + assert_eq!(*attempts, expected, "attempts accrue across ticks") + } + other => panic!("expected Failed after failing tick, got {other:?}"), + } + } + assert_eq!(fetcher.calls.lock().unwrap().len(), 3, "retried each tick"); + + // A now-succeeding fetch flips the record to Held (attempts no longer relevant). + let ok_fetcher = Arc::new(RecordingFetcher { + calls: Mutex::new(vec![]), + fail: false, + }); + let ok_deps = SubscriptionDeps { + chain: d.chain.clone(), + fetcher: ok_fetcher, + held: d.held.clone(), + }; + reconcile_tick(&mut sub, &ok_deps).await; + assert_eq!(sub.tip_status(&root), Some(&TipSync::Held)); + } + + /// **Proves:** membership syncs from the persisted set (add + drop), preserving existing + /// subscription state. + #[test] + fn membership_add_and_drop() { + let mut subs = Subscriptions::new(); + let mut set = SubscriptionSet::new(); + set.add(&Bytes32([1; 32]).to_hex()).unwrap(); + set.add(&Bytes32([2; 32]).to_hex()).unwrap(); + subs.sync_membership(&set); + assert_eq!(subs.len(), 2); + + // Unsubscribe store 1. + set.remove(&Bytes32([1; 32]).to_hex()).unwrap(); + subs.sync_membership(&set); + assert_eq!(subs.len(), 1); + assert!(subs.get(&Bytes32([2; 32]).to_hex()).is_some()); + } + + /// **Proves:** the `Subscriptions` driver reconciles a whole watched set in one pass, + /// returning a per-store outcome (fail-closed skip AND confirmed backfill in the same tick). + #[tokio::test] + async fn reconcile_all_drives_the_whole_set() { + let lineage = Lineage::try_new(vec![cap(1, 10), cap(1, 20)]).unwrap(); + let mut chain = BTreeMap::new(); + chain.insert(Bytes32([1; 32]).to_hex(), Ok(Some(lineage))); + chain.insert(Bytes32([2; 32]).to_hex(), Err("chain down".into())); + let fetcher = Arc::new(RecordingFetcher { + calls: Mutex::new(vec![]), + fail: false, + }); + let d = deps(MockChain(chain), Held(vec![]), fetcher.clone()); + + let mut subs = Subscriptions::new(); + assert!(subs.is_empty()); + let mut set = SubscriptionSet::new(); + set.add(&Bytes32([1; 32]).to_hex()).unwrap(); + set.add(&Bytes32([2; 32]).to_hex()).unwrap(); + subs.sync_membership(&set); + + let outcomes = subs.reconcile_all(&d).await; + assert_eq!(outcomes.len(), 2); + assert_eq!( + outcomes[0], + ( + Bytes32([1; 32]), + TickOutcome::Synced { + attempted: 2, + fetched: 2 + } + ), + "store 1: confirmed lineage backfilled" + ); + assert_eq!( + outcomes[1], + ( + Bytes32([2; 32]), + TickOutcome::Skipped(SkipReason::ChainError) + ), + "store 2: fail-closed skip" + ); + // Status is queryable via the live set + iter. + assert_eq!(subs.iter().count(), 2); + let s1 = subs.get(&Bytes32([1; 32]).to_hex()).unwrap(); + assert_eq!(s1.current_tip(), Some(cap(1, 20))); + } + + /// **Proves:** the interval env var is parsed + floored (a bogus/zero value never floods). + #[test] + fn interval_floor_and_default() { + std::env::remove_var("DIG_WATCH_INTERVAL"); + assert_eq!( + watch_interval_from_env(), + Duration::from_secs(DEFAULT_WATCH_INTERVAL_SECS) + ); + std::env::set_var("DIG_WATCH_INTERVAL", "0"); + assert_eq!( + watch_interval_from_env(), + Duration::from_secs(DEFAULT_WATCH_INTERVAL_SECS) + ); + std::env::set_var("DIG_WATCH_INTERVAL", "5"); + assert_eq!(watch_interval_from_env(), Duration::from_secs(5)); + std::env::remove_var("DIG_WATCH_INTERVAL"); + } +} diff --git a/crates/digstore-subscription/src/seams.rs b/crates/digstore-subscription/src/seams.rs new file mode 100644 index 0000000..4a99a66 --- /dev/null +++ b/crates/digstore-subscription/src/seams.rs @@ -0,0 +1,67 @@ +//! The four injected **seams** the Subscription core drives. +//! +//! The Subscription primitive owns the *policy* (which capsules to fetch, in what order, +//! how reorgs are handled) but none of the *mechanism* (chain reads, network `.dig` +//! fetches, held-module checks, disk I/O). Those are traits, provided by the consumer — +//! the same injected-seam pattern dig-node already uses (`AnchoredRootResolver`, +//! `GapFiller`, `HeldCheck`). This keeps the core networkless, runtime-agnostic, and +//! unit-testable with deterministic mocks. + +use async_trait::async_trait; +use digstore_core::{Bytes32, Capsule}; + +use crate::lineage::Lineage; +use crate::set::SubscriptionSet; + +/// The chain's answer for one store this tick. +/// +/// The FAIL-CLOSED contract lives here: an implementation MUST return `Err` when the chain +/// read failed and `Ok(None)` when the store has no confirmed generation. Only a +/// chain-CONFIRMED walk yields `Ok(Some(lineage))`. The reconcile core never fetches on +/// `Err`/`Ok(None)`, so an unconfirmable root is never acted on. +pub type LineageResult = Result, String>; + +/// Resolve a store's confirmed capsule lineage (eve → tip) from the chain. +/// +/// This generalizes dig-node's `AnchoredRootResolver` from "the single confirmed tip root" +/// to "the full confirmed lineage" — the extra history is what enables full-history +/// backfill. Production impls wrap a coinset walk (dig-store's `sync_datastore_with_history`). +#[async_trait] +pub trait ChainWatch: Send + Sync { + /// Resolve the confirmed lineage for `store_id`, or the fail-closed no-op signals. + async fn lineage(&self, store_id: &Bytes32) -> LineageResult; +} + +/// Find + sync + verify + land the `.dig` for one capsule on the network. +/// +/// This IS the "search the network for the `.dig` matching this tip and sync it down" +/// leg (#979 clauses 2 + 3), generalized from dig-node's `GapFiller` to ANY capsule in the +/// lineage rather than only the latest tip. It MUST verify the fetched module against the +/// chain-anchored root before landing it, and be idempotent (a call for an already-held +/// capsule is a cheap success). +#[async_trait] +pub trait CapsuleFetcher: Send + Sync { + /// Pull + verify + cache the `.dig` for `capsule`. `Ok(())` on a verified, landed + /// module; `Err` describes the failure (the loop records it + retries next tick). + async fn fetch(&self, capsule: Capsule) -> Result<(), String>; +} + +/// Whether the `.dig` for a capsule is already held locally. +/// +/// A thin seam over the consumer's inventory (dig-node's `module_exists`) so the "is this +/// generation missing?" check is injectable in tests. +pub trait HeldCheck: Send + Sync { + /// `true` iff the `.dig` for `capsule` is present locally. + fn is_held(&self, capsule: &Capsule) -> bool; +} + +/// Load + save the persisted [`SubscriptionSet`]. +/// +/// The set's add/remove/codec policy is pure (in [`crate::set`]); this seam is the thin +/// disk (or other durable store) I/O the consumer owns, keeping this crate filesystem-free. +pub trait Persistence: Send + Sync { + /// Load the persisted set. A missing/corrupt store MUST yield an empty set, never an error. + fn load(&self) -> SubscriptionSet; + /// Persist the set durably (atomic write recommended). + fn save(&self, set: &SubscriptionSet) -> Result<(), String>; +} diff --git a/crates/digstore-subscription/src/set.rs b/crates/digstore-subscription/src/set.rs new file mode 100644 index 0000000..edf2a20 --- /dev/null +++ b/crates/digstore-subscription/src/set.rs @@ -0,0 +1,280 @@ +//! The persisted **subscription set** — a node/client's own set of store ids it intends +//! to follow (watch + sync + backfill). +//! +//! This is the canonical, extracted form of the dig-node `subscription.rs` set logic. +//! [`SubscriptionSet`] is a plain, order-preserving, de-duplicated set of 64-hex store +//! ids with all the add/remove/list policy; it is pure and directly unit-testable. The +//! JSON codec ([`decode`]/[`encode`]) round-trips it through a schema-versioned document. +//! Actual disk I/O is NOT here — it lives behind the [`Persistence`](crate::Persistence) +//! seam in the consumer, so this crate stays networkless and filesystem-free. +//! +//! The subscription set is DISTINCT from the durable capsule inventory (the `.dig` +//! modules a client holds). The inventory answers "what do I currently hold?"; the +//! subscription set answers "what do I intend to keep current?" — it is the worklist the +//! watch loop iterates. + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// The persisted subscription document: a schema-versioned list of subscribed store ids. +/// +/// Versioned so a future additive field (per-store options — pin depth, priority) is a +/// backwards-compatible read: an older reader ignores unknown fields, a newer reader +/// defaults them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SubscriptionsDoc { + /// Schema version (currently `1`). Additive-only; a bump never removes/repurposes a field. + pub version: u32, + /// The subscribed store ids, lower-case 64-hex, in insertion order (stable for the UI + tests). + pub stores: Vec, +} + +impl Default for SubscriptionsDoc { + fn default() -> Self { + SubscriptionsDoc { + version: 1, + stores: Vec::new(), + } + } +} + +/// An in-memory, order-preserving, de-duplicated set of subscribed store ids (lower-case 64-hex). +/// +/// All add/remove/list policy lives here so it is pure + directly unit-testable. Store ids +/// are normalized to lower-case on insert so the same id in mixed case is one subscription, +/// never two. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct SubscriptionSet { + stores: Vec, +} + +/// Whether `s` is a canonical 64-char hex store id. +fn is_hex64(s: &str) -> bool { + s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit()) +} + +/// The canonical form of a store id as stored in the set: trimmed + lower-cased. This is +/// the SINGLE normalization [`SubscriptionSet::add`]/[`remove`](SubscriptionSet::remove)/ +/// [`contains`](SubscriptionSet::contains) apply, exposed so a caller (an RPC echo) can +/// report EXACTLY the id that was persisted — never the raw input. +pub fn normalize_store_id(store_id: &str) -> String { + store_id.trim().to_ascii_lowercase() +} + +impl SubscriptionSet { + /// An empty set. + pub fn new() -> Self { + SubscriptionSet::default() + } + + /// Build a set from a persisted document, normalizing (lower-case) + de-duplicating + + /// dropping any malformed (non-64-hex) entry so a hand-edited/corrupt file can never + /// inject a bad id. + pub fn from_doc(doc: &SubscriptionsDoc) -> Self { + let mut set = SubscriptionSet::new(); + for s in &doc.stores { + let _ = set.add(s); + } + set + } + + /// Serialize to a persisted document (schema version 1). + pub fn to_doc(&self) -> SubscriptionsDoc { + SubscriptionsDoc { + version: 1, + stores: self.stores.clone(), + } + } + + /// Subscribe to `store_id`. Returns `Ok(true)` if it was newly added, `Ok(false)` if it + /// was already subscribed (idempotent), or `Err` if `store_id` is not a valid 64-hex id. + /// The id is normalized to lower-case so mixed-case duplicates collapse to one. + pub fn add(&mut self, store_id: &str) -> Result { + let id = normalize_store_id(store_id); + if !is_hex64(&id) { + return Err(format!("store_id must be 64-hex, got {store_id:?}")); + } + if self.stores.iter().any(|s| s == &id) { + return Ok(false); + } + self.stores.push(id); + Ok(true) + } + + /// Unsubscribe from `store_id`. Returns `Ok(true)` if it was present + removed, + /// `Ok(false)` if it was not subscribed (idempotent), or `Err` on a malformed id. + pub fn remove(&mut self, store_id: &str) -> Result { + let id = normalize_store_id(store_id); + if !is_hex64(&id) { + return Err(format!("store_id must be 64-hex, got {store_id:?}")); + } + let before = self.stores.len(); + self.stores.retain(|s| s != &id); + Ok(self.stores.len() != before) + } + + /// Whether `store_id` (case-insensitive) is subscribed. + pub fn contains(&self, store_id: &str) -> bool { + let id = normalize_store_id(store_id); + self.stores.iter().any(|s| s == &id) + } + + /// The subscribed store ids in insertion order (lower-case 64-hex). + pub fn stores(&self) -> &[String] { + &self.stores + } + + /// How many stores are subscribed. + pub fn len(&self) -> usize { + self.stores.len() + } + + /// Whether no stores are subscribed. + pub fn is_empty(&self) -> bool { + self.stores.is_empty() + } +} + +/// Encode a subscription set to the pretty-printed JSON bytes a [`Persistence`](crate::Persistence) +/// impl writes to disk. Pure (no I/O) so the on-disk schema is directly assertable in a test. +pub fn encode(set: &SubscriptionSet) -> Vec { + serde_json::to_vec_pretty(&set.to_doc()).unwrap_or_default() +} + +/// Decode a subscription set from persisted JSON `text`. An empty or unparseable input is an +/// EMPTY set (never an error) — a client simply has no subscriptions yet. A legacy bare +/// `{ "stores": [...] }` shape (no `version`) is tolerated. Malformed entries inside a valid +/// document are dropped by [`SubscriptionSet::from_doc`]. +pub fn decode(text: &str) -> SubscriptionSet { + match serde_json::from_str::(text) { + Ok(doc) => SubscriptionSet::from_doc(&doc), + Err(_) => match serde_json::from_str::(text) { + Ok(v) => { + let stores = v + .get("stores") + .and_then(Value::as_array) + .map(|a| { + a.iter() + .filter_map(|s| s.as_str().map(str::to_string)) + .collect::>() + }) + .unwrap_or_default(); + SubscriptionSet::from_doc(&SubscriptionsDoc { version: 1, stores }) + } + Err(_) => SubscriptionSet::new(), + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn id(n: u8) -> String { + hex::encode([n; 32]) + } + + /// **Proves:** add is idempotent + insertion-ordered; a second add of the same id is a + /// no-op (`Ok(false)`) and does not duplicate. + #[test] + fn add_is_idempotent_and_ordered() { + let mut set = SubscriptionSet::new(); + assert_eq!(set.add(&id(1)), Ok(true), "first add is new"); + assert_eq!(set.add(&id(2)), Ok(true)); + assert_eq!(set.add(&id(1)), Ok(false), "re-add is a no-op"); + assert_eq!(set.stores(), &[id(1), id(2)], "insertion order preserved"); + assert_eq!(set.len(), 2); + } + + /// **Proves:** mixed-case ids normalize to one lower-case subscription (not two). + #[test] + fn add_normalizes_case() { + let mut set = SubscriptionSet::new(); + let lower = id(0xab); + let upper = lower.to_ascii_uppercase(); + assert_eq!(set.add(&upper), Ok(true)); + assert_eq!(set.add(&lower), Ok(false), "same id, different case"); + assert_eq!(set.stores(), std::slice::from_ref(&lower)); + assert!(set.contains(&upper), "contains is case-insensitive"); + } + + /// **Proves:** remove is idempotent (`Ok(false)` when absent) and unsubscribes when present. + #[test] + fn remove_is_idempotent() { + let mut set = SubscriptionSet::new(); + set.add(&id(1)).unwrap(); + assert_eq!(set.remove(&id(9)), Ok(false), "not subscribed"); + assert_eq!(set.remove(&id(1)), Ok(true), "removed"); + assert!(set.is_empty()); + assert_eq!(set.remove(&id(1)), Ok(false), "already gone"); + } + + /// **Proves:** a malformed store id is rejected on add AND remove (never persisted). + #[test] + fn rejects_malformed_ids() { + let mut set = SubscriptionSet::new(); + assert!(set.add("not-hex").is_err()); + assert!(set.add("abcd").is_err(), "too short"); + assert!(set.add(&"zz".repeat(32)).is_err(), "non-hex chars"); + assert!(set.remove("not-hex").is_err()); + assert!(set.is_empty()); + } + + /// **Proves:** a persisted document round-trips through [`SubscriptionSet`] byte-stably, + /// and [`from_doc`] drops malformed entries from a corrupt file. + #[test] + fn doc_roundtrip_and_sanitization() { + let mut set = SubscriptionSet::new(); + set.add(&id(1)).unwrap(); + set.add(&id(2)).unwrap(); + let doc = set.to_doc(); + assert_eq!(doc.version, 1); + assert_eq!(SubscriptionSet::from_doc(&doc), set, "clean round-trip"); + + let corrupt = SubscriptionsDoc { + version: 1, + stores: vec!["garbage".to_string(), id(3)], + }; + let cleaned = SubscriptionSet::from_doc(&corrupt); + assert_eq!(cleaned.stores(), &[id(3)], "malformed entry dropped"); + } + + /// **Proves:** decode of empty/garbage input is an empty set (never an error), and + /// encode→decode round-trips through the real JSON. + #[test] + fn codec_roundtrip_and_empty() { + assert!(decode("").is_empty(), "empty text → empty set"); + assert!(decode("}{not json").is_empty(), "garbage → empty set"); + + let mut set = SubscriptionSet::new(); + set.add(&id(7)).unwrap(); + set.add(&id(8)).unwrap(); + let bytes = encode(&set); + let decoded = decode(std::str::from_utf8(&bytes).unwrap()); + assert_eq!(decoded, set, "encode → decode round-trips"); + } + + /// **Proves:** `normalize_store_id` trims AND lower-cases, matching what add/remove persist. + #[test] + fn normalize_matches_stored_form() { + let raw = format!(" {} ", id(0xcd).to_ascii_uppercase()); + let norm = normalize_store_id(&raw); + assert_eq!(norm, id(0xcd), "trimmed + lower-cased"); + let mut set = SubscriptionSet::new(); + set.add(&raw).unwrap(); + assert_eq!(set.stores(), std::slice::from_ref(&norm)); + } + + /// **Proves:** decode tolerates a legacy bare `{ "stores": [...] }` shape (no `version`), + /// dropping bad ids. + #[test] + fn decode_tolerates_legacy_shape() { + let legacy = serde_json::json!({ "stores": [id(4), "bad", id(5)] }).to_string(); + let loaded = decode(&legacy); + assert_eq!( + loaded.stores(), + &[id(4), id(5)], + "legacy shape, bad id dropped" + ); + } +} diff --git a/crates/digstore-subscription/src/subscription.rs b/crates/digstore-subscription/src/subscription.rs new file mode 100644 index 0000000..67acfb9 --- /dev/null +++ b/crates/digstore-subscription/src/subscription.rs @@ -0,0 +1,349 @@ +//! The [`Subscription`] — a per-store managed object owning the history-tracking state +//! and per-capsule sync status (#979 clause 4: a distinct, isolated behaviour). +//! +//! It tracks the store's last-known confirmed lineage, the [`TipSync`] status of each +//! capsule, and — critically — the capsules that have been SUPERSEDED by a reorg. The +//! reorg/permanence invariant (§5.1) is structural: [`observe_lineage`](Subscription::observe_lineage) +//! never removes a status entry and this type exposes NO eviction API, so a superseded +//! root that was already held stays recorded as held and its `.dig` is never dropped. + +use std::collections::BTreeMap; + +use digstore_core::{Bytes32, Capsule}; + +use crate::decide::{decide, SyncAction}; +use crate::lineage::Lineage; +use crate::seams::HeldCheck; + +/// The sync state of one capsule (one historical or current tip) in a subscription. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TipSync { + /// The `.dig` for this capsule is held locally. + Held, + /// The `.dig` is not held and no fetch has been attempted yet — a backfill worklist item. + Missing, + /// A fetch is in flight this tick. Carries the count of failed attempts SO FAR (before + /// this in-flight one) so the retry count accrues across ticks even though the + /// [`Pending`](TipSync::Pending) transition sits between [`Failed`](TipSync::Failed) + /// records in the real reconcile flow. + Pending { attempts: u32 }, + /// The last fetch failed; retried on a later tick. Carries the attempt count + last error. + Failed { attempts: u32, last_error: String }, +} + +/// A subscription to one store: its tracked lineage, per-capsule sync status, and the +/// capsules superseded by a reorg (retained, never evicted). +#[derive(Debug, Clone)] +pub struct Subscription { + store_id: Bytes32, + /// The last-observed confirmed lineage, eve → tip. + history: Vec, + /// Per-capsule sync status, keyed by root hex (root hashes are content-addressed). + status: BTreeMap, + /// Capsules that were in a previously-observed lineage but dropped out of the current + /// canonical one (a reorg superseded them). Retained so an already-held superseded + /// `.dig` is never forgotten — §5.1 permanence. + orphaned: Vec, +} + +impl Subscription { + /// A fresh subscription with no observed history yet. + pub fn new(store_id: Bytes32) -> Self { + Subscription { + store_id, + history: Vec::new(), + status: BTreeMap::new(), + orphaned: Vec::new(), + } + } + + /// The store this subscription follows. + pub fn store_id(&self) -> Bytes32 { + self.store_id + } + + /// The current tip capsule (the newest observed generation), or `None` before the first + /// confirmed lineage is observed. + pub fn current_tip(&self) -> Option { + self.history.last().copied() + } + + /// The full ordered lineage view, eve → tip (#979 clause 3). + pub fn history(&self) -> &[Capsule] { + &self.history + } + + /// The capsules superseded by a reorg — dropped from the canonical lineage but retained + /// (their `.dig`, if held, is permanent). + pub fn orphaned(&self) -> &[Capsule] { + &self.orphaned + } + + /// The sync status of the capsule with this root, if tracked. + pub fn tip_status(&self, root: &Bytes32) -> Option<&TipSync> { + self.status.get(&root.to_hex()) + } + + /// The backfill worklist: capsules in the current lineage whose `.dig` is not held + /// (status [`TipSync::Missing`] or [`TipSync::Failed`]), tip-first then oldest → newest. + pub fn missing(&self) -> Vec { + let tip = self.current_tip(); + let is_missing = |c: &Capsule| { + matches!( + self.status.get(&c.root_hash.to_hex()), + Some(TipSync::Missing) | Some(TipSync::Failed { .. }) + ) + }; + let mut out: Vec = Vec::new(); + if let Some(t) = tip { + if is_missing(&t) { + out.push(t); + } + } + for c in &self.history { + if Some(*c) != tip && is_missing(c) && !out.iter().any(|q| q.root_hash == c.root_hash) { + out.push(*c); + } + } + out + } + + /// Observe a freshly-confirmed lineage: detect tip progression AND reorg, and refresh + /// per-capsule status against the held inventory. + /// + /// Reorg/permanence (§5.1): any capsule in the PRIOR lineage that is absent from the new + /// one is moved to [`orphaned`](Subscription::orphaned) — its status entry is kept (a held + /// superseded root stays [`TipSync::Held`]), and nothing is ever deleted. New capsules are + /// seeded [`TipSync::Held`] or [`TipSync::Missing`] from the held check; an entry already + /// recorded as [`TipSync::Failed`] is left intact (so its retry count survives) unless the + /// `.dig` is now held. + /// + /// Rejects a lineage for a DIFFERENT store (a buggy [`ChainWatch`](crate::ChainWatch) + /// returning another store's walk): a foreign lineage is ignored rather than repointing + /// this subscription's history — the subscription is bound to exactly one store id. + pub fn observe_lineage(&mut self, lineage: &Lineage, held: &dyn HeldCheck) { + if lineage.store_id() != self.store_id { + tracing::warn!( + subscription = %self.store_id.to_hex(), + lineage = %lineage.store_id().to_hex(), + "observe_lineage: ignoring a foreign store's lineage (ChainWatch bug)" + ); + return; + } + let new_roots: Vec = lineage + .capsules() + .iter() + .map(|c| c.root_hash.to_hex()) + .collect(); + + // Capsules dropped from the canonical lineage by a reorg → orphan them (retain). + for prior in &self.history { + let still_canonical = new_roots.contains(&prior.root_hash.to_hex()); + let already_orphan = self.orphaned.iter().any(|o| o.root_hash == prior.root_hash); + if !still_canonical && !already_orphan { + self.orphaned.push(*prior); + } + } + + // Refresh status for every capsule in the new lineage. + for capsule in lineage.capsules() { + let key = capsule.root_hash.to_hex(); + if held.is_held(capsule) { + self.status.insert(key, TipSync::Held); + } else { + match self.status.get(&key) { + // Preserve an in-flight retry record; only downgrade to Missing if unknown. + Some(TipSync::Failed { .. }) | Some(TipSync::Pending { .. }) => {} + _ => { + self.status.insert(key, TipSync::Missing); + } + } + } + } + + self.history = lineage.capsules().to_vec(); + } + + /// Mark the capsule's fetch as in-flight this tick, carrying the count of prior FAILED + /// attempts into the [`TipSync::Pending`] record so the retry count accrues across ticks + /// (the real reconcile flow marks pending immediately before recording the result, so the + /// prior `Failed` count must survive the transition). + pub fn mark_pending(&mut self, capsule: &Capsule) { + let attempts = self.attempts_so_far(&capsule.root_hash.to_hex()); + self.status + .insert(capsule.root_hash.to_hex(), TipSync::Pending { attempts }); + } + + /// The count of failed attempts recorded so far for a root, across a `Failed` OR an + /// in-flight `Pending` record (0 for a first attempt / unknown root). + fn attempts_so_far(&self, key: &str) -> u32 { + match self.status.get(key) { + Some(TipSync::Failed { attempts, .. }) => *attempts, + Some(TipSync::Pending { attempts }) => *attempts, + _ => 0, + } + } + + /// Record the outcome of a fetch: [`TipSync::Held`] on success, else [`TipSync::Failed`] + /// with an incremented attempt count (so retries accumulate across ticks — the prior count + /// is read from either the in-flight `Pending` record or a persisted `Failed` record). + pub fn record_fetch_result(&mut self, capsule: &Capsule, result: Result<(), String>) { + let key = capsule.root_hash.to_hex(); + match result { + Ok(()) => { + self.status.insert(key, TipSync::Held); + } + Err(e) => { + let attempts = self.attempts_so_far(&key) + 1; + self.status.insert( + key, + TipSync::Failed { + attempts, + last_error: e, + }, + ); + } + } + } + + /// The ordered fetch worklist for the current lineage (delegates to the pure + /// [`decide`](crate::decide) core). Empty before any lineage is observed. + pub fn plan(&self, held: &dyn HeldCheck) -> Vec { + match Lineage::try_new(self.history.clone()) { + Ok(lineage) => decide(&lineage, held), + Err(_) => Vec::new(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cap(root: u8) -> Capsule { + Capsule { + store_id: Bytes32([1; 32]), + root_hash: Bytes32([root; 32]), + } + } + + fn lineage(roots: &[u8]) -> Lineage { + Lineage::try_new(roots.iter().map(|&r| cap(r)).collect()).unwrap() + } + + /// A held-check over an explicit set of held roots. + struct Held(Vec); + impl HeldCheck for Held { + fn is_held(&self, capsule: &Capsule) -> bool { + self.0.contains(&capsule.root_hash.0[0]) + } + } + + /// **Proves:** observing a lineage seeds status (held vs missing) + exposes tip + history. + #[test] + fn observe_seeds_status_and_views() { + let mut sub = Subscription::new(Bytes32([1; 32])); + sub.observe_lineage(&lineage(&[10, 20, 30]), &Held(vec![10])); + assert_eq!(sub.current_tip(), Some(cap(30))); + assert_eq!(sub.history(), &[cap(10), cap(20), cap(30)]); + assert_eq!(sub.tip_status(&Bytes32([10; 32])), Some(&TipSync::Held)); + assert_eq!(sub.tip_status(&Bytes32([20; 32])), Some(&TipSync::Missing)); + assert_eq!( + sub.missing(), + vec![cap(30), cap(20)], + "tip-first backfill worklist" + ); + } + + /// **Proves:** a new tip progresses the lineage and appends the new missing capsule. + #[test] + fn tip_progression_appends() { + let mut sub = Subscription::new(Bytes32([1; 32])); + sub.observe_lineage(&lineage(&[10]), &Held(vec![10])); + sub.observe_lineage(&lineage(&[10, 20]), &Held(vec![10])); + assert_eq!(sub.current_tip(), Some(cap(20))); + assert_eq!(sub.tip_status(&Bytes32([20; 32])), Some(&TipSync::Missing)); + } + + /// **Proves the reorg/permanence invariant (§5.1):** when a reorg drops an already-HELD + /// capsule out of the canonical lineage, it is moved to `orphaned`, its status stays + /// `Held`, and it is NEVER evicted. + /// **Catches:** a tracker that forgets or downgrades a superseded held `.dig`. + #[test] + fn reorg_never_evicts_a_held_capsule() { + let mut sub = Subscription::new(Bytes32([1; 32])); + // First lineage: eve=10, tip=20, both held. + sub.observe_lineage(&lineage(&[10, 20]), &Held(vec![10, 20])); + assert_eq!(sub.tip_status(&Bytes32([20; 32])), Some(&TipSync::Held)); + + // Reorg: 20 is superseded by a fork 10 → 30 (20 leaves the canonical lineage). + sub.observe_lineage(&lineage(&[10, 30]), &Held(vec![10, 20])); + + assert_eq!(sub.current_tip(), Some(cap(30)), "new canonical tip"); + assert_eq!( + sub.tip_status(&Bytes32([20; 32])), + Some(&TipSync::Held), + "superseded root stays Held — never evicted (§5.1)" + ); + assert!( + sub.orphaned().contains(&cap(20)), + "superseded capsule retained as orphaned, not deleted" + ); + } + + /// **Proves:** a failed fetch record survives a re-observe (retry count is not reset), + /// and a success flips it to Held. + #[test] + fn failed_record_survives_reobserve_then_succeeds() { + let mut sub = Subscription::new(Bytes32([1; 32])); + sub.observe_lineage(&lineage(&[10]), &Held(vec![])); + sub.record_fetch_result(&cap(10), Err("no peer".into())); + assert_eq!( + sub.tip_status(&Bytes32([10; 32])), + Some(&TipSync::Failed { + attempts: 1, + last_error: "no peer".into() + }) + ); + // Re-observe (still not held) must preserve the Failed record. + sub.observe_lineage(&lineage(&[10]), &Held(vec![])); + assert!(matches!( + sub.tip_status(&Bytes32([10; 32])), + Some(TipSync::Failed { attempts: 1, .. }) + )); + // A second failure increments attempts. + sub.record_fetch_result(&cap(10), Err("still no peer".into())); + assert!(matches!( + sub.tip_status(&Bytes32([10; 32])), + Some(TipSync::Failed { attempts: 2, .. }) + )); + // Success flips to Held. + sub.record_fetch_result(&cap(10), Ok(())); + assert_eq!(sub.tip_status(&Bytes32([10; 32])), Some(&TipSync::Held)); + } + + /// **Proves:** `plan` before any observed lineage is empty (no unconfirmable work). + #[test] + fn plan_empty_before_first_observe() { + let sub = Subscription::new(Bytes32([1; 32])); + assert!(sub.plan(&Held(vec![])).is_empty()); + } + + /// **Proves:** a lineage for a DIFFERENT store is ignored — a buggy ChainWatch can't + /// repoint this subscription's history to another store's walk. + /// **Catches:** an `observe_lineage` that folds a foreign store's lineage. + #[test] + fn ignores_foreign_store_lineage() { + let mut sub = Subscription::new(Bytes32([1; 32])); + // A lineage for store 9 (not this subscription's store 1). + let foreign = Lineage::try_new(vec![Capsule { + store_id: Bytes32([9; 32]), + root_hash: Bytes32([99; 32]), + }]) + .unwrap(); + sub.observe_lineage(&foreign, &Held(vec![])); + assert!(sub.current_tip().is_none(), "foreign lineage not folded"); + assert!(sub.history().is_empty()); + assert!(sub.tip_status(&Bytes32([99; 32])).is_none()); + } +}