From bca9ae3c4d398e9a0d5c0ee4b48a83a4f5bbfd3c Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 06:41:36 -0700 Subject: [PATCH 1/8] chore(digsex): open the dig-sex wiring lane Stub commit so the branch, PR and issue log exist before implementation. Co-Authored-By: Claude --- crates/dig-node-core/src/lib.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index b0a7f1e3..f56c8c14 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -15767,3 +15767,5 @@ mod tests { } } } + +// WIP(loop/batch-digsex): wiring dig-sex conduct/admission/acquisition/reward into the node. From 4546e33daba6267792b4e5e813bea3de677650de Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 07:01:40 -0700 Subject: [PATCH 2/8] feat(admission): meter inbound peer work per authenticated identity (#269) Wires `dig_sex::admission` (SPEC 8.5) into the mTLS peer surface. The meter was implemented, tested and gating nothing: the node did the work first and had no admission step at all. `PeerAdmission` holds one node-wide `AdmissionMeter` and is consulted at the top of `NodeResponder::handle_json_rpc` and `handle_availability`, ahead of the method allowlist and every dispatch, so a refused request costs a hex decode rather than a read, a decode or a DHT lookup. Two properties the shape enforces rather than documents: - The meter key is the mTLS-verified `peer_id`, decoded via the existing `hex64`. A session with no verified identity is REFUSED, never admitted unmetered and never metered under a placeholder -- a constant key collapses every requestor into one bucket, so one peer would exhaust the allowance for everybody. - `AdmissionGuard` releases on `Drop`, so allowance returns on every exit path including the error paths a hand-written release is forgotten on. Co-Authored-By: Claude --- crates/dig-node-core/src/lib.rs | 37 ++ crates/dig-node-core/src/peer.rs | 38 ++ .../src/seams/dig_peer/admission.rs | 332 ++++++++++++++++++ .../dig-node-core/src/seams/dig_peer/mod.rs | 1 + 4 files changed, 408 insertions(+) create mode 100644 crates/dig-node-core/src/seams/dig_peer/admission.rs diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index f56c8c14..dd3fd3f2 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -386,6 +386,12 @@ pub struct Node { /// registry. The registry's distinct-generation cap ([`crate::seams::dig_peer::DEFAULT_MAX_CONCURRENT_WARMS`]) /// therefore bounds concurrent acquisitions across BOTH legs, not each in isolation. capsule_acquisition: Arc, + /// Inbound admission for the mTLS peer surface (dig-sex SPEC 8.5, #269). + /// + /// One meter for the whole NODE, not one per connection or per responder: the ceiling it enforces + /// is node-wide, and a per-connection meter would let a peer buy more allowance simply by opening + /// more connections. + peer_admission: Arc, /// A WEAK self-reference, installed by the standalone peer-network bring-up (which holds the /// `Arc`), so a `&self` read handler can spawn a detached background task that needs an owned /// `Arc` — the capsule backfill (§14.3). `Weak` (not `Arc`) so the node's refcount is @@ -4425,6 +4431,13 @@ impl Node { /// ([`Node::maybe_backfill_capsule`]) and the #1576 reshare warm. Handed to /// [`crate::download::NodeContent::wire_capsule_reshare`] so both legs claim the same registry and a /// read triggers at most one whole-capsule acquisition. + /// The node-wide inbound admission meter (dig-sex SPEC 8.5, #269). + pub(crate) fn peer_admission( + &self, + ) -> &Arc { + &self.peer_admission + } + pub(crate) fn capsule_acquisition_gate(&self) -> Arc { self.capsule_acquisition.clone() } @@ -4506,6 +4519,9 @@ impl Node { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -4797,6 +4813,9 @@ pub(crate) mod test_support { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5585,6 +5604,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5717,6 +5739,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5783,6 +5808,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5874,6 +5902,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5947,6 +5978,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -8622,6 +8656,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index ba718cbb..4f33bf7d 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -1183,6 +1183,20 @@ where } } +/// The JSON-RPC error body for inbound work refused at the admission boundary (#269). +/// +/// `-32000` (server error) rather than a method/param error: the request was well-formed and the node +/// simply declined to spend on it now. The reason names the LIMIT, never the peer's standing, so shed +/// load stays distinguishable from a ban by the peer and from an outage by the operator. +fn admission_refused( + id: Value, + refusal: crate::seams::dig_peer::admission::AdmissionRefusal, +) -> Value { + tracing::debug!(reason = refusal.reason(), "peer serve: inbound work refused at admission"); + json!({"jsonrpc":"2.0","id":id, + "error":{"code":-32000,"message":"request refused","data":{"reason":refusal.reason()}}}) +} + /// Whether `method` may be answered over the **mTLS peer surface** (other DIG nodes). /// /// The allowlist itself lives in ONE place — [`dig_rpc_protocol::Method::is_peer_reachable`], @@ -1325,6 +1339,18 @@ impl PeerRpcResponder for NodeResponder { async fn handle_json_rpc(&self, req: Value, conn_key: &str) -> Value { let method = req.get("method").and_then(Value::as_str).unwrap_or(""); let id = req.get("id").cloned().unwrap_or(json!(1)); + // ADMISSION (dig-sex SPEC 8.5, #269) — ahead of the allowlist and every dispatch below, so a + // refused request costs a hex decode and a counter bump rather than a read, a decode or a DHT + // lookup. The guard is held for the whole method: it releases on `Drop`, including on the + // early `return`s below, which is why no path here needs to remember to. + let _admitted = match self.node.peer_admission().admit( + conn_key, + dig_sex::WorkKind::Own, + 1, + ) { + Ok(guard) => guard, + Err(refusal) => return admission_refused(id, refusal), + }; // PEER-SURFACE ALLOWLIST (audit #179 CRITICAL). The mTLS verifier accepts any self-signed // leaf, so an "authenticated" peer is merely "some peer_id", NOT an authorized admin. Route // ONLY the intended L7 read/discovery/announce methods to the shared dispatch; return -32601 @@ -1388,6 +1414,18 @@ impl PeerRpcResponder for NodeResponder { } async fn handle_availability(&self, items: Value, conn_key: &str) -> Value { + // ADMISSION (dig-sex SPEC 8.5, #269) — before the items are even read. `items.len()` is the + // attacker-chosen quantity this request asks for, so it is what gets clamped at the boundary + // (`AdmissionLimits::max_request_units`); clamping it deeper in would already have paid for it. + let requested_units = u32::try_from(items.as_array().map_or(0, Vec::len)).unwrap_or(u32::MAX); + let _admitted = match self.node.peer_admission().admit( + conn_key, + dig_sex::WorkKind::Own, + requested_units, + ) { + Ok(guard) => guard, + Err(refusal) => return admission_refused(json!(1), refusal), + }; let items = items.as_array().cloned().unwrap_or_default(); // The verified mTLS peer_id (`conn_key`) keys the per-requestor miss-lookup budget, identical // to the range-stream miss on this same peer surface (dig_ecosystem#2007). diff --git a/crates/dig-node-core/src/seams/dig_peer/admission.rs b/crates/dig-node-core/src/seams/dig_peer/admission.rs new file mode 100644 index 00000000..a0f56c9d --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/admission.rs @@ -0,0 +1,332 @@ +//! Inbound admission for the mTLS peer surface (dig-sex SPEC §8.5, dig-node#269). +//! +//! `dig_sex::admission` decides whether inbound work is admitted; this module is the node's half — +//! it derives the authenticated identity to meter against, holds the meter, and makes the paired +//! release structural. +//! +//! # Admit BEFORE the work, not after +//! +//! The value of an admission meter is that it refuses before spending anything. A check placed after +//! the read/decode/fetch has already paid the cost it exists to avoid, so [`PeerAdmission::admit`] is +//! called at the top of each responder method, ahead of every dispatch. +//! +//! # Metered by the AUTHENTICATED identity, never by anything the caller chooses +//! +//! The meter key is the mTLS-verified `peer_id` the session derived — never a wire field, never a +//! connection counter, never a constant. A key a caller can choose freely turns a per-peer limit into +//! a limit on whichever bucket the caller picks, and a CONSTANT key collapses every requestor into one +//! shared bucket, so a single peer exhausts the allowance for everybody. That failure looks exactly +//! like working DoS protection until the moment it matters, which is why +//! [`dig_sex::AuthenticatedPeer`] is a newtype and why this module refuses rather than substituting a +//! placeholder when no verified identity exists. +//! +//! # An unauthenticated peer-surface request is REFUSED, not admitted unmetered +//! +//! In production every session reaching the responder carries a verified `peer_id` (the peer surface is +//! mTLS-only), so an absent one means the session carried no authenticated caller at all. Admitting it +//! unmetered would make "present no identity" the cheapest way to escape the meter — the guard would +//! be optional at the attacker's discretion. It is therefore refused as [`Refusal::MeterFull`]'s +//! sibling case, [`AdmissionRefusal::Unauthenticated`]. Nothing on the loopback admin / in-process FFI +//! path routes through here (that is `crate::handle_rpc`), so refusing costs no legitimate caller. +//! +//! # Release is structural, not remembered +//! +//! [`AdmissionGuard`] releases on `Drop`. A `release` skipped on an error path leaks allowance until +//! the node refuses everything, and an error path is precisely the one a hand-written release is +//! forgotten on — so the guard makes the omission unrepresentable rather than reviewable. + +use std::sync::Mutex; + +use dig_sex::{AdmissionLimits, AdmissionMeter, AuthenticatedPeer, Refusal, WorkKind}; + +use super::dht::hex64; + +/// Why inbound work was refused at the boundary. +/// +/// Wraps the crate's [`Refusal`] so the node can name the one case the crate cannot: a request with no +/// authenticated identity to meter against. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdmissionRefusal { + /// The session carried no mTLS-verified `peer_id`, so there is no identity to meter against. + Unauthenticated, + /// The crate's meter refused; the variant says which limit was reached. + Limited(Refusal), +} + +impl AdmissionRefusal { + /// A short, stable reason string for the JSON-RPC error body and the serve log. + /// + /// Deliberately names the LIMIT rather than the peer's standing: shed load must be + /// distinguishable from an outage by an operator reading a log, and from a ban by the peer + /// reading the response. + #[must_use] + pub const fn reason(self) -> &'static str { + match self { + AdmissionRefusal::Unauthenticated => "unauthenticated", + AdmissionRefusal::Limited(Refusal::GlobalCeiling) => "node at capacity", + AdmissionRefusal::Limited(Refusal::PeerShare) => "peer at capacity", + AdmissionRefusal::Limited(Refusal::RelayBudget) => "relay budget exhausted", + AdmissionRefusal::Limited(Refusal::MeterFull) => "meter full", + AdmissionRefusal::Limited(Refusal::RequestTooLarge) => "request too large", + } + } +} + +/// The authenticated identity to meter against, derived from the session's verified `peer_id`. +/// +/// `conn_key` is the mTLS-verified peer id as lowercase 64-hex, empty on a caller-less session. Only a +/// well-formed 64-hex value yields an identity: anything else is not a verified peer id, and coercing +/// it into one would meter distinct callers into whatever bucket the malformed value happened to hash +/// to. +#[must_use] +pub fn authenticated_peer(conn_key: &str) -> Option { + hex64(conn_key).map(AuthenticatedPeer::from_verified_session) +} + +/// One admitted unit of inbound work. Releases its slot on `Drop`, on every exit path. +/// +/// Held by value across the work it admits; the borrow checker then makes "return before releasing" +/// impossible rather than merely discouraged. +pub struct AdmissionGuard<'a> { + admission: &'a PeerAdmission, + peer: AuthenticatedPeer, + kind: WorkKind, +} + +impl Drop for AdmissionGuard<'_> { + fn drop(&mut self) { + // A poisoned meter is recovered rather than propagated: refusing to RELEASE on the panic path + // would leak the very allowance this guard exists to return, permanently. + let mut meter = self + .admission + .meter + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + meter.release(self.peer, self.kind); + } +} + +/// The node's inbound admission meter, shared across every peer session. +/// +/// One meter for the whole node, because the ceiling it enforces is node-wide: a per-connection meter +/// would let a peer buy more allowance by opening more connections, which is the same collapse as a +/// caller-chosen key wearing a different shape. +#[derive(Debug)] +pub struct PeerAdmission { + meter: Mutex, +} + +impl Default for PeerAdmission { + fn default() -> Self { + Self::new(AdmissionLimits::default()) + } +} + +impl PeerAdmission { + /// A meter with no work in flight, admitting under `limits`. + #[must_use] + pub fn new(limits: AdmissionLimits) -> Self { + Self { + meter: Mutex::new(AdmissionMeter::new(limits)), + } + } + + /// Admit one unit of inbound work for the session identified by `conn_key`, BEFORE performing it. + /// + /// `requested_units` is the attacker-chosen quantity the request asks for, clamped here at the + /// boundary rather than deeper in where the cost would already be committed. + /// + /// # Errors + /// + /// [`AdmissionRefusal::Unauthenticated`] when the session carried no verified peer id, or + /// [`AdmissionRefusal::Limited`] when a limit is reached. In both cases NO work has been done. + pub fn admit( + &self, + conn_key: &str, + kind: WorkKind, + requested_units: u32, + ) -> Result, AdmissionRefusal> { + let peer = authenticated_peer(conn_key).ok_or(AdmissionRefusal::Unauthenticated)?; + let mut meter = self + .meter + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + meter + .admit(peer, kind, requested_units) + .map_err(AdmissionRefusal::Limited)?; + drop(meter); + Ok(AdmissionGuard { + admission: self, + peer, + kind, + }) + } + + /// Units of work currently in flight node-wide. For tests and the operator status surface. + #[must_use] + pub fn in_flight_total(&self) -> u32 { + self.meter + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .in_flight_total() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A 64-hex conn_key that is distinct per `n` — the shape a real mTLS session yields. + fn conn_key(n: u8) -> String { + hex::encode([n; 32]) + } + + /// Limits with a per-peer share small enough to exhaust in a test, and a global ceiling well + /// ABOVE it — so exhausting one peer's share can only be the PER-PEER limit, never the node-wide + /// one. A test whose two ceilings are equal cannot tell those apart and would pass on a meter that + /// had collapsed every peer into one bucket, which is the exact defect this file exists to prevent. + fn limits() -> AdmissionLimits { + AdmissionLimits { + global_ceiling: 64, + per_peer_share: 2, + relay_ceiling: 16, + max_tracked_peers: 1024, + max_request_units: 8, + } + } + + /// **Proves (#269):** the meter is keyed on the AUTHENTICATED identity, so exhausting one peer's + /// share refuses that peer and leaves a second peer entirely unaffected. + /// + /// The second peer is the load-bearing half. Asserting only that an over-quota peer is refused + /// passes identically on a meter keyed by a CONSTANT — the shared-bucket mistake — because that + /// meter also refuses at the limit. It just refuses everybody. Varying one actor while keeping an + /// honest control is what distinguishes a per-peer limit from a global one wearing its clothes. + #[test] + fn exhausting_one_peers_share_does_not_refuse_a_different_peer() { + let admission = PeerAdmission::new(limits()); + let noisy = conn_key(0xaa); + let quiet = conn_key(0xbb); + + let _first = admission + .admit(&noisy, WorkKind::Own, 1) + .expect("first unit is within the share"); + let _second = admission + .admit(&noisy, WorkKind::Own, 1) + .expect("second unit is within the share"); + + assert_eq!( + admission.admit(&noisy, WorkKind::Own, 1).unwrap_err(), + AdmissionRefusal::Limited(Refusal::PeerShare), + "a third unit exceeds this peer's share and must be refused" + ); + assert!( + admission.admit(&quiet, WorkKind::Own, 1).is_ok(), + "a DIFFERENT peer must be unaffected — if it is refused, every requestor shares one \ + bucket and a single peer can exhaust the allowance for everybody" + ); + } + + /// **Proves (#269):** the guard releases on `Drop`, so allowance returns on every exit path. + /// + /// A `release` skipped on an error path leaks allowance until the node refuses everything, and the + /// error path is the one a hand-written release is forgotten on. Dropping the guard here stands in + /// for every such path, because `Drop` cannot distinguish them. + #[test] + fn a_dropped_guard_returns_its_allowance() { + let admission = PeerAdmission::new(limits()); + let peer = conn_key(0xcc); + + { + let _a = admission.admit(&peer, WorkKind::Own, 1).expect("first"); + let _b = admission.admit(&peer, WorkKind::Own, 1).expect("second"); + assert_eq!(admission.in_flight_total(), 2); + assert!( + admission.admit(&peer, WorkKind::Own, 1).is_err(), + "the share must be exhausted while the guards are held, or the release below \ + proves nothing" + ); + } + + assert_eq!( + admission.in_flight_total(), + 0, + "both guards went out of scope; neither slot may still be held" + ); + assert!( + admission.admit(&peer, WorkKind::Own, 1).is_ok(), + "the same peer must be admissible again once its work finished" + ); + } + + /// **Proves (#269):** a session with no verified peer id is REFUSED, not admitted unmetered. + /// + /// Admitting it would make "present no identity" the cheapest way to escape the meter, and the + /// alternative mistake — substituting a placeholder identity — is the shared bucket again. Both + /// non-64-hex shapes are covered: absent, and present-but-malformed. + #[test] + fn an_unauthenticated_session_is_refused_rather_than_metered_under_a_placeholder() { + let admission = PeerAdmission::new(limits()); + for bad in ["", "not-hex", &"ab".repeat(31), &"zz".repeat(32)] { + assert_eq!( + admission.admit(bad, WorkKind::Own, 1).unwrap_err(), + AdmissionRefusal::Unauthenticated, + "{bad:?} is not a verified peer id and must yield no admission" + ); + } + assert_eq!( + admission.in_flight_total(), + 0, + "a refused request must consume no allowance" + ); + } + + /// **Proves (#269):** the attacker-chosen request size is clamped AT the boundary. + /// + /// `max_request_units` is 8 here, so 9 is over and 8 is at the bound. Pinning both sides matters: + /// a bound tested only from below can only confirm itself, and a clamp that refused everything + /// would pass a one-sided test while denying all legitimate work. + #[test] + fn an_oversized_request_is_refused_and_the_at_bound_request_is_admitted() { + let admission = PeerAdmission::new(limits()); + let peer = conn_key(0xdd); + + assert_eq!( + admission.admit(&peer, WorkKind::Own, 9).unwrap_err(), + AdmissionRefusal::Limited(Refusal::RequestTooLarge), + "one unit over max_request_units must be refused" + ); + assert!( + admission.admit(&peer, WorkKind::Own, 8).is_ok(), + "the at-bound request must still be admitted, or the clamp denies legitimate work" + ); + } + + /// **Proves (#269):** relayed work draws on its own ceiling, so work done on other nodes' behalf + /// cannot consume the whole node-wide allowance (SPEC 6.1.8). + #[test] + fn relayed_work_exhausts_the_relay_ceiling_while_own_work_still_admits() { + let admission = PeerAdmission::new(AdmissionLimits { + global_ceiling: 64, + per_peer_share: 8, + relay_ceiling: 1, + max_tracked_peers: 1024, + max_request_units: 8, + }); + let peer = conn_key(0xee); + + let _relayed = admission + .admit(&peer, WorkKind::Relayed, 1) + .expect("the first relayed unit is within the relay ceiling"); + assert_eq!( + admission.admit(&peer, WorkKind::Relayed, 1).unwrap_err(), + AdmissionRefusal::Limited(Refusal::RelayBudget), + "the relay ceiling is 1 and must be reached" + ); + assert!( + admission.admit(&peer, WorkKind::Own, 1).is_ok(), + "OWN work draws on a separate budget — a spent relay allowance must not stop this node \ + serving its own callers" + ); + } +} diff --git a/crates/dig-node-core/src/seams/dig_peer/mod.rs b/crates/dig-node-core/src/seams/dig_peer/mod.rs index 9531af35..ca8cf4f3 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -11,6 +11,7 @@ //! (matching W1b-0's pattern) but is out of scope for this trait carve. pub mod address_book; +pub mod admission; pub mod ask_routing; pub mod bootstrap; pub mod capsule_fallback; From 58e989ea425438d7a01c808bc7059f2db0aeabe4 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 07:14:55 -0700 Subject: [PATCH 3/8] feat(conduct): observe peer conduct and let it gate the dial set (#268) Wires `dig_sex::conduct` (SPEC 8.2A), which was implemented, tested and received not one observation from dig-node. `ConductState` keys `ConductRecord`s on `RoutedPeer` -- the mTLS-verified peer_id the ask router already ranks on -- so conduct can neither be attributed to nor escaped by a self-chosen identity, and is bounded by pool membership rather than a TTL. Reputation stays node-local and is never gossiped. The forwarded-ask loop now feeds it the outcome it already classifies, and `dialable()` filters the pool BEFORE `decide_forward` ranks it: ranking a peer this node has proven dishonest would still spend dials on it whenever the ranking favoured it. The threshold is "share > 0.0", which is not arbitrary -- `dial_share` returns exactly 0.0 for a proven fault and floors non-performance above zero. So the filter excludes precisely the verifiably faulty and can never evict a merely-slow peer, which is what stops induced distress being an eviction primitive. VACUITY, stated rather than implied: only `HonestAnswer` and `NonPerformance` are produced today. `ProvenLie` needs a per-peer verification verdict that `dig-download` owns and does not surface (module_transport.rs:1587), and `SelfContradiction` needs an announce/answer correlation the node does not keep. Both are follow-ups; neither is faked from a transport error, because branding an honest peer on unverifiable evidence is the conflation SPEC 8.2A exists to prevent. The exclusion path is therefore correct, tested, and dormant in production. Co-Authored-By: Claude --- crates/dig-node-core/src/download.rs | 62 +++- .../src/seams/dig_peer/admission.rs | 1 + .../src/seams/dig_peer/ask_routing.rs | 2 +- .../src/seams/dig_peer/conduct.rs | 317 ++++++++++++++++++ .../dig-node-core/src/seams/dig_peer/mod.rs | 1 + 5 files changed, 380 insertions(+), 3 deletions(-) create mode 100644 crates/dig-node-core/src/seams/dig_peer/conduct.rs diff --git a/crates/dig-node-core/src/download.rs b/crates/dig-node-core/src/download.rs index 6c5c4e84..f3006af4 100644 --- a/crates/dig-node-core/src/download.rs +++ b/crates/dig-node-core/src/download.rs @@ -856,6 +856,15 @@ pub struct NodeContent { /// at all, because pool membership is its liveness gate. Folding them together would make one /// structure whose key, lifetime and eviction rule all mean two different things at once. ask_routing: AskRoutingState, + /// What this node has observed each pool peer DO, and the dial share that earns them (#268). + /// + /// Held beside [`Self::ask_routing`] because the two read the same verified identity and are + /// bounded by the same pool: routing ranks the peers worth asking FIRST, conduct decides which are + /// worth asking AT ALL. Node-local and never gossiped — a shared reputation channel would be a + /// defamation primitive. + conduct: crate::seams::dig_peer::conduct::ConductState, + /// The instant conduct ticks are measured from. Wall-clock-free and monotonic. + conduct_epoch: std::time::Instant, } /// What a holder search ESTABLISHED — the records it found AND whether an empty result is a fact. @@ -1357,6 +1366,8 @@ impl NodeContent { ask_seen: AskSeenSet::new(), onion_relay: std::sync::atomic::AtomicBool::new(onion_relay_from_env()), ask_routing, + conduct: crate::seams::dig_peer::conduct::ConductState::new(), + conduct_epoch: std::time::Instant::now(), }) } @@ -1842,15 +1853,22 @@ impl NodeContent { .filter_map(|(peer, _)| RoutedPeer::from_pool_key(peer)) .collect(); + // CONDUCT gates who is dialable at all (#268, SPEC 8.3); routing then ranks what remains. + // The order matters: ranking a peer this node has PROVEN dishonest would still spend a dial on + // it whenever the ranking happened to favour it. A peer excluded here has a verifiable fault + // against it — a lie or a self-contradiction — and never merely a slow or silent history, + // which `dial_share` floors above zero precisely so distress cannot evict an honest holder. + let dialable = self.conduct.dialable(&routable, self.conduct_ticks()); + // Ranked by what THIS node has observed, not by the pool `HashMap`'s arbitrary order — and the - // observations of peers no longer in `routable` are dropped in the same call, so a cycled-away + // observations of peers no longer in `dialable` are dropped in the same call, so a cycled-away // peer leaves this node's memory when it leaves the pool. let decision = self.ask_routing.decide( &config, asker, budget.remaining(config.hop_cap), me, - &routable, + &dialable, self.relay_rate_limiter.check(requestor), ); @@ -1938,6 +1956,30 @@ impl NodeContent { // The ONLY writer of this node's routing memory, fed an outcome this node classified from // an exchange it issued and saw complete (dig_ecosystem#3129). self.ask_routing.record(routed, &outcome, started.elapsed()); + // The SAME exchange, classified for conduct (#268). An outcome where the peer answered — + // including an honest "I do not have it" — is an `HonestAnswer`, because SPEC 8.2A + // requires that answering is never worse than staying silent. A refusal, a timeout and an + // unreachable peer are `NonPerformance`: unverifiable, decaying, and floored, since none + // of them can be distinguished from distress an attacker induced in an honest peer. + // + // Neither VERIFIABLE class is produced here, and deliberately so. A `ProvenLie` needs + // bytes that failed verification against the anchor, attributed to the peer that supplied + // them; that attribution happens inside `dig-download`'s engine against `chunk_hashes` and + // is not surfaced per-peer to this node (see the report on #268). Claiming one from a + // transport error would brand an honest peer on unverifiable evidence, which is the exact + // conflation SPEC 8.2A exists to prevent. + self.conduct.observe( + routed, + match outcome { + AskOutcome::Answered(_) | AskOutcome::AnsweredInconclusive(_) => { + dig_sex::ConductEvidence::HonestAnswer + } + AskOutcome::Refused | AskOutcome::TimedOut | AskOutcome::Unreachable => { + dig_sex::ConductEvidence::NonPerformance + } + }, + self.conduct_ticks(), + ); match outcome { AskOutcome::Answered(records) => answers.records.extend(records), // The peer answered and told us its OWN subtree did not finish. Its records are @@ -1974,6 +2016,22 @@ impl NodeContent { .collect() } + /// This node's monotonic conduct clock, in seconds since process start. + /// + /// Conduct decay is measured in elapsed ticks and nothing else, so the clock must advance on its + /// own — a counter incremented per exchange would mean a peer nobody dials never ages out of its + /// penalty, and the recovery SPEC 8.2A requires would be unreachable for exactly the peer being + /// punished. + fn conduct_ticks(&self) -> u64 { + self.conduct_epoch.elapsed().as_secs() + } + + /// What this node has observed its pool peers do (#268), so a test can drive the forwarded ask + /// and then ask what it left behind. + pub(crate) fn conduct(&self) -> &crate::seams::dig_peer::conduct::ConductState { + &self.conduct + } + /// This node's routing memory, so a test can drive the forwarded ask and then ask what the ask /// LEFT BEHIND. Without it the recording leg would only be observable through its effect on a /// later round, and a test that could not see the write directly could not tell a missing write diff --git a/crates/dig-node-core/src/seams/dig_peer/admission.rs b/crates/dig-node-core/src/seams/dig_peer/admission.rs index a0f56c9d..f59d32af 100644 --- a/crates/dig-node-core/src/seams/dig_peer/admission.rs +++ b/crates/dig-node-core/src/seams/dig_peer/admission.rs @@ -87,6 +87,7 @@ pub fn authenticated_peer(conn_key: &str) -> Option { /// /// Held by value across the work it admits; the borrow checker then makes "return before releasing" /// impossible rather than merely discouraged. +#[derive(Debug)] pub struct AdmissionGuard<'a> { admission: &'a PeerAdmission, peer: AuthenticatedPeer, diff --git a/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs b/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs index 15717bee..628c903e 100644 --- a/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs +++ b/crates/dig-node-core/src/seams/dig_peer/ask_routing.rs @@ -69,7 +69,7 @@ const TICK: Duration = Duration::from_secs(1); /// /// See the module docs: this type existing at all is the security boundary, because it is the only /// way a `dig-sex` routing key can be minted and it can only be minted from a verified identity. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub(crate) struct RoutedPeer([u8; 32]); impl RoutedPeer { diff --git a/crates/dig-node-core/src/seams/dig_peer/conduct.rs b/crates/dig-node-core/src/seams/dig_peer/conduct.rs new file mode 100644 index 00000000..01bfe0a8 --- /dev/null +++ b/crates/dig-node-core/src/seams/dig_peer/conduct.rs @@ -0,0 +1,317 @@ +//! Peer conduct, and the dial share it earns (dig-sex SPEC §8.2A, dig-node#268). +//! +//! `dig_sex::conduct` classifies what this node observed a peer do and answers how much of the dial +//! budget that peer has earned. It was implemented, tested, and received not one observation from +//! dig-node. This module is the node's half: it holds the per-peer records, feeds them the outcomes +//! the ask loop already classifies, and exposes [`ConductState::dial_share`] so the ranking actually +//! spends fewer dials on peers that do not answer. +//! +//! # Why the classes are kept apart, and why that is a security property +//! +//! A **proven lie** (bytes that fail verification against the chain-anchored root) and a +//! **self-contradiction** (a peer that claimed to hold content, then answered absence) are both +//! *verifiable*: arithmetic and the peer's own words, checkable without trusting anyone. They carry a +//! durable penalty. +//! +//! **Non-performance** — a timeout, a reset, silence, a truncation — is *not* verifiable. It is +//! indistinguishable from genuine distress, and an attacker can manufacture it in an honest third +//! party by loading it up. So it decays on elapsed time alone, is capped, and MUST NOT reduce a dial +//! share to zero: if it could, inducing distress would become a way to evict an honest holder from +//! everybody's routing table. The crate enforces all three of those; this module's job is not to +//! undo them by feeding the wrong class. +//! +//! # Reputation is node-LOCAL and is never gossiped +//! +//! Nothing here is advertised, exchanged, or written to a peer-visible surface. A shared reputation +//! channel is a defamation primitive — one node's claim that a peer lied becomes every node's belief, +//! with no way to check it — so a record earned here influences only this node's own dialling. +//! +//! # Bounds: the pool is the liveness gate, so there is no TTL +//! +//! [`ConductState::retain`] is called with the CURRENT pool before every read, exactly as +//! [`AskRoutingState`](super::ask_routing) does for its observations. The map is therefore keyed only +//! by peers this node holds a verified session to and cannot grow while the pool does not — an +//! attacker cannot inflate it by minting identities it never connects with. + +use std::collections::HashMap; +use std::sync::Mutex; + +use dig_sex::{ConductEvidence, ConductRecord}; + +use super::ask_routing::RoutedPeer; + +/// This node's memory of how its pool peers have behaved. +/// +/// Keyed by [`RoutedPeer`] — the mTLS-verified `peer_id`, the same identity the ask router ranks on — +/// so a peer's conduct cannot be attributed to, or escaped by, an identity it chose for itself. +#[derive(Debug, Default)] +pub(crate) struct ConductState { + records: Mutex>, +} + +impl ConductState { + /// An empty conduct memory. + pub(crate) fn new() -> Self { + Self::default() + } + + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.records + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Fold one observation about `peer` into its record. + /// + /// An unseen peer starts from [`ConductRecord::neutral`], never from a penalised one: a peer this + /// node has no history with must be indistinguishable from one that has behaved, or a fresh + /// identity would be *better* than an honest long-lived one and the ranking would reward churn. + pub(crate) fn observe(&self, peer: RoutedPeer, evidence: ConductEvidence, now_ticks: u64) { + let mut records = self.lock(); + let record = records.get(&peer).copied().unwrap_or_else(ConductRecord::neutral); + records.insert(peer, dig_sex::observe(record, evidence, now_ticks)); + } + + /// The share of the dial budget `peer` has earned, in `0.0..=1.0`. + /// + /// Decay is applied at READ time rather than on a timer, because the crate's decay is a pure + /// function of elapsed ticks — so a peer whose non-performance has aged out recovers **without + /// this node having to talk to it**. That direction is the point: a peer punished for silence + /// that could only be forgiven by answering could never recover, because it is not being asked. + pub(crate) fn dial_share(&self, peer: RoutedPeer, now_ticks: u64) -> f64 { + let record = self + .lock() + .get(&peer) + .copied() + .unwrap_or_else(ConductRecord::neutral); + dig_sex::dial_share(dig_sex::decay(record, now_ticks)) + } + + /// The subset of `pool` this node will still spend a dial on, worst conduct excluded. + /// + /// The threshold is "a share above zero", and it is not an arbitrary cut: `dig_sex::dial_share` + /// returns exactly `0.0` for a peer with a PROVEN fault and floors every non-performance penalty + /// at [`MIN_NON_PERFORMANCE_DIAL_SHARE`](dig_sex::MIN_NON_PERFORMANCE_DIAL_SHARE). So this + /// excludes precisely the peers that lied or contradicted themselves — SPEC 8.3's durable + /// exclusion, earned by verifiable evidence — and can never exclude a peer that was merely slow, + /// however slow it was. Picking any threshold above the floor instead would hand an attacker the + /// eviction primitive the floor exists to deny. + /// + /// `retain` runs first, so a peer that left the pool is neither ranked nor remembered. + pub(crate) fn dialable(&self, pool: &[RoutedPeer], now_ticks: u64) -> Vec { + self.retain(pool); + pool.iter() + .copied() + .filter(|peer| self.dial_share(*peer, now_ticks) > 0.0) + .collect() + } + + /// Drop the records of peers no longer in `pool`, so this map is bounded by pool membership. + pub(crate) fn retain(&self, pool: &[RoutedPeer]) { + self.lock().retain(|peer, _| pool.contains(peer)); + } + + /// How many peers this node currently holds conduct for. Test-only: the bound it proves is the + /// pool's, not a number this code chooses. + #[cfg(test)] + pub(crate) fn len(&self) -> usize { + self.lock().len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn peer(n: u8) -> RoutedPeer { + RoutedPeer::from_pool_key(&hex::encode([n; 32])).expect("64-hex is a pool key") + } + + /// **Proves (#268):** a peer that delivers bytes failing verification loses dial share, while a + /// peer observed answering honestly at the same moment keeps its full share. + /// + /// The honest peer is the load-bearing half. Asserting only that the liar drops passes on an + /// implementation that penalises EVERY peer — including the one that behaved — because that + /// implementation also drops the liar. Varying one actor against a truthful control is what + /// separates "conduct is read" from "a number went down". + #[test] + fn a_proven_lie_costs_dial_share_while_an_honest_peer_keeps_its_own() { + let conduct = ConductState::new(); + let liar = peer(0xa1); + let honest = peer(0xb2); + + conduct.observe(liar, ConductEvidence::ProvenLie, 0); + conduct.observe(honest, ConductEvidence::HonestAnswer, 0); + + let liar_share = conduct.dial_share(liar, 0); + let honest_share = conduct.dial_share(honest, 0); + + assert!( + liar_share < honest_share, + "a proven lie must cost dial share (liar {liar_share}, honest {honest_share})" + ); + assert!( + (honest_share - 1.0).abs() < f64::EPSILON, + "the honest peer must be unpenalised, or every peer is being penalised alike \ + (got {honest_share})" + ); + } + + /// **Proves (#268):** a proven lie is DURABLE — it does not decay, however much time passes. + /// + /// This is the direction that distinguishes the two verifiable classes from non-performance, and + /// it is asserted at a tick far beyond the non-performance decay window so a decay applied to the + /// wrong field would show up here rather than passing silently. + #[test] + fn a_proven_lie_does_not_decay_with_elapsed_time() { + let conduct = ConductState::new(); + let liar = peer(0xa2); + conduct.observe(liar, ConductEvidence::ProvenLie, 0); + + let immediately = conduct.dial_share(liar, 0); + let much_later = conduct.dial_share(liar, dig_sex::NON_PERFORMANCE_DECAY_TICKS * 100); + + assert!( + (immediately - much_later).abs() < f64::EPSILON, + "a verifiable fault is a fact about what the peer did and must not age away \ + ({immediately} then {much_later})" + ); + } + + /// **Proves (#268):** non-performance costs some share and then recovers on ELAPSED TIME ALONE — + /// the peer is never spoken to between the two reads. + /// + /// That is the property that keeps an attacker from evicting an honest holder by manufacturing + /// distress in it. A recovery that required a fresh successful exchange would be unreachable for a + /// peer nobody is dialling any more, which is precisely the peer being punished. + #[test] + fn non_performance_decays_on_elapsed_time_without_the_peer_being_talked_to() { + let conduct = ConductState::new(); + let distressed = peer(0xc3); + + conduct.observe(distressed, ConductEvidence::NonPerformance, 0); + let penalised = conduct.dial_share(distressed, 0); + assert!( + penalised < 1.0, + "a timeout must cost something, or the observation is not being read ({penalised})" + ); + + // No `observe` between these two reads: nothing happened except the clock moving. + let recovered = conduct.dial_share(distressed, dig_sex::NON_PERFORMANCE_DECAY_TICKS * 4); + assert!( + recovered > penalised, + "non-performance must decay on elapsed time alone ({penalised} then {recovered})" + ); + } + + /// **Proves (#268):** sustained non-performance never reduces a dial share to zero. + /// + /// Far more observations than the crate's ceiling, so an implementation that let the penalty run + /// unbounded would reach zero here. A zero share means an honest peer that a hostile third party + /// merely made SLOW can be removed from this node's dialling entirely — the eviction the floor + /// exists to prevent. + #[test] + fn sustained_non_performance_never_silences_a_peer_completely() { + let conduct = ConductState::new(); + let distressed = peer(0xd4); + for _ in 0..(u32::from(dig_sex::NON_PERFORMANCE_CEILING) * 10) { + conduct.observe(distressed, ConductEvidence::NonPerformance, 0); + } + + let share = conduct.dial_share(distressed, 0); + assert!( + share >= dig_sex::MIN_NON_PERFORMANCE_DIAL_SHARE, + "unverifiable distress must never drive a share below the floor (got {share})" + ); + assert!(share > 0.0, "a distressed honest peer must remain dialable"); + } + + /// **Proves (#268):** an unobserved peer is treated as neutral, not as suspect. + /// + /// Guards the direction where a missing record is read as a bad one — which would make every newly + /// connected peer worse than a known-mediocre one and quietly freeze the routing table. + #[test] + fn an_unobserved_peer_starts_neutral_rather_than_penalised() { + let conduct = ConductState::new(); + let share = conduct.dial_share(peer(0xe5), 0); + assert!( + (share - 1.0).abs() < f64::EPSILON, + "a peer with no history must be indistinguishable from one that behaved (got {share})" + ); + } + + /// **Proves (#268):** conduct is bounded by pool membership — a departed peer leaves this node's + /// memory when it leaves the pool, so the map cannot grow while the pool does not. + #[test] + fn retain_drops_peers_that_left_the_pool() { + let conduct = ConductState::new(); + let stays = peer(0x01); + let goes = peer(0x02); + conduct.observe(stays, ConductEvidence::NonPerformance, 0); + conduct.observe(goes, ConductEvidence::ProvenLie, 0); + assert_eq!(conduct.len(), 2); + + conduct.retain(&[stays]); + + assert_eq!(conduct.len(), 1, "the departed peer must be forgotten"); + assert!( + (conduct.dial_share(goes, 0) - 1.0).abs() < f64::EPSILON, + "a forgotten peer returns to neutral — its durable fault was dropped WITH its session, \ + which is the bound's cost and is deliberate: the alternative is an unbounded map keyed \ + by identities an attacker mints for free" + ); + } + + /// **Proves (#268):** the dial filter excludes a proven liar from the peers this node will spend + /// a dial on, and leaves a merely-distressed peer in. + /// + /// This is the assertion that makes the whole wiring load-bearing rather than merely reachable: + /// it is about the SET the node dials, not about a number a function returned. The distressed + /// peer is the control, and it is the half that catches an over-eager filter — a threshold set + /// anywhere above the non-performance floor would drop it too, handing an attacker exactly the + /// eviction primitive the floor exists to deny. + #[test] + fn a_proven_liar_leaves_the_dial_set_while_a_merely_slow_peer_stays_in_it() { + let conduct = ConductState::new(); + let liar = peer(0x11); + let slow = peer(0x22); + let quiet = peer(0x33); + let pool = [liar, slow, quiet]; + + conduct.observe(liar, ConductEvidence::ProvenLie, 0); + // Far past the ceiling: as much non-performance as an attacker could ever manufacture. + for _ in 0..(u32::from(dig_sex::NON_PERFORMANCE_CEILING) * 10) { + conduct.observe(slow, ConductEvidence::NonPerformance, 0); + } + + let dialable = conduct.dialable(&pool, 0); + + assert!( + !dialable.contains(&liar), + "a peer with a verifiable fault must not be dialled" + ); + assert!( + dialable.contains(&slow), + "unverifiable distress must NEVER remove a peer from the dial set, however sustained — otherwise loading an honest holder is enough to evict it" + ); + assert!( + dialable.contains(&quiet), + "an unobserved peer must remain dialable" + ); + } + + /// **Proves (#268):** a self-contradiction is treated as verifiable, like a lie and unlike a + /// timeout — the peer contradicted its OWN claim, which needs no trust to check. + #[test] + fn a_self_contradiction_is_durable_like_a_lie_not_transient_like_a_timeout() { + let conduct = ConductState::new(); + let contradictor = peer(0x44); + conduct.observe(contradictor, ConductEvidence::SelfContradiction, 0); + + assert_eq!( + conduct.dial_share(contradictor, dig_sex::NON_PERFORMANCE_DECAY_TICKS * 100), + 0.0, + "a peer that announced content then denied holding it earned a durable exclusion" + ); + } +} diff --git a/crates/dig-node-core/src/seams/dig_peer/mod.rs b/crates/dig-node-core/src/seams/dig_peer/mod.rs index ca8cf4f3..5e4263dd 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -15,6 +15,7 @@ pub mod admission; pub mod ask_routing; pub mod bootstrap; pub mod capsule_fallback; +pub mod conduct; // The self-verifying tier-0 preimage resolver (#2033, PR-2). Its surface is exercised by its own // tests but not yet CALLED by production code — the tier-0 fetch loop wires it in PR-3 — so the // as-yet-unconsumed client + resolver would otherwise trip `dead_code`. The allow is removed when From 3caf7b6d07fcfb2a1c3232d292897c7f1af1acb4 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 07:22:03 -0700 Subject: [PATCH 4/8] chore(release): dig-node 0.190.0, dig-node-core 0.65.0 MINOR: two new capabilities, both backwards compatible. Inbound peer work is now admitted per authenticated identity before it is performed (#269), and peer conduct gates which peers a forwarded ask will dial (#268). No public API was removed or renamed and no wire format changed; an existing caller sees the same surface. Also drops the WIP marker the lane opened with. Co-Authored-By: Claude --- Cargo.lock | 2 +- crates/dig-node-core/Cargo.toml | 2 +- crates/dig-node-core/src/lib.rs | 1 - 3 files changed, 2 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d80efae9..d824252e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2969,7 +2969,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.64.0" +version = "0.65.0" dependencies = [ "async-trait", "axum", diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index bcc9bd91..10e7f31a 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -30,7 +30,7 @@ name = "dig-node-core" # dig-node#276/#296). Changing a public return type is BREAKING for an out-of-workspace implementor; # this crate is consumed in-workspace only and is pre-1.0, so it is a MINOR bump under SemVer's 0.x # rule -- recorded here rather than letting the number imply the locator surface held still. -version = "0.64.0" +version = "0.65.0" edition = "2021" license = "GPL-2.0-only" description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works." diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index dd3fd3f2..980f785e 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -15805,4 +15805,3 @@ mod tests { } } -// WIP(loop/batch-digsex): wiring dig-sex conduct/admission/acquisition/reward into the node. From a28b39357e17ec61bd3261e9e4a5d9cccdc0b45c Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 07:40:52 -0700 Subject: [PATCH 5/8] test(admission): prove the meter is live on the responder, and satisfy the gates (#269) Adds the WIRING assertion the unit tests cannot make: `admission.rs`'s tests prove the meter behaves, not that anything calls it, and a meter nothing calls is the defect #269 describes. The new test drives `NodeResponder::handle_json_rpc` and asserts an unauthenticated session is refused with -32000 rather than the allowlist's -32601 -- which is how we know admission ran FIRST -- with an authenticated control proving the responder is not simply refusing everybody. `node_responder_returns_method_not_found_for_management_methods` now passes a 64-hex conn_key. That is the shape every production session supplies, since both listeners derive it from the verified client leaf and no production path reaches the responder caller-less. Without it the test would answer -32000 and stop exercising the allowlist at all; with it, the property under test is unchanged. Also drops an unused accessor and two redundant u32 conversions for clippy -D warnings. Co-Authored-By: Claude --- crates/dig-node-core/src/download.rs | 6 -- crates/dig-node-core/src/lib.rs | 49 +++++-------- crates/dig-node-core/src/peer.rs | 71 ++++++++++++++++--- .../src/seams/dig_peer/conduct.rs | 9 ++- 4 files changed, 84 insertions(+), 51 deletions(-) diff --git a/crates/dig-node-core/src/download.rs b/crates/dig-node-core/src/download.rs index f3006af4..2c795c6e 100644 --- a/crates/dig-node-core/src/download.rs +++ b/crates/dig-node-core/src/download.rs @@ -2026,12 +2026,6 @@ impl NodeContent { self.conduct_epoch.elapsed().as_secs() } - /// What this node has observed its pool peers do (#268), so a test can drive the forwarded ask - /// and then ask what it left behind. - pub(crate) fn conduct(&self) -> &crate::seams::dig_peer::conduct::ConductState { - &self.conduct - } - /// This node's routing memory, so a test can drive the forwarded ask and then ask what the ask /// LEFT BEHIND. Without it the recording leg would only be observable through its effect on a /// later round, and a test that could not see the write directly could not tell a missing write diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index 980f785e..ac860aba 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -4431,17 +4431,15 @@ impl Node { /// ([`Node::maybe_backfill_capsule`]) and the #1576 reshare warm. Handed to /// [`crate::download::NodeContent::wire_capsule_reshare`] so both legs claim the same registry and a /// read triggers at most one whole-capsule acquisition. - /// The node-wide inbound admission meter (dig-sex SPEC 8.5, #269). - pub(crate) fn peer_admission( - &self, - ) -> &Arc { - &self.peer_admission - } - pub(crate) fn capsule_acquisition_gate(&self) -> Arc { self.capsule_acquisition.clone() } + /// The node-wide inbound admission meter (dig-sex SPEC 8.5, #269). + pub(crate) fn peer_admission(&self) -> &Arc { + &self.peer_admission + } + /// Build a node from the environment (cache dir/cap, §21 identity, upstream). /// Used by both the standalone bin's [`run`] and the in-process `dig-runtime`. pub fn from_env() -> Arc { @@ -4519,9 +4517,7 @@ impl Node { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -4813,9 +4809,7 @@ pub(crate) mod test_support { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5604,9 +5598,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5739,9 +5731,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5808,9 +5798,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5902,9 +5890,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -5978,9 +5966,9 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new( + crate::seams::dig_peer::admission::PeerAdmission::default(), + ), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -8656,9 +8644,7 @@ mod tests { content_cache: std::sync::Mutex::new(ContentCache::default()), inventory_refresher: OnceLock::new(), capsule_acquisition: Arc::new(crate::seams::dig_peer::WarmRegistry::new()), - peer_admission: Arc::new( - crate::seams::dig_peer::admission::PeerAdmission::default(), - ), + peer_admission: Arc::new(crate::seams::dig_peer::admission::PeerAdmission::default()), verification_ledger: verification_ledger::VerificationLedger::new(), self_ref: OnceLock::new(), gossip: OnceLock::new(), @@ -15804,4 +15790,3 @@ mod tests { } } } - diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index 4f33bf7d..e7906fc0 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -1192,7 +1192,10 @@ fn admission_refused( id: Value, refusal: crate::seams::dig_peer::admission::AdmissionRefusal, ) -> Value { - tracing::debug!(reason = refusal.reason(), "peer serve: inbound work refused at admission"); + tracing::debug!( + reason = refusal.reason(), + "peer serve: inbound work refused at admission" + ); json!({"jsonrpc":"2.0","id":id, "error":{"code":-32000,"message":"request refused","data":{"reason":refusal.reason()}}}) } @@ -1343,11 +1346,11 @@ impl PeerRpcResponder for NodeResponder { // refused request costs a hex decode and a counter bump rather than a read, a decode or a DHT // lookup. The guard is held for the whole method: it releases on `Drop`, including on the // early `return`s below, which is why no path here needs to remember to. - let _admitted = match self.node.peer_admission().admit( - conn_key, - dig_sex::WorkKind::Own, - 1, - ) { + let _admitted = match self + .node + .peer_admission() + .admit(conn_key, dig_sex::WorkKind::Own, 1) + { Ok(guard) => guard, Err(refusal) => return admission_refused(id, refusal), }; @@ -1417,7 +1420,8 @@ impl PeerRpcResponder for NodeResponder { // ADMISSION (dig-sex SPEC 8.5, #269) — before the items are even read. `items.len()` is the // attacker-chosen quantity this request asks for, so it is what gets clamped at the boundary // (`AdmissionLimits::max_request_units`); clamping it deeper in would already have paid for it. - let requested_units = u32::try_from(items.as_array().map_or(0, Vec::len)).unwrap_or(u32::MAX); + let requested_units = + u32::try_from(items.as_array().map_or(0, Vec::len)).unwrap_or(u32::MAX); let _admitted = match self.node.peer_admission().admit( conn_key, dig_sex::WorkKind::Own, @@ -5214,8 +5218,17 @@ pub(crate) mod tests { // End-to-end over the responder: a peer JSON-RPC frame naming a management/mutation // method is answered with -32601 (method not found) WITHOUT ever reaching the // node's `handle_rpc` dispatch (which would run the mutation). getPeers still works. + // + // The conn_key is a well-formed 64-hex peer id — the shape every production session supplies, + // since both listeners derive it from the verified client leaf. It became load-bearing when + // admission landed (#269): the meter runs AHEAD of this allowlist and refuses a session with + // no verified identity, so an empty key would now answer -32000 and this test would stop + // exercising the allowlist at all. Using a real one keeps the property under test unchanged — + // an AUTHENTICATED peer is still merely "some peer_id", never an authorized admin, which is + // the whole point of audit #179. let (node, _td) = crate::test_support::test_node_for_peer_surface(); let responder = NodeResponder::without_pool(node); + let authenticated = "ab".repeat(32); for m in [ "cache.clear", "cache.setCapBytes", @@ -5224,7 +5237,7 @@ pub(crate) mod tests { "dig.stage", ] { let req = json!({"jsonrpc":"2.0","id":1,"method":m,"params":{}}); - let resp = responder.handle_json_rpc(req, "").await; + let resp = responder.handle_json_rpc(req, &authenticated).await; assert_eq!( resp["error"]["code"], json!(-32601), @@ -5239,7 +5252,7 @@ pub(crate) mod tests { let ok = responder .handle_json_rpc( json!({"jsonrpc":"2.0","id":1,"method":"dig.getNetworkInfo"}), - "", + &authenticated, ) .await; assert!( @@ -5248,11 +5261,49 @@ pub(crate) mod tests { ); // getPeers is answered from the (empty) pool view, not -32601. let peers = responder - .handle_json_rpc(json!({"jsonrpc":"2.0","id":1,"method":"dig.getPeers"}), "") + .handle_json_rpc( + json!({"jsonrpc":"2.0","id":1,"method":"dig.getPeers"}), + &authenticated, + ) .await; assert!(peers["result"]["peers"].is_array()); } + /// **Proves (#269):** the admission meter is LIVE on the responder — not merely implemented + /// beside it — and it refuses a session that carries no verified identity BEFORE the method + /// allowlist is consulted. + /// + /// This is the wiring assertion. `admission.rs`'s unit tests prove the meter behaves; they cannot + /// prove anything calls it, and a meter nothing calls is the exact defect this ticket describes. + /// The authenticated control is what makes the refusal meaningful: without it, a responder that + /// refused EVERY request would pass. + #[tokio::test] + async fn the_responder_refuses_an_unauthenticated_session_before_consulting_the_allowlist() { + let (node, _td) = crate::test_support::test_node_for_peer_surface(); + let responder = NodeResponder::without_pool(node); + let method = json!({"jsonrpc":"2.0","id":1,"method":"dig.getNetworkInfo"}); + + // No verified peer id: refused at admission, and NOT with the allowlist's -32601 — which is + // how we know the meter ran first rather than the request simply failing later. + let refused = responder.handle_json_rpc(method.clone(), "").await; + assert_eq!( + refused["error"]["code"], + json!(-32000), + "an unauthenticated peer session must be refused at admission" + ); + assert!( + refused.get("result").is_none(), + "a refused request must produce no result — the work must not have been done" + ); + + // CONTROL: the same method, from an authenticated session, is served. + let served = responder.handle_json_rpc(method, &"cd".repeat(32)).await; + assert!( + served.get("result").is_some(), + "an authenticated peer must still be served, or the meter is refusing everybody" + ); + } + // -- OUTGOING-BANDWIDTH THROTTLE on the peer range-stream (dig_ecosystem #30) -------------------- // // `stream_range` is the busiest node-to-node egress path (multi-source downloaders fan ranges diff --git a/crates/dig-node-core/src/seams/dig_peer/conduct.rs b/crates/dig-node-core/src/seams/dig_peer/conduct.rs index 01bfe0a8..c5c579cc 100644 --- a/crates/dig-node-core/src/seams/dig_peer/conduct.rs +++ b/crates/dig-node-core/src/seams/dig_peer/conduct.rs @@ -68,7 +68,10 @@ impl ConductState { /// identity would be *better* than an honest long-lived one and the ranking would reward churn. pub(crate) fn observe(&self, peer: RoutedPeer, evidence: ConductEvidence, now_ticks: u64) { let mut records = self.lock(); - let record = records.get(&peer).copied().unwrap_or_else(ConductRecord::neutral); + let record = records + .get(&peer) + .copied() + .unwrap_or_else(ConductRecord::neutral); records.insert(peer, dig_sex::observe(record, evidence, now_ticks)); } @@ -214,7 +217,7 @@ mod tests { fn sustained_non_performance_never_silences_a_peer_completely() { let conduct = ConductState::new(); let distressed = peer(0xd4); - for _ in 0..(u32::from(dig_sex::NON_PERFORMANCE_CEILING) * 10) { + for _ in 0..(dig_sex::NON_PERFORMANCE_CEILING * 10) { conduct.observe(distressed, ConductEvidence::NonPerformance, 0); } @@ -280,7 +283,7 @@ mod tests { conduct.observe(liar, ConductEvidence::ProvenLie, 0); // Far past the ceiling: as much non-performance as an attacker could ever manufacture. - for _ in 0..(u32::from(dig_sex::NON_PERFORMANCE_CEILING) * 10) { + for _ in 0..(dig_sex::NON_PERFORMANCE_CEILING * 10) { conduct.observe(slow, ConductEvidence::NonPerformance, 0); } From 4d010ca37503e5ffb2d63e81e4437beddcef5e6b Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 11:15:38 -0700 Subject: [PATCH 6/8] fix(admission): admit the batch size the node advertises, and correct three false docs The admission clamp used dig-sex's default `max_request_units` = 256 while this crate's own `MAX_AVAILABILITY_ITEMS` = 512, so a 257-512 item `dig.getAvailability` batch was refused `-32000 "request too large"` by a node whose own `availability_batch` stood ready to answer all 512. `admission::node_limits()` now DERIVES the clamp from `MAX_AVAILABILITY_ITEMS` rather than restating it, so the two numbers cannot drift apart again. The clamp is kept -- it is the admission metering #269 exists to provide. A batch past 512 is now refused whole at the boundary instead of answered as a truncated prefix; `availability_batch`'s truncation remains as the in-process last line of defence, and both doc claims that said otherwise now say what the code does. The existing 513-item test calls `availability_batch` BELOW the responder that decides, so it passes identically whether the clamp admits, refuses, or is unwired. It is relabelled to say what it actually pins. The new pair sits at the RESPONDER level and asserts both sides: a batch AT the advertised limit is answered in full, one past it is refused with reason `request too large`. Also corrects `serve_peer_session`'s doc, which claimed a caller-less session still serves the JSON-RPC/availability paths. Since #269 it serves neither -- both meter against the mTLS-verified peer_id and refuse an absent one -- and the doc now states that on the `pub` item. Co-Authored-By: Claude --- crates/dig-node-core/src/lib.rs | 10 ++- crates/dig-node-core/src/peer.rs | 66 ++++++++++++++++++- .../src/seams/dig_peer/admission.rs | 37 ++++++++++- 3 files changed, 109 insertions(+), 4 deletions(-) diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index ac860aba..30ebfe05 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -3926,7 +3926,11 @@ impl Node { /// a per-request walk of the whole cache is a cost amplifier a peer controls. /// /// The batch is CAPPED at [`MAX_AVAILABILITY_ITEMS`] — the item count is caller-controlled — with - /// the excess simply not answered (the result array is aligned to the answered prefix). + /// the excess simply not answered (the result array is aligned to the answered prefix). That + /// truncation is the LAST line of defence, reached only by in-process callers: on the peer + /// surface an oversized batch never gets here, because + /// `NodeResponder::handle_availability` (peer.rs) meters the requested item + /// count against the same limit and refuses the whole request at the boundary (#269). /// /// `requestor` keys the per-item not-held → DHT `find_providers` enrichment against its /// per-requestor miss-lookup budget (dig_ecosystem#2007), so a large batch of not-held items from @@ -10266,6 +10270,10 @@ mod tests { assert_eq!(arr[2]["available"], false, "unknown capsule is a miss"); } + /// Pins the IN-PROCESS truncation only. A peer-surface batch this size is refused whole at + /// admission long before it reaches here, and this test cannot see that: it calls the batch + /// BELOW the responder that decides. The responder-level pair lives in `peer.rs` + /// (`the_responder_serves_a_batch_at_the_advertised_limit_and_refuses_one_past_it`). #[tokio::test] async fn availability_batch_caps_the_item_count() { let (node, _td) = test_node(None); diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index e7906fc0..77af346b 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -854,6 +854,11 @@ pub trait PeerRpcResponder: Send + Sync { /// by the SAME per-requestor miss-lookup budget as the single-item legs (dig_ecosystem#2007). A /// batch is the LARGEST amplification vector (up to `MAX_AVAILABILITY_ITEMS` lookups per request), /// so it must key by the ASKING peer, never one shared peer bucket every peer could exhaust. + /// + /// A batch LARGER than `MAX_AVAILABILITY_ITEMS` is refused whole at the admission boundary + /// (`-32000`, reason `request too large`) rather than answered as a truncated prefix: the node + /// admits exactly the batch size it advertises, and a caller that asked for more is told so + /// instead of being handed a short answer it must diff against its request to notice. async fn handle_availability(&self, items: Value, conn_key: &str) -> Value; /// Stream a `dig.fetchRange` response for `req` (the RangeRequest value) by writing framed @@ -992,13 +997,22 @@ async fn reconcile_and_flood( /// [`PeerRpcResponder::handle_availability`], a range fetch via [`PeerRpcResponder::stream_range`]. /// Returns when the peer closes the connection. The caller has already verified the remote `peer_id` /// (dig-nat enforces it during the mTLS handshake), so every stream here is from an authenticated peer. +/// +/// This entry point threads NO caller identity through to the responder, and since #269 that is +/// observable: the JSON-RPC and availability paths meter against the mTLS-verified `peer_id` and +/// refuse a session without one (`-32000`, reason `unauthenticated`). Callers that have the verified +/// identity — every production listener does — MUST use [`serve_peer_session_from`], which is what +/// keeps those two paths served. pub async fn serve_peer_session( mut session: dig_nat::mux::PeerSession, responder: Arc, ) { // No authenticated caller threaded here (the mTLS-verified caller is supplied by the listener via - // `serve_peer_session_from`); a caller-less session still serves the JSON-RPC/range/availability - // paths — only DHT routing-table population needs the caller. + // `serve_peer_session_from`). Since #269 a caller-less session serves the RANGE and module-range + // paths only: JSON-RPC and availability now meter against the mTLS-verified peer_id, and an + // absent one is refused `-32000` (`unauthenticated`) rather than admitted unmetered — otherwise + // presenting no identity would be the cheapest way out of the meter. DHT routing-table + // population likewise needs the caller. Every production listener supplies one. serve_peer_session_from(None, &mut session, responder).await } @@ -5303,6 +5317,54 @@ pub(crate) mod tests { "an authenticated peer must still be served, or the meter is refusing everybody" ); } + /// **Proves (#269):** the admission clamp and the availability batch's own advertised limit are + /// the SAME number, measured through the responder that actually decides. + /// + /// The clamp lives in [`NodeResponder::handle_availability`], above + /// [`crate::Node::availability_batch`]. `lib.rs`'s cap test calls the batch directly, BELOW that + /// decision, so it passes identically whether the clamp admits this batch, refuses it, or is not + /// wired at all — a test below the decision passes under the defect. This one asks the responder. + /// + /// Both sides are pinned, because a bound tested only from below can only confirm itself: a batch + /// AT [`crate::MAX_AVAILABILITY_ITEMS`] must be ANSWERED (a clamp set lower than the advertised + /// limit would refuse work this node says it serves), and one item past it must be REFUSED at + /// admission (a clamp set higher would let the meter admit work the batch will not answer). + #[tokio::test] + async fn the_responder_serves_a_batch_at_the_advertised_limit_and_refuses_one_past_it() { + let (node, _td) = crate::test_support::test_node_for_peer_surface(); + let responder = NodeResponder::without_pool(node); + let authenticated = "ef".repeat(32); + let batch = |n: usize| -> Value { + (0..n) + .map(|_| json!({ "store_id": "ee".repeat(32) })) + .collect::>() + .into() + }; + + let at_bound = responder + .handle_availability(batch(crate::MAX_AVAILABILITY_ITEMS), &authenticated) + .await; + assert_eq!( + at_bound["items"].as_array().map(Vec::len), + Some(crate::MAX_AVAILABILITY_ITEMS), + "a batch at the advertised limit must be ANSWERED in full — the node states it serves \ + this many items, so refusing it denies work of its own contract" + ); + + let past_bound = responder + .handle_availability(batch(crate::MAX_AVAILABILITY_ITEMS + 1), &authenticated) + .await; + assert_eq!( + past_bound["error"]["data"]["reason"], + json!("request too large"), + "one item past the advertised limit must be refused AT admission, before the batch is \ + read — not silently truncated after the cost is committed" + ); + assert!( + past_bound.get("items").is_none(), + "a refused batch must produce no answers at all" + ); + } // -- OUTGOING-BANDWIDTH THROTTLE on the peer range-stream (dig_ecosystem #30) -------------------- // diff --git a/crates/dig-node-core/src/seams/dig_peer/admission.rs b/crates/dig-node-core/src/seams/dig_peer/admission.rs index f59d32af..c31b9eb5 100644 --- a/crates/dig-node-core/src/seams/dig_peer/admission.rs +++ b/crates/dig-node-core/src/seams/dig_peer/admission.rs @@ -117,9 +117,26 @@ pub struct PeerAdmission { meter: Mutex, } +/// The limits this node admits under. +/// +/// dig-sex's own defaults are used for every dimension EXCEPT `max_request_units`, which is raised +/// to [`crate::MAX_AVAILABILITY_ITEMS`] — the largest batch this node advertises it answers. The +/// crate's default of 256 is half that, and a clamp set below the advertised limit refuses work the +/// node's own contract says it serves: a 257-512 item `dig.getAvailability` batch would have been +/// answered `-32000 "request too large"` while [`crate::Node::availability_batch`] stood ready to +/// answer all 512. The clamp and the limit it clamps to must be the SAME number, so this derives one +/// from the other rather than restating it. +#[must_use] +pub fn node_limits() -> AdmissionLimits { + AdmissionLimits { + max_request_units: crate::MAX_AVAILABILITY_ITEMS as u32, + ..AdmissionLimits::default() + } +} + impl Default for PeerAdmission { fn default() -> Self { - Self::new(AdmissionLimits::default()) + Self::new(node_limits()) } } @@ -303,6 +320,24 @@ mod tests { ); } + /// **Proves (#269):** the node's admitted request size is the availability batch's OWN advertised + /// limit, not the crate's smaller default. + /// + /// Asserting a literal 512 here would pass on a hard-coded constant that had drifted from + /// `MAX_AVAILABILITY_ITEMS`; asserting the derivation is what keeps the two numbers one number. + #[test] + fn the_node_admits_a_request_as_large_as_the_availability_batch_it_advertises() { + assert_eq!( + node_limits().max_request_units, + crate::MAX_AVAILABILITY_ITEMS as u32, + "the admission clamp must equal the advertised batch limit, or the node refuses batches it says it serves" + ); + assert!( + node_limits().max_request_units > AdmissionLimits::default().max_request_units, + "the crate default is the smaller of the two — if this ever stops holding, the override is doing nothing and the comment above it is false" + ); + } + /// **Proves (#269):** relayed work draws on its own ceiling, so work done on other nodes' behalf /// cannot consume the whole node-wide allowance (SPEC 6.1.8). #[test] From caf48319dcb357b33a9222a07a065539219fa058 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 11:17:49 -0700 Subject: [PATCH 7/8] chore(release): 0.194.0 -- rebased past #458, which took 0.190.0 Clean rebase, no conflict, no `dropping` -- the bump commit survives and simply stops meaning anything against the new main. Read from the file, not the log. 0.191-0.193 are claimed by siblings still in flight. --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 1005fc3d..da356942 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.190.0" +version = "0.194.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over From f662dda7bea9a1dd9cf295cfbaa40438c99cdf46 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 11:21:06 -0700 Subject: [PATCH 8/8] chore: refresh Cargo.lock for the 0.194.0 re-bump CI runs `cargo nextest --locked`, which will not update the lock itself. A bump touching only Cargo.toml fails the whole test job with `cannot update the lock file ... because --locked was passed` -- an error that names the lock, so it reads as a dependency problem rather than a stale version. --- Cargo.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index d824252e..d228393a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3031,7 +3031,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.190.0" +version = "0.194.0" dependencies = [ "async-trait", "axum",