diff --git a/Cargo.lock b/Cargo.lock index 739cec2..b3937b3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2067,7 +2067,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.13.5" +version = "0.13.6" dependencies = [ "async-trait", "axum", @@ -2115,7 +2115,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.47.5" +version = "0.47.6" dependencies = [ "axum", "axum-server", diff --git a/Cargo.toml b/Cargo.toml index a785244..079fbe6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.47.5" +version = "0.47.6" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index ee7f43d..d6d74e3 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dig-node-core" -version = "0.13.5" +version = "0.13.6" edition = "2021" license = "GPL-2.0-only" description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works." diff --git a/crates/dig-node-core/src/chainwatch.rs b/crates/dig-node-core/src/chainwatch.rs index 01cd1da..6447785 100644 --- a/crates/dig-node-core/src/chainwatch.rs +++ b/crates/dig-node-core/src/chainwatch.rs @@ -23,7 +23,7 @@ use std::sync::Arc; use std::time::Duration; use crate::subscription::SubscriptionSet; -use crate::{AnchoredRootResolver, Bytes32, ChainSource}; +use crate::{AnchoredRootResolver, Bytes32, CapsuleStore, ChainSource}; /// Default interval between chain-watch polls of the subscribed store set. Deliberately modest — a new /// generation confirms on-chain in tens of seconds to minutes, so a ~30 s poll detects it promptly diff --git a/crates/dig-node-core/src/download.rs b/crates/dig-node-core/src/download.rs index 4a2af0f..1fb182f 100644 --- a/crates/dig-node-core/src/download.rs +++ b/crates/dig-node-core/src/download.rs @@ -1024,73 +1024,6 @@ impl crate::Node { self.p2p_content.get() } - /// Background CAPSULE BACKFILL (SPEC §5.6): when a resource read for `(store_hex, root_hex)` is - /// being satisfied FROM ANOTHER NODE (a redirect or a fetch-through miss), also pull the WHOLE - /// `.dig` capsule for that generation in the background and cache it, so the NEXT read of this - /// store is served locally. Configurable (`DIG_NODE_BACKFILL_ON_MISS`, default ON). - /// - /// Fire-and-forget: it spawns a detached task and returns immediately so the current read is never - /// delayed. It is a NO-OP when: backfill is disabled; there is no P2P content engine (the - /// in-process FFI consumer — it has no upstream/peer network to pull a whole capsule from); the - /// capsule is already held locally; or a backfill for this exact capsule is already in flight - /// (deduped via [`Node::backfilling`], so a burst of resource reads for the same not-yet-held store - /// triggers ONE whole-`.dig` pull, not one per read). The pull reuses - /// [`gap_fill_generation`](crate::Node::gap_fill_generation) — the authenticated §21 whole-store - /// sync, chain-anchored-root pinned + DHT-announced — so a backfilled capsule is verified exactly - /// like every other cached generation. - pub(crate) fn maybe_backfill_capsule(&self, store_hex: &str, root_hex: &str) { - // Config gate (default on) + only where a peer network / upstream exists to pull from. - if !backfill_on_miss_enabled() || self.p2p_content().is_none() { - return; - } - // Need an owned `Arc` to spawn the detached pull. Installed by the standalone - // peer-network bring-up; `None` on the FFI path (which also has no p2p_content, so we already - // returned above) or during teardown. - let Some(node) = self.arc_self() else { - return; - }; - // Need a concrete, valid (store, root). `hex64` validates AND decodes; a rootless/`"latest"` - // read (no concrete capsule) or a malformed value yields `None` and is skipped — the read - // path resolves the tip separately. - let (Some(store_id), Some(root_bytes)) = - (crate::dht::hex64(store_hex), crate::dht::hex64(root_hex)) - else { - return; - }; - // Already held → nothing to warm up. - if crate::module_exists(self.cache_dir_path(), store_hex, root_hex) { - return; - } - let key = format!("{store_hex}:{root_hex}"); - // Dedup: claim the in-flight slot; if another read already claimed it, do nothing (a burst of - // resource reads for the same not-yet-held store triggers ONE whole-capsule pull). - { - let mut inflight = self.backfilling.lock().unwrap_or_else(|p| p.into_inner()); - if !inflight.insert(key.clone()) { - return; // a backfill for this capsule is already running - } - } - let root = crate::Bytes32(root_bytes); - tokio::spawn(async move { - match node.gap_fill_generation(store_id, root).await { - Ok(()) => tracing::debug!( - capsule = %key, - "backfill: cached the whole capsule after a resource read from another node" - ), - Err(e) => tracing::debug!( - capsule = %key, - error = %e, - "backfill: whole-capsule pull did not complete (will re-attempt on the next miss)" - ), - } - // Release the in-flight slot so a later miss can re-attempt if this one failed. - node.backfilling - .lock() - .unwrap_or_else(|p| p.into_inner()) - .remove(&key); - }); - } - /// Decide the #165 miss outcome for `content` at redirect depth `depth`: fetch-through when /// configured (falling back to redirect if the fetch fails), else locate + redirect within the /// hop budget, else not-found. NEVER a silent 404 while a provider exists. diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index 6437055..bfcd892 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -52,6 +52,10 @@ pub mod peer; /// every existing `crate::net`/`crate::pex`/… reference working unchanged (W1b-0 is a pure /// relocation — no behaviour change, no caller updates required). pub mod seams; +/// The `CapsuleStore` trait is seam 6's public surface (#1285 W1b-4) — bring it into scope to call +/// `cache_list_cached`/`cache_remove_cached`/`cache_fetch_and_cache`/`gap_fill_generation`/ +/// `maybe_backfill_capsule`/`set_self_ref`/`arc_self` on a `Node`. +pub use seams::capsule::CapsuleStore; pub(crate) use seams::chia_peer::{default_anchored_resolver, resolution_coinset}; /// The `ChainSource` trait is seam 1's public surface (#1285 W1b-3) — bring it into scope to call /// `anchored_root_resolver_arc` on a `Node`. `CoinsetResolver` is public (production impl callers @@ -335,21 +339,6 @@ pub struct Node { pub(crate) type InventoryRefresher = Box std::pin::Pin + Send>> + Send + Sync>; -impl Node { - /// Install the WEAK self-reference (the standalone peer-network bring-up calls this once with the - /// `Arc` it holds). Enables `&self` read handlers to spawn owned-`Arc` background tasks — the - /// capsule backfill (§14.3). Idempotent; never set on the FFI path. - pub(crate) fn set_self_ref(&self, weak: std::sync::Weak) { - let _ = self.self_ref.set(weak); - } - - /// Upgrade the weak self-reference to an owned `Arc`, if the standalone bring-up installed - /// one and the node is still alive. `None` on the FFI path / before bring-up / during teardown. - pub(crate) fn arc_self(&self) -> Option> { - self.self_ref.get().and_then(std::sync::Weak::upgrade) - } -} - /// The CANONICAL (shared) cache dir — the one the DIG Browser's in-process /// dig-node AND the standalone dig-node/dig-companion both resolve to, so they /// share a `.dig` cache by construction (#96). Precedence: @@ -1722,190 +1711,6 @@ impl Node { // `module_path(store_hex, root_hex)` = `/modules//.module`, // so listing/removing/fetching are all keyed by capsule identity. - /// List every cached capsule (`storeId:rootHash`) with its on-disk size and - /// last-used time. Walks `/modules//.module` - /// (the same layout `module_path`/`serve_local`/`sync_module_from` use), - /// reusing the directory-enumerate pattern from [`cache_used_bytes`] and - /// [`Node::evict_if_needed`]. `last_used_unix_ms` is the file mtime (the LRU - /// recency stamp bumped by [`touch`] on every local serve), in Unix epoch ms. - pub async fn cache_list_cached(&self) -> Vec { - let modules_root = self.cache_dir.join("modules"); - let mut out = Vec::new(); - // Outer level: one directory per store id (hex). Inner: `.module`. - let Ok(stores) = std::fs::read_dir(&modules_root) else { - return out; // no modules cached yet - }; - for store_entry in stores.flatten() { - if !store_entry.path().is_dir() { - continue; - } - let Some(store_hex) = store_entry.file_name().to_str().map(str::to_string) else { - continue; - }; - let Ok(modules) = std::fs::read_dir(store_entry.path()) else { - continue; - }; - for m in modules.flatten() { - let path = m.path(); - // A capsule module is `.module`; skip anything else. - let Some(root_hex) = path - .file_name() - .and_then(|f| f.to_str()) - .and_then(|f| f.strip_suffix(".module")) - .map(str::to_string) - else { - continue; - }; - let Ok(md) = m.metadata() else { continue }; - let last_used_unix_ms = md - .modified() - .ok() - .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) - .map(|d| d.as_millis() as u64) - .unwrap_or(0); - out.push(CachedCapsule { - store_id: store_hex.clone(), - root: root_hex, - size_bytes: md.len(), - last_used_unix_ms, - }); - } - } - out - } - - /// Remove one cached capsule's module by `(store_id_hex, root_hex)`. Returns - /// `Ok(true)` if a module was unlinked, `Ok(false)` if it was already absent - /// (idempotent), or `Err` for invalid input. - /// - /// PATH-TRAVERSAL DEFENSE: the hex inputs are validated 64-hex (mirroring the - /// `response_key`/`sync_eligible` sanitization), then the resolved path is - /// canonicalized and asserted to live UNDER the cache dir before any unlink — - /// so a crafted `store_id`/`root` can never delete a file outside the cache. - /// Holds the existing `cache_lock` for the unlink so it can't race eviction. - /// (Async because that lock is a `tokio::sync::Mutex`, acquired with `.await`.) - pub async fn cache_remove_cached( - &self, - store_id_hex: &str, - root_hex: &str, - ) -> Result { - fn is_hex64(s: &str) -> bool { - s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit()) - } - if !is_hex64(store_id_hex) { - return Err(format!("invalid store_id (want 64-hex): {store_id_hex}")); - } - if !is_hex64(root_hex) { - return Err(format!("invalid root (want 64-hex): {root_hex}")); - } - let path = module_path(&self.cache_dir, store_id_hex, root_hex); - - let _guard = self.cache_lock.lock().await; - if !path.exists() { - return Ok(false); // nothing to remove — idempotent no-op - } - // Canonicalize and confirm the target is contained by the cache dir. With - // 64-hex inputs this always holds; the check is defense-in-depth so the - // unlink can never reach outside the cache even if the layout changes. - let canon = std::fs::canonicalize(&path).map_err(|e| e.to_string())?; - let cache_canon = std::fs::canonicalize(&self.cache_dir).map_err(|e| e.to_string())?; - if !canon.starts_with(&cache_canon) { - return Err("refusing to remove a path outside the cache dir".to_string()); - } - std::fs::remove_file(&canon).map_err(|e| e.to_string())?; - // Drop any in-memory decoded content for this capsule so a removed module can never still be - // served from the content cache (audit #179). - self.invalidate_content_cache(store_id_hex, root_hex); - Ok(true) - } - - /// Fetch and cache one capsule on demand over the §21 authenticated - /// whole-store sync path (the same `sync_module_from` / `DigClient::clone_store` - /// the local-first miss path uses, signed with the startup `identity_seed`). - /// Returns `(size_bytes, served_root_hex)` on success. - /// - /// If the capsule is already cached it returns its size without re-downloading - /// (the RPC reports `already_cached`). The cache write itself happens inside - /// `sync_module_from`, which already serializes via the module path; this also - /// holds the `cache_lock` around the call so concurrent on-demand fetches of - /// the same capsule don't race each other. - pub async fn cache_fetch_and_cache( - &self, - store_id_hex: &str, - root_hex: &str, - ) -> Result<(u64, String), String> { - // Already cached → report its size, no network. - let existing = module_path(&self.cache_dir, store_id_hex, root_hex); - if let Ok(md) = std::fs::metadata(&existing) { - return Ok((md.len(), root_hex.to_string())); - } - // Serialize on-demand writes so two fetches of the same capsule don't race. - let _guard = self.cache_lock.lock().await; - // sync_module_from returns true only when the served root == requested - // root; either way the module lands under its SERVED root, so we read the - // file back to report size + confirm the capsule is now present. - let matched = self - .sync_module_from(&self.upstream, store_id_hex, root_hex) - .await; - let path = module_path(&self.cache_dir, store_id_hex, root_hex); - match std::fs::metadata(&path) { - Ok(md) => Ok((md.len(), root_hex.to_string())), - Err(_) if matched => { - // matched but no file: should not happen, surface it. - Err("sync reported a match but the module is not cached".to_string()) - } - Err(_) => Err(format!( - "could not fetch capsule {store_id_hex}:{root_hex} (no §21 identity, \ - not authorized, or served root differs)" - )), - } - } - - /// GAP-FILL one missing generation (SPEC §14.3): pull the whole `.dig` module for - /// `(store_id, root)` down from other nodes, verify it against the chain-anchored root, land it in - /// the local cache, and (best-effort) refresh the DHT provider records so peers immediately find - /// this node as a NEW holder of the just-synced capsule (§14.1). Idempotent — an already-held - /// generation is a cheap success with no network. - /// - /// The pull reuses the authenticated whole-store sync ([`Self::cache_fetch_and_cache`] → - /// `sync_module_from`), which lands the module keyed by capsule `(store, root)`. The - /// VERIFICATION INVARIANT (SPEC §14.3) is upheld at every SERVE: a gap-filled module is never served - /// as current unless its root equals the chain-anchored tip (the read-path pin, §14.4), so a - /// tampered or wrong-generation pull can never be served — the same guarantee whether the module - /// arrived via a client read, a §21 sync, or this proactive gap-fill. - /// - /// `root` is passed as [`Bytes32`] (the chain-anchored tip the watcher resolved), so gap-fill - /// always targets a chain-confirmed generation — never a caller-chosen root. - pub async fn gap_fill_generation( - &self, - store_id: [u8; 32], - root: Bytes32, - ) -> Result<(), String> { - let store_hex = hex::encode(store_id); - let root_hex = root.to_hex(); - // Already held → nothing to pull (idempotent). - if module_exists(&self.cache_dir, &store_hex, &root_hex) { - return Ok(()); - } - // Pull + cache the whole module under (store, root) via the authenticated §21 whole-store sync. - // `cache_fetch_and_cache` serializes concurrent pulls of the same capsule and reports the - // failure reason (no identity / not authorized / served root differs) on error. - self.cache_fetch_and_cache(&store_hex, &root_hex).await?; - - // Confirm the generation actually landed (a sync whose served root differed leaves it absent). - if !module_exists(&self.cache_dir, &store_hex, &root_hex) { - return Err(format!( - "gap-fill for {store_hex}:{root_hex} pulled a module but not at the confirmed root" - )); - } - - // Best-effort: refresh the DHT provider records so peers find this node as a holder of the - // newly-synced capsule (§14.1). The peer-network bring-up installs the announce hook; when no - // peer network is running (FFI path) this is a no-op. - self.refresh_dht_inventory().await; - Ok(()) - } - // -- L7 peer RPC (PHASE-2b, #162) — serving the node's LOCAL inventory ------ // // The node serves the SAME content over the peer network that it serves over §21 / the HTTP read diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index cb7cc5e..5e88406 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -61,7 +61,7 @@ use std::sync::Arc; use serde_json::{json, Value}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use crate::{CachedCapsule, PeerNetwork}; +use crate::{CachedCapsule, CapsuleStore, PeerNetwork}; // -- Constants --------------------------------------------------------------------------------------- diff --git a/crates/dig-node-core/src/seams/capsule/capsule_store.rs b/crates/dig-node-core/src/seams/capsule/capsule_store.rs new file mode 100644 index 0000000..e41cc21 --- /dev/null +++ b/crates/dig-node-core/src/seams/capsule/capsule_store.rs @@ -0,0 +1,308 @@ +//! Seam 6's public surface (#1285/#1303) — the on-disk capsule cache operations the RPC/control +//! surface (`dig-node-service`'s `control.rs`, the in-process `dig-wallet` FFI surface, the L7 peer +//! RPC inventory answers) and the chain-watch gap-filler drive on [`Node`]. +//! +//! `CapsuleStore` is implemented by [`Node`] with its EXISTING method bodies (carved unchanged from +//! `lib.rs`/`download.rs`, #1285 W1b-4) — a behaviour-preserving trait extraction, not a new +//! implementation. `async_trait`-boxed (matching the other seam traits) so it stays dyn-compatible +//! for the future `Arc` handle (W1c). +//! +//! This is also where the W1b-2-deferred self-reference bring-up hooks (`set_self_ref`/`arc_self`) +//! land, per the locked plan's tangle (b): they exist ONLY to let `&self` capsule read handlers +//! (`maybe_backfill_capsule`) spawn an owned-`Arc` background pull — genuinely a capsule-store +//! concern, not a peer-network one. They stay a plain `Weak`/`Arc` pair for this +//! behaviour-preserving pass (full struct decomposition into `Arc` is W1c's job). + +use std::sync::{Arc, Weak}; + +use digstore_core::Bytes32; + +use crate::{module_exists, CachedCapsule, Node, PeerNetwork}; + +/// Seam 6 (capsule management) — the node's on-disk `.dig` capsule cache: list/remove/fetch a held +/// capsule, gap-fill a missing chain-confirmed generation, and the self-reference plumbing that lets +/// `&self` read handlers spawn an owned background backfill. +#[async_trait::async_trait] +pub trait CapsuleStore: Send + Sync { + /// List every cached capsule (`storeId:rootHash`) with its on-disk size and + /// last-used time. Walks `/modules//.module` + /// (the same layout `module_path`/`serve_local`/`sync_module_from` use), + /// reusing the directory-enumerate pattern from [`cache_used_bytes`](crate::cache_used_bytes) and + /// [`Node::evict_if_needed`]. `last_used_unix_ms` is the file mtime (the LRU + /// recency stamp bumped by [`touch`] on every local serve), in Unix epoch ms. + async fn cache_list_cached(&self) -> Vec; + + /// Remove one cached capsule's module by `(store_id_hex, root_hex)`. Returns + /// `Ok(true)` if a module was unlinked, `Ok(false)` if it was already absent + /// (idempotent), or `Err` for invalid input. + /// + /// PATH-TRAVERSAL DEFENSE: the hex inputs are validated 64-hex (mirroring the + /// `response_key`/`sync_eligible` sanitization), then the resolved path is + /// canonicalized and asserted to live UNDER the cache dir before any unlink — + /// so a crafted `store_id`/`root` can never delete a file outside the cache. + /// Holds the existing `cache_lock` for the unlink so it can't race eviction. + /// (Async because that lock is a `tokio::sync::Mutex`, acquired with `.await`.) + async fn cache_remove_cached(&self, store_id_hex: &str, root_hex: &str) + -> Result; + + /// Fetch and cache one capsule on demand over the §21 authenticated + /// whole-store sync path (the same `sync_module_from` / `DigClient::clone_store` + /// the local-first miss path uses, signed with the startup `identity_seed`). + /// Returns `(size_bytes, served_root_hex)` on success. + /// + /// If the capsule is already cached it returns its size without re-downloading + /// (the RPC reports `already_cached`). The cache write itself happens inside + /// `sync_module_from`, which already serializes via the module path; this also + /// holds the `cache_lock` around the call so concurrent on-demand fetches of + /// the same capsule don't race each other. + async fn cache_fetch_and_cache( + &self, + store_id_hex: &str, + root_hex: &str, + ) -> Result<(u64, String), String>; + + /// GAP-FILL one missing generation (SPEC §14.3): pull the whole `.dig` module for + /// `(store_id, root)` down from other nodes, verify it against the chain-anchored root, land it in + /// the local cache, and (best-effort) refresh the DHT provider records so peers immediately find + /// this node as a NEW holder of the just-synced capsule (§14.1). Idempotent — an already-held + /// generation is a cheap success with no network. + /// + /// The pull reuses the authenticated whole-store sync ([`Self::cache_fetch_and_cache`] → + /// `sync_module_from`), which lands the module keyed by capsule `(store, root)`. The + /// VERIFICATION INVARIANT (SPEC §14.3) is upheld at every SERVE: a gap-filled module is never served + /// as current unless its root equals the chain-anchored tip (the read-path pin, §14.4), so a + /// tampered or wrong-generation pull can never be served — the same guarantee whether the module + /// arrived via a client read, a §21 sync, or this proactive gap-fill. + /// + /// `root` is passed as [`Bytes32`] (the chain-anchored tip the watcher resolved), so gap-fill + /// always targets a chain-confirmed generation — never a caller-chosen root. + async fn gap_fill_generation(&self, store_id: [u8; 32], root: Bytes32) -> Result<(), String>; + + /// Background CAPSULE BACKFILL (SPEC §5.6): when a resource read for `(store_hex, root_hex)` is + /// being satisfied FROM ANOTHER NODE (a redirect or a fetch-through miss), also pull the WHOLE + /// `.dig` capsule for that generation in the background and cache it, so the NEXT read of this + /// store is served locally. Configurable (`DIG_NODE_BACKFILL_ON_MISS`, default ON). + /// + /// Fire-and-forget: it spawns a detached task and returns immediately so the current read is never + /// delayed. It is a NO-OP when: backfill is disabled; there is no P2P content engine (the + /// in-process FFI consumer — it has no upstream/peer network to pull a whole capsule from); the + /// capsule is already held locally; or a backfill for this exact capsule is already in flight + /// (deduped via `Node::backfilling`, so a burst of resource reads for the same not-yet-held store + /// triggers ONE whole-`.dig` pull, not one per read). The pull reuses + /// [`Self::gap_fill_generation`] — the authenticated §21 whole-store sync, chain-anchored-root + /// pinned + DHT-announced — so a backfilled capsule is verified exactly like every other cached + /// generation. + fn maybe_backfill_capsule(&self, store_hex: &str, root_hex: &str); + + /// Install the WEAK self-reference (the standalone peer-network bring-up calls this once with the + /// `Arc` it holds). Enables `&self` read handlers to spawn owned-`Arc` background tasks — the + /// capsule backfill (§14.3). Idempotent; never set on the FFI path. + fn set_self_ref(&self, weak: Weak); + + /// Upgrade the weak self-reference to an owned `Arc`, if the standalone bring-up installed + /// one and the node is still alive. `None` on the FFI path / before bring-up / during teardown. + fn arc_self(&self) -> Option>; +} + +#[async_trait::async_trait] +impl CapsuleStore for Node { + async fn cache_list_cached(&self) -> Vec { + let modules_root = self.cache_dir.join("modules"); + let mut out = Vec::new(); + // Outer level: one directory per store id (hex). Inner: `.module`. + let Ok(stores) = std::fs::read_dir(&modules_root) else { + return out; // no modules cached yet + }; + for store_entry in stores.flatten() { + if !store_entry.path().is_dir() { + continue; + } + let Some(store_hex) = store_entry.file_name().to_str().map(str::to_string) else { + continue; + }; + let Ok(modules) = std::fs::read_dir(store_entry.path()) else { + continue; + }; + for m in modules.flatten() { + let path = m.path(); + // A capsule module is `.module`; skip anything else. + let Some(root_hex) = path + .file_name() + .and_then(|f| f.to_str()) + .and_then(|f| f.strip_suffix(".module")) + .map(str::to_string) + else { + continue; + }; + let Ok(md) = m.metadata() else { continue }; + let last_used_unix_ms = md + .modified() + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + out.push(CachedCapsule { + store_id: store_hex.clone(), + root: root_hex, + size_bytes: md.len(), + last_used_unix_ms, + }); + } + } + out + } + + async fn cache_remove_cached( + &self, + store_id_hex: &str, + root_hex: &str, + ) -> Result { + fn is_hex64(s: &str) -> bool { + s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit()) + } + if !is_hex64(store_id_hex) { + return Err(format!("invalid store_id (want 64-hex): {store_id_hex}")); + } + if !is_hex64(root_hex) { + return Err(format!("invalid root (want 64-hex): {root_hex}")); + } + let path = crate::module_path(&self.cache_dir, store_id_hex, root_hex); + + let _guard = self.cache_lock.lock().await; + if !path.exists() { + return Ok(false); // nothing to remove — idempotent no-op + } + // Canonicalize and confirm the target is contained by the cache dir. With + // 64-hex inputs this always holds; the check is defense-in-depth so the + // unlink can never reach outside the cache even if the layout changes. + let canon = std::fs::canonicalize(&path).map_err(|e| e.to_string())?; + let cache_canon = std::fs::canonicalize(&self.cache_dir).map_err(|e| e.to_string())?; + if !canon.starts_with(&cache_canon) { + return Err("refusing to remove a path outside the cache dir".to_string()); + } + std::fs::remove_file(&canon).map_err(|e| e.to_string())?; + // Drop any in-memory decoded content for this capsule so a removed module can never still be + // served from the content cache (audit #179). + self.invalidate_content_cache(store_id_hex, root_hex); + Ok(true) + } + + async fn cache_fetch_and_cache( + &self, + store_id_hex: &str, + root_hex: &str, + ) -> Result<(u64, String), String> { + // Already cached → report its size, no network. + let existing = crate::module_path(&self.cache_dir, store_id_hex, root_hex); + if let Ok(md) = std::fs::metadata(&existing) { + return Ok((md.len(), root_hex.to_string())); + } + // Serialize on-demand writes so two fetches of the same capsule don't race. + let _guard = self.cache_lock.lock().await; + // sync_module_from returns true only when the served root == requested + // root; either way the module lands under its SERVED root, so we read the + // file back to report size + confirm the capsule is now present. + let matched = self + .sync_module_from(&self.upstream, store_id_hex, root_hex) + .await; + let path = crate::module_path(&self.cache_dir, store_id_hex, root_hex); + match std::fs::metadata(&path) { + Ok(md) => Ok((md.len(), root_hex.to_string())), + Err(_) if matched => { + // matched but no file: should not happen, surface it. + Err("sync reported a match but the module is not cached".to_string()) + } + Err(_) => Err(format!( + "could not fetch capsule {store_id_hex}:{root_hex} (no §21 identity, \ + not authorized, or served root differs)" + )), + } + } + + async fn gap_fill_generation(&self, store_id: [u8; 32], root: Bytes32) -> Result<(), String> { + let store_hex = hex::encode(store_id); + let root_hex = root.to_hex(); + // Already held → nothing to pull (idempotent). + if module_exists(&self.cache_dir, &store_hex, &root_hex) { + return Ok(()); + } + // Pull + cache the whole module under (store, root) via the authenticated §21 whole-store sync. + // `cache_fetch_and_cache` serializes concurrent pulls of the same capsule and reports the + // failure reason (no identity / not authorized / served root differs) on error. + self.cache_fetch_and_cache(&store_hex, &root_hex).await?; + + // Confirm the generation actually landed (a sync whose served root differed leaves it absent). + if !module_exists(&self.cache_dir, &store_hex, &root_hex) { + return Err(format!( + "gap-fill for {store_hex}:{root_hex} pulled a module but not at the confirmed root" + )); + } + + // Best-effort: refresh the DHT provider records so peers find this node as a holder of the + // newly-synced capsule (§14.1). The peer-network bring-up installs the announce hook; when no + // peer network is running (FFI path) this is a no-op. + self.refresh_dht_inventory().await; + Ok(()) + } + + fn maybe_backfill_capsule(&self, store_hex: &str, root_hex: &str) { + // Config gate (default on) + only where a peer network / upstream exists to pull from. + if !crate::download::backfill_on_miss_enabled() || self.p2p_content().is_none() { + return; + } + // Need an owned `Arc` to spawn the detached pull. Installed by the standalone + // peer-network bring-up; `None` on the FFI path (which also has no p2p_content, so we already + // returned above) or during teardown. + let Some(node) = self.arc_self() else { + return; + }; + // Need a concrete, valid (store, root). `hex64` validates AND decodes; a rootless/`"latest"` + // read (no concrete capsule) or a malformed value yields `None` and is skipped — the read + // path resolves the tip separately. + let (Some(store_id), Some(root_bytes)) = + (crate::dht::hex64(store_hex), crate::dht::hex64(root_hex)) + else { + return; + }; + // Already held → nothing to warm up. + if crate::module_exists(self.cache_dir_path(), store_hex, root_hex) { + return; + } + let key = format!("{store_hex}:{root_hex}"); + // Dedup: claim the in-flight slot; if another read already claimed it, do nothing (a burst of + // resource reads for the same not-yet-held store triggers ONE whole-capsule pull). + { + let mut inflight = self.backfilling.lock().unwrap_or_else(|p| p.into_inner()); + if !inflight.insert(key.clone()) { + return; // a backfill for this capsule is already running + } + } + let root = Bytes32(root_bytes); + tokio::spawn(async move { + match node.gap_fill_generation(store_id, root).await { + Ok(()) => tracing::debug!( + capsule = %key, + "backfill: cached the whole capsule after a resource read from another node" + ), + Err(e) => tracing::debug!( + capsule = %key, + error = %e, + "backfill: whole-capsule pull did not complete (will re-attempt on the next miss)" + ), + } + // Release the in-flight slot so a later miss can re-attempt if this one failed. + node.backfilling + .lock() + .unwrap_or_else(|p| p.into_inner()) + .remove(&key); + }); + } + + fn set_self_ref(&self, weak: Weak) { + let _ = self.self_ref.set(weak); + } + + fn arc_self(&self) -> Option> { + self.self_ref.get().and_then(Weak::upgrade) + } +} diff --git a/crates/dig-node-core/src/seams/capsule/mod.rs b/crates/dig-node-core/src/seams/capsule/mod.rs new file mode 100644 index 0000000..63c5963 --- /dev/null +++ b/crates/dig-node-core/src/seams/capsule/mod.rs @@ -0,0 +1,9 @@ +//! Seam 6 — capsule management (#1285/#1303). Houses the [`CapsuleStore`] trait (the seam's +//! public surface — list/remove/fetch/gap-fill/backfill the on-disk `.dig` capsule cache, +//! carved unchanged from `lib.rs`/`download.rs`, #1285 W1b-4). The concrete `.dig` format +//! reader/writer stays external (`digstore-core`); this seam is the NODE's cache-management +//! surface over it. + +mod capsule_store; + +pub use capsule_store::CapsuleStore; diff --git a/crates/dig-node-core/src/seams/content/content_serve.rs b/crates/dig-node-core/src/seams/content/content_serve.rs index 2ac9aef..57ec868 100644 --- a/crates/dig-node-core/src/seams/content/content_serve.rs +++ b/crates/dig-node-core/src/seams/content/content_serve.rs @@ -28,7 +28,7 @@ use digstore_core::wire::ContentResponse; use digstore_core::{resource_leaf, Bytes32, SecretSalt, Urn, CHAIN, DEFAULT_RESOURCE_KEY}; use serde_json::{json, Value}; -use crate::{decide_pin, pin_enforced, Node, PinDecision}; +use crate::{decide_pin, pin_enforced, CapsuleStore, Node, PinDecision}; /// JSON-RPC-style code for a serve that fetched bytes but could not verify/decrypt/reach them — /// distinct from a clean content miss (`NotFound`) and from the anchored-root pin (`RootError`). diff --git a/crates/dig-node-core/src/seams/mod.rs b/crates/dig-node-core/src/seams/mod.rs index 52804e1..30ea205 100644 --- a/crates/dig-node-core/src/seams/mod.rs +++ b/crates/dig-node-core/src/seams/mod.rs @@ -4,6 +4,7 @@ //! across the W1b sub-PR sequence (see #1285); a seam not yet listed here still lives at //! the crate root pending its carve. +pub mod capsule; pub mod chia_peer; pub mod content; pub mod dig_peer; diff --git a/crates/dig-node-service/src/control.rs b/crates/dig-node-service/src/control.rs index 7dfc2d0..2ed0696 100644 --- a/crates/dig-node-service/src/control.rs +++ b/crates/dig-node-service/src/control.rs @@ -49,7 +49,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; -use dig_node_core::Node; +use dig_node_core::{CapsuleStore, Node}; use serde_json::{json, Value}; use crate::meta::ErrorCode; diff --git a/crates/dig-wallet/src/lib.rs b/crates/dig-wallet/src/lib.rs index 17b7340..e558ce5 100644 --- a/crates/dig-wallet/src/lib.rs +++ b/crates/dig-wallet/src/lib.rs @@ -24,6 +24,11 @@ use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, OnceLock}; +// `CapsuleStore` is seam 6's public surface (#1285 W1b-4) — brings `cache_list_cached` / +// `cache_remove_cached` / `cache_fetch_and_cache` into scope for the fully-qualified +// `dig_node_core::Node` calls below. +use dig_node_core::CapsuleStore; + use axum::{ extract::State, http::{header, HeaderMap, HeaderValue, StatusCode},