diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index 53728f35..090c55a0 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -10227,6 +10227,65 @@ mod tests { ); } + /// **Proves:** `funded.observed_at` dates the CONSULTATION (the oldest per-item chain read), + /// never the assembly (dig_ecosystem#3323). The handler's pre-loop clock predates every read + /// `port.distributor_report` performs, so stamping it there understates staleness; the fix + /// folds the reports' own `observed_at` and keeps the oldest (a collection is only as fresh + /// as its stalest member). `claimable.observed_at` is untouched — no read happens for it, so + /// the handler's own clock remains the honest answer for that `NotConsulted` arm. + #[test] + fn list_reward_distributors_funded_observed_at_is_the_oldest_report_stamp() { + let state_dir = tempfile::tempdir().unwrap(); + let registry = + crate::rewards::funded::FundedDistributorRegistry::with_state_dir(state_dir.path()); + for launcher_id in [[0x55u8; 32], [0x66u8; 32]] { + assert_eq!( + registry.record(&crate::rewards::funded::FundedDistributor { + launcher_id, + store_id: None, + }), + crate::rewards::funded::RecordOutcome::Recorded + ); + } + let (node, _td) = test_node(None); + assert!(node.install_funded_distributor_registry(registry)); + + let report_a = sample_distributor_report(0x55, vec![]); + let report_b = sample_distributor_report(0x66, vec![]); + assert!( + node.install_reward_chain_port(Arc::new(FakeRewardsChainPort { + reports: std::collections::HashMap::from([ + ([0x55u8; 32], Ok(report_a.clone())), + ([0x66u8; 32], Ok(report_b.clone())), + ]), + })) + ); + + let resp = rt().block_on(handle_rpc( + &node, + json!({"jsonrpc":"2.0","id":1,"method":"dig.listRewardDistributors"}), + crate::download::ReadOrigin::Local, + crate::download::RequestProvenance::FirstParty, + )); + assert_eq!(resp["result"]["funded"]["outcome"], json!("consulted")); + assert_eq!( + resp["result"]["funded"]["observed_at"], + json!(1_700_200_000u64 + 0x55) + ); + assert_eq!( + resp["result"]["funded"]["observed_at"], + json!(report_a.observed_at) + ); + assert_ne!( + resp["result"]["funded"]["observed_at"], + json!(report_b.observed_at) + ); + assert_eq!( + resp["result"]["claimable"]["outcome"], + json!("not_consulted") + ); + } + /// **Proves:** a funded identity whose per-item chain report fails refuses the WHOLE call /// (the same `ChainPortError` response the sibling reward-distributor handlers use), rather /// than emitting a partial list or a fabricated ref (dig_ecosystem#3308/#3309). diff --git a/crates/dig-node-core/src/rewards/funded.rs b/crates/dig-node-core/src/rewards/funded.rs index b19afb3b..56dcd1e2 100644 --- a/crates/dig-node-core/src/rewards/funded.rs +++ b/crates/dig-node-core/src/rewards/funded.rs @@ -213,7 +213,7 @@ impl FundedDistributorRegistry { }, 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 }, + Err(LoadFailure::Io(error)) => self.report_io_failed(path, error), } } @@ -231,7 +231,7 @@ impl FundedDistributorRegistry { }, Ok(None) => Vec::new(), Err(LoadFailure::Corrupt(reason)) => return self.refuse_corrupt(&path, &reason), - Err(LoadFailure::Io(error)) => return RecordOutcome::IoFailed { path, error }, + Err(LoadFailure::Io(error)) => return self.refuse_io_failed(path, error), }; let outcome = match merge(&mut set, distributor) { @@ -243,7 +243,7 @@ impl FundedDistributorRegistry { } match self.save(&path, &set) { Ok(()) => outcome, - Err(error) => RecordOutcome::IoFailed { path, error }, + Err(error) => self.refuse_io_failed(path, error), } } @@ -327,6 +327,31 @@ impl FundedDistributorRegistry { } } + /// Log and report an unreadable/unwritable record to a READER. Mirrors [`Self::report_corrupt`] + /// so an operator sees the same signal for either fault: `PersistedStateCorrupt` has logged + /// since v0.257.0, but `IoFailed` was constructed bare beside it, and `dispatch.rs` collapses + /// both into one wire `NotConsulted` with no log of its own (dig_ecosystem#3324) — so the + /// operator saw nothing for a permissions error, a missing mount, or any other I/O fault. + fn report_io_failed(&self, path: PathBuf, error: String) -> FundedDistributorsRead { + tracing::error!( + path = %path.display(), + error, + "the funded-distributor record could not be read" + ); + FundedDistributorsRead::IoFailed { path, error } + } + + /// Log and report an unreadable/unwritable record to a WRITER. Mirrors [`Self::refuse_corrupt`]; + /// see [`Self::report_io_failed`] for why this arm was silent (dig_ecosystem#3324). + fn refuse_io_failed(&self, path: PathBuf, error: String) -> RecordOutcome { + tracing::error!( + path = %path.display(), + error, + "the funded-distributor record could not be read or written" + ); + RecordOutcome::IoFailed { path, error } + } + /// 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 { @@ -817,6 +842,69 @@ mod tests { ); } + /// **Proves:** an `IoFailed` read logs the path and the error, mirroring + /// [`FundedDistributorRegistry::report_corrupt`]'s `tracing::error!` (dig_ecosystem#3324 — the + /// corrupt arm has logged since v0.257.0; the I/O arm was constructed bare, so an operator saw + /// nothing for a permissions error or a missing mount). Forces `IoFailed` by making the record + /// PATH a directory, so `read_to_string` fails with a non-`NotFound` error — the wildcard + /// `PersistedStateCorrupt`/`IoFailed` collapse this fixture must not trip is + /// `dispatch.rs`'s wire mapping, untouched here; this test drives the registry directly (not + /// `handle_rpc`), because `rt()` may run the handler on a worker thread a thread-scoped + /// subscriber never sees. + #[test] + fn an_io_failed_read_is_logged_with_its_path_and_error() { + /// An in-memory sink a `tracing_subscriber::fmt` layer writes formatted records into. + #[derive(Clone, Default)] + struct LogCapture(std::sync::Arc>>); + + impl std::io::Write for LogCapture { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for LogCapture { + type Writer = LogCapture; + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } + } + + let dir = TempDir::new().expect("temp dir"); + std::fs::create_dir(record_path(&dir)).expect("make the record path a directory"); + let registry = FundedDistributorRegistry::with_state_dir(dir.path()); + + let buffer = LogCapture::default(); + let subscriber = tracing_subscriber::fmt() + .with_ansi(false) // plain text: the assertions read the fields as an operator would + .with_writer(buffer.clone()) + .finish(); + + // Scoped to this thread only, deliberately: `read()` runs synchronously here, never on a + // worker thread a thread-scoped subscriber would miss (dig_ecosystem#3324's note on why + // this test does not go through `handle_rpc`). + let read = tracing::subscriber::with_default(subscriber, || registry.read()); + + assert!( + matches!(read, FundedDistributorsRead::IoFailed { .. }), + "a directory at the record path must fail as IoFailed, got {read:?}" + ); + let logged = String::from_utf8(buffer.0.lock().unwrap().clone()).unwrap(); + let expected_path = record_path(&dir); + assert!( + logged.contains(&expected_path.display().to_string()), + "log did not name the record path: {logged}" + ); + assert!( + logged.contains("error"), + "log did not name the error field: {logged}" + ); + } + /// **Catches:** a regression on the not-an-answer half of [`FundedDistributorsRead`] listed /// below reporting a renderable set. It does NOT by itself catch a future variant added to the /// enum — that would need adding here too. The guard that actually forces the issue is diff --git a/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs b/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs index 4f1ecf55..d30d6eae 100644 --- a/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs +++ b/crates/dig-node-core/src/seams/dig_rpc/dispatch.rs @@ -1167,6 +1167,12 @@ impl RpcDispatch for Node { }; let mut funded_refs = Vec::with_capacity(identities.len()); + // `observed_at` must date the CONSULTATION, never the assembly (dig_ecosystem#3323, + // dig-rpc-protocol SPEC §4.4.2's first bullet): each `report.observed_at` postdates + // the pre-loop `now` above (the chain port stamps it after its own read, uncached), + // so stamping the handler's clock here understates staleness. Fold the reports' own + // stamps and keep the OLDEST — a collection is only as fresh as its stalest member. + let mut oldest_observed_at: Option = None; for identity in identities { let Some(port) = node.reward_chain_port() else { return reward_chain_port_absent_response(&id); @@ -1179,6 +1185,10 @@ impl RpcDispatch for Node { Ok(report) => report, Err(e) => return reward_chain_port_error_response(&id, &e), }; + oldest_observed_at = Some(match oldest_observed_at { + Some(oldest) => oldest.min(report.observed_at), + None => report.observed_at, + }); funded_refs.push(dig_rpc_protocol::types::RewardDistributorRef { launcher_id: hex::encode(report.launcher_id), store_id: hex::encode(report.store_id), @@ -1188,7 +1198,10 @@ impl RpcDispatch for Node { let result = dig_rpc_protocol::types::ListRewardDistributorsResult { funded: dig_rpc_protocol::types::Half::Consulted { - observed_at: now, + // No reads happened for `FundsNothing` (empty `identities`), so the + // pre-loop handler clock is the honest stamp for that case — it IS the + // consultation. Both `NotConsulted` arms above keep `now` unchanged. + observed_at: oldest_observed_at.unwrap_or(now), items: funded_refs, }, claimable: dig_rpc_protocol::types::Half::NotConsulted { observed_at: now },