feat(digstore-subscription): canonical chain-watch Subscription with full-history backfill (#979)

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]

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"

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.

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<Option<Lineage>, 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<Capsule>`) 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.

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<SyncAction> {
+    let tip = lineage.tip();
+    let mut actions = Vec::new();
+    let mut queued: Vec<Capsule> = Vec::new();
+
+    let mut push_if_needed = |capsule: Capsule, actions: &mut Vec<SyncAction>| {
+        // 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<u8>);
+    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<u8> {
+        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"
+        );
+    }
+}