diff --git a/Cargo.lock b/Cargo.lock index 0462e8dd..31abbbb9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3041,7 +3041,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.256.0" +version = "0.257.0" dependencies = [ "async-trait", "axum", @@ -3079,6 +3079,7 @@ dependencies = [ "notify", "num-bigint", "reqwest", + "ring", "rpassword", "rustls", "serde", diff --git a/Cargo.toml b/Cargo.toml index a694e447..a3a8d7f2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ edition = "2021" # 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.256.0" +version = "0.257.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over # it, so silent wrapping in release would turn a length bug into a memory/logic hazard. diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index 53cb6ba6..087610ef 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1647,3 +1647,4 @@ Two things worth carrying forward: difference is not scrutiny of the record — it is whether a wrong record can ever, by itself, produce a wrong outcome. Here it cannot: `Unbonded`/`Unverified` from the re-check discards the candidate no matter how confidently the record states it. + diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index d582f173..50e211a7 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -576,6 +576,20 @@ pub struct Node { /// [`Node::register_reward_prover_status`], the same read starts returning it with no /// dispatch-side change. reward_prover_statuses: Arc>>, + /// This node's durable record of WHICH reward distributors it funds + /// (dig_ecosystem#3285) — identity only, never an amount; see + /// [`rewards::funded`]'s module doc. + /// + /// A slot rather than a constructor argument for the same reason [`Node::mirror_pointers`] is + /// one: the FFI/browser path has no state directory and must keep constructing a `Node` + /// without one. Nothing installs it in production yet — nothing in dig-node funds a + /// distributor today (`rewards::port`'s module doc, blocker 2), and the startup wiring that + /// would call [`Node::install_funded_distributor_registry`] with the node's state directory + /// belongs to dig_ecosystem#3268. Until then the slot stays empty, and + /// [`Node::funded_distributors_read`] answers + /// [`rewards::funded::NotConfiguredReason::NoStateDirectory`] — UNKNOWN, deliberately never an + /// empty funded set. + funded_distributors: OnceLock, } impl Node { @@ -607,6 +621,47 @@ impl Node { .map(rewards::state::StatusHandle::snapshot) .collect() } + + /// Install this node's funder-ownership registry (dig_ecosystem#3285), once. Returns `false` + /// if a registry is already installed, in which case NOTHING changed — a second install must + /// not be able to swap a live registry for an inert one behind a caller's back. + /// + /// Called from tests today: the startup path that would install a real one lives in + /// dig_ecosystem#3268's files, so clippy's non-test lib target sees no production caller yet. + /// `allow(dead_code)` stands in for that missing caller — remove it when #3268 wires the call. + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn install_funded_distributor_registry( + &self, + registry: rewards::funded::FundedDistributorRegistry, + ) -> bool { + self.funded_distributors.set(registry).is_ok() + } + + /// The installed funder-ownership registry, or `None` when nothing has installed one. + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn funded_distributor_registry( + &self, + ) -> Option<&rewards::funded::FundedDistributorRegistry> { + self.funded_distributors.get() + } + + /// Read which distributors this node funds, through the installed registry. + /// + /// With no registry installed the answer is + /// [`rewards::funded::NotConfiguredReason::NoStateDirectory`], the same UNKNOWN an inert + /// registry reports — never [`rewards::funded::FundedDistributorsRead::FundsNothing`]. A + /// caller rendering `dig.listRewardDistributors` must distinguish the two: an unknown rendered + /// as `[]` tells an operator it funds nothing when it may fund plenty + /// (`dig-rewards-coin` SPEC §2.4 clause 1). + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn funded_distributors_read(&self) -> rewards::funded::FundedDistributorsRead { + match self.funded_distributor_registry() { + Some(registry) => registry.read(), + None => rewards::funded::FundedDistributorsRead::NotConfigured( + rewards::funded::NotConfiguredReason::NoStateDirectory, + ), + } + } } /// A boxed async hook that reconciles the node's DHT provider records with its current cache @@ -4855,6 +4910,7 @@ impl Node { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }) } @@ -5195,6 +5251,7 @@ pub(crate) mod test_support { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }; (Arc::new(node), td) } @@ -5953,6 +6010,63 @@ mod tests { ); } + /// dig_ecosystem#3285: a node with no funder registry installed — which is EVERY production + /// node until #3268 wires one — must read UNKNOWN, never an empty funded set. + /// **Catches:** a `funded_distributors_read` that defaults to `FundsNothing`, or a `Vec`/ + /// `Option` return that a caller would render as `[]`. + #[test] + fn a_node_with_no_installed_funder_registry_reads_not_configured() { + let (node, _td) = test_node(None); + + assert!( + node.funded_distributor_registry().is_none(), + "nothing installs the registry in production yet (see the field doc)" + ); + assert_eq!( + node.funded_distributors_read(), + crate::rewards::funded::FundedDistributorsRead::NotConfigured( + crate::rewards::funded::NotConfiguredReason::NoStateDirectory + ), + "a node with no registry installed must read UNKNOWN, never an empty funded set" + ); + } + + /// dig_ecosystem#3285: the node's read goes through the installed registry all the way to + /// disk, and a second install cannot swap a live registry for an inert one. + /// **Catches:** an accessor reading some other (empty) source, and a `set`-ignoring install. + #[test] + fn a_node_reads_through_the_installed_funder_registry() { + let state_dir = tempfile::tempdir().unwrap(); + let funded = crate::rewards::funded::FundedDistributor { + launcher_id: [0x21u8; 32], + store_id: Some([0x22u8; 32]), + }; + let registry = + crate::rewards::funded::FundedDistributorRegistry::with_state_dir(state_dir.path()); + assert_eq!( + registry.record(&funded), + crate::rewards::funded::RecordOutcome::Recorded + ); + let (node, _td) = test_node(None); + + assert!( + node.install_funded_distributor_registry(registry), + "the first install must take" + ); + + assert_eq!( + node.funded_distributors_read(), + crate::rewards::funded::FundedDistributorsRead::Funded(vec![funded]), + "the node's read must go through the installed registry to disk" + ); + assert!( + !node.install_funded_distributor_registry( + crate::rewards::funded::FundedDistributorRegistry::disabled() + ), + "a second install must be refused rather than silently replace the live registry" + ); + } + fn test_node(identity_seed: Option<[u8; 32]>) -> (Node, tempfile::TempDir) { test_node_with_resolver(identity_seed, MockResolver::always(Ok(None))) } @@ -5991,6 +6105,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }; (node, td) } @@ -6126,6 +6241,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }; // Missing before the pull. @@ -6195,6 +6311,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }); // Build the loop's deps from the PRODUCTION seams, with a fixed one-store subscription set. @@ -6295,6 +6412,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }); assert!(!module_exists(&node.cache_dir, &store_hex, &root.to_hex())); @@ -6373,6 +6491,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }); assert!(!module_exists(&node.cache_dir, &store_hex, &root.to_hex())); @@ -9854,6 +9973,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), }; let before = handle_rpc( @@ -17010,6 +17130,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), ..node }; // A holder for this EXACT content is known via the DHT. @@ -17062,6 +17183,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), ..node }; // A P2P engine is attached but the DHT knows of NO holder for this content — the graceful @@ -17113,6 +17235,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), ..node }; @@ -17146,6 +17269,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), ..node }; let cid = ContentId::resource(store.0, tip.0, rk); @@ -17188,6 +17312,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), ..node }; let cid = ContentId::resource(store.0, tip.0, rk); @@ -17232,6 +17357,7 @@ mod tests { node_peer_id: OnceLock::new(), mirror_pointers: OnceLock::new(), reward_prover_statuses: Arc::new(std::sync::RwLock::new(Vec::new())), + funded_distributors: OnceLock::new(), ..node }; let cid = ContentId::resource(store.0, tip.0, rk); diff --git a/crates/dig-node-core/src/rewards/funded.rs b/crates/dig-node-core/src/rewards/funded.rs new file mode 100644 index 00000000..8f4ea39e --- /dev/null +++ b/crates/dig-node-core/src/rewards/funded.rs @@ -0,0 +1,848 @@ +//! The funder-ownership registry: WHICH reward distributors this node funds (dig_ecosystem#3285). +//! +//! [`port`](super::port)'s module doc names this as blocker 2 of two: +//! `RewardsChainPort::funded_distributors` needs an identity set to start from, because there is no +//! chain-wide "list every distributor and filter to mine" call. This module is that identity set, +//! and nothing else. +//! +//! # Identity only — never an amount +//! +//! A record here is a launcher id plus, when the funding act knew it, the store id it rewards. It +//! carries no reserve balance, no accrued figure and no paid-out total, and there is deliberately +//! nowhere in [`FundedDistributor`] to put one, for two independent reasons: +//! +//! 1. Every money figure in this subsystem is chain-derived (see +//! [`port::DistributorChainState`](super::port::DistributorChainState), which is read, never +//! stored) and goes stale the moment the chain moves. A persisted amount is a wrong number with +//! a convincing timestamp. +//! 2. dig_ecosystem#3286: upstream `chia-sdk-driver`'s `withdraw_incentives` multiplies +//! `rewards * withdrawal_share_bps` in `u64` and wraps in release builds, so a figure crossing +//! this boundary can already be wrong. Durable storage would make such a figure permanent. +//! +//! The store id is identity; the merkle ROOT that +//! [`port::DistributorRef`](super::port::DistributorRef) also carries is not — it names one +//! generation of a store and is superseded on every update, so persisting it would be persisting a +//! value guaranteed to go stale. A caller that needs the current root reads it from the chain. +//! +//! # Persistence mirrors the claim engine, and holds no state between calls +//! +//! An optional directory, exactly like `rewards_claim::engine::ClaimEngine`'s +//! `fee_window_state_dir`: `None` keeps this registry inert, so tests and every default build need +//! no disk. The set itself is NEVER cached on [`FundedDistributorRegistry`] — every read re-reads +//! the file and every write is a read-modify-write of it, the discipline that engine's F16/F18 +//! notes arrived at after one mechanism (a per-call value held as process-lifetime state) produced +//! three separate defects. With no field to go stale, an operator who repairs the file underneath +//! a running node is observed on the very next read rather than only on the next restart. +//! +//! # A corrupt, missing or unreadable record MUST NOT read as "funds nothing" +//! +//! `dig-rewards-coin`'s SPEC §2.4 clause 1 ("absence is not silence") applies here in the place it +//! costs most: `dig.listRewardDistributors` is how an operator sees which distributors it funds, so +//! rendering a failed read as `[]` would make a funded distributor invisible and tell the operator +//! it funds none. [`FundedDistributorsRead`] therefore names the legitimate empty case +//! ([`FundedDistributorsRead::FundsNothing`]) separately from every not-an-answer case, the same +//! way `rewards_claim::types::ClaimOutcome` names its legitimate not-paid cases separately from +//! `Faulted`. + +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use super::port::Bytes32; + +/// The record file, inside the registry's state directory. +pub const FUNDED_DISTRIBUTORS_FILE: &str = "funded-distributors.json"; + +/// Where a corrupt record is COPIED for the operator, next to the record itself. A copy, not a +/// move: see [`FundedDistributorRegistry::read`] for why moving it aside would recreate the exact +/// "a funded distributor became invisible" failure this module exists to prevent. +pub const FUNDED_DISTRIBUTORS_QUARANTINE_FILE: &str = "funded-distributors.json.corrupt"; + +/// On-disk format version. A record written by a different version is CORRUPT to this one — +/// unreadable is unreadable, and guessing at a format we do not know is how a wrong answer gets +/// rendered confidently. +const RECORD_FORMAT_VERSION: u32 = 1; + +/// One distributor this node funds — identity only. See the module doc for why there is nowhere +/// here to record an amount. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FundedDistributor { + /// The distributor singleton's launcher id: the one identifier that never changes. + pub launcher_id: Bytes32, + /// The store this distributor rewards, when the funding act knew it. `None` means "not + /// recorded", never "no store". + pub store_id: Option, +} + +/// Why a read could not answer with a set. Never a stand-in for "the set is empty". +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NotConfiguredReason { + /// The registry has no state directory, so persistence is off and this node has no record to + /// consult. The ordinary state of a default build and of every test that wants no disk. + NoStateDirectory, + /// A state directory is configured but does not exist on disk. The answer lives somewhere this + /// node cannot see, which is unknown, not empty. + StateDirectoryMissing, + /// The state directory exists and holds no record file: nothing has ever been recorded through + /// this registry. Distinct from [`FundedDistributorsRead::FundsNothing`], which is a record + /// that exists and says "none". + NoRecordWritten, +} + +/// The closed outcome of reading the registry. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum FundedDistributorsRead { + /// A record exists and names these distributors, in recorded order. Never empty — an empty + /// record is [`Self::FundsNothing`]. + Funded(Vec), + /// A record exists, is intact, and names no distributor: this node has funded none. The one + /// outcome a caller may render as an empty list. + FundsNothing, + /// No record could be consulted. The answer is UNKNOWN. + NotConfigured(NotConfiguredReason), + /// The record exists and could not be trusted — unparseable, malformed, or written by a format + /// version this build does not know. `quarantined_to` is where its bytes were copied for the + /// operator, or `None` if even the copy failed (which changes nothing about the verdict). + PersistedStateCorrupt { + path: PathBuf, + quarantined_to: Option, + }, + /// The record could not be read at all. + IoFailed { path: PathBuf, error: String }, +} + +impl FundedDistributorsRead { + /// The funded set when — and only when — this read actually determined one: `Some(&[])` for + /// [`Self::FundsNothing`], `None` for every outcome that did not answer. + /// + /// A caller that renders a list MUST distinguish `None` from `Some(&[])`: `None` is "unknown", + /// and rendering it as an empty list is the failure this module's doc opens with. + #[cfg_attr(not(test), allow(dead_code))] + #[must_use] + pub fn determined(&self) -> Option<&[FundedDistributor]> { + match self { + Self::Funded(set) => Some(set), + Self::FundsNothing => Some(&[]), + Self::NotConfigured(_) | Self::PersistedStateCorrupt { .. } | Self::IoFailed { .. } => { + None + } + } + } +} + +/// The closed outcome of recording a funding act. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RecordOutcome { + /// A launcher id this record had not seen was appended. + Recorded, + /// This launcher id was already recorded with the same identity; the file is unchanged. + AlreadyRecorded, + /// This launcher id was already recorded and its `store_id` was learned (`None` -> `Some`). + /// Refining identity is allowed; contradicting it is [`Self::IdentityConflict`]. + IdentityRefined, + /// The record already names this launcher id with a DIFFERENT store id. One of the two is wrong + /// and this registry cannot tell which, so it overwrites neither. + IdentityConflict { recorded: Bytes32, offered: Bytes32 }, + /// Persistence is off: nothing was recorded and nothing will be readable later. + NotConfigured(NotConfiguredReason), + /// The record on disk is corrupt, so it was NOT overwritten — the same refusal + /// `rewards_claim::engine::ClaimEngine::persist_fee_window` makes, for the same reason: writing + /// over corruption produces a file that looks clean and has silently lost whatever it held. + PersistedStateCorrupt { + path: PathBuf, + quarantined_to: Option, + }, + /// The record could not be read or written. + IoFailed { path: PathBuf, error: String }, +} + +/// The durable record of which distributors this node funds. +/// +/// Holds a path and nothing else — see the module doc's "holds no state between calls". +#[derive(Debug, Clone)] +pub struct FundedDistributorRegistry { + /// `None` = persistence off; every read answers [`NotConfiguredReason::NoStateDirectory`] and + /// every write records nothing. + state_dir: Option, +} + +impl FundedDistributorRegistry { + /// A registry with persistence off. The default for any build with no state directory to give + /// it, including the FFI/browser path. + #[must_use] + pub fn disabled() -> Self { + Self { state_dir: None } + } + + /// A registry persisting to `dir`. The directory is created on first write, not here, so + /// constructing one is infallible and side-effect free. + #[cfg_attr(not(test), allow(dead_code))] + #[must_use] + pub fn with_state_dir(dir: &Path) -> Self { + Self { + state_dir: Some(dir.to_path_buf()), + } + } + + /// The record file path, when persistence is on. + fn record_path(&self) -> Option { + self.state_dir + .as_ref() + .map(|dir| dir.join(FUNDED_DISTRIBUTORS_FILE)) + } + + /// Read the funded set fresh from disk. + /// + /// # A corrupt record is quarantined by COPY, and stays where it is + /// Moving the corrupt file aside would leave the next read finding no file at all — i.e. + /// reporting [`NotConfiguredReason::NoRecordWritten`] and, one honest-looking render later, an + /// empty list. So the bytes are copied to [`FUNDED_DISTRIBUTORS_QUARANTINE_FILE`] for the + /// operator and the original is left in place, which keeps every subsequent read reporting + /// [`FundedDistributorsRead::PersistedStateCorrupt`] until a human resolves it. That is the + /// same "leave the corrupt file exactly as it is on disk" posture + /// `rewards_claim::engine::ClaimEngine::persist_fee_window` takes, plus a forensic copy. + #[cfg_attr(not(test), allow(dead_code))] + #[must_use] + pub fn read(&self) -> FundedDistributorsRead { + let Some(path) = self.record_path() else { + return FundedDistributorsRead::NotConfigured(NotConfiguredReason::NoStateDirectory); + }; + match self.load(&path) { + Ok(Some(record)) => match record.into_distributors() { + Ok(set) if set.is_empty() => FundedDistributorsRead::FundsNothing, + Ok(set) => FundedDistributorsRead::Funded(set), + Err(reason) => self.report_corrupt(&path, &reason), + }, + Ok(None) => FundedDistributorsRead::NotConfigured(self.absent_record_reason()), + Err(LoadFailure::Corrupt(reason)) => self.report_corrupt(&path, &reason), + Err(LoadFailure::Io(error)) => FundedDistributorsRead::IoFailed { path, error }, + } + } + + /// Record that this node funds `distributor`, creating the state directory and the record file + /// if they do not exist. Idempotent per launcher id. + #[cfg_attr(not(test), allow(dead_code))] + pub fn record(&self, distributor: &FundedDistributor) -> RecordOutcome { + let Some(path) = self.record_path() else { + return RecordOutcome::NotConfigured(NotConfiguredReason::NoStateDirectory); + }; + let mut set = match self.load(&path) { + Ok(Some(record)) => match record.into_distributors() { + Ok(set) => set, + Err(reason) => return self.refuse_corrupt(&path, &reason), + }, + Ok(None) => Vec::new(), + Err(LoadFailure::Corrupt(reason)) => return self.refuse_corrupt(&path, &reason), + Err(LoadFailure::Io(error)) => return RecordOutcome::IoFailed { path, error }, + }; + + let outcome = match merge(&mut set, distributor) { + Ok(outcome) => outcome, + Err(conflict) => return conflict, + }; + if matches!(outcome, RecordOutcome::AlreadyRecorded) { + return outcome; + } + match self.save(&path, &set) { + Ok(()) => outcome, + Err(error) => RecordOutcome::IoFailed { path, error }, + } + } + + /// Whether an absent record file means "directory gone" or "nothing recorded yet" — two + /// different unknowns, and neither of them "funds nothing". + fn absent_record_reason(&self) -> NotConfiguredReason { + match &self.state_dir { + None => NotConfiguredReason::NoStateDirectory, + Some(dir) if !dir.is_dir() => NotConfiguredReason::StateDirectoryMissing, + Some(_) => NotConfiguredReason::NoRecordWritten, + } + } + + /// Read and parse the record. `Ok(None)` = no record file (the directory may or may not exist; + /// [`Self::absent_record_reason`] tells those apart). + fn load(&self, path: &Path) -> Result, LoadFailure> { + let text = match std::fs::read_to_string(path) { + Ok(text) => text, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(e) => return Err(LoadFailure::Io(e.to_string())), + }; + let record: PersistedRecord = + serde_json::from_str(&text).map_err(|e| LoadFailure::Corrupt(e.to_string()))?; + if record.version != RECORD_FORMAT_VERSION { + return Err(LoadFailure::Corrupt(format!( + "record format version {} is not {RECORD_FORMAT_VERSION}", + record.version + ))); + } + Ok(Some(record)) + } + + /// Write the set ATOMICALLY: to a temp file beside the record, then renamed over it, so a crash + /// mid-write cannot leave a torn file the next read would have to call corrupt. The same + /// pattern `rewards_claim::config::RewardsClaimConfig::save_to` uses for the claim side's + /// persisted state. + fn save(&self, path: &Path, set: &[FundedDistributor]) -> Result<(), String> { + let dir = path.parent().ok_or_else(|| { + format!( + "the funded-distributor record path {} has no parent directory", + path.display() + ) + })?; + std::fs::create_dir_all(dir).map_err(|e| e.to_string())?; + let record = PersistedRecord::from_distributors(set); + let text = serde_json::to_string_pretty(&record).map_err(|e| e.to_string())?; + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, text.as_bytes()).map_err(|e| e.to_string())?; + std::fs::rename(&tmp, path).map_err(|e| e.to_string()) + } + + /// Log, quarantine-copy, and report a corrupt record to a READER. + fn report_corrupt(&self, path: &Path, reason: &str) -> FundedDistributorsRead { + let quarantined_to = self.quarantine(path); + tracing::error!( + path = %path.display(), + reason, + quarantined_to = ?quarantined_to, + "the funded-distributor record is corrupt; reporting corrupt rather than an empty \ + funded set" + ); + FundedDistributorsRead::PersistedStateCorrupt { + path: path.to_path_buf(), + quarantined_to, + } + } + + /// Log, quarantine-copy, and report a corrupt record to a WRITER, which leaves the file alone. + fn refuse_corrupt(&self, path: &Path, reason: &str) -> RecordOutcome { + let quarantined_to = self.quarantine(path); + tracing::error!( + path = %path.display(), + reason, + quarantined_to = ?quarantined_to, + "the funded-distributor record is corrupt; refusing to overwrite it with a new funding \ + record" + ); + RecordOutcome::PersistedStateCorrupt { + path: path.to_path_buf(), + quarantined_to, + } + } + + /// Copy the corrupt record beside itself for the operator, leaving the original in place. + /// `None` when the copy failed — the corrupt verdict does not depend on it. + fn quarantine(&self, path: &Path) -> Option { + let target = path.with_file_name(FUNDED_DISTRIBUTORS_QUARANTINE_FILE); + match std::fs::copy(path, &target) { + Ok(_) => Some(target), + Err(e) => { + tracing::warn!( + path = %path.display(), + target = %target.display(), + error = %e, + "the corrupt funded-distributor record could not be copied to quarantine" + ); + None + } + } + } +} + +/// Add `distributor` to `set`, or refine the identity already there. `Err` carries the conflict +/// outcome, so a caller cannot forget to stop. +fn merge( + set: &mut Vec, + distributor: &FundedDistributor, +) -> Result { + let Some(existing) = set + .iter_mut() + .find(|d| d.launcher_id == distributor.launcher_id) + else { + set.push(distributor.clone()); + return Ok(RecordOutcome::Recorded); + }; + match (existing.store_id, distributor.store_id) { + (Some(recorded), Some(offered)) if recorded != offered => { + Err(RecordOutcome::IdentityConflict { recorded, offered }) + } + (None, Some(offered)) => { + existing.store_id = Some(offered); + Ok(RecordOutcome::IdentityRefined) + } + _ => Ok(RecordOutcome::AlreadyRecorded), + } +} + +/// Why [`FundedDistributorRegistry::load`] could not hand back a record. +enum LoadFailure { + /// The file exists and cannot be trusted. + Corrupt(String), + /// The file could not be read. + Io(String), +} + +/// The on-disk shape: a version plus hex-string ids, so an operator can read and repair the file by +/// hand. `[u8; 32]` would serialize as 32 JSON numbers, which nobody can check by eye. +#[derive(Debug, Serialize, Deserialize)] +struct PersistedRecord { + version: u32, + distributors: Vec, +} + +/// One record line: hex ids, `store_id` omitted entirely when it was never learned. +#[derive(Debug, Serialize, Deserialize)] +struct PersistedDistributor { + launcher_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + store_id: Option, +} + +impl PersistedRecord { + fn from_distributors(set: &[FundedDistributor]) -> Self { + Self { + version: RECORD_FORMAT_VERSION, + distributors: set + .iter() + .map(|d| PersistedDistributor { + launcher_id: hex::encode(d.launcher_id), + store_id: d.store_id.map(hex::encode), + }) + .collect(), + } + } + + /// `Err` carries why the record is corrupt. A malformed id is corruption, never an entry to + /// skip: silently dropping one would under-report the funded set, which is the same lie as + /// reporting it empty, only harder to notice. + fn into_distributors(self) -> Result, String> { + self.distributors + .into_iter() + .map(|d| { + Ok(FundedDistributor { + launcher_id: parse_id(&d.launcher_id, "launcher_id")?, + store_id: d + .store_id + .as_deref() + .map(|s| parse_id(s, "store_id")) + .transpose()?, + }) + }) + .collect() + } +} + +/// Parse one 32-byte hex id, naming the field in the error so a corrupt-record log points at the +/// thing to fix. +fn parse_id(text: &str, field: &str) -> Result { + let bytes = hex::decode(text).map_err(|e| format!("{field} is not hex: {e}"))?; + let len = bytes.len(); + bytes + .try_into() + .map_err(|_| format!("{field} is {len} bytes, not 32")) +} + +#[cfg(test)] +mod tests { + //! Every persistence test round-trips against a REAL temporary directory + //! (`tempfile::TempDir`, removed on drop), never a mock: the thing under test is what survives + //! a restart, and a mock filesystem cannot answer that. "Restart" is simulated the only way it + //! can be without spawning a process — by dropping the registry that wrote and constructing a + //! FRESH one over the same directory, which is exactly the state a new process starts from, + //! since [`FundedDistributorRegistry`] caches nothing. + + use std::path::PathBuf; + + use tempfile::TempDir; + + use super::*; + + /// A distinguishable 32-byte id. + fn id(seed: u8) -> Bytes32 { + [seed; 32] + } + + fn record_path(dir: &TempDir) -> PathBuf { + dir.path().join(FUNDED_DISTRIBUTORS_FILE) + } + + /// **Catches:** a write that persists nothing, or a read that drops the store id. + #[test] + fn records_then_reads_back_the_same_identity() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + let funded = FundedDistributor { + launcher_id: id(1), + store_id: Some(id(2)), + }; + + assert_eq!(registry.record(&funded), RecordOutcome::Recorded); + + assert_eq!( + registry.read(), + FundedDistributorsRead::Funded(vec![funded]), + "a recorded distributor must read back with its identity intact" + ); + } + + /// The ticket's actual requirement: the set survives the process that recorded it. + /// **Catches:** an in-memory-only registry, or a write that never reached disk. + #[test] + fn a_fresh_registry_over_the_same_directory_still_sees_the_set() { + let dir = TempDir::new().expect("temp dir"); + let with_store = FundedDistributor { + launcher_id: id(3), + store_id: Some(id(4)), + }; + let without_store = FundedDistributor { + launcher_id: id(5), + store_id: None, + }; + { + // Scoped so the writing registry is dropped before the reading one exists: nothing but + // the directory carries information across the boundary, which is what a restart is. + let writer = FundedDistributorRegistry::with_state_dir(dir.path()); + assert_eq!(writer.record(&with_store), RecordOutcome::Recorded); + assert_eq!(writer.record(&without_store), RecordOutcome::Recorded); + } + + let after_restart = FundedDistributorRegistry::with_state_dir(dir.path()); + + assert_eq!( + after_restart.read(), + FundedDistributorsRead::Funded(vec![with_store, without_store]), + "the funded set must survive the process that recorded it, in recorded order" + ); + } + + /// **Catches:** an amount finding its way into the durable record, and ids persisted as raw + /// byte arrays no operator can check by eye. + #[test] + fn the_persisted_record_is_hex_and_carries_no_amount() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + registry.record(&FundedDistributor { + launcher_id: id(0xab), + store_id: Some(id(0xcd)), + }); + + let text = std::fs::read_to_string(record_path(&dir)).expect("record readable"); + + assert!( + text.contains(&"ab".repeat(32)) && text.contains(&"cd".repeat(32)), + "ids must persist as operator-readable hex, got: {text}" + ); + for money in [ + "amount", + "mojos", + "base_units", + "reserve", + "accrued", + "paid_out", + "balance", + ] { + assert!( + !text.contains(money), + "the record must carry no money figure, found {money:?} in: {text}" + ); + } + } + + /// **Catches:** a duplicate record line per funding act, and a refinement that is silently + /// dropped instead of persisted. + #[test] + fn recording_the_same_launcher_twice_is_idempotent_and_refines_identity() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + let unknown_store = FundedDistributor { + launcher_id: id(7), + store_id: None, + }; + let learned_store = FundedDistributor { + launcher_id: id(7), + store_id: Some(id(8)), + }; + + assert_eq!(registry.record(&unknown_store), RecordOutcome::Recorded); + assert_eq!( + registry.record(&unknown_store), + RecordOutcome::AlreadyRecorded + ); + assert_eq!( + registry.record(&learned_store), + RecordOutcome::IdentityRefined + ); + assert_eq!( + registry.record(&learned_store), + RecordOutcome::AlreadyRecorded + ); + + assert_eq!( + registry.read(), + FundedDistributorsRead::Funded(vec![learned_store]), + "one launcher id must occupy one record line, with the identity it refined to" + ); + } + + /// **Catches:** a second, contradicting store id overwriting recorded identity. + #[test] + fn a_contradicting_store_id_is_a_conflict_and_overwrites_nothing() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + let recorded = FundedDistributor { + launcher_id: id(9), + store_id: Some(id(10)), + }; + registry.record(&recorded); + + let outcome = registry.record(&FundedDistributor { + launcher_id: id(9), + store_id: Some(id(11)), + }); + + assert_eq!( + outcome, + RecordOutcome::IdentityConflict { + recorded: id(10), + offered: id(11), + } + ); + assert_eq!( + registry.read(), + FundedDistributorsRead::Funded(vec![recorded]), + "a conflicting offer must leave the recorded identity exactly as it was" + ); + } + + /// The requirement everything else serves. + /// **Catches:** a corrupt read rendering as an empty funded set, a quarantine that does not + /// preserve the bytes, and a quarantine that MOVES the record so the next read reads empty. + #[test] + fn a_corrupt_record_reports_corrupt_and_quarantines_and_is_never_empty() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + registry.record(&FundedDistributor { + launcher_id: id(12), + store_id: None, + }); + let path = record_path(&dir); + let corrupt_bytes = b"{\"version\": 1, \"distributors\": [ truncated"; + std::fs::write(&path, corrupt_bytes).expect("corrupt the record"); + + let read = registry.read(); + + let FundedDistributorsRead::PersistedStateCorrupt { + path: reported, + quarantined_to, + } = &read + else { + panic!("a corrupt record must report PersistedStateCorrupt, got {read:?}"); + }; + assert_eq!(reported, &path); + let quarantine = quarantined_to + .as_ref() + .expect("the corrupt record must be quarantined"); + assert_eq!( + quarantine, + &dir.path().join(FUNDED_DISTRIBUTORS_QUARANTINE_FILE) + ); + assert_eq!( + std::fs::read(quarantine).expect("quarantine readable"), + corrupt_bytes, + "quarantine must preserve the corrupt bytes verbatim" + ); + assert_eq!( + std::fs::read(&path).expect("original still readable"), + corrupt_bytes, + "the original must stay in place so the NEXT read is corrupt too, not empty" + ); + assert_eq!( + read.determined(), + None, + "corrupt must never present as a determined (and therefore renderable) set" + ); + assert_ne!(read, FundedDistributorsRead::FundsNothing); + assert!( + matches!( + registry.read(), + FundedDistributorsRead::PersistedStateCorrupt { .. } + ), + "quarantining must not let the following read decay into an empty answer" + ); + } + + /// **Catches:** a parser that skips a malformed entry, which under-reports the funded set. + #[test] + fn a_malformed_id_inside_a_parseable_record_is_corrupt_not_a_skipped_entry() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + std::fs::write( + record_path(&dir), + r#"{"version": 1, "distributors": [{"launcher_id": "beef"}]}"#, + ) + .expect("write a short id"); + + let read = registry.read(); + + assert!( + matches!(read, FundedDistributorsRead::PersistedStateCorrupt { .. }), + "a 2-byte launcher id must be corruption, not an entry to drop, got {read:?}" + ); + assert_eq!(read.determined(), None); + } + + /// **Catches:** a future format version read as an empty set by a version-blind parser. + #[test] + fn an_unknown_format_version_is_corrupt_not_empty() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + std::fs::write(record_path(&dir), r#"{"version": 2, "distributors": []}"#) + .expect("write a future record"); + + let read = registry.read(); + + assert!( + matches!(read, FundedDistributorsRead::PersistedStateCorrupt { .. }), + "a version this build cannot read must be corrupt, got {read:?}" + ); + assert_eq!(read.determined(), None); + } + + /// **Catches:** a write that papers over corruption with a clean-looking file, losing whatever + /// the corrupt record held. + #[test] + fn a_corrupt_record_is_never_overwritten_by_a_new_funding_record() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + let path = record_path(&dir); + let corrupt_bytes = b"not json at all"; + std::fs::write(&path, corrupt_bytes).expect("seed a corrupt record"); + + let outcome = registry.record(&FundedDistributor { + launcher_id: id(13), + store_id: None, + }); + + assert!( + matches!(outcome, RecordOutcome::PersistedStateCorrupt { .. }), + "recording over corruption must refuse, got {outcome:?}" + ); + assert_eq!( + std::fs::read(&path).expect("original still readable"), + corrupt_bytes, + "the corrupt record must be left exactly as it was found" + ); + } + + /// **Catches:** persistence-off reading as an empty funded set, and an inert registry claiming + /// it recorded something. + #[test] + fn no_state_directory_reports_not_configured_never_empty() { + let registry = FundedDistributorRegistry::disabled(); + + let read = registry.read(); + + assert_eq!( + read, + FundedDistributorsRead::NotConfigured(NotConfiguredReason::NoStateDirectory) + ); + assert_eq!( + read.determined(), + None, + "persistence off is unknown, not an empty funded set" + ); + assert_ne!(read, FundedDistributorsRead::FundsNothing); + assert_eq!( + registry.record(&FundedDistributor { + launcher_id: id(14), + store_id: None, + }), + RecordOutcome::NotConfigured(NotConfiguredReason::NoStateDirectory), + "an inert registry must say it recorded nothing rather than pretend it did" + ); + } + + /// **Catches:** a vanished state directory rendering as an empty funded set. + #[test] + fn a_missing_state_directory_reports_not_configured_distinctly_from_empty() { + let dir = TempDir::new().expect("temp dir"); + let gone = dir.path().join("never-created"); + let registry = FundedDistributorRegistry::with_state_dir(&gone); + + let read = registry.read(); + + assert_eq!( + read, + FundedDistributorsRead::NotConfigured(NotConfiguredReason::StateDirectoryMissing), + "a directory this node cannot see is unknown, not empty" + ); + assert_eq!(read.determined(), None); + assert_ne!(read, FundedDistributorsRead::FundsNothing); + } + + /// **Catches:** "nothing written yet" collapsed into the one renderable empty answer. + #[test] + fn an_existing_directory_with_no_record_is_not_configured_not_empty() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + + let read = registry.read(); + + assert_eq!( + read, + FundedDistributorsRead::NotConfigured(NotConfiguredReason::NoRecordWritten), + "nothing ever written is a different unknown from a record that says none" + ); + assert_eq!(read.determined(), None); + assert_ne!(read, FundedDistributorsRead::FundsNothing); + } + + /// **Catches:** a genuinely-empty record that cannot be told apart from a failure, which would + /// make an honest empty list unrenderable. + #[test] + fn a_genuinely_empty_record_reports_funds_nothing_and_is_renderable() { + let dir = TempDir::new().expect("temp dir"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + std::fs::write(record_path(&dir), r#"{"version": 1, "distributors": []}"#) + .expect("write an empty record"); + + let read = registry.read(); + + assert_eq!( + read, + FundedDistributorsRead::FundsNothing, + "an intact record naming nobody is the ONE legitimate empty answer" + ); + assert_eq!( + read.determined(), + Some(&[][..]), + "funds-nothing is the only outcome a caller may render as an empty list" + ); + } + + /// **Catches:** a future variant added to the not-an-answer half of + /// [`FundedDistributorsRead`] that `determined` reports as a renderable set. + #[test] + fn every_not_an_answer_outcome_is_undetermined() { + let undetermined = [ + FundedDistributorsRead::NotConfigured(NotConfiguredReason::NoStateDirectory), + FundedDistributorsRead::NotConfigured(NotConfiguredReason::StateDirectoryMissing), + FundedDistributorsRead::NotConfigured(NotConfiguredReason::NoRecordWritten), + FundedDistributorsRead::PersistedStateCorrupt { + path: PathBuf::from("x"), + quarantined_to: None, + }, + FundedDistributorsRead::IoFailed { + path: PathBuf::from("x"), + error: "denied".to_owned(), + }, + ]; + + for read in undetermined { + assert_eq!( + read.determined(), + None, + "{read:?} must not be renderable as a funded set" + ); + } + } +} diff --git a/crates/dig-node-core/src/rewards/mod.rs b/crates/dig-node-core/src/rewards/mod.rs index d80a21e7..787d0c7e 100644 --- a/crates/dig-node-core/src/rewards/mod.rs +++ b/crates/dig-node-core/src/rewards/mod.rs @@ -41,6 +41,7 @@ pub mod admission; pub mod challenge; pub mod cycle; +pub mod funded; pub mod gate; pub mod port; pub mod spec_constants; diff --git a/crates/dig-node-core/src/rewards/port.rs b/crates/dig-node-core/src/rewards/port.rs index d9844fe9..7eb2d5a0 100644 --- a/crates/dig-node-core/src/rewards/port.rs +++ b/crates/dig-node-core/src/rewards/port.rs @@ -1,12 +1,64 @@ //! The chain port — the seam this whole engine is built against instead of `dig-rewards-coin`. //! -//! `dig-rewards-coin` is SPEC-only as of the tag this lane read: `src/lib.rs` is a documented -//! placeholder and `pub mod distributor {}` is empty. Implementing the driver is -//! DIG-Network/dig_ecosystem#3249, a sibling lane. So the prover engine is built COMPLETELY against -//! a narrow trait derived from the SPEC's own described surface (not from the driver's internals, -//! so it is stable across #3249 landing), tested with an in-memory fake, and the production -//! adapter — until #3249 ships — reports [`ChainPortError::Unavailable`] and runs no cycles. See -//! [`unavailable`] for that adapter. +//! `dig-rewards-coin` was SPEC-only as of the tag this lane first read it: `src/lib.rs` was a +//! documented placeholder and `pub mod distributor {}` was empty. So the prover engine is built +//! COMPLETELY against a narrow trait derived from the SPEC's own described surface (not from the +//! driver's internals, so it is stable across the driver landing), tested with an in-memory fake, and +//! the production adapter reports [`ChainPortError::Unavailable`] and runs no cycles. See +//! [`UnavailableChainPort`] for that adapter. +//! +//! # dig_ecosystem#3269 unit 2 — the driver shipped, but still with no reader (blocking finding) +//! +//! `dig-rewards-coin` 0.2.0 is published and adds real types — `DistributorSnapshot` / +//! `DistributorSlots` (its `state` module) plus `clawback`, `comment`, `constants`, `eligibility`, +//! `entries`, `epoch`, `fund`, `launch`, `payout`. **It still ships no chain reader — blocker 1.** +//! 0.2.0's own `state.rs:1-31` module doc says so directly: SPEC §12.1's `read_distributor` "does not +//! publish one, deliberately" — the implementation that existed applied +//! `RewardDistributor::from_parent_spend` to the eve coin's spend (the launch inner puzzle) instead +//! of `from_eve_coin_spend`, so every read reported `Malformed`; the correct hop additionally needs +//! `reserve_parent_id`/`reserve_lineage_proof` provenance a reader starting from a launcher id cannot +//! currently discover. That is tracked as real design work at +//! — as of this unit, open, with a PR up +//! (`DIG-Network/dig-rewards-coin#6`, `feat/3267-chain-reader`, +950/-80, targeting `0.3.0`) — and +//! 0.2.0's own doc states the rule the future reader must honour: "every `ChainSource` error MUST +//! become `RewardsError::ChainUnavailable` … a distributor whose read failed MUST NOT render as 'no +//! entries' or 'nothing accrued'". Read this adapter against `0.3.0`'s actual reader shape when it +//! ships, not against this description. +//! +//! **Blocker 2, independent of #3267:** nothing in this codebase today records which distributors +//! this node funds. `funded_distributors` (below) needs that identity set as its starting point — +//! there is no chain-wide "list every distributor and filter to mine" call this crate can make. The +//! only adjacent registry is the CLAIM side's `ClaimChainPort::discover_distributors` in +//! `dig-node-service`'s `rewards_claim::port` — a **different trait**, filtering by mirror-admission +//! (which distributors this node might claim FROM), not by funder ownership (which distributors this +//! node funds); it is not a substitute. A repo-wide search for a funder-ownership registry — +//! `grep -rln "funded_launcher_ids\|FundedDistributor\|reward_distributor_registry\|create_distributor\|launch_distributor" crates/ --include=*.rs` +//! — returned **no matches** as of this unit's tip (worth re-running before assuming this is still +//! true; a negative search is a claim about a point in time, not a permanent fact). No launch flow, no +//! config, no persisted launcher-id list exists in this crate or in `dig-node-service` today. Tracked +//! as a separate ticket, parallel to #3267 (not downstream of it): a working reader tells a caller HOW +//! to read one distributor; it does not tell the caller WHICH launcher ids are its own. Both must land +//! before any of `dig.getRewardDistributor` / `dig.listRewardDistributorCommitments` / +//! `dig.listRewardDistributors`'s `funded` half can answer honestly. +//! +//! So a "real" `RewardsChainPort` adapter over 0.2.0 cannot honestly answer ANY of the four trait +//! methods with live chain data yet: `funded_distributors` has no identity source, and +//! `distributor_state`/`submit_entry_writes`/`spend_new_epoch` all need the withheld reader (a spend +//! needs the live singleton coin `read_distributor` would supply). Writing one anyway — either by +//! reimplementing `read_distributor` myself or by inventing a funded-distributor registry with no +//! writer — would be exactly the kind of restated, unreviewed money-shape work SPEC §0.1 clause 1 and +//! this crate's own withholding of a broken reader argue against, and is the shape fork this ticket's +//! kernel invariant 6 says to escalate rather than guess. Escalated to the L1, and settled: no new +//! adapter and no dispatch arm land until a reader (0.3.0+) and the funder-ownership registry both +//! exist. **`dig.listRewardDistributors` stays `-32601` deliberately** — serving it through +//! `UnavailableChainPort` was considered and rejected: it would be a false capability signal (a +//! feature-probe or `rpc.discover` reading the method as implemented when it always errors) and the +//! exact "dispatch surface with no function behind it" pattern DIG-Network/dig-node#593 was the last +//! PR allowed to land on. `UnavailableChainPort` remains the only production adapter for now — still +//! correct, since every real call would fail for one of the two reasons above regardless. No +//! `dig-rewards-coin` dependency is added by this unit: an unused dependency with no consumer is +//! inert weight and would want whichever version ships the reader (0.3.0+), not 0.2 — add it in the +//! unit that actually consumes it. use super::admission::AdmittedPeer; use async_trait::async_trait; diff --git a/crates/dig-node-service/Cargo.toml b/crates/dig-node-service/Cargo.toml index e340a965..ed7e3f94 100644 --- a/crates/dig-node-service/Cargo.toml +++ b/crates/dig-node-service/Cargo.toml @@ -116,6 +116,12 @@ dig-mirror-coin = "0.9" # a dissenting THIRD source and did not distinguish source CLASSES from bare source strings. dig-stun = "0.2" +# The claim loop's cadence jitter (DIG-Network/dig_ecosystem#3268, SPEC §8.6) draws from the OS +# CSPRNG, never a global/thread RNG -- `ring::rand::SystemRandom` is the same primitive this +# crate's own signing paths already use for randomness (`seams::dig_peer::holdings`, in +# `dig-node-core`), so this is not a second RNG choice entering the tree. +ring = "0.17" + # The canonical `ChainSource` trait `dig-mirror-coin`'s census is generic over. Declared, not # implemented: `chia-query` already provides the implementation this node uses # (`ChiaQueryProvider`), reached through `dig-wallet`'s one shared chain transport. Caret-matched diff --git a/crates/dig-node-service/src/rewards_claim/driver.rs b/crates/dig-node-service/src/rewards_claim/driver.rs new file mode 100644 index 00000000..858b978f --- /dev/null +++ b/crates/dig-node-service/src/rewards_claim/driver.rs @@ -0,0 +1,1381 @@ +//! Wires the claim engine onto a real background cadence, reachable from the node's actual +//! startup path (DIG-Network/dig_ecosystem#3268). Mirrors [`crate::self_heal`]'s split exactly: +//! a private injected-tick [`drive`] (testable under `#[tokio::test(start_paused = true)]`) behind +//! a pure, tested gate ([`decide_claim_driver`] / [`spawn_claim_driver_if`]) that `server.rs` calls +//! exactly once. +//! +//! # `enabled = true` must stop being a false statement +//! Before this module, nothing in the codebase ever constructed a [`super::ClaimEngine`] outside +//! its own tests (see [`super`]'s module doc, now updated). After it, a background task always +//! exists whenever `rewards_claim.enabled` and `enable_chain_sync` are both true, drives a cycle +//! every `cadence_seconds + jitter`, and its outcome is readable in-process via [`handle`] as a +//! NAMED [`super::ClaimLoopState`] — see [`ClaimLoopHandle`]. +//! +//! # Diverges from `self_heal::drive` on purpose: the FIRST pass waits for the interval +//! `self_heal::drive` fires its pass immediately, then once per fixed tick — right for a +//! maintenance sweep with no anti-silence surface. This driver's whole point (A2, the ticket's +//! headline acceptance item) is that "scheduler running, zero cycles ever fired" must be +//! DISTINGUISHABLE from "it ran" via a monotonic cycle counter that reads `0` before any interval +//! has elapsed. Running a pass at spawn, before the counter could ever read `0` under observation, +//! would defeat that on every startup. So [`drive`] sleeps `cadence_seconds + jitter` FIRST, then +//! runs a cycle, then repeats — the counter is genuinely `0` until the first interval elapses. +//! +//! # No RPC surface here (SCOPE) +//! [`handle`] is an IN-PROCESS accessor only — a future RPC (blocked on DIG-Network/dig_ecosystem#3249 +//! re-deriving the `ClaimStatus` wire semantics) can read it; this module puts nothing on the wire +//! and adds no RPC method, dispatch-table row or handler. +//! +//! # The only production adapter is [`super::UnavailableClaimChainPort`] +//! #3249 has not landed, so every real cycle this driver runs reports [`super::ClaimLoopState::ChainSourceUnavailable`] +//! and submits nothing — the honest state, not an invented adapter. + +use std::path::Path; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Mutex, OnceLock}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use chia_protocol::Bytes32; + +use super::cadence::{next_interval_seconds, JitterSource, CLAIM_JITTER_SECONDS_DEFAULT}; +use super::config::{RewardsClaimConfig, CLAIM_CADENCE_SECONDS_DEFAULT}; +use super::engine::ClaimEngine; +use super::hints::{DistributorHintSource, NoHintSource}; +use super::port::{ClaimChainPort, UnavailableClaimChainPort}; +use super::types::{ClaimLoopState, ClaimStatus}; + +/// The in-process accessor onto the running claim loop (SCOPE: never exposed over the wire here). +/// Cheap to clone -- every field is an `Arc`-backed handle onto the same shared state. +/// +/// # A2, the anti-silence test +/// [`Self::cycles_driven`] is the counter a reader compares against [`Self::status`]'s +/// [`super::ClaimLoopState`] to tell "constructed and spawned but never drove a cycle" (`0`, +/// `Idle`) apart from "ran and reported a real outcome" (`> 0`, whatever [`super::ClaimEngine`] +/// computed). Neither field alone would do it: `status()` before the first cycle is already +/// `Idle` BY DESIGN (see [`super::types::ClaimLoopState::Idle`]'s doc, "no cycle has ever been +/// attempted yet") -- that is the correct, honest reading, not a defect, and a test that only +/// checked `status()` for `Idle` could not tell a scheduler that never fires apart from one that +/// correctly reports nothing pending. The count is the only thing here that is monotonic and can +/// never be read as "healthy" by a writer describing itself. +#[derive(Clone, Default)] +pub struct ClaimLoopHandle { + status: std::sync::Arc>, + cycles_driven: std::sync::Arc, + refusal: std::sync::Arc>>, +} + +/// Why the driver never reached [`drive`]'s loop at all -- distinct from anything +/// [`super::ClaimLoopState`] can say, because every one of ITS states presupposes an engine that +/// exists and a cycle that was at least attempted. Without this, "disabled", "chain sync is off" +/// and "no operator wallet, so there is nothing to build an engine with" all collapse into the +/// same reassuring `Idle` + zero-count reading -- three different truths about whether this peer +/// is being paid, indistinguishable to an operator or a future `dig.getRewardClaimStatus`. Kept on +/// the DRIVER's own handle, never added to [`super::types::ClaimStatus`] (read-only, and it is the +/// wrong home: it is a fact about whether an engine exists, not about a cycle one ran). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClaimDriverRefusal { + /// `rewards_claim.enabled = false` -- the ordinary, deliberate off state. + Disabled, + /// `enabled = true` but `enable_chain_sync = false` -- see [`ClaimDriverDecision::ChainSyncDisabled`]'s doc. + ChainSyncDisabled, + /// `enabled = true`, chain sync is on, but this node has no operator wallet to derive + /// [`own_payout_puzzle_hash`] from -- there is no puzzle hash to build an engine with at all. + NoOperatorWallet, +} + +impl ClaimLoopHandle { + /// The most recent [`ClaimStatus`] any cycle has produced, or [`ClaimStatus::default`]'s + /// `Idle` state before the first one ever runs. + #[must_use] + pub fn status(&self) -> ClaimStatus { + *self + .status + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// How many times [`super::ClaimEngine::run_cycle`] has been invoked through this handle -- + /// incremented on EVERY invocation, whatever it returned (a refused, faulted or empty cycle + /// still counts: A1 requires an OBSERVED CYCLE COUNT, never "the task was spawned"). + #[must_use] + pub fn cycles_driven(&self) -> u64 { + self.cycles_driven.load(Ordering::SeqCst) + } + + /// Why no engine was ever built for this handle, or `None` when one was (whether or not it has + /// driven a cycle yet -- see [`ClaimDriverRefusal`]'s doc for the three-way collapse this + /// exists to prevent). + #[must_use] + pub fn refusal(&self) -> Option { + *self + .refusal + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Record why no engine will ever be built on this handle. Called only from + /// [`spawn_claim_driver_if`]'s non-`Spawn` branches and [`run_claim_driver`]'s + /// no-operator-wallet path. + fn set_refusal(&self, reason: ClaimDriverRefusal) { + *self + .refusal + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(reason); + } + + /// Record that one cycle was driven and publish its resulting status. Called only from + /// [`drive`], once per cycle, after `run_cycle` returns. + fn record(&self, status: ClaimStatus) { + *self + .status + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = status; + self.cycles_driven.fetch_add(1, Ordering::SeqCst); + } +} + +/// The process-wide handle to the running (or never-spawned) claim loop -- one per node process, +/// mirroring how [`crate::state::state_dir`] and friends are process-wide singletons. Initialized +/// lazily to its `Idle`/zero default so a reader (a future RPC, a test) never has to handle +/// "not spawned yet" as a THIRD state distinct from `Idle` -- it is the same state, honestly. +static HANDLE: OnceLock = OnceLock::new(); + +/// The in-process accessor a future RPC (blocked on #3249) reads. Never wired onto the wire here. +#[must_use] +pub fn handle() -> ClaimLoopHandle { + HANDLE.get_or_init(ClaimLoopHandle::default).clone() +} + +/// Drive the claim cadence: sleep `cadence_seconds + jitter` (drawn from `jitter`), run one cycle, +/// record it on `handle`, repeat forever. `now` and `jitter` are injected -- never a global RNG or +/// the clock read directly here -- so the schedule is deterministic and falsifiable under +/// `#[tokio::test(start_paused = true)]` (see this module's doc for why the FIRST pass waits +/// rather than firing immediately, unlike [`crate::self_heal::drive`]). +async fn drive( + mut engine: ClaimEngine, + cadence_seconds: u64, + jitter_seconds: u64, + jitter: &dyn JitterSource, + mut now: impl FnMut() -> u64, + handle: ClaimLoopHandle, +) where + P: ClaimChainPort, + H: DistributorHintSource, +{ + loop { + let interval = next_interval_seconds(cadence_seconds, jitter_seconds, jitter); + tokio::time::sleep(Duration::from_secs(interval)).await; + let t = now(); + engine.run_cycle(t).await; + let status = engine.status(); + handle.record(status); + log_cycle(&status, handle.cycles_driven()); + } +} + +/// Emit the ONE record that makes a driven cycle observable in a running node. +/// +/// Without this, the whole status surface has no reader in a shipped binary: [`handle`] is +/// in-process only and deliberately carries no RPC (deferred to DIG-Network/dig_ecosystem#3249), +/// so a node whose claim loop can never claim a single reward would produce output IDENTICAL to a +/// healthy one -- silence. A status nobody can read is a doc claim, not a measurement. +/// +/// [`ClaimLoopState::Nominal`] is the routine case (`info`). Every other state means this peer is +/// earning nothing and names why, which on a money surface is a warning, not chatter. +fn log_cycle(status: &ClaimStatus, cycles_driven: u64) { + if status.state == ClaimLoopState::Nominal { + tracing::info!( + target: "rewards_claim", + state = ?status.state, + cycles_driven, + distributors_known = status.distributors_known, + distributors_claimable = status.distributors_claimable, + claims_submitted = status.claims_submitted, + "claim cycle complete" + ); + } else { + tracing::warn!( + target: "rewards_claim", + state = ?status.state, + cycles_driven, + distributors_known = status.distributors_known, + distributors_claimable = status.distributors_claimable, + claims_submitted = status.claims_submitted, + concat!( + "claim cycle complete but this node is NOT claiming rewards -- see the named ", + "state for why" + ) + ); + } +} + +/// A jitter source drawing from the OS CSPRNG (`ring::rand::SystemRandom`, the same primitive +/// [`crate::mirror`]'s signing paths use for randomness in this crate) -- never a global/thread +/// RNG. A CSPRNG failure (the underlying OS call erroring) fails to jitter `0` rather than +/// panicking the driver: the worst case is every node's cadence landing exactly on +/// `cadence_seconds` with no spread, not a crashed claim loop. +struct OsJitter; + +impl JitterSource for OsJitter { + fn jitter_seconds(&self, bound: u64) -> u64 { + if bound == 0 { + return 0; + } + use ring::rand::SecureRandom; + let rng = ring::rand::SystemRandom::new(); + let mut buf = [0u8; 8]; + if rng.fill(&mut buf).is_err() { + return 0; + } + // `bound.saturating_add(1)` rather than `bound + 1`: `jitter_seconds` comes from the + // persisted config unclamped, so `u64::MAX` reaches here and `+ 1` would overflow-panic + // inside the detached driver task -- killing the claim loop silently for the process + // lifetime. Saturating keeps the draw in `0..=bound` for every input. + u64::from_le_bytes(buf) % bound.saturating_add(1) + } +} + +fn unix_now_seconds() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +/// A6, money-correctness: this node's own payout puzzle hash as the claim-chain entry slot +/// compares it. `$DIG` is a CAT, so the operator's coins -- and therefore the puzzle hash an +/// `InitiatePayout` should be admitted under -- sit at the canonical CAT wrapping of the owner's +/// inner puzzle hash, never the bare inner hash itself: the SAME derivation +/// [`crate::mirror::lifecycle`]'s `reclaimed_coin_id` and [`crate::mirror::funding::dig_cat_puzzle_hash`]'s +/// own doc use ("the operator's ordinary $DIG coins... sit at the canonical CAT wrapping... never +/// the bare owner puzzle hash"). Picking the unwrapped hash here would make every distributor +/// refuse this node's claims (`claims_refused_payout_mismatch`) while [`super::ClaimLoopState::compute_state`] +/// still reads `Nominal` when nothing is claimable at all -- exactly the misdirection this epic has +/// already measured. See this module's tests for the two-sided proof (a distributor paying to this +/// derivation claims; one paying the unwrapped hash is refused). +#[must_use] +pub fn own_payout_puzzle_hash(owner_inner_puzzle_hash: Bytes32) -> Bytes32 { + crate::mirror::funding::dig_cat_puzzle_hash(owner_inner_puzzle_hash) +} + +/// The real, detached claim-loop task: derive this node's own payout puzzle hash from its operator +/// wallet (the same public, no-unseal-required derivation [`crate::server::spawn_mirror_passes`] +/// falls back to), load [`RewardsClaimConfig`], build a [`ClaimEngine`] against the only +/// production adapter that exists ([`UnavailableClaimChainPort`] -- see this module's doc), and +/// drive it forever. +/// +/// Never called directly by `server.rs` -- see [`spawn_claim_driver_if`], the tested gate that +/// decides WHETHER to call this. `handle` is INJECTED (never the [`handle`] singleton read +/// directly) so a test can drive this against a private, non-shared handle instead of the +/// process-wide one. +async fn run_claim_driver(handle: ClaimLoopHandle) { + let paths = dig_wallet::autoseed::default_paths(); + let Some(owner_inner_puzzle_hash) = dig_wallet::operator_wallet::operator_puzzle_hash(&paths) + else { + tracing::warn!( + target: "rewards_claim", + "no operator wallet is available, so this node has no payout puzzle hash to claim \ + against; the claim loop is NOT started -- rewards_claim.enabled stays true but no \ + cycle will ever run until an operator wallet exists" + ); + handle.set_refusal(ClaimDriverRefusal::NoOperatorWallet); + return; + }; + let own_payout_puzzle_hash = own_payout_puzzle_hash(owner_inner_puzzle_hash); + + run_claim_driver_in( + &crate::state::state_dir(), + own_payout_puzzle_hash, + UnavailableClaimChainPort, + handle, + ) + .await; +} + +/// The whole production body of the claim loop, with every process global it used to read taken as +/// an argument: the state directory it loads [`RewardsClaimConfig`] from, this node's own payout +/// puzzle hash, and the chain `port`. Split out of [`run_claim_driver`] on the same `load` / +/// `load_from` pattern [`RewardsClaimConfig`] itself uses, for one reason: the joint between the +/// tested gate and the tested [`drive`] loop was previously the only UNTESTED link in the chain, +/// and an untested joint is exactly how #594's claim engine shipped complete and inert. +/// +/// Generic over `P` so a test can drive this real body against a fake port; production always +/// passes [`UnavailableClaimChainPort`] (see the module doc -- there is deliberately no second +/// production adapter until #3249 lands). +/// The largest schedule value this driver will honour, in seconds: 31 days. Chosen to sit +/// comfortably above every documented default -- [`CLAIM_CADENCE_SECONDS_DEFAULT`] is 86,400s +/// (1 day) and [`CLAIM_JITTER_SECONDS_DEFAULT`] is 3,600s (1 hour) -- and above any plausible +/// operator choice ("claim monthly" is 30 days), while excluding every value that means NEVER. +/// +/// # Why a ceiling exists at all +/// [`next_interval_seconds`] saturates rather than panicking, so a persisted +/// `jitter_seconds = u64::MAX` (or a cadence of the same shape) no longer crashes the node -- it +/// schedules the next cycle roughly 585 billion years out. That is strictly WORSE than a panic for +/// this ticket: the claim loop never fires again, so no cycle, no `log_cycle` line, and the +/// cycle counter reads a permanent, reassuring `0`. #594 shipped an engine that was inert and +/// green; a config value must not be able to put this driver back in that state silently. +const CLAIM_SCHEDULE_SECONDS_MAX: u64 = 31 * 24 * 60 * 60; + +/// Replace a schedule value that would switch the loop off (or spin it) with its documented +/// default, saying so at `WARN` -- never silently accept it, and never silently accept the +/// default either. Returns `(cadence_seconds, jitter_seconds)` fit to schedule with. +/// +/// A zero cadence is rejected for the opposite reason to a huge one: it would busy-loop the claim +/// engine as fast as the runtime can poll it. A zero JITTER is legitimate (it means "no jitter") +/// and is left alone. +fn sanitized_schedule(cadence_seconds: u64, jitter_seconds: u64) -> (u64, u64) { + let cadence = if cadence_seconds == 0 || cadence_seconds > CLAIM_SCHEDULE_SECONDS_MAX { + tracing::warn!( + target: "rewards_claim", + field = "cadence_seconds", + rejected = cadence_seconds, + substituted = CLAIM_CADENCE_SECONDS_DEFAULT, + max = CLAIM_SCHEDULE_SECONDS_MAX, + "{}", + concat!( + "rewards_claim.cadence_seconds is outside the honoured range and was IGNORED; ", + "the documented default is used instead -- a value that large would stop the ", + "claim loop from ever firing again, and zero would busy-loop it" + ) + ); + CLAIM_CADENCE_SECONDS_DEFAULT + } else { + cadence_seconds + }; + + let jitter = if jitter_seconds > CLAIM_SCHEDULE_SECONDS_MAX { + tracing::warn!( + target: "rewards_claim", + field = "jitter_seconds", + rejected = jitter_seconds, + substituted = CLAIM_JITTER_SECONDS_DEFAULT, + max = CLAIM_SCHEDULE_SECONDS_MAX, + "{}", + concat!( + "rewards_claim.jitter_seconds is outside the honoured range and was IGNORED; ", + "the documented default is used instead -- a value that large saturates the ", + "next interval and the claim loop would never fire again" + ) + ); + CLAIM_JITTER_SECONDS_DEFAULT + } else { + jitter_seconds + }; + + (cadence, jitter) +} + +async fn run_claim_driver_in

( + state_dir: &Path, + own_payout_puzzle_hash: Bytes32, + port: P, + handle: ClaimLoopHandle, +) where + P: ClaimChainPort, +{ + let cfg = RewardsClaimConfig::load_from(state_dir); + // A4/F8: a corrupt config is not a reason to refuse to SPAWN -- `ClaimEngine::run_cycle` + // already fails closed and reports `PersistedStateCorrupt` by name on every cycle until an + // operator fixes or removes the file (see `engine.rs`'s `run_cycle` doc). Refusing to spawn + // here instead would report NOTHING at all, which is the exact silent failure this ticket + // exists to prevent -- a corrupt file must stay visible, not vanish into "never started". + + // The config is operator-writable and unclamped at rest (`config.rs` deliberately reports what + // is on disk). Sanitize HERE, at the read, before either value can reach the scheduler. + let (cadence_seconds, jitter_seconds) = + sanitized_schedule(cfg.cadence_seconds, cfg.jitter_seconds); + + let engine = ClaimEngine::new( + port, + NoHintSource, + own_payout_puzzle_hash, + cfg.max_fee_mojos, + cfg.max_cycle_fee_budget_mojos, + dig_mirror_coin::DIG_ASSET_ID, + ) + .with_rotation_cursor(cfg.rotation_cursor) + .with_persisted_fee_window(state_dir, cadence_seconds); + + drive( + engine, + cadence_seconds, + jitter_seconds, + &OsJitter, + unix_now_seconds, + handle, + ) + .await; +} + +/// Spawn the real claim-loop task, detached, against `handle` -- injected, never the [`handle`] +/// singleton read from inside, so the only place the process-wide singleton is named is +/// [`spawn_claim_driver_from_config`]. +fn spawn_claim_driver(handle: ClaimLoopHandle) { + tokio::spawn(run_claim_driver(handle)); +} + +/// Why [`spawn_claim_driver_if`] declined to spawn -- named so the caller can log a reason instead +/// of silence (A3). `Disabled` is the ordinary, expected off state (`rewards_claim.enabled = +/// false`); `ChainSyncDisabled` is the one that matters most, because it is reachable with +/// `enabled = true` -- exactly the shape this ticket exists to close: an operator who reads +/// `enabled: true` and believes claims are running, on a node where `enable_chain_sync` is off. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ClaimDriverDecision { + Spawn, + Disabled, + ChainSyncDisabled, +} + +/// The pure decision behind [`spawn_claim_driver_if`] -- no I/O, no logging, so a test can assert +/// every branch directly. `enabled` gates on its own (SPEC-level opt-out); `enable_chain_sync` is +/// gated the same way `spawn_collateral_census` and `mirror::bond_verify::spawn_bond_verifier_install` +/// already are in `server.rs` -- that flag already means "this node talks to the Chia network", and +/// an integration harness sets it false precisely so nothing dials. +fn decide_claim_driver(enabled: bool, enable_chain_sync: bool) -> ClaimDriverDecision { + if !enabled { + return ClaimDriverDecision::Disabled; + } + if !enable_chain_sync { + return ClaimDriverDecision::ChainSyncDisabled; + } + ClaimDriverDecision::Spawn +} + +/// The single wiring seam `server.rs`'s `serve_with_shutdown` calls (A1's "exact precedent": +/// `self_heal::spawn_driver_if`). `spawn` is invoked exactly when [`decide_claim_driver`] returns +/// `Spawn`; every other branch logs its reason instead of spawning silently (A3) and leaves +/// `handle` at its already-honest `Idle` default -- never a third, undocumented state. +/// +/// `handle` is INJECTED rather than read from the process-wide [`handle`] singleton, so a test can +/// assert the recorded [`ClaimDriverRefusal`] of each branch in-process, on a private handle, with +/// no cross-test interference from a `OnceLock` that outlives the test that touched it. +fn spawn_claim_driver_if( + enabled: bool, + enable_chain_sync: bool, + handle: &ClaimLoopHandle, + spawn: impl FnOnce(), +) { + match decide_claim_driver(enabled, enable_chain_sync) { + ClaimDriverDecision::Spawn => spawn(), + ClaimDriverDecision::Disabled => { + tracing::debug!( + target: "rewards_claim", + "rewards_claim.enabled=false; the claim loop is not started" + ); + handle.set_refusal(ClaimDriverRefusal::Disabled); + } + ClaimDriverDecision::ChainSyncDisabled => { + tracing::warn!( + target: "rewards_claim", + "rewards_claim.enabled=true but enable_chain_sync=false; the claim loop is NOT \ + started -- rewards_claim.enabled is a false statement on this node until chain \ + sync is enabled" + ); + handle.set_refusal(ClaimDriverRefusal::ChainSyncDisabled); + } + } +} + +/// Reads [`RewardsClaimConfig::load`] (the node's own state-dir config) and `enable_chain_sync`, +/// and calls [`spawn_claim_driver_if`] -- the exact one call `serve_with_shutdown` makes. +pub fn spawn_claim_driver_from_config(enable_chain_sync: bool) { + let cfg = RewardsClaimConfig::load(); + // The ONE place the process-wide singleton is read: everything below it takes an injected + // handle so it stays testable in-process. + let process_handle = handle(); + let driver_handle = process_handle.clone(); + spawn_claim_driver_if(cfg.enabled, enable_chain_sync, &process_handle, move || { + spawn_claim_driver(driver_handle); + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + use std::sync::atomic::AtomicUsize; + use std::sync::Arc; + + use super::super::port::ClaimPortError; + use super::super::types::{DiscoveredDistributor, OwnEntry}; + + // ---- decide_claim_driver / spawn_claim_driver_if (A3) ---------------------------------- + + #[test] + fn disabled_never_spawns() { + assert_eq!( + decide_claim_driver(false, true), + ClaimDriverDecision::Disabled + ); + assert_eq!( + decide_claim_driver(false, false), + ClaimDriverDecision::Disabled + ); + } + + #[test] + fn enabled_but_chain_sync_off_refuses_named() { + assert_eq!( + decide_claim_driver(true, false), + ClaimDriverDecision::ChainSyncDisabled + ); + } + + #[test] + fn enabled_and_chain_sync_on_spawns() { + assert_eq!(decide_claim_driver(true, true), ClaimDriverDecision::Spawn); + } + + #[test] + fn gate_invokes_spawn_only_on_the_spawn_decision() { + let spawned = Arc::new(AtomicUsize::new(0)); + let handle = ClaimLoopHandle::default(); + + let s = spawned.clone(); + spawn_claim_driver_if(false, true, &handle, || { + s.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!(spawned.load(Ordering::SeqCst), 0, "enabled=false: no spawn"); + + let s = spawned.clone(); + spawn_claim_driver_if(true, false, &handle, || { + s.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!( + spawned.load(Ordering::SeqCst), + 0, + "enabled=true, chain sync off: no spawn" + ); + + let s = spawned.clone(); + spawn_claim_driver_if(true, true, &handle, || { + s.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!( + spawned.load(Ordering::SeqCst), + 1, + "enabled+chain sync: spawns" + ); + } + + /// ACCEPTANCE A3: the gate's two refusals are READABLE off the injected handle and distinct + /// from each other -- not both collapsed into the same zero-cycle `Idle` reading. The third + /// refusal (`NoOperatorWallet`) is proven in + /// `a_missing_operator_wallet_is_a_distinct_named_refusal` below; the fourth truth, + /// "spawned and running but never ticked", is `refusal() == None` with `cycles_driven() == 0`, + /// asserted here and driven past zero in + /// `zero_cycles_before_the_interval_elapses_then_a_counted_number_after`. + #[test] + fn each_refusal_is_readable_and_distinct_on_the_injected_handle() { + let disabled = ClaimLoopHandle::default(); + assert_eq!( + disabled.refusal(), + None, + "nothing refused before the gate runs" + ); + spawn_claim_driver_if(false, true, &disabled, || {}); + assert_eq!(disabled.refusal(), Some(ClaimDriverRefusal::Disabled)); + + let chain_off = ClaimLoopHandle::default(); + spawn_claim_driver_if(true, false, &chain_off, || {}); + assert_eq!( + chain_off.refusal(), + Some(ClaimDriverRefusal::ChainSyncDisabled) + ); + + let spawned = ClaimLoopHandle::default(); + spawn_claim_driver_if(true, true, &spawned, || {}); + assert_eq!( + spawned.refusal(), + None, + "a spawned driver has refused nothing: the fourth truth, `refusal() == None` with a zero cycle count" + ); + assert_eq!(spawned.cycles_driven(), 0); + + // The four readings are pairwise distinct, which is the whole point of A3: three refusals + // plus "running but never ticked" are four different answers to "is this peer being paid". + let readings = [ + disabled.refusal(), + chain_off.refusal(), + Some(ClaimDriverRefusal::NoOperatorWallet), + spawned.refusal(), + ]; + for (i, a) in readings.iter().enumerate() { + for b in &readings[i + 1..] { + assert_ne!(a, b, "two driver refusals must never read the same"); + } + } + } + + /// ACCEPTANCE A3 (third refusal): the no-operator-wallet path records its OWN named reason on + /// the handle rather than leaving `Idle` + zero cycles, and drives no cycle. `run_claim_driver` + /// reads the real default wallet paths, so this asserts the refusal only when this machine + /// genuinely has no operator wallet; where one exists the driver legitimately proceeds and the + /// refusal stays `None` -- either way the reading is a NAMED one, never a silent `Idle`. + #[tokio::test] + async fn a_missing_operator_wallet_is_a_distinct_named_refusal() { + let paths = dig_wallet::autoseed::default_paths(); + if dig_wallet::operator_wallet::operator_puzzle_hash(&paths).is_some() { + return; // this machine HAS an operator wallet; the refusal branch is unreachable here + } + let handle = ClaimLoopHandle::default(); + run_claim_driver(handle.clone()).await; + assert_eq!( + handle.refusal(), + Some(ClaimDriverRefusal::NoOperatorWallet), + "no operator wallet must be a named refusal, not a reassuring Idle" + ); + assert_eq!(handle.cycles_driven(), 0, "and it must drive no cycle"); + } + + // ---- A1 + A2: the anti-silence cycle counter through the real drive() loop ------------- + + /// A fake port whose every call succeeds with an empty/zero answer -- enough to let + /// `run_cycle` reach `Nominal` every time, so the driven-cycle counter is exercised against a + /// REAL completed cycle, not just an early `ChainSourceUnavailable` return. + struct EmptyPort; + + #[async_trait] + impl ClaimChainPort for EmptyPort { + async fn discover_distributors( + &self, + ) -> Result, ClaimPortError> { + Ok(Vec::new()) + } + async fn resolve_launch_comment( + &self, + _launcher_id: Bytes32, + ) -> Result, ClaimPortError> { + Ok(None) + } + async fn reserve_asset_id(&self, _launcher_id: Bytes32) -> Result { + Ok(Bytes32::from([0u8; 32])) + } + async fn payout_threshold(&self, _launcher_id: Bytes32) -> Result { + Ok(0) + } + async fn own_entry( + &self, + _launcher_id: Bytes32, + _payout_puzzle_hash: Bytes32, + ) -> Result, ClaimPortError> { + Ok(None) + } + async fn required_fee_mojos(&self, _launcher_id: Bytes32) -> Result { + Ok(0) + } + async fn submit_initiate_payout( + &self, + _launcher_id: Bytes32, + _payout_puzzle_hash: Bytes32, + _fee_mojos: u64, + ) -> Result<(), ClaimPortError> { + Ok(()) + } + } + + fn empty_engine() -> ClaimEngine { + ClaimEngine::new( + EmptyPort, + NoHintSource, + Bytes32::from([1u8; 32]), + 1, + 10, + Bytes32::from([2u8; 32]), + ) + } + + async fn settle() { + for _ in 0..8 { + tokio::task::yield_now().await; + } + } + + /// ACCEPTANCE A1 + A2 (the anti-silence test): a scheduler that is running but whose interval + /// has never elapsed must read `cycles_driven() == 0` -- NOT "spawn returned", an observed + /// count. Advancing the clock past `cadence_seconds + jitter` must then drive `run_cycle` a + /// counted number of times. + #[tokio::test(start_paused = true)] + async fn zero_cycles_before_the_interval_elapses_then_a_counted_number_after() { + let cadence = 100u64; + let handle = ClaimLoopHandle::default(); + let h = handle.clone(); + let driver = tokio::spawn(async move { + drive( + empty_engine(), + cadence, + 0, + &super::super::cadence::FixedJitter(0), + { + let mut t = 0u64; + move || { + t += cadence; + t + } + }, + h, + ) + .await; + }); + + settle().await; + assert_eq!( + handle.cycles_driven(), + 0, + "THE ANTI-SILENCE TEST: a scheduler that is running but has never fired a cycle must \ + report a driven-cycle count of 0, not silence and not a false 'ran' reading" + ); + + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + assert_eq!( + handle.cycles_driven(), + 1, + "one interval elapsed, one cycle driven" + ); + assert_eq!( + handle.status().state, + super::super::types::ClaimLoopState::Nominal, + "the driven cycle's real outcome is readable, not just its count" + ); + + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + assert_eq!( + handle.cycles_driven(), + 2, + "a second interval drives a second cycle" + ); + + driver.abort(); + } + + // ---- A3: enabled=false vs. enabled=true+gate-refused are both zero-cycle, named states - + + #[tokio::test(start_paused = true)] + async fn disabled_config_never_drives_a_cycle_via_the_configured_seam() { + let dir = tempfile::tempdir().unwrap(); + let cfg = RewardsClaimConfig { + enabled: false, + ..RewardsClaimConfig::default() + }; + cfg.save_to(dir.path()).unwrap(); + // The gate itself (not the full production seam, which reads the process-wide state dir) + // is what's under test here -- see `gate_invokes_spawn_only_on_the_spawn_decision` above + // for the direct proof that `enabled=false` never calls `spawn`. + assert_eq!( + decide_claim_driver(cfg.enabled, true), + ClaimDriverDecision::Disabled + ); + } + + // ---- A6: own_payout_puzzle_hash is the CAT-wrapped hash, proven against the engine ------ + + /// A distributor whose recorded entry is keyed to THIS node's own payout derivation is + /// claimable; one keyed to the bare, unwrapped owner puzzle hash is refused + /// (`PayoutPuzzleHashMismatch`) -- proving `own_payout_puzzle_hash` computes the CAT-wrapped + /// hash the engine's `own_entry` comparison expects, not the raw inner hash. + struct OneDistributorPort { + entry_keyed_to: Bytes32, + dig_asset_id: Bytes32, + } + + #[async_trait] + impl ClaimChainPort for OneDistributorPort { + async fn discover_distributors( + &self, + ) -> Result, ClaimPortError> { + Ok(vec![DiscoveredDistributor { + launcher_id: Bytes32::from([9u8; 32]), + store_id: Bytes32::from([0u8; 32]), + root: Bytes32::from([0u8; 32]), + }]) + } + async fn resolve_launch_comment( + &self, + _launcher_id: Bytes32, + ) -> Result, ClaimPortError> { + Ok(None) + } + async fn reserve_asset_id(&self, _launcher_id: Bytes32) -> Result { + Ok(self.dig_asset_id) + } + async fn payout_threshold(&self, _launcher_id: Bytes32) -> Result { + Ok(1) + } + async fn own_entry( + &self, + _launcher_id: Bytes32, + // Ignored ON PURPOSE (see the NOTE below): this fake always hands back the entry keyed + // to `entry_keyed_to`, so the ENGINE's own comparison decides claimable vs. refused. + _payout_puzzle_hash: Bytes32, + ) -> Result, ClaimPortError> { + Ok(Some(OwnEntry { + payout_puzzle_hash: self.entry_keyed_to, + counter: 0, + accrued_base_units: 1_000, + })) + // NOTE: `_payout_puzzle_hash` (the argument the engine passed in, this node's own + // derivation) is ignored on purpose -- this fake always hands back the entry keyed to + // `entry_keyed_to`, so the engine's OWN comparison (`entry.payout_puzzle_hash != + // self.own_payout_puzzle_hash`) is what decides claimable vs. refused, exactly the + // real chain behaviour this proves against. + } + async fn required_fee_mojos(&self, _launcher_id: Bytes32) -> Result { + Ok(0) + } + async fn submit_initiate_payout( + &self, + _launcher_id: Bytes32, + _payout_puzzle_hash: Bytes32, + _fee_mojos: u64, + ) -> Result<(), ClaimPortError> { + Ok(()) + } + } + + #[tokio::test] + async fn a_distributor_paying_this_nodes_derivation_is_claimable() { + let owner_inner = Bytes32::from([7u8; 32]); + let wrapped = own_payout_puzzle_hash(owner_inner); + let asset_id = Bytes32::from([3u8; 32]); + let mut engine = ClaimEngine::new( + OneDistributorPort { + entry_keyed_to: wrapped, + dig_asset_id: asset_id, + }, + NoHintSource, + wrapped, + 1_000_000, + 10_000_000, + asset_id, + ); + let outcomes = engine.run_cycle(1).await; + assert_eq!(outcomes.len(), 1); + assert!( + matches!( + outcomes[0], + super::super::types::ClaimOutcome::Submitted { .. } + ), + "a distributor keyed to the CAT-wrapped derivation must be claimable, got {:?}", + outcomes[0] + ); + } + + #[tokio::test] + async fn a_distributor_paying_the_unwrapped_hash_is_refused() { + let owner_inner = Bytes32::from([7u8; 32]); + let asset_id = Bytes32::from([3u8; 32]); + let wrapped = own_payout_puzzle_hash(owner_inner); + let mut engine = ClaimEngine::new( + OneDistributorPort { + // Keyed to the RAW inner hash -- the wrong derivation -- not the wrapped one. + entry_keyed_to: owner_inner, + dig_asset_id: asset_id, + }, + NoHintSource, + wrapped, + 1_000_000, + 10_000_000, + asset_id, + ); + let outcomes = engine.run_cycle(1).await; + assert_eq!(outcomes.len(), 1); + assert!( + matches!( + outcomes[0], + super::super::types::ClaimOutcome::PayoutPuzzleHashMismatch { .. } + ), + "a distributor keyed to the unwrapped hash must be refused, got {:?}", + outcomes[0] + ); + } + + // ---- A5: restart safety + clock movement, through the real persisted-config path -------- + + #[tokio::test] + async fn restart_with_a_recent_completion_skips_via_cadence_not_elapsed() { + let dir = tempfile::tempdir().unwrap(); + let cfg = RewardsClaimConfig { + cadence_seconds: 1_000, + last_cycle_completed_at: Some(500), + ..RewardsClaimConfig::default() + }; + cfg.save_to(dir.path()).unwrap(); + + let mut engine = ClaimEngine::new( + EmptyPort, + NoHintSource, + Bytes32::from([1u8; 32]), + 1, + 10, + Bytes32::from([2u8; 32]), + ) + .with_persisted_fee_window(dir.path(), cfg.cadence_seconds); + + let outcomes = engine.run_cycle(900).await; // 900 - 500 = 400 < 1_000 + assert!(outcomes.is_empty()); + assert_eq!( + engine.status().state, + super::super::types::ClaimLoopState::CadenceNotElapsed, + "a crash-restart loop must not immediately re-run a cycle that already ran" + ); + } + + #[tokio::test] + async fn restart_with_an_elapsed_completion_runs_a_cycle() { + let dir = tempfile::tempdir().unwrap(); + let cfg = RewardsClaimConfig { + cadence_seconds: 1_000, + last_cycle_completed_at: Some(500), + ..RewardsClaimConfig::default() + }; + cfg.save_to(dir.path()).unwrap(); + + let mut engine = ClaimEngine::new( + EmptyPort, + NoHintSource, + Bytes32::from([1u8; 32]), + 1, + 10, + Bytes32::from([2u8; 32]), + ) + .with_persisted_fee_window(dir.path(), cfg.cadence_seconds); + + let outcomes = engine.run_cycle(2_000).await; // 2_000 - 500 = 1_500 >= 1_000 + assert!(outcomes.is_empty(), "nothing to claim, but the cycle RAN"); + assert_eq!( + engine.status().state, + super::super::types::ClaimLoopState::Nominal + ); + } + + #[tokio::test] + async fn a_future_dated_completion_fails_closed_not_underflowed() { + let dir = tempfile::tempdir().unwrap(); + let cfg = RewardsClaimConfig { + cadence_seconds: 1_000, + last_cycle_completed_at: Some(10_000), // in the future relative to `now` below + ..RewardsClaimConfig::default() + }; + cfg.save_to(dir.path()).unwrap(); + + let mut engine = ClaimEngine::new( + EmptyPort, + NoHintSource, + Bytes32::from([1u8; 32]), + 1, + 10, + Bytes32::from([2u8; 32]), + ) + .with_persisted_fee_window(dir.path(), cfg.cadence_seconds); + + let outcomes = engine.run_cycle(100).await; // now < last_cycle_completed_at + assert!(outcomes.is_empty()); + assert_eq!( + engine.status().state, + super::super::types::ClaimLoopState::PersistedStateCorrupt, + "a future-dated clock must fail CLOSED, never compute a negative/underflowed interval" + ); + } + + // ---- A4: corrupt = true (via a torn file through load_from), never engine::corrupt set -- + + #[tokio::test] + async fn a_torn_config_file_never_runs_a_cycle() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("rewards-claim.json"), b"{ not json").unwrap(); + + let cfg = RewardsClaimConfig::load_from(dir.path()); + assert!( + cfg.corrupt, + "load_from must observe the torn file as corrupt" + ); + + let mut engine = ClaimEngine::new( + EmptyPort, + NoHintSource, + Bytes32::from([1u8; 32]), + 1, + 10, + Bytes32::from([2u8; 32]), + ) + .with_persisted_fee_window(dir.path(), 1_000); + + let outcomes = engine.run_cycle(1).await; + assert!(outcomes.is_empty()); + assert_eq!( + engine.status().state, + super::super::types::ClaimLoopState::PersistedStateCorrupt + ); + } + // ---- The JOINT: the production body itself, not the halves around it ------------------- + + /// Write a `rewards-claim.json` with a fixed cadence and NO jitter, so a composition test can + /// advance the clock by an exact number of seconds and know precisely how many cycles that + /// buys. + fn write_config(dir: &Path, cadence_seconds: u64) { + RewardsClaimConfig { + enabled: true, + cadence_seconds, + jitter_seconds: 0, + ..RewardsClaimConfig::default() + } + .save_to(dir) + .unwrap(); + } + + /// THE COMPOSITION TEST. `decide_claim_driver` was tested, `drive` was tested -- and the + /// production body that joins them (`run_claim_driver_in`: load the config from the state dir, + /// construct the engine, reach `drive`) was tested by NOTHING. That is the same shape as #594, + /// which shipped a complete, fully-tested, entirely INERT claim engine: if this body returned + /// early, built the engine wrong, or never reached `drive`, every other test on this change + /// would still pass and a real node would still never claim. + /// + /// So this drives the REAL body -- the one production calls -- and asserts the anti-silence + /// property through it: zero cycles before the configured interval elapses, then an exactly + /// COUNTED number after. + #[tokio::test(start_paused = true)] + async fn the_production_body_drives_counted_cycles_from_a_written_config() { + let cadence = 100u64; + let dir = tempfile::tempdir().unwrap(); + write_config(dir.path(), cadence); + + let handle = ClaimLoopHandle::default(); + let h = handle.clone(); + let state_dir = dir.path().to_path_buf(); + let driver = tokio::spawn(async move { + run_claim_driver_in(&state_dir, Bytes32::from([1u8; 32]), EmptyPort, h).await; + }); + + settle().await; + assert_eq!( + handle.cycles_driven(), + 0, + "the production body must honour the configured interval: no cycle before it elapses" + ); + + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + assert_eq!( + handle.cycles_driven(), + 1, + concat!( + "one configured interval elapsed: the production body drove exactly one cycle, ", + "proving the joint between the tested gate and the tested drive loop is live" + ) + ); + + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + assert_eq!( + handle.cycles_driven(), + 2, + "and it keeps driving, one cycle per configured interval" + ); + + driver.abort(); + } + + /// The same production body against the port production ACTUALLY passes it + /// ([`UnavailableClaimChainPort`], the only adapter until #3249) reports + /// [`ClaimLoopState::ChainSourceUnavailable`] by name once a cycle has been driven -- the + /// honest state of a real node today. Proves the real adapter path is reached, not only a fake + /// one: a counted cycle whose outcome names the missing chain source, never a reassuring + /// `Nominal` and never silence. + #[tokio::test(start_paused = true)] + async fn the_production_adapter_reports_chain_source_unavailable_by_name() { + let cadence = 100u64; + let dir = tempfile::tempdir().unwrap(); + write_config(dir.path(), cadence); + + let handle = ClaimLoopHandle::default(); + let h = handle.clone(); + let state_dir = dir.path().to_path_buf(); + let driver = tokio::spawn(async move { + run_claim_driver_in( + &state_dir, + Bytes32::from([1u8; 32]), + UnavailableClaimChainPort, + h, + ) + .await; + }); + + // Let the spawned body reach its first `sleep` before advancing: under paused time an + // `advance` that lands before the timer is registered buys no cycle at all. + settle().await; + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + assert_eq!(handle.cycles_driven(), 1, "one cycle was driven"); + assert_eq!( + handle.status().state, + super::super::types::ClaimLoopState::ChainSourceUnavailable, + "with no chain adapter wired, the driven cycle must name ChainSourceUnavailable" + ); + assert_eq!( + handle.refusal(), + None, + "the loop RAN: an unavailable chain source is a cycle outcome, not a refusal to start" + ); + + driver.abort(); + } + + // ---- the cycle log: the only reader of the status surface in a shipped binary ---------- + + /// An in-memory sink a `tracing_subscriber::fmt` layer renders records into, so a test can + /// assert what a running node would actually print (the same pattern `never_log.rs` and + /// `server.rs` use for their log assertions). + #[derive(Clone)] + struct CapturedLogs(Arc>>); + + impl std::io::Write for CapturedLogs { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0 + .lock() + .expect("the capture buffer") + .extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs { + type Writer = CapturedLogs; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + impl CapturedLogs { + fn rendered(&self) -> String { + String::from_utf8(self.0.lock().expect("the capture buffer").clone()) + .expect("the rendered lines are utf-8") + } + } + + /// Install a capturing subscriber for the duration of the returned guard. `set_default` is + /// thread-local, and `#[tokio::test]` runs a current-thread runtime, so the driver task + /// spawned below is polled on this very thread and its records land in the buffer. + fn capture_logs() -> (CapturedLogs, tracing::subscriber::DefaultGuard) { + let buffer = CapturedLogs(Arc::new(Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::fmt() + .with_writer(buffer.clone()) + .with_ansi(false) + .without_time() + .finish(); + let guard = tracing::subscriber::set_default(subscriber); + (buffer, guard) + } + + /// THE ACCEPTANCE BAR: a driven cycle is OBSERVABLE, not merely readable through an + /// in-process handle nothing in the shipped binary calls. A healthy cycle says so at `INFO`, + /// naming its state and its cycle count. + #[tokio::test(start_paused = true)] + async fn a_driven_cycle_emits_an_event_naming_its_state() { + let cadence = 100u64; + let (logs, _guard) = capture_logs(); + let handle = ClaimLoopHandle::default(); + let h = handle.clone(); + let driver = tokio::spawn(async move { + drive( + empty_engine(), + cadence, + 0, + &super::super::cadence::FixedJitter(0), + { + let mut t = 0u64; + move || { + t += cadence; + t + } + }, + h, + ) + .await; + }); + + settle().await; + assert_eq!( + logs.rendered(), + "", + "no interval has elapsed, so there is nothing to report yet" + ); + + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + driver.abort(); + + let rendered = logs.rendered(); + assert!( + rendered.contains("Nominal"), + "the event must NAME the state a reader has to act on; got: {rendered}" + ); + assert!( + rendered.contains("cycles_driven=1"), + "and the cycle count that distinguishes a running loop from a stalled one; got: {}", + rendered + ); + assert!( + rendered.contains("rewards_claim"), + "under the module's own target, so it can be filtered on; got: {rendered}" + ); + } + + /// A cycle that CANNOT claim -- today's real production path, with no chain adapter wired -- + /// must be a WARNING naming the state, not an `INFO` line that reads like health. This is the + /// defect the whole ticket exists to remove: silence covering a permanent inability to earn. + #[tokio::test(start_paused = true)] + async fn a_cycle_that_cannot_claim_warns_and_names_why() { + let cadence = 100u64; + let (logs, _guard) = capture_logs(); + let handle = ClaimLoopHandle::default(); + let h = handle.clone(); + let driver = tokio::spawn(async move { + drive( + ClaimEngine::new( + UnavailableClaimChainPort, + NoHintSource, + Bytes32::from([1u8; 32]), + 1, + 10, + Bytes32::from([2u8; 32]), + ), + cadence, + 0, + &super::super::cadence::FixedJitter(0), + { + let mut t = 0u64; + move || { + t += cadence; + t + } + }, + h, + ) + .await; + }); + + settle().await; + tokio::time::advance(Duration::from_secs(cadence)).await; + settle().await; + driver.abort(); + + let rendered = logs.rendered(); + assert!( + rendered.contains("WARN"), + "earning nothing is a warning, not routine chatter; got: {rendered}" + ); + assert!( + rendered.contains("ChainSourceUnavailable"), + "and it must name WHY this node is not claiming; got: {rendered}" + ); + } + + /// `jitter_seconds` is read from the persisted config WITHOUT a clamp, so the maximum `u64` + /// reaches `OsJitter`. An overflowing `bound + 1` there panics the detached driver task, + /// which never restarts -- the claim loop would die silently for the process lifetime. + #[test] + fn an_unclamped_max_jitter_bound_does_not_panic_the_driver() { + let bound = std::hint::black_box(u64::MAX); + let offset = OsJitter.jitter_seconds(bound); + assert!( + offset <= bound, + "the draw must stay within 0..=bound; got {offset}" + ); + } + + // ---- the config-read sanitizer: a value must not be able to switch the loop off --------- + + /// Write a config with BOTH schedule fields chosen by the caller, so a test can persist a + /// value production would otherwise honour to the letter. + fn write_schedule_config(dir: &Path, cadence_seconds: u64, jitter_seconds: u64) { + RewardsClaimConfig { + enabled: true, + cadence_seconds, + jitter_seconds, + ..RewardsClaimConfig::default() + } + .save_to(dir) + .unwrap(); + } + + /// An out-of-range `cadence_seconds` must be REPLACED by the documented default, and the + /// substitution must be visible: a value this large means "never fire again", and silently + /// honouring it reopens #594's inert-but-green shape one level up, in the config file. + #[test] + fn an_out_of_range_cadence_is_replaced_by_the_default_and_warned() { + let (logs, _guard) = capture_logs(); + + let rejected = std::hint::black_box(u64::MAX); + let (cadence, jitter) = sanitized_schedule(rejected, 0); + + assert_eq!( + cadence, CLAIM_CADENCE_SECONDS_DEFAULT, + "an out-of-range cadence must fall back to the documented default" + ); + assert_eq!(jitter, 0, "a legitimate zero jitter is left alone"); + + let rendered = logs.rendered(); + assert!( + rendered.contains("WARN"), + "ignoring a configured value is a warning, not routine chatter; got: {rendered}" + ); + assert!( + rendered.contains("cadence_seconds"), + "the warning must name the FIELD that was ignored; got: {rendered}" + ); + assert!( + rendered.contains(&rejected.to_string()) + && rendered.contains(&CLAIM_CADENCE_SECONDS_DEFAULT.to_string()), + "and both the rejected and the substituted value; got: {rendered}" + ); + } + + /// The same for `jitter_seconds` -- and, through the REAL production body, that the loop still + /// drives counted cycles instead of never firing again. Without the sanitizer, + /// `next_interval_seconds` saturates on this value and no cycle is ever driven: green, silent + /// and unpaid. + #[tokio::test(start_paused = true)] + async fn an_out_of_range_jitter_is_replaced_and_the_loop_still_drives_cycles() { + let (logs, _guard) = capture_logs(); + + let cadence = 100u64; + let dir = tempfile::tempdir().unwrap(); + write_schedule_config(dir.path(), cadence, std::hint::black_box(u64::MAX)); + + let handle = ClaimLoopHandle::default(); + let h = handle.clone(); + let state_dir = dir.path().to_path_buf(); + let driver = tokio::spawn(async move { + run_claim_driver_in(&state_dir, Bytes32::from([1u8; 32]), EmptyPort, h).await; + }); + + settle().await; + // The substituted jitter is the DEFAULT hour, so one interval is at most cadence + 3600s. + tokio::time::advance(Duration::from_secs(cadence + CLAIM_JITTER_SECONDS_DEFAULT)).await; + settle().await; + + assert!( + handle.cycles_driven() >= 1, + concat!( + "an out-of-range jitter must not switch the claim loop off: with the default ", + "substituted, at least one cycle is driven within cadence + the default jitter" + ) + ); + + let rendered = logs.rendered(); + assert!( + rendered.contains("WARN") && rendered.contains("jitter_seconds"), + "the ignored jitter field must be named at WARN; got: {rendered}" + ); + assert!( + rendered.contains(&CLAIM_JITTER_SECONDS_DEFAULT.to_string()), + "and the substituted default must be readable; got: {rendered}" + ); + + driver.abort(); + } +} diff --git a/crates/dig-node-service/src/rewards_claim/mod.rs b/crates/dig-node-service/src/rewards_claim/mod.rs index 5d5d4a57..7804be4c 100644 --- a/crates/dig-node-service/src/rewards_claim/mod.rs +++ b/crates/dig-node-service/src/rewards_claim/mod.rs @@ -32,19 +32,22 @@ //! prevent (SPEC §2.4): with the unavailable adapter wired, zero claims IS the true state, so the //! status surface must say so by name, not by omission. //! -//! # Not yet wired into node startup (Defect D — stated, not fixed here) -//! Nothing in this codebase constructs a [`ClaimEngine`] outside this module's own tests: there is -//! no scheduler that drives [`ClaimEngine::run_cycle`] on a cadence, and no RPC method exposes -//! [`ClaimStatus`] to an operator, even though [`RewardsClaimConfig::enabled`] defaults to `true`. -//! Wiring this into node startup — picking a concrete [`ClaimChainPort`] adapter, starting the -//! cadence loop, and exposing `ClaimStatus` over RPC — is a separate unit of work with its own -//! review surface, deferred out of this PR on purpose: the only production adapter available today -//! is [`UnavailableClaimChainPort`], and the real one arrives with -//! DIG-Network/dig_ecosystem#3249. Until that wiring lands, this module compiles, is fully tested -//! against the fake chain port, and does nothing in a running node. +//! # Wired into node startup (DIG-Network/dig_ecosystem#3268) +//! [`driver::spawn_claim_driver_from_config`] is the one call `dig-node-service::server`'s +//! `serve_with_shutdown` makes: it is gated on `RewardsClaimConfig::enabled` AND +//! `Config::enable_chain_sync` (the same flag `spawn_collateral_census` and +//! `mirror::bond_verify::spawn_bond_verifier_install` already gate on), and when both are true it +//! spawns a detached task that drives [`ClaimEngine::run_cycle`] on a jittered cadence forever. +//! [`driver::handle`] is the IN-PROCESS accessor a future RPC can read once DIG-Network/dig_ecosystem#3249 +//! lands a real [`ClaimChainPort`] adapter and the `ClaimStatus` wire semantics are re-derived +//! against it — this module puts nothing on the wire itself (see `driver`'s own module doc for +//! why). Until #3249 lands, the only production adapter is still [`UnavailableClaimChainPort`], so +//! every real cycle reports [`ClaimLoopState::ChainSourceUnavailable`] and submits nothing — the +//! honest state, not a silent no-op. mod cadence; mod config; +mod driver; mod engine; mod hints; mod parser; @@ -56,6 +59,7 @@ pub use config::{ RewardsClaimConfig, CLAIM_CADENCE_SECONDS_DEFAULT, CLAIM_CYCLE_FEE_BUDGET_MOJOS_DEFAULT, CLAIM_FEE_CEILING_MOJOS_DEFAULT, }; +pub use driver::{handle, spawn_claim_driver_from_config, ClaimDriverRefusal, ClaimLoopHandle}; pub use engine::ClaimEngine; pub use hints::{DistributorHint, DistributorHintSource, NoHintSource}; pub use parser::parse_launch_comment; diff --git a/crates/dig-node-service/src/server.rs b/crates/dig-node-service/src/server.rs index a646f962..cec18b25 100644 --- a/crates/dig-node-service/src/server.rs +++ b/crates/dig-node-service/src/server.rs @@ -2204,6 +2204,17 @@ where // (a tested unit, #1864) so it cannot be silently flipped to always- or never-spawn. crate::self_heal::spawn_driver_if_service(); + // The peer reward-claim loop (DIG-Network/dig_ecosystem#3268, #3251): drives + // `rewards_claim::ClaimEngine::run_cycle` on a jittered cadence so + // `RewardsClaimConfig::enabled = true` stops being a false statement. Gated the same way the + // census and bond-verifier spawns above are -- `enable_chain_sync` already means "this node + // talks to the Chia network", and a harness sets it false precisely so nothing dials. The + // service-gate lives inside the seam (a tested unit, mirroring `self_heal::spawn_driver_if`) + // so it cannot be silently flipped to always- or never-spawn. The only production chain + // adapter until DIG-Network/dig_ecosystem#3249 lands is `UnavailableClaimChainPort`, so every + // real cycle reports `ChainSourceUnavailable` and submits nothing -- the honest state. + crate::rewards_claim::spawn_claim_driver_from_config(config.enable_chain_sync); + // Best-effort wallet mTLS listener (#368, Sage byte-parity, node-class clients, §5.3). Binds // loopback only on [`DEFAULT_MTLS_PORT`], which is deliberately NOT Sage's own RPC port // (dig-node#260). A bind failure is NON-FATAL — the wallet stays reachable over the plain-HTTP