diff --git a/Cargo.lock b/Cargo.lock index cb126131..f0d4954e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2782,9 +2782,9 @@ dependencies = [ [[package]] name = "dig-download" -version = "0.19.1" +version = "0.19.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3677ea834b43e91713ce56ea0a84b0ff9618b8466ca854f9fd03be7b46310171" +checksum = "02a51e9858830dcc3a631177ba1c8c30cbd8cce5e13bfd32e0f62dc544caf6cb" dependencies = [ "async-trait", "dig-constants 0.10.1", @@ -3011,7 +3011,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.54.2" +version = "0.55.0" dependencies = [ "async-trait", "axum", @@ -3073,7 +3073,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.145.1" +version = "0.146.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index fc6e38c9..9121f332 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.145.1" +version = "0.146.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/SPEC.md b/SPEC.md index 7975ee79..9f22881d 100644 --- a/SPEC.md +++ b/SPEC.md @@ -3012,6 +3012,7 @@ method runs, and it MUST NOT be conflated with the wallet's own `-32043` egress | -32010 | `UPSTREAM_ERROR` | shell | The blind-passthrough relay failed (unreachable / non-JSON). | | -32015 | `METADATA_TOO_LARGE` | node | `dig.getMetadata` refused: the publisher metadata section is too large or too complex to render safely. Refused when the ENCODED section exceeds `METADATA_SECTION_MAX_BYTES` (3 MiB) or its `custom` exceeds `MAX_CUSTOM_ENTRIES`/`MAX_CUSTOM_JSON_DEPTH`/`MAX_CUSTOM_JSON_ELEMENTS` (both checked BEFORE decode, #2160), or the RENDERED body exceeds `METADATA_RESPONSE_MAX_BYTES` (3 MiB, #2145). This section is rendered WHOLE — it cannot be windowed like `dig.getCapsule` — and `custom`/`links` are publisher-controlled, so an oversized/hostile capsule is refused with this bounded error rather than expanded ~16× in memory or blasted into a ~100 MB response (§5.5.1). A normal (kilobyte) metadata section is served unchanged. | | -32017 | `CONTENT_MISS_INCONCLUSIVE` | peer | No holder was named AND the search could not establish that there is none: a consulted leg timed out, was unreachable, or refused uninformatively (§10.4.5). The OPPOSITE instruction to a plain not-found — a not-found says stop looking, this says the question was not answered and the request MAY be retried. Collapsing the two let ONE slow peer manufacture an authoritative absence and, since a hop relays its answer, propagate it downwards. DEFINED by `dig-rpc-protocol` as `ErrorCode::ContentMissInconclusive` with origin `Peer` — the failure arises in the discovery layer from an unconsultable hop, mirroring `PeerUnreachable`. dig-node adopts the variant and does not assign the number. | +| -32017 | `ContentMissInconclusive` | — | Availability not ESTABLISHED. Also the answer a hop gives while it is still RELAYING a capsule on the requestor's behalf, in which case `error.data.relay_staged_bytes` carries the hop's staged byte count (§21.1). The field is ADDITIVE: a reader that ignores it sees an ordinary inconclusive miss and retries. | | -32020 | *(reserved: onion `onion_circuit_unavailable`)* | — | Reserved for the onion-routing contract; NOT minted by the control plane. | | -32021 | *(reserved: onion `privacy_requires_local_node`)* | — | Reserved for the onion-routing contract. | | -32022 | *(reserved: onion `onion_hops_out_of_range`)* | — | Reserved for the onion-routing contract. | @@ -3425,6 +3426,14 @@ boolean, default `false`). It MUST NOT trigger on `push` to `main`. The guard MUST be falsifiable — a self-test MUST assert that it FAILS an asset list carrying only the native packages. + **"Verified complete" is a statement about BYTES, not about names.** An asset is complete only when + it is present AND its upload state reports it as fully uploaded (GitHub: `state == "uploaded"`); an + asset still being written reports `state == "starting"`. The row is created when the upload BEGINS, + so every expected asset name can be present while bytes are still in flight, and a verification that + counts names alone promotes a release whose binaries are truncated or absent. A reimplementation that + reads only the name list satisfies the letter of the clause above and reintroduces exactly the race + it exists to prevent, which is why the state is stated here rather than left to the implementation. + 11.1a. **Doc-only commits never release** (the version is unchanged → the tag exists → the stable job is a no-op). The manual-dispatch `workflow_dispatch` on `release.yml` is a build-only "does main still build?" canary — it never publishes (publish is gated on a tag ref). @@ -6238,6 +6247,67 @@ stream is closed, so that its descriptor memo is populated and the re-ask is ans total a requestor spends on one holder MUST be bounded; an ask that is ANSWERED negatively MUST NOT be re-asked, since re-asking cannot change a refusal and would delay trying the next holder. +**A hop that is RELAYING MUST acknowledge, not block (MUST, dig-node#333).** A node asked with +`proxy: true` for a capsule it does not hold pulls that capsule from a holder on the requestor's +behalf. That pull is a whole-capsule transfer and takes arbitrarily long — minutes for a 135 MB +capsule on an ordinary link — while the descriptor ask that triggered it is bounded in tens of +seconds. A hop MUST NOT hold the ask open for the length of the transfer: it MUST answer within a +short grace, and if the capsule has not landed by then it MUST answer +`ContentMissInconclusive` (`-32017`) carrying `error.data.relay_staged_bytes`, its own count of the +bytes it has staged so far, and MUST leave the pull running. A capsule that lands inside the grace is +answered with the ordinary descriptor, indistinguishably from a holder's. + +The code is the taxonomy's existing inconclusive-miss code and MUST NOT be a new number: a running +relay is exactly the condition that code names — the availability answer is UNKNOWN and a retry is +meaningful — as opposed to `RESOURCE_UNAVAILABLE`, which settles the question. A requestor that does +not understand `relay_staged_bytes` therefore behaves correctly by default: it retries later. + +**A requestor MUST bound a relay wait by PROGRESS and by a ceiling (MUST).** On receiving that answer +a requestor MAY wait, re-asking the same hop on an interval. It MUST continue only while the reported +staged count STRICTLY ADVANCES, MUST abandon the hop after a bounded stall window in which it does +not, and MUST abandon it at a hard ceiling however healthy the progress appears. + +Both bounds are required and the ceiling is a SECURITY bound. `relay_staged_bytes` is a hop's claim +about itself, so a hostile hop can fabricate a counter that rises forever and would never stall; only +the ceiling makes the worst case finite. + +**The ceiling MUST be accompanied by a per-PULL budget.** A ceiling bounds a wait on ONE hop, and a +pull asks many: a puller's worst case is `descriptor attempts × holders × the per-ask bound`, so +raising the per-ask bound from the descriptor ladder to the relay ceiling multiplies by the holder +count. Where the provider set includes merely-CONNECTED peers rather than only announced holders, that +count is the whole connected pool, and hops each fabricating a byte of progress per poll would hold one +pull open for hours while every one of them stayed inside its individual ceiling. A requestor MUST +therefore charge relay waiting against a budget scoped to the CAPSULE being pulled, and MUST NOT grant +each hop a fresh allowance. Shrinking the per-hop ceiling is NOT an acceptable substitute: an honest hop +relaying a large capsule genuinely needs the full ceiling, and that case is what this path exists for. + +**The budget's lifetime MUST be the PULL's.** It MUST be released when the pull ends — in success or +failure — so that a later pull of the same capsule starts with the whole of it. A budget that persists +beyond its pull makes a capsule whose first pull spent it permanently ineligible for the relay path, +and, because the exhaustion is reported as a failure against whichever peer was being asked, it +attributes this node's own earlier spend to that peer. A requestor MUST NOT report a budget exhaustion +in a form that names a peer as its cause. + +An idle timeout MUST NOT be used in place of the release. Relay time is charged when a wait ENDS, so a +wait in progress is indistinguishable from an idle entry for up to the whole per-hop ceiling: a timeout +shorter than that ceiling can expire a live pull's budget mid-wait and silently restore the +per-hop multiplication, while one at or above it withholds the budget from the next pull for as long as +the condition it was meant to prevent. The pull boundary is therefore reported by the caller that +drives the pull, never inferred. A requestor MUST NOT treat the count as evidence about the +bytes: the capsule that eventually arrives is verified against the chain-anchored root exactly as a +direct holder's would be (§21.2), so a hop that fabricates its way through a wait still cannot produce +content that passes. + +**A relay ask is a SECOND-PASS escalation, and both passes MUST fit in one request (MUST, +dig-node#322).** A requestor MUST spend a PLAIN descriptor round on a `(capsule, peer)` pair before it +sets `proxy: true` for that pair — asking every connected peer to fetch a capsule on this node's +behalf before establishing that no reachable holder exists is the amplification the two-phase design +exists to bound. When that plain round is ANSWERED and the answer is no, the requestor MUST escalate +within the SAME invocation rather than deferring to a later one; a user issuing the documented single +command MUST NOT have to issue it twice. A plain round that was never answered at all MUST NOT be +escalated: a peer that could not answer a plain ask will not answer a relay ask, and re-asking it +doubles the invocation's bound for nothing. + ### 21.2. The anchor verifier is the ONLY root of trust (MUST) Every check before the anchor gate compares peer-supplied bytes against peer-supplied hashes. Those diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index 8ddb86d2..a341704e 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.54.2" +version = "0.55.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." @@ -112,10 +112,15 @@ dig-chainsource-interface = "0.1" # The two `ChainSourceProvider` traits then fail to unify and `registry.register` stops # compiling, so a caret `0.5` silently breaks this crate the moment the lock refreshes. # 0.5.1 is the last version whose `dig-chainsource-interface` agrees with chia-peer's. -# Unpinning (to 0.6, alongside dig-wallet) needs a `chia-peer` release built against -# dig-chainsource-interface 0.2 FIRST (release-first, §4.1) — a separate repo, tracked -# separately. dig-wallet is on chia-query 0.6 meanwhile: it does not exchange `chia_query` -# types across this crate boundary, so the two lines coexist without unifying. +# THE BLOCKER NAMED HERE UNTIL 2026-08-23 WAS WRONG, AND THE WRONG NAME IS WHY NOBODY UNPINNED. +# It said the unpin awaited a `chia-peer` release built against dig-chainsource-interface 0.2. +# No such release was ever needed: `chia-peer` 0.2.0 skipped 0.2 entirely and is built against +# dci ^0.3 + chia-protocol ^0.36.1, and `chia-query` (now 0.15.0) is on the same dci ^0.3 / +# chia 0.36 line. The two `ChainSourceProvider` traits therefore ALREADY unify again. +# What actually blocks the unpin is that they unify at CHIA 0.36, while this crate is still on +# the 0.26 line — so adopting them is the chia-0.36 cascade (dig_ecosystem#3152), not a +# one-line bump. Verify against the index before acting on this comment; a stale blocker is +# worse than none, because it stops people looking. chia-query = "=0.5.1" # The ecosystem's canonical address-family / IPv6-first primitive (CLAUDE.md §5.2). The advertise # path keys + orders candidate addresses via `dig_ip::Family` and aggregates them source-tagged + diff --git a/crates/dig-node-core/src/capsule_warm_locator_tests.rs b/crates/dig-node-core/src/capsule_warm_locator_tests.rs index 3188fab5..34ba77b2 100644 --- a/crates/dig-node-core/src/capsule_warm_locator_tests.rs +++ b/crates/dig-node-core/src/capsule_warm_locator_tests.rs @@ -153,6 +153,7 @@ fn warmer_over( CapsuleWarmer::new( content.warm_provider_locator(), transport, + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::InMemoryStateStore::new()), Arc::new(AnchoringChain), WarmPaths { @@ -187,9 +188,17 @@ async fn a_warm_reaches_a_holder_that_only_the_connected_pool_can_name() { warmer.warm(&hex32(STORE), &hex32(ROOT)).await; + // The assertion is on the SET of peers asked, not on the sequence. Both halves of the property + // survive that: a pull that asked NOBODY leaves the set empty, and a pull that reached past the + // pool puts a second id in it. What the set deliberately does NOT pin is HOW MANY TIMES the + // holder was asked, because that is dig-download's retry budget and not this test's subject -- + // 0.19.2 added an across-round descriptor re-ask (dig-download#37), so this holder, which never + // answers, is now asked `MAX_DESCRIPTOR_ATTEMPTS` times. Pinning the count made a legitimate + // downstream retry fix look like a locator regression here. + let asked: std::collections::BTreeSet = transport.asked().into_iter().collect(); assert_eq!( - transport.asked(), - vec![holder], + asked, + std::collections::BTreeSet::from([holder]), "the connected pool peer is the ONLY candidate the DHT could not name, so a pull that \ asked it proves the warm locator unioned the pool - and a pull that asked nobody is the \ shipped defect" diff --git a/crates/dig-node-core/src/download.rs b/crates/dig-node-core/src/download.rs index 3a214861..fa2edb19 100644 --- a/crates/dig-node-core/src/download.rs +++ b/crates/dig-node-core/src/download.rs @@ -2249,7 +2249,11 @@ impl NodeContent { )); self.set_capsule_warmer(crate::seams::dig_peer::CapsuleWarmer::new( self.warm_provider_locator(), - transport, + // The SAME transport, handed over under both of its roles: the thing that talks to + // holders, and the thing holding state whose lifetime is one pull. Two different objects + // here would leave the relay-wait ledger never released. + Arc::clone(&transport) as Arc, + transport as Arc, self.state_store.clone(), anchor_resolver, crate::seams::dig_peer::WarmPaths { @@ -4783,6 +4787,7 @@ pub(crate) mod tests { vec![], 8, )), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(FileStateStore::new(td.path().join("warm-state"))), Arc::new(HangingResolver), crate::seams::dig_peer::WarmPaths { diff --git a/crates/dig-node-core/src/lib.rs b/crates/dig-node-core/src/lib.rs index eb083b80..96207b44 100644 --- a/crates/dig-node-core/src/lib.rs +++ b/crates/dig-node-core/src/lib.rs @@ -2812,13 +2812,15 @@ impl Node { // operator opted in and within its proxy allowance, makes this node pull the whole capsule // from a holder and describe it from its own cache. Every gate is inside `relay_capsule`; a // refusal simply leaves `info` as `None` and the honest not-held answer below stands. - if info.is_none() - && seams::dig_peer::module_relay::relay_capsule( + let mut relay = seams::dig_peer::module_relay::RelayStatus::Refused; + if info.is_none() { + relay = seams::dig_peer::module_relay::relay_capsule( self, &store_hex, &root_hex, params, requestor, ) - .await - { - info = self.describe_held_module(&store_hex, &root_hex).await; + .await; + if relay == seams::dig_peer::module_relay::RelayStatus::Landed { + info = self.describe_held_module(&store_hex, &root_hex).await; + } } // The serve log records both outcomes with sentinelled ids, so "was this holder asked for the // descriptor, and did it have it?" is answerable from the log alone (#1595). @@ -2833,11 +2835,18 @@ impl Node { Ok(value) => json!({"jsonrpc":"2.0","id":id,"result": value}), Err(_) => rpc_err(&id, -32000, "could not encode the module descriptor"), }, - None => rpc_err( - &id, - download::RESOURCE_UNAVAILABLE, - "module not held locally at the requested root", - ), + // A relay this node is still RUNNING is not a miss, and answering it as one is what made + // the capability unreachable (dig-node#333). See [`relay_pending_err`]. + None => match relay { + seams::dig_peer::module_relay::RelayStatus::Pending { staged_bytes } => { + relay_pending_err(&id, staged_bytes) + } + _ => rpc_err( + &id, + download::RESOURCE_UNAVAILABLE, + "module not held locally at the requested root", + ), + }, } } @@ -2889,15 +2898,17 @@ impl Node { // MISS -> the RELAY leg (dig-node#276), exactly as on the descriptor above: a relayed window // is read from the same cache, through the same reader, so it is byte-identical to the one a // genuine holder would have served and the requestor needs no second code path. - if window.is_none() - && seams::dig_peer::module_relay::relay_capsule( + let mut relay = seams::dig_peer::module_relay::RelayStatus::Refused; + if window.is_none() { + relay = seams::dig_peer::module_relay::relay_capsule( self, &store_hex, &root_hex, params, requestor, ) - .await - { - window = self - .read_held_module_window(&store_hex, &root_hex, offset, length) - .await; + .await; + if relay == seams::dig_peer::module_relay::RelayStatus::Landed { + window = self + .read_held_module_window(&store_hex, &root_hex, offset, length) + .await; + } } match window { @@ -2917,11 +2928,16 @@ impl Node { } None => { module_serve::module_range_outcome("", &store_hex, &root_hex, offset, None); - rpc_err( - &id, - download::RESOURCE_UNAVAILABLE, - "module not held locally at the requested root", - ) + match relay { + seams::dig_peer::module_relay::RelayStatus::Pending { staged_bytes } => { + relay_pending_err(&id, staged_bytes) + } + _ => rpc_err( + &id, + download::RESOURCE_UNAVAILABLE, + "module not held locally at the requested root", + ), + } } } } @@ -4256,6 +4272,46 @@ fn rpc_err(id: &Value, code: i64, message: &str) -> Value { json!({"jsonrpc":"2.0","id":id,"error":{"code":code,"message":message}}) } +/// The answer a hop gives while it is STILL RELAYING the capsule it was asked for (dig-node#333): +/// the availability question is genuinely unsettled, and this node is the reason it is unsettled. +/// +/// # Why this reuses the inconclusive-miss code rather than declaring a new one +/// +/// The error taxonomy is owned by `dig-rpc-protocol` and adopted, never restated +/// (`SYSTEM.md`) — a node that invents a number is one release away from disagreeing with the +/// taxonomy it claims to speak, which this repo has already paid for twice (see +/// [`content_miss_inconclusive`](download::content_miss_inconclusive)). `ContentMissInconclusive` +/// already means exactly what a running relay means: the answer is UNKNOWN and a retry is +/// meaningful, as opposed to a settled not-found that tells the caller to stop. So the CODE is the +/// canonical one and the new fact rides in `error.data`, which is additive and needs no cascade. +/// +/// # What is in `data`, and what it is worth +/// +/// [`RELAY_PROGRESS_FIELD`] carries this node's staged byte count. A requestor uses it to tell +/// PROGRESS from a STALL while it waits, and for nothing else — it is this hop's claim about itself, +/// so it can be inflated at will. The bytes it describes still have to pass the requestor's own +/// merkle verification against the chain-anchored root before any of them count (NC-12), and a +/// requestor that waits on it must bound that wait independently. +/// +/// A requestor too old to read the field sees an inconclusive miss and retries later — which is +/// correct, and is what the field observations of #333 recorded people doing by hand. +fn relay_pending_err(id: &Value, staged_bytes: u64) -> Value { + json!({ + "jsonrpc": "2.0", + "id": id, + "error": { + "code": download::content_miss_inconclusive(), + "message": "relaying the requested capsule on your behalf; not yet complete", + "data": { RELAY_PROGRESS_FIELD: staged_bytes }, + } + }) +} + +/// The `error.data` key a relaying hop reports its staged byte count under, and the key a waiting +/// requestor reads to tell a progressing relay from a stalled one. Named once, because a producer and +/// a consumer that spell a wire key separately are one typo away from a silent mismatch. +pub(crate) const RELAY_PROGRESS_FIELD: &str = "relay_staged_bytes"; + /// Core JSON-RPC dispatch — the actual DIG node. Takes the request Value and /// returns the response Value. This is the single source of truth shared by the /// service shell's HTTP transport (`dig-node-service`) AND the in-process FFI @@ -6223,6 +6279,7 @@ mod tests { module.clone(), 8, )), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::InMemoryStateStore::new()), MockResolver::one(&store_a_hex, Bytes32(root)), crate::seams::dig_peer::WarmPaths { @@ -7193,6 +7250,7 @@ mod tests { Arc::new(dig_download::testkit::MockModuleTransport::serving( store, root, module, 8, )), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::InMemoryStateStore::new()), MockResolver::one(store, Bytes32::from_hex(root).expect("64-hex root")), crate::seams::dig_peer::WarmPaths { diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index d90355ff..ebb427ac 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -1445,26 +1445,33 @@ impl PeerRpcResponder for NodeResponder { // and serves the window from its own cache. All three gates live in `relay_capsule`; a refusal // leaves the not-held frame below exactly as it was, so the requestor stays free to ask // another hop (NC-12: a hop's "not found" may be a lie, including ours). - if window.is_none() - && crate::seams::dig_peer::module_relay::relay_capsule( + let mut relay = crate::seams::dig_peer::module_relay::RelayStatus::Refused; + if window.is_none() { + relay = crate::seams::dig_peer::module_relay::relay_capsule( &self.node, &store, &root, ¶ms, &crate::rate_limit::RequestorId::Peer(conn_key.to_string()), ) - .await - { - window = read_window().await; + .await; + if relay == crate::seams::dig_peer::module_relay::RelayStatus::Landed { + window = read_window().await; + } } let Some(window) = window else { module_serve::module_range_outcome(conn_key, &store, &root, offset, None); - return write_framed( - out, - &module_serve::module_unavailable_frame(crate::download::RESOURCE_UNAVAILABLE), - ) - .await; + // A relay STILL RUNNING is not a miss: the availability question is unsettled and this + // node is the reason, so the requestor is told to wait rather than to look elsewhere + // (dig-node#333). + let frame = match relay { + crate::seams::dig_peer::module_relay::RelayStatus::Pending { staged_bytes } => { + module_serve::module_relay_pending_frame(staged_bytes) + } + _ => module_serve::module_unavailable_frame(crate::download::RESOURCE_UNAVAILABLE), + }; + return write_framed(out, &frame).await; }; // OUTGOING-BANDWIDTH THROTTLE (#30/#1616): the module-range serve is the whole-capsule pull — 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 0e63cdb9..8648c8b1 100644 --- a/crates/dig-node-core/src/seams/dig_peer/mod.rs +++ b/crates/dig-node-core/src/seams/dig_peer/mod.rs @@ -66,6 +66,9 @@ pub use module_reshare::{ WarmRegistry, }; pub use module_transport::NatModuleTransport; +#[cfg(test)] +pub(crate) use module_transport::NoPullState; +pub(crate) use module_transport::PullLifecycle; pub use peer_network::PeerNetwork; pub(crate) use pool_locator::{ConnectedPool, PoolProviderLocator}; pub(crate) use selector_adapter::SelectorAdapter; diff --git a/crates/dig-node-core/src/seams/dig_peer/module_relay.rs b/crates/dig-node-core/src/seams/dig_peer/module_relay.rs index 4bffb3d2..8dfb7cdf 100644 --- a/crates/dig-node-core/src/seams/dig_peer/module_relay.rs +++ b/crates/dig-node-core/src/seams/dig_peer/module_relay.rs @@ -49,58 +49,103 @@ use serde_json::Value; use crate::download::proxy_requested; use crate::rate_limit::RequestorId; -use crate::seams::dig_peer::module_reshare::WarmOutcome; use crate::Node; -/// Try to make this node able to serve `(store_hex, root_hex)` on `requestor`'s behalf, returning -/// whether the capsule is now in the local cache and may be read from. +/// What a relay ask can honestly be told about `(store_hex, root_hex)`. /// -/// Awaited, not spawned: the caller has a module window to answer RIGHT NOW, and the answer depends -/// on the pull. This is the store-and-forward cost, paid once per capsule — a second window of the -/// same capsule finds it cached and returns immediately. +/// Three outcomes, because the two that used to be one — *this node will not relay* and *this node is +/// relaying and has not finished* — carry OPPOSITE instructions to the requestor. Collapsing them into +/// a single `false` is what made a cold relayed fetch look like a settled miss while the hop was in +/// the middle of answering it (dig-node#333). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RelayStatus { + /// The capsule is in this node's cache and its windows may be read now. + Landed, + /// A relay pull is RUNNING and has staged this many bytes so far. The requestor may wait. + /// + /// The byte count is this node's own measurement of its own staging file, offered as a liveness + /// signal so a waiting requestor can tell progress from a stall. It says nothing about + /// correctness: every one of those bytes still has to pass the requestor's merkle verification + /// against the chain-anchored root, exactly as a direct holder's would (NC-12). + Pending { staged_bytes: u64 }, + /// No relay: the requestor did not ask, the operator did not opt in, the requestor is outside its + /// proxy allowance, or this build has no capsule warmer. Indistinguishable to the requestor from + /// a plain miss, deliberately — a refusal must not narrate which gate refused. + Refused, +} + +/// How long a hop waits for its own relay pull before answering [`RelayStatus::Pending`]. +/// +/// Sized to be comfortably INSIDE the requestor's first descriptor rung (5 s), so a capsule that +/// lands quickly is still answered in one round trip and the pre-#333 behaviour is preserved for the +/// case that already worked. It must never grow toward the length of a bulk transfer: a hop that +/// holds a requestor's stream for minutes is the held-slot cost the descriptor ladder exists to bound, +/// and lengthening it here would simply move that cost rather than remove it. +const RELAY_GRACE: std::time::Duration = std::time::Duration::from_secs(3); + +/// Try to make this node able to serve `(store_hex, root_hex)` on `requestor`'s behalf. /// -/// Every refusal is silent and returns `false`; the caller's own `RESOURCE_UNAVAILABLE` then stands. -/// Never a silent success, and never an unbounded fetch: the pull is the ordinary -/// [`CapsuleWarmer`](super::CapsuleWarmer) one, byte-capped and chain-anchored, and it does NOT make -/// this node a holder ([`HolderClaim::Suppress`](super::module_reshare::HolderClaim)). +/// The pull runs in the BACKGROUND and this call waits only [`RELAY_GRACE`] for it. That is the whole +/// of dig-node#333: awaiting a 135 MB third-party transfer inside the requestor's descriptor ask meant +/// the ask always expired first, so the relay completed minutes after the only caller who wanted it +/// had given up. A hop that ACKs instead of blocking lets the requestor wait on this node's PROGRESS +/// rather than on a wall clock it has no way to size. +/// +/// Every refusal is silent and returns [`RelayStatus::Refused`]; the caller's own +/// `RESOURCE_UNAVAILABLE` then stands. Never a silent success, and never an unbounded fetch: the pull +/// is the ordinary [`CapsuleWarmer`](super::CapsuleWarmer) one, byte-capped and chain-anchored, and it +/// does NOT make this node a holder ([`HolderClaim::Suppress`](super::module_reshare::HolderClaim)). pub(crate) async fn relay_capsule( node: &Node, store_hex: &str, root_hex: &str, params: &Value, requestor: &RequestorId, -) -> bool { +) -> RelayStatus { // (1) The requestor must ASK. Checked first because it is free and because it is the only gate // whose absence means "this request never wanted a relay" rather than "this node refuses". if !proxy_requested(params) { - return false; + return RelayStatus::Refused; } let Some(content) = node.p2p_content() else { - return false; + return RelayStatus::Refused; }; // (2) The OPERATOR must have opted in. if !content.onion_relay_enabled() { - return false; + return RelayStatus::Refused; } // (3) The requestor must be inside its PROXY-class allowance — the expensive-egress bucket. if !content.allow_proxy_fetch(requestor) { - return false; + return RelayStatus::Refused; } - let Some(warmer) = content.capsule_warmer() else { + let Some(warmer) = content.capsule_warmer().cloned() else { // No warmer wired (the FFI/base path): there is no whole-capsule pull to drive, so there is // no relay. A read behaves identically with or without the leg. - return false; + return RelayStatus::Refused; }; tracing::debug!( store = %super::serve_log::SafeId::new(store_hex), root = %super::serve_log::SafeId::new(root_hex), "module relay: pulling a capsule this node does not hold, on a requestor's behalf" ); - // `AlreadyHeld` is admitted alongside `Held` because a concurrent warm may have landed the - // capsule between the caller's miss and this call — the question this function answers is "can the - // window be read now?", not "did I personally pull it?". - matches!( - warmer.warm_relayed(store_hex, root_hex).await, - WarmOutcome::Held { .. } | WarmOutcome::AlreadyHeld - ) + // Re-entrant by construction: `WarmRegistry` admits one warm per generation, so a second requestor + // — or this same requestor polling — joins the running pull rather than starting a rival one. + // + // SPAWNED, not awaited, and that changes who bears the cost. An awaited pull died with the + // request; a spawned one outlives it, so a requestor that gives up leaves this node still + // pulling. The registry's cap is GLOBAL and SHARED with this node's own `spawn_capsule_warm`, + // so an abandoning peer can hold a warm slot that a local read wanted. Bounded (the cap), opt-in + // (gate 2) and allowance-limited (gate 3) — but a real cost of running the leg, recorded on + // `CapsuleWarmer::warm_relayed` as well so it is visible from either end. + super::module_reshare::spawn_relayed_capsule_warm( + std::sync::Arc::clone(&warmer), + store_hex.to_string(), + root_hex.to_string(), + ); + if warmer.await_landing(store_hex, root_hex, RELAY_GRACE).await { + return RelayStatus::Landed; + } + RelayStatus::Pending { + staged_bytes: warmer.staged_bytes(store_hex, root_hex), + } } diff --git a/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs b/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs index 4e577db6..6dcfbf51 100644 --- a/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs +++ b/crates/dig-node-core/src/seams/dig_peer/module_reshare.rs @@ -364,11 +364,61 @@ const RELAY_MARKER_BODY: &[u8] = b"relayed: held on another node's behalf; this node is not a holder (dig-node#276) "; -/// Discard a warm's staging artifacts, so a failed pull leaves nothing behind that a later run (or a -/// GC sweep) could mistake for progress. -fn discard_staging(staged: &Path) { - let _ = std::fs::remove_file(staged); - let _ = std::fs::remove_file(dig_download::staging_path_for(staged)); +/// How often [`CapsuleWarmer::await_landing`] re-checks the cache while its grace runs. Short enough +/// that a capsule landing early is served promptly, long enough that the wait is not a spin. +const LANDING_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100); + +/// A discarded pull's whole on-disk footprint, so the two halves cannot be erased separately. +/// +/// A partial capsule is stored as TWO facts in two places — the staged bytes under the staging dir, +/// and the resume checkpoint in the [`StateStore`](dig_download::StateStore) — and they are only ever +/// true together. Passing them as one value is what stops a caller erasing one and leaving the other, +/// which is precisely the state dig-node#332 measured: a checkpoint claiming chunks that no longer +/// existed on disk. +struct StagedPull<'a> { + /// The path dig-download's `FileSink` finalizes onto (its in-progress bytes live beside it, at + /// [`staging_path_for`](dig_download::staging_path_for)). + staged: &'a Path, + /// Where the resume checkpoint lives. + state_store: &'a dyn dig_download::StateStore, + /// The checkpoint's key — dig-download's own, never a second derivation of it. + key: String, +} + +impl<'a> StagedPull<'a> { + /// Name the pull of `(store_hex, root_hex)` staging at `staged`. + fn new( + staged: &'a Path, + state_store: &'a dyn dig_download::StateStore, + store_hex: &str, + root_hex: &str, + ) -> Self { + StagedPull { + staged, + state_store, + key: dig_download::module_download_key(store_hex, root_hex), + } + } + + /// Erase a warm's staging artifacts — bytes AND checkpoint — so a failed pull leaves nothing + /// behind that a later run (or a GC sweep) could mistake for progress. + /// + /// # Why the checkpoint goes with the bytes + /// + /// A surviving checkpoint describes chunks that are no longer on disk, so the next warm resumes + /// against them, fails to read them back, and re-attributes each one. That is harmless to + /// correctness — the re-attribution re-fetches — but it emits the per-chunk warning an operator + /// reads as *a peer served me bytes that did not match*, i.e. an attack indicator, on a routine + /// discard. A monitoring signal that fires on a benign path is worse than none, because the real + /// event stops being distinguishable from the noise (dig-node#332). + /// + /// Every step is best-effort: this runs on a path that is ALREADY failing, and a cleanup error + /// must not replace the failure the caller has to report. + async fn erase(&self) { + let _ = std::fs::remove_file(self.staged); + let _ = std::fs::remove_file(dig_download::staging_path_for(self.staged)); + let _ = self.state_store.clear(&self.key).await; + } } /// What a FAILED pull leaves behind in the staging area. @@ -432,15 +482,62 @@ impl StagingDisposition { } } - /// Apply this disposition to `staged`. - fn apply(self, staged: &Path) { + /// Apply this disposition to `pull`. + /// + /// [`Preserve`](Self::Preserve) touches NOTHING — not the bytes and not the checkpoint. The two + /// are what a resume is made of, and dropping either half is what made the resume machinery + /// structurally unreachable in the field (dig-node#328). + async fn apply(self, pull: &StagedPull<'_>) { match self { StagingDisposition::Preserve => {} - StagingDisposition::Discard => discard_staging(staged), + StagingDisposition::Discard => pull.erase().await, + } + } +} + +/// Reports the end of a pull when it goes out of scope, so the boundary is reported on EVERY exit — +/// including an unwinding panic. +/// +/// # Why a guard and not a statement after the `await` +/// +/// A plain call after `downloader.download(...).await` is skipped when that await PANICS, and a +/// panic here is reachable rather than theoretical: the tier-0 precache path runs +/// `run_round -> fetch_and_cache -> warm_capped -> download` inside a real +/// [`catch_unwind`](crate::shared::panic_guard) added precisely so a panic there is survived, and this +/// crate does not build with `panic = "abort"`. So the task continues and the ledger entry leaks. +/// +/// The leak is not a tidiness matter — it restores the two defects this budget exists to remove: the +/// capsule becomes permanently ineligible for the relay path, and the resulting exhaustion is composed +/// into a transport error NAMING A PEER, which `SPEC.md` §21.1 forbids in as many words. +/// +/// `Drop` runs during unwinding, so the guard reports on the panic path and the returning paths +/// through one mechanism. That is the same lesson this file has already paid for twice: prefer making +/// the omission unexpressible over asserting that it does not occur. +struct PullBoundary<'a> { + /// Told the pull is over. Borrowed, so the guard cannot outlive the warmer driving the pull. + lifecycle: &'a dyn super::PullLifecycle, + /// The capsule whose pull this is. + store_hex: &'a str, + root_hex: &'a str, +} + +impl<'a> PullBoundary<'a> { + /// Open the boundary. It closes when the value drops. + fn new(lifecycle: &'a dyn super::PullLifecycle, store_hex: &'a str, root_hex: &'a str) -> Self { + PullBoundary { + lifecycle, + store_hex, + root_hex, } } } +impl Drop for PullBoundary<'_> { + fn drop(&mut self) { + self.lifecycle.pull_finished(self.store_hex, self.root_hex); + } +} + /// Everything a capsule warm needs, resolved once at composition time. /// /// `anchor_resolver` is what makes the whole reshare path trustworthy: the generation root the pull is @@ -451,6 +548,12 @@ pub struct CapsuleWarmer { locator: Arc, /// Talks `dig.getModuleInfo` / `dig.fetchModuleRange` to them. transport: Arc, + /// Told when each pull ends, so the transport can release state scoped to that pull. + /// + /// Usually the SAME object as `transport`, handed over twice under its two roles. It is a + /// separate field because the pull boundary is this warmer's knowledge and the state is the + /// transport's, and neither can see the other's half. + pull_lifecycle: Arc, /// Resume checkpoints, so an interrupted warm does not restart from zero. state_store: Arc, /// The CHAIN's view of each store's anchored root — the pull's only root of trust. @@ -488,6 +591,7 @@ impl CapsuleWarmer { pub(crate) fn new( locator: Arc, transport: Arc, + pull_lifecycle: Arc, state_store: Arc, anchor_resolver: Arc, paths: WarmPaths, @@ -499,6 +603,7 @@ impl CapsuleWarmer { Arc::new(CapsuleWarmer { locator, transport, + pull_lifecycle, state_store, anchor_resolver, paths, @@ -526,6 +631,67 @@ impl CapsuleWarmer { .await } + /// Whether this node can serve `(store_hex, root_hex)` from its cache RIGHT NOW. + /// + /// The cache path's existence IS the answer everywhere else in this module (it is what a holder + /// claim is made of), so this asks the same question the same way rather than adding a second, + /// driftable notion of "landed". + pub(crate) fn holds(&self, store_hex: &str, root_hex: &str) -> bool { + CapsuleKey::parse(store_hex, root_hex) + .is_some_and(|capsule| self.paths.cached_module(&capsule).exists()) + } + + /// How many bytes of `(store_hex, root_hex)` are staged on disk — this node's own honest + /// measure of how far an in-flight pull has got. + /// + /// Zero for a pull that has not started, has staged nothing yet, or names a non-canonical + /// generation. It is a LIVENESS signal for a waiting requestor and nothing else: it says how far + /// this hop has got, never that any byte is correct — that is settled by the merkle verification + /// every one of these bytes must still pass (NC-12). + pub(crate) fn staged_bytes(&self, store_hex: &str, root_hex: &str) -> u64 { + let Some(capsule) = CapsuleKey::parse(store_hex, root_hex) else { + return 0; + }; + let staged = self.paths.staged_module(&capsule); + let in_progress = dig_download::staging_path_for(&staged); + // The in-progress file is where a running pull writes; the finalized one appears only at the + // very end. Whichever exists is the pull's current extent. + [in_progress, staged] + .iter() + .filter_map(|p| std::fs::metadata(p).ok()) + .map(|m| m.len()) + .max() + .unwrap_or(0) + } + + /// Wait up to `grace` for a background pull of `(store_hex, root_hex)` to reach the cache, + /// reporting whether it did. + /// + /// # Why a grace at all, rather than answering immediately + /// + /// A capsule small enough to land inside the grace is answered by the hop in ONE round trip, + /// exactly as it was before dig-node#333 — so the case that already worked keeps working and no + /// requestor pays a poll interval for it. The grace is deliberately far inside the requestor's + /// own descriptor rungs; what it must never do is stretch to cover a bulk transfer, which is the + /// held-slot cost the rungs exist to bound. + pub(crate) async fn await_landing( + &self, + store_hex: &str, + root_hex: &str, + grace: std::time::Duration, + ) -> bool { + let deadline = tokio::time::Instant::now() + grace; + loop { + if self.holds(store_hex, root_hex) { + return true; + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(LANDING_POLL_INTERVAL.min(grace)).await; + } + } + /// [`warm`](Self::warm) for a capsule pulled on ANOTHER node's behalf (dig-node#276): identical in /// every trust step — chain anchor, merkle verification, promote-recheck, cache bound — except that /// this node does **not** announce itself as a holder of the result. @@ -533,6 +699,20 @@ impl CapsuleWarmer { /// The capsule still lands in the cache, because that is what lets the relayed module windows be /// served from the same code path a genuine holder serves from, byte-identically. What it does not /// do is make a stranger's choice of content into this node's advertised inventory. + /// + /// # This competes with local warms for the same slots + /// + /// The pull runs under the SHARED [`WarmRegistry`], which admits + /// [`DEFAULT_MAX_CONCURRENT_WARMS`] generations across this whole node — relayed and local alike. + /// So a relay started for a stranger occupies a slot this node's own reads would otherwise use, + /// and since dig-node#333 the pull is SPAWNED rather than awaited, meaning it keeps that slot even + /// after the requestor that asked for it has given up and disconnected. + /// + /// That is bounded and deliberate rather than a hole — the cap is global, the leg is opt-in + /// (`DIG_NODE_ONION_RELAY`, default OFF), and the requestor must additionally be inside its + /// proxy-class allowance — but it IS a real cost of enabling the relay, and an operator who + /// enables it should know that a peer which abandons its request can still hold a warm slot until + /// the pull finishes or fails. pub async fn warm_relayed(self: &Arc, store_hex: &str, root_hex: &str) -> WarmOutcome { self.warm_claiming(store_hex, root_hex, HolderClaim::Suppress) .await @@ -646,6 +826,8 @@ impl CapsuleWarmer { // `truncate` + `read_at` the engine's promotion probe requires, so there is no bespoke sink // here to accidentally inherit a default from. let staged = self.paths.staged_module(&capsule); + // Bytes + checkpoint as ONE value, so no failure path below can erase half of a partial. + let pull = StagedPull::new(&staged, self.state_store.as_ref(), store_hex, root_hex); let sink = dig_download::FileSink::new(&staged); let downloader = dig_download::ModuleDownloader::new( Arc::clone(&self.locator), @@ -655,7 +837,14 @@ impl CapsuleWarmer { config, ); - let pulled = downloader.download(store_hex, root_hex, &sink).await; + // The boundary is scoped to the download and nothing else, so it is reported the moment the + // pull ends — on success, on failure, and on a PANIC that unwinds through it. A failed pull is + // exactly the one that spent relay budget on hops that did not deliver, and leaving its ledger + // entry behind would charge the NEXT pull of this capsule for it (dig-node#333 review). + let pulled = { + let _boundary = PullBoundary::new(self.pull_lifecycle.as_ref(), store_hex, root_hex); + downloader.download(store_hex, root_hex, &sink).await + }; // 3. ONLY `Ok` may lead to a holder claim. Not finalize-observed, not partial staging, not an // `Err` that happened to leave bytes behind. @@ -673,7 +862,7 @@ impl CapsuleWarmer { // resume machinery structurally unreachable for capsule warms (#328): the checkpoint // survived while the bytes it described did not, so every retry restarted from offset 0. let disposition = StagingDisposition::for_failure(&error); - disposition.apply(&staged); + disposition.apply(&pull).await; tracing::info!( store = %super::serve_log::SafeId::new(store_hex), root = %super::serve_log::SafeId::new(root_hex), @@ -692,7 +881,7 @@ impl CapsuleWarmer { let cached = self.paths.cached_module(&capsule); match promote_into_cache(&staged, &cached, &verifier, claim) { Ok(promoted) => { - discard_staging(&staged); + pull.erase().await; // The ONE step a relayed warm skips. Everything above it — the chain anchor, the // merkle verification, the promote-recheck — ran identically, so the bytes are equally // trustworthy; what differs is whether this node CLAIMS them (see [`HolderClaim`]). @@ -710,7 +899,7 @@ impl CapsuleWarmer { WarmOutcome::Held { bytes: promoted } } Err(failure) => { - discard_staging(&staged); + pull.erase().await; tracing::warn!( store = %super::serve_log::SafeId::new(store_hex), root = %super::serve_log::SafeId::new(root_hex), @@ -758,6 +947,22 @@ pub fn spawn_capsule_warm(warmer: Arc, store_hex: String, root_he }); } +/// [`spawn_capsule_warm`]'s relay twin: pull `(store_hex, root_hex)` ON A REQUESTOR'S BEHALF, in the +/// background, claiming nothing. +/// +/// The claim is the whole difference — a relayed capsule is cached so the hop can serve its windows +/// and is NOT announced, so relaying never turns this node into an advertised holder of a stranger's +/// choosing ([`HolderClaim::Suppress`]). +/// +/// Spawned rather than awaited because the requestor's ask must not be held open for the length of a +/// third-party transfer (dig-node#333). Re-entrant: [`WarmRegistry`] admits one warm per generation, +/// so a second request for the same capsule joins the running pull instead of starting a rival one. +pub fn spawn_relayed_capsule_warm(warmer: Arc, store_hex: String, root_hex: String) { + tokio::spawn(async move { + warmer.warm_relayed(&store_hex, &root_hex).await; + }); +} + /// Decode a canonical 64-hex id into 32 raw bytes. fn decode_id(hex: &str) -> Option<[u8; 32]> { if hex.len() != 64 { @@ -1068,6 +1273,7 @@ mod tests { CapsuleWarmer::new( Arc::new(NoHolders), Arc::new(UnusedTransport), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::FileStateStore::new(dir.join("state"))), resolver, WarmPaths { @@ -1094,6 +1300,7 @@ mod tests { let warmer = CapsuleWarmer::new( Arc::new(NoHolders), Arc::new(UnusedTransport), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::FileStateStore::new(dir.join("state"))), Arc::new(ConfirmingResolver), WarmPaths { @@ -1185,6 +1392,7 @@ mod tests { let warmer = CapsuleWarmer::new( Arc::new(OneHolder), Arc::new(RefusingHolder), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::FileStateStore::new(dir.join("state"))), Arc::new(ConfirmingResolver), WarmPaths { @@ -1271,6 +1479,7 @@ mod tests { let warmer = CapsuleWarmer::new( locator, transport, + Arc::new(crate::seams::dig_peer::NoPullState), // In-memory, not `FileStateStore`: the resume-checkpoint backing store is orthogonal to // what this test proves (staged->cache promotion + announce-once), and `FileStateStore`'s // hex-doubled `module:<64hex>:<64hex>` key exceeds Windows' ~255-char filename limit @@ -1432,6 +1641,7 @@ mod tests { dig_download::testkit::mock_providers(1, &content), )), Arc::clone(transport) as Arc, + Arc::new(crate::seams::dig_peer::NoPullState), Arc::clone(state), Arc::new(ConfirmingResolver), WarmPaths { @@ -1580,12 +1790,17 @@ mod tests { ) .with_corrupt_module_hash(), ); + // Held, not inlined: dig-node#332 is about what SURVIVES in this store after the failure, and + // an inlined store cannot be read back. + let state: Arc = + Arc::new(dig_download::InMemoryStateStore::new()); let warmer = CapsuleWarmer::new( Arc::new(dig_download::testkit::MockProviderLocator::fixed( dig_download::testkit::mock_providers(1, &content), )), transport, - Arc::new(dig_download::InMemoryStateStore::new()), + Arc::new(crate::seams::dig_peer::NoPullState), + Arc::clone(&state), Arc::new(ConfirmingResolver), WarmPaths { staging_dir: dir.join("staging"), @@ -1615,6 +1830,17 @@ mod tests { !cached_module_path(&dir).exists(), "nothing may reach the cache path" ); + // dig-node#332: the checkpoint is the OTHER half of the same partial. Left behind, it claims + // chunks that are no longer on disk, so the next warm re-attributes each one and logs the + // per-chunk warning an operator reads as hostile content — on a routine discard. + assert!( + state + .load(&dig_download::module_download_key(&store_hex, &root_hex)) + .await + .expect("the state store is readable") + .is_none(), + "a discarded partial must leave no resume checkpoint behind" + ); let _ = std::fs::remove_dir_all(&dir); } @@ -1666,6 +1892,7 @@ mod tests { Arc::new(dig_download::testkit::MockModuleTransport::serving( &store_hex, &root_hex, module, 8, )), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::InMemoryStateStore::new()), Arc::new(ConfirmingResolver), WarmPaths { @@ -1688,6 +1915,208 @@ mod tests { .join(format!("{root_hex}.dig")) } + /// Records every pull boundary the warmer reports, so a test can assert the CALL rather than the + /// ledger it happens to drive. + #[derive(Default)] + struct RecordingLifecycle { + finished: std::sync::Mutex>, + } + + impl crate::seams::dig_peer::PullLifecycle for RecordingLifecycle { + fn pull_finished(&self, store_id: &str, root: &str) { + self.finished + .lock() + .unwrap_or_else(|p| p.into_inner()) + .push((store_id.to_string(), root.to_string())); + } + } + + /// **Proves (dig-node#333 re-gate):** the WARMER reports the end of every pull — the succeeding + /// one and, more importantly, the failing one. + /// + /// **Why this test and not only the ledger's own:** `RelayWaitBudget::release` being correct says + /// nothing about whether anything calls it, and a per-pull ledger nobody ends is exactly the + /// daemon-lifetime bug this fixes. This drives the real `warm` entry point and asserts the seam is + /// invoked, with the capsule's own ids. + /// + /// **Why the FAILING pull is the load-bearing half:** a failed pull is precisely the one that + /// spent relay budget on hops that did not deliver, so an implementation that reports only on + /// success leaves the worst case unreleased. It is also the easy mistake — a `?` or an early + /// return placed above the report. + #[tokio::test] + async fn a_warmer_reports_the_end_of_every_pull_including_a_failed_one() { + let (store_hex, root_hex) = (hex32(STORE), hex32(chain_root())); + + // FAILING pull: no holder can be located at all. + let failed_dir = temp_dir("lifecycle-failed"); + let failed_spy = Arc::new(RecordingLifecycle::default()); + let failed = CapsuleWarmer::new( + Arc::new(dig_download::testkit::MockProviderLocator::fixed(Vec::new())), + Arc::new(dig_download::testkit::MockModuleTransport::serving( + &store_hex, + &root_hex, + Vec::new(), + 8, + )), + Arc::clone(&failed_spy) as Arc, + Arc::new(dig_download::InMemoryStateStore::new()), + Arc::new(ConfirmingResolver), + WarmPaths { + staging_dir: failed_dir.join("staging"), + cache_dir: failed_dir.join("cache"), + }, + Arc::new(AnnounceSpy::default()), + Arc::new(WarmRegistry::new()), + dig_download::ModuleDownloadConfig::default(), + Arc::new(crate::tier0_live::NoopModulesEvictor), + ); + + let outcome = failed.warm(&store_hex, &root_hex).await; + assert!( + matches!(outcome, WarmOutcome::Refused(_)), + "the control: this pull must genuinely FAIL, or the report below is the success path in disguise: {outcome:?}" + ); + assert_eq!( + failed_spy + .finished + .lock() + .expect("lifecycle lock") + .as_slice(), + &[(store_hex.clone(), root_hex.clone())], + "a FAILED pull must still report its end, naming the capsule — it is the pull most likely to have spent relay budget on hops that delivered nothing" + ); + + // SUCCEEDING pull: the same report, so the seam is not failure-only either. + let ok_dir = temp_dir("lifecycle-ok"); + let ok_spy = Arc::new(RecordingLifecycle::default()); + let module = module_committing(STORE, chain_root()); + let content = dig_download::module_content_id(&store_hex, &root_hex) + .expect("canonical ids yield a content id"); + let succeeded = CapsuleWarmer::new( + Arc::new(dig_download::testkit::MockProviderLocator::fixed( + dig_download::testkit::mock_providers(1, &content), + )), + Arc::new(dig_download::testkit::MockModuleTransport::serving( + &store_hex, &root_hex, module, 8, + )), + Arc::clone(&ok_spy) as Arc, + Arc::new(dig_download::InMemoryStateStore::new()), + Arc::new(ConfirmingResolver), + WarmPaths { + staging_dir: ok_dir.join("staging"), + cache_dir: ok_dir.join("cache"), + }, + Arc::new(AnnounceSpy::default()), + Arc::new(WarmRegistry::new()), + dig_download::ModuleDownloadConfig::default(), + Arc::new(crate::tier0_live::NoopModulesEvictor), + ); + + let outcome = succeeded.warm(&store_hex, &root_hex).await; + assert!( + matches!(outcome, WarmOutcome::Held { .. }), + "the control: this pull must genuinely SUCCEED: {outcome:?}" + ); + assert_eq!( + ok_spy.finished.lock().expect("lifecycle lock").len(), + 1, + "a successful pull reports its end exactly once" + ); + + let _ = std::fs::remove_dir_all(&failed_dir); + let _ = std::fs::remove_dir_all(&ok_dir); + } + + /// A holder whose descriptor ask PANICS — the shape a real bug in the pull takes. + struct PanickingHolder; + + #[async_trait::async_trait] + impl dig_download::ModuleTransport for PanickingHolder { + async fn get_module_info( + &self, + _provider_peer_id: &str, + _store_id: &str, + _root: &str, + ) -> Result { + panic!("a bug inside the pull"); + } + + async fn fetch_module_range( + &self, + _provider_peer_id: &str, + _store_id: &str, + _root: &str, + _offset: u64, + _length: u64, + ) -> Result, dig_download::DownloadError> { + unreachable!("the descriptor panics first"); + } + } + + /// **Proves (dig-node#333 re-gate 3):** a pull that PANICS still reports its end, so the relay-wait + /// ledger cannot leak an entry. + /// + /// **Why this is reachable rather than theoretical:** the tier-0 precache path runs + /// `run_round -> fetch_and_cache -> warm_capped -> download` inside + /// [`crate::shared::panic_guard::catch_iteration`], added so a panic there is SURVIVED, and this + /// crate does not build with `panic = "abort"`. A plain statement after the `.await` is skipped by + /// the unwind, the task then continues, and the entry is never released. + /// + /// **What the leak costs, which is why this is not tidiness:** the capsule becomes permanently + /// ineligible for the relay path, and the resulting exhaustion is composed into a transport error + /// NAMING A PEER — which `SPEC.md` §21.1, as amended by this same change, says MUST NOT happen. + /// + /// **The fixture is the production guard, driven by a production panic**, through the same + /// `AssertUnwindSafe(...).catch_unwind()` the tier-0 loop uses, so it exercises the unwind rather + /// than simulating it. + #[tokio::test] + async fn a_pull_that_panics_still_reports_its_end() { + use futures::FutureExt; + + let (store_hex, root_hex) = (hex32(STORE), hex32(chain_root())); + let dir = temp_dir("lifecycle-panic"); + let spy = Arc::new(RecordingLifecycle::default()); + let content = dig_download::module_content_id(&store_hex, &root_hex) + .expect("canonical ids yield a content id"); + let warmer = CapsuleWarmer::new( + Arc::new(dig_download::testkit::MockProviderLocator::fixed( + dig_download::testkit::mock_providers(1, &content), + )), + Arc::new(PanickingHolder), + Arc::clone(&spy) as Arc, + Arc::new(dig_download::InMemoryStateStore::new()), + Arc::new(ConfirmingResolver), + WarmPaths { + staging_dir: dir.join("staging"), + cache_dir: dir.join("cache"), + }, + Arc::new(AnnounceSpy::default()), + Arc::new(WarmRegistry::new()), + dig_download::ModuleDownloadConfig::default(), + Arc::new(crate::tier0_live::NoopModulesEvictor), + ); + + // The panic message is expected; keep it out of the test log so a real panic stays visible. + let previous = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + let outcome = std::panic::AssertUnwindSafe(warmer.warm(&store_hex, &root_hex)) + .catch_unwind() + .await; + std::panic::set_hook(previous); + + assert!( + outcome.is_err(), + "the control: the pull must genuinely PANIC and be caught, or the report below is just the ordinary return path" + ); + assert_eq!( + spy.finished.lock().expect("lifecycle lock").as_slice(), + &[(store_hex.clone(), root_hex.clone())], + "an unwinding pull must still report its end — a statement after the await is skipped by the unwind, and the tier-0 catch_unwind then continues with the entry leaked" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + /// **Proves (dig-node#276, unit 4):** a capsule pulled ON A STRANGER'S BEHALF lands in the cache — /// so the relayed windows can be served from it — and announces NOTHING, while the *same pull, /// through the same holder, of the same bytes*, driven for this node's OWN sake announces exactly @@ -1784,6 +2213,7 @@ mod tests { Arc::new(dig_download::testkit::MockModuleTransport::serving( &store_hex, &root_hex, module, 8, )), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::InMemoryStateStore::new()), Arc::new(ConfirmingResolver), WarmPaths { @@ -1982,6 +2412,7 @@ mod tests { let warmer = CapsuleWarmer::new( locator, transport, + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::InMemoryStateStore::new()), Arc::new(ConfirmingResolver), WarmPaths { @@ -2017,6 +2448,7 @@ mod tests { let warmer = CapsuleWarmer::new( Arc::new(NoHolders), Arc::new(UnusedTransport), + Arc::new(crate::seams::dig_peer::NoPullState), Arc::new(dig_download::FileStateStore::new(dir.join("state"))), Arc::new(ConfirmingResolver), WarmPaths { @@ -2116,6 +2548,46 @@ mod tests { ); } + /// **Proves (dig-node#333 review, finding 3):** a RELAYED warm and this node's OWN warm draw from + /// the SAME bounded pool of concurrent-warm slots, so a relay started for a stranger can deny a + /// slot to a local read. + /// + /// **Why this is worth a test rather than a comment:** the relay pull changed from awaited to + /// SPAWNED, so it now outlives the request that asked for it — a peer that gives up leaves the + /// slot held. The behaviour is bounded and opt-in and is not a hole, but "relaying for strangers + /// can starve your own warms" is a property an operator enabling the leg needs to be able to + /// discover, and an undocumented, untested property is one that quietly changes. + /// + /// **Fixture design:** the cap is filled by RELAY-shaped claims and the local warm is the one + /// refused, which is the direction that matters. Asserting the reverse — that a local warm can + /// deny a relay — would be satisfied by any shared cap and would not distinguish the two pools + /// being one from the two pools merely both existing. + #[test] + fn a_relayed_warm_and_a_local_warm_compete_for_the_same_slots() { + let registry = Arc::new(WarmRegistry::new()); + + // Every slot taken by capsules a STRANGER asked this node to relay. + let relayed: Vec<_> = (0..DEFAULT_MAX_CONCURRENT_WARMS) + .map(|n| { + registry + .claim(format!("relayed:{n}")) + .expect("a relay claim within the cap is granted") + }) + .collect(); + + assert!( + registry.claim("local:mine".into()).is_none(), + "this node's OWN warm was refused a slot because relays hold them all — the two share one cap, which is the cost of running the relay leg" + ); + + // A finished relay hands the slot back; nothing about the claim is relay-specific. + drop(relayed.into_iter().next().expect("at least one claim")); + assert!( + registry.claim("local:mine".into()).is_some(), + "the control: the local warm must succeed once a relay slot frees, or the refusal above proves only that the registry was broken" + ); + } + /// **Proves:** breadth across DISTINCT generations is capped — the (N+1)th distinct generation is /// SKIPPED, not queued, once `max_concurrent` are already in flight; a slot freed by a finished warm /// makes the registry warmable again. diff --git a/crates/dig-node-core/src/seams/dig_peer/module_serve.rs b/crates/dig-node-core/src/seams/dig_peer/module_serve.rs index 8a804bf1..634faf82 100644 --- a/crates/dig-node-core/src/seams/dig_peer/module_serve.rs +++ b/crates/dig-node-core/src/seams/dig_peer/module_serve.rs @@ -253,6 +253,24 @@ pub fn module_unavailable_frame(code: i64) -> Value { json!({"error": {"code": code, "message": "this node does not hold the requested .dig module"}}) } +/// The frame a hop sends while it is STILL RELAYING the requested capsule (dig-node#333). +/// +/// Same shape as [`module_unavailable_frame`] — an `error` object with a canonical code — because a +/// requestor decodes exactly one refusal shape. What differs is the CODE and the `data`: the code is +/// the taxonomy's inconclusive miss (the answer is unknown, a retry is meaningful) rather than a +/// settled not-held, and the data carries how far this hop has got. +/// +/// The byte count is this node's claim about its own staging file. A requestor may use it to tell a +/// progressing relay from a stalled one; it is never evidence about the bytes themselves, which are +/// merkle-verified against the chain-anchored root like every other peer's (NC-12). +pub fn module_relay_pending_frame(staged_bytes: u64) -> Value { + json!({"error": { + "code": crate::download::content_miss_inconclusive(), + "message": "relaying the requested capsule on your behalf; not yet complete", + "data": { crate::RELAY_PROGRESS_FIELD: staged_bytes }, + }}) +} + /// Lower-case hex of 32 raw bytes. fn hex32(bytes: &[u8; 32]) -> String { bytes.iter().map(|b| format!("{b:02x}")).collect() diff --git a/crates/dig-node-core/src/seams/dig_peer/module_transport.rs b/crates/dig-node-core/src/seams/dig_peer/module_transport.rs index 964b4dd2..82d4f71f 100644 --- a/crates/dig-node-core/src/seams/dig_peer/module_transport.rs +++ b/crates/dig-node-core/src/seams/dig_peer/module_transport.rs @@ -41,6 +41,8 @@ use dig_download::{ use dig_peer::DigPeer; use dig_rpc_protocol::types::{FetchModuleRangeParams, GetModuleInfoParams, ModuleInfo}; +use descriptor_ask::DescriptorAnswer; + use super::pool_locator::ConnectedPool; use crate::download::BestEffort; @@ -66,6 +68,9 @@ pub struct NatModuleTransport { /// Which `(capsule, peer)` pairs have already had a plain descriptor round, so the relay opt-in /// is an escalation rather than a default. See [`RelayEscalation`]. escalation: RelayEscalation, + /// How much relay waiting each CAPSULE has already cost, so the bound is per pull rather than + /// per peer. See [`RelayWaitBudget`]. + relay_budget: RelayWaitBudget, } impl NatModuleTransport { @@ -86,6 +91,7 @@ impl NatModuleTransport { connected, locator, escalation: RelayEscalation::default(), + relay_budget: RelayWaitBudget::default(), } } @@ -293,6 +299,168 @@ impl RelayEscalation { } } +/// How much total relay WAITING one pull of one capsule may spend, across every hop it asks. +/// +/// # Why the bound has to be per PULL and not per peer (the amplification the ceiling alone misses) +/// +/// [`descriptor_ask::RELAY_MAX_WAIT`] bounds a wait on ONE hop, which is the right instrument for +/// the question it answers — how long to believe a single peer that says it is fetching. It is the +/// wrong instrument for the question the PULL asks, because the pull asks many peers. +/// +/// dig-download documents its own worst case as +/// `MAX_DESCRIPTOR_ATTEMPTS × holders × the transport's per-ask timeout`, and this leg raises that +/// last term from the descriptor ladder's 65 s to the relay ceiling's 30 minutes. The warm locator +/// unions the CONNECTED POOL into the provider set, so "holders" is every connected peer rather than +/// every announced one — and ten peers each fabricating one byte of progress per poll would hold a +/// single pull for roughly fifteen hours, with each peer individually inside its ceiling the whole +/// time. Nothing is provably lying; the pull simply never ends. +/// +/// **Shrinking the per-hop ceiling is NOT the fix** and would undo dig-node#333: an honest hop +/// relaying a large capsule over a slow link genuinely needs minutes, and that case is the entire +/// reason this path exists. So the per-hop ceiling stays exactly as it is, and the PULL gets its own +/// budget which every hop draws from. One honest hop may still spend the whole of it — the large +/// capsule keeps working — while ten liars share it rather than each being granted it. +const RELAY_WAIT_BUDGET_PER_PULL: std::time::Duration = descriptor_ask::RELAY_MAX_WAIT; + +/// Tracks relay waiting already spent per capsule, so the budget above is enforced across the hops +/// of one pull rather than granted afresh to each. +/// +/// Keyed by `(store_id, root)` — the generation, which is exactly what one `ModuleDownloader::download` +/// call is about, so the key IS the pull. Entries are this node's own pulls, so growth tracks its own +/// activity and not anything a peer controls. +/// +/// # The FIFO cap is a backstop, not the working mechanism +/// +/// Entries are released when their pull ends ([`PullLifecycle`]), and `WarmRegistry` admits at most +/// [`DEFAULT_MAX_CONCURRENT_WARMS`](super::module_reshare::DEFAULT_MAX_CONCURRENT_WARMS) generations +/// node-wide, so only a handful of entries are ever LIVE at once. Reaching [`Self::MAX_ENTRIES`] would +/// therefore take a release path that had stopped working, and evicting a LIVE entry would need the +/// 1023 older ones to be live too — which the warm cap makes impossible. The cap exists so that a +/// future leak is bounded, not because entries are expected to accumulate. +#[derive(Default)] +struct RelayWaitBudget { + /// Relay time already spent per capsule. + spent: std::sync::Mutex>, + /// Insertion order, so eviction is FIFO without a timestamp per entry. + order: std::sync::Mutex>, +} + +impl RelayWaitBudget { + /// The ledger ceiling, matching [`RelayEscalation::MAX_ENTRIES`]: both track this node's own + /// concurrent pulls, so one bound serves both. + const MAX_ENTRIES: usize = 1024; + + /// The capsule's key in this ledger. + fn key(store_id: &str, root: &str) -> String { + format!("{store_id}:{root}") + } + + /// How much of the pull's relay-wait budget is LEFT for `(store_id, root)`. + /// + /// Saturating, so an over-charge (a wait that overran its own ceiling by a scheduling margin) + /// yields zero rather than wrapping into a fresh budget. + fn remaining(&self, store_id: &str, root: &str) -> std::time::Duration { + let spent = self + .spent + .lock() + .unwrap_or_else(|p| p.into_inner()) + .get(&Self::key(store_id, root)) + .copied() + .unwrap_or(std::time::Duration::ZERO); + RELAY_WAIT_BUDGET_PER_PULL.saturating_sub(spent) + } + + /// Release `(store_id, root)`'s entry, because its PULL is over. + /// + /// Called by whoever drove `ModuleDownloader::download` — see [`PullLifecycle`]. Without it the + /// key would be the capsule for the DAEMON's lifetime rather than the pull's, and a capsule whose + /// first honest pull spent the budget could never take the relay path again until restart. Worse, + /// the refusal would be composed as a transport error naming a PEER, blaming that peer for this + /// node's own earlier spend. + fn release(&self, store_id: &str, root: &str) { + let key = Self::key(store_id, root); + if self + .spent + .lock() + .unwrap_or_else(|p| p.into_inner()) + .remove(&key) + .is_some() + { + self.order + .lock() + .unwrap_or_else(|p| p.into_inner()) + .retain(|k| k != &key); + } + } + + /// Charge `waited` against `(store_id, root)`. + /// + /// A zero charge records nothing: the overwhelming majority of descriptor asks involve no relay + /// at all, and an entry per ask would evict the entries that matter. + fn charge(&self, store_id: &str, root: &str, waited: std::time::Duration) { + if waited.is_zero() { + return; + } + let key = Self::key(store_id, root); + let mut spent = self.spent.lock().unwrap_or_else(|p| p.into_inner()); + let mut order = self.order.lock().unwrap_or_else(|p| p.into_inner()); + if !spent.contains_key(&key) { + if order.len() >= Self::MAX_ENTRIES { + if let Some(oldest) = order.pop_front() { + spent.remove(&oldest); + } + } + order.push_back(key.clone()); + } + let entry = spent.entry(key).or_insert(std::time::Duration::ZERO); + *entry = entry.saturating_add(waited); + } +} + +/// Told when a whole-capsule pull has ENDED, so state scoped to that pull can be released. +/// +/// # Why the transport cannot work this out for itself +/// +/// [`ModuleTransport`] is asked about ONE HOLDER at a time. It sees `get_module_info` for holder A, +/// then B, then C, and nothing distinguishes "C is the next holder of the same pull" from "C is the +/// first holder of a new pull of the same capsule". Only the caller driving +/// `ModuleDownloader::download` knows where the boundary is, so only the caller can say. +/// +/// # Why a TTL was rejected rather than being simpler +/// +/// A relay wait is charged when it ENDS, so during a wait — up to the whole per-hop ceiling — nothing +/// is charged and the entry looks idle. Any idle-TTL shorter than that ceiling can therefore expire a +/// LIVE pull's budget mid-wait, which silently restores the `holders × ceiling` multiplication the +/// budget exists to remove; and a TTL at or above the ceiling leaves a finished pull's spend blocking +/// the next one for exactly as long as the harm it was meant to fix. There is no safe window, so the +/// boundary has to be reported rather than guessed. +pub(crate) trait PullLifecycle: Send + Sync { + /// The pull of `(store_id, root)` has finished, in success or failure. + /// + /// MUST be called on EVERY exit from the pull, including failures — a pull that ends badly is + /// exactly the one that spent budget on hops that did not deliver. + fn pull_finished(&self, store_id: &str, root: &str); +} + +impl PullLifecycle for NatModuleTransport { + fn pull_finished(&self, store_id: &str, root: &str) { + self.relay_budget.release(store_id, root); + } +} + +/// A [`PullLifecycle`] that does nothing. +/// +/// TEST-ONLY, and deliberately so: production has exactly one warmer wiring and it MUST report pull +/// boundaries, so a no-op available to production code would be a way to switch the budget off by +/// accident. Tests use it because most of them are not about the budget at all. +#[cfg(test)] +pub(crate) struct NoPullState; + +#[cfg(test)] +impl PullLifecycle for NoPullState { + fn pull_finished(&self, _store_id: &str, _root: &str) {} +} + /// The largest framed JSON body accepted for a module DESCRIPTOR answer. /// /// The generic peer-request reader ([`crate::peer::read_framed`]) caps at 64 KiB, which is right for @@ -408,31 +576,813 @@ const DESCRIPTOR_ASK_DEADLINES: [std::time::Duration; 3] = [ std::time::Duration::from_secs(45), ]; -/// Run `ask` under each of `deadlines` in turn, returning the first answer. +/// The descriptor ask, as ONE indivisible unit: the deadline ladder, the second-pass relay +/// escalation, and the progress-bounded wait on a relaying hop. +/// +/// # Why this is a MODULE and not three functions in the parent +/// +/// Because a call site that can reach the middle of a decision will eventually be rewritten to do +/// exactly that. Measured on this very PR: an earlier revision unified the three steps into one +/// function, which made them inseparable from EACH OTHER but left the inner ladder helper reachable +/// from the parent -- and replacing the call site's body with a direct call to that helper deleted +/// BOTH the escalation and the relay wait from the shipped path with **968 tests still passing**. /// -/// `None` means every rung elapsed or every attempt failed — the caller turns that into ONE transport -/// error in this node's own vocabulary. Extracted from the dial so the LADDER's behaviour is testable -/// on simulated time without a peer, a socket, or a capsule. -async fn ask_within_deadlines(deadlines: &[std::time::Duration], mut ask: A) -> Option -where - A: FnMut() -> F, - F: std::future::Future>, -{ - for (rung, deadline) in deadlines.iter().enumerate() { - match tokio::time::timeout(*deadline, ask()).await { - Ok(Some(answer)) => return Some(answer), - // An attempt that ANSWERED "no" is a refusal, not slowness: re-asking cannot change it, - // and spending the remaining rungs on it would make every genuine miss cost the full - // ladder. Only an elapsed rung is retried. - Ok(None) => return None, - Err(_elapsed) => tracing::debug!( - rung = rung + 1, - deadline_secs = deadline.as_secs(), - "module pull: descriptor ask exceeded its deadline; re-asking on the next rung" - ), +/// The ladder lives in here and is PRIVATE, so no code OUTSIDE this file can climb it: the only way +/// in is [`descriptor_for_holder`], and the parent supplies only a single one-shot ask carrying no +/// policy at all. Rust privacy is module-scoped, so this is a boundary against the rest of the crate +/// and NOT against an edit made inside this file — an in-file bypass is still expressible, it is +/// merely detected rather than prevented. What catches that remaining case is +/// `the_production_get_module_info_climbs_the_whole_ladder` (which pins the elapsed ladder time +/// through the real method) together with +/// `a_cold_first_invocation_escalates_through_the_production_transport` (which pins the second round +/// through the real method's own locator). Both fail, bounded, in milliseconds. +/// +/// Prefer narrowing what a defect can be expressed in over asserting that it is absent. +mod descriptor_ask { + use super::{ModuleInfo, DESCRIPTOR_ASK_DEADLINES}; + + /// How a climb of the deadline ladder ended. + /// + /// The two failures are kept apart because they license different NEXT moves. A peer that answered + /// has proved it can answer, so asking it a different question is worth a round trip; a peer that + /// never answered has proved nothing except that it is unresponsive. + #[derive(Debug, PartialEq, Eq)] + enum LadderEnd { + /// The peer answered within a rung. + Answered(T), + /// The peer ANSWERED, and the answer was no. + Refused, + /// Every rung elapsed, or every attempt failed before it could answer. + Exhausted, + } + + /// Run `ask` under each of `deadlines` in turn, returning the first answer. + /// + /// Extracted from the dial so the LADDER's behaviour is testable on simulated time without a peer, a + /// socket, or a capsule. + async fn ask_within_deadlines( + deadlines: &[std::time::Duration], + mut ask: A, + ) -> LadderEnd + where + A: FnMut() -> F, + F: std::future::Future>, + { + for (rung, deadline) in deadlines.iter().enumerate() { + match tokio::time::timeout(*deadline, ask()).await { + Ok(Some(answer)) => return LadderEnd::Answered(answer), + // An attempt that ANSWERED "no" is a refusal, not slowness: re-asking cannot change it, + // and spending the remaining rungs on it would make every genuine miss cost the full + // ladder. Only an elapsed rung is retried. + Ok(None) => return LadderEnd::Refused, + Err(_elapsed) => tracing::debug!( + rung = rung + 1, + deadline_secs = deadline.as_secs(), + "module pull: descriptor ask exceeded its deadline; re-asking on the next rung" + ), + } + } + LadderEnd::Exhausted + } + + /// Obtain the descriptor for one holder: ask at the pair's current phase, escalate to a RELAY ask + /// within the same invocation if the plain round was answered with a no, and wait out a hop that + /// answers that it is relaying. + /// + /// # Why all three steps live in ONE function + /// + /// They are one decision -- what this node does about a holder that did not simply hand over the + /// descriptor -- and splitting them would leave a caller free to take the first step and skip the + /// rest. That is not hypothetical: a helper extracted purely for testability is a helper a call + /// site can quietly stop using, and its tests keep passing while the behaviour is gone. + /// + /// # Why the escalation happens here rather than on the next invocation (dig-node#322) + /// + /// The two-phase escalation itself is deliberate and stays: a requestor must not ask the whole + /// connected pool to fetch a capsule on its behalf before establishing that no reachable holder + /// exists, and that bound is what keeps the relay from being an amplification primitive. What was + /// wrong was where the second phase happened. A cold requestor spent its FIRST invocation entirely on + /// the plain round, so the documented single command could never relay and only an identical second + /// command worked -- which a user reads as flakiness, with nothing in the output to say otherwise. + /// Escalating here preserves the bound exactly (the plain round still goes first, and it is its + /// emptiness that unlocks the second) while making one command sufficient. + /// + /// # Why only a REFUSAL escalates + /// + /// A peer that could not answer a plain ask within the whole ladder will not answer a relay ask + /// either -- it travels the same stream to the same process. Escalating an exhausted ladder would + /// double this method's wall clock in precisely the case where the extra time buys nothing, so + /// [`LadderEnd::Exhausted`] ends the invocation and [`LadderEnd::Refused`] is what licenses phase two. + pub(super) async fn descriptor_for_holder( + already_escalated: bool, + relay_ceiling: std::time::Duration, + mut ask_once: A, + ) -> DescriptorOutcome + where + A: FnMut(bool) -> F, + F: std::future::Future>, + { + let mut answer = + ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || ask_once(already_escalated)).await; + if !already_escalated && matches!(answer, LadderEnd::Refused) { + answer = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || ask_once(true)).await; + } + match answer { + LadderEnd::Answered(DescriptorAnswer::Descriptor(info)) => DescriptorOutcome { + descriptor: Ok(*info), + relay_waited: std::time::Duration::ZERO, + }, + // The hop is FETCHING the capsule for us, so the ask is not over -- see + // [`wait_for_relayed_descriptor`] for why the wait that follows is bounded by PROGRESS. + LadderEnd::Answered(DescriptorAnswer::RelayPending { staged_bytes }) => { + let started = tokio::time::Instant::now(); + let waited = + wait_for_relayed_descriptor(staged_bytes, relay_ceiling, ask_once).await; + DescriptorOutcome { + descriptor: waited.map_err(DescriptorFailure::RelayWait), + relay_waited: started.elapsed(), + } + } + LadderEnd::Refused | LadderEnd::Exhausted => DescriptorOutcome { + descriptor: Err(DescriptorFailure::NoAnswer), + relay_waited: std::time::Duration::ZERO, + }, + } + } + + /// What one holder's descriptor ask produced, and what it COST in relay waiting. + /// + /// The cost is reported rather than inferred because the caller charges it against a per-PULL budget + /// (see `RelayWaitBudget`), and only this function can tell relay waiting apart from ladder time. + pub(super) struct DescriptorOutcome { + /// The descriptor, or this node's own name for why there is none. + pub(super) descriptor: Result, + /// How long was spent waiting on a RELAYING hop -- zero when no hop was relaying. + pub(super) relay_waited: std::time::Duration, + } + + /// Why a descriptor ask produced no descriptor. Named so the caller composes ONE transport error in + /// this node's own vocabulary, and so the relay endings stay distinguishable in a log. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub(super) enum DescriptorFailure { + /// No holder answered, or the answer was no. + NoAnswer, + /// A hop was relaying and the wait on it ended without a descriptor. + RelayWait(RelayWaitEnd), + } + + impl DescriptorFailure { + /// This node's own words for the failure. Never the peer's -- the crate sanitizes at its + /// `Display` layer and upstream must not defeat it (#1603). + pub(super) fn reason(self) -> &'static str { + match self { + DescriptorFailure::NoAnswer => "getModuleInfo failed", + DescriptorFailure::RelayWait(end) => end.reason(), + } + } + } + + /// What a descriptor ask came back with. + /// + /// The two are different FACTS, not two spellings of failure: a descriptor ends the ask, while a + /// relay in progress says the hop is mid-way through answering it and the requestor may wait. Before + /// dig-node#333 the second was indistinguishable from a miss, so a requestor abandoned a hop that was + /// actively fetching on its behalf. + #[derive(Debug, Clone, PartialEq, Eq)] + pub(super) enum DescriptorAnswer { + /// The holder (or a hop that has finished relaying) described the capsule. + /// + /// Boxed because a [`ModuleInfo`] carries one hash and one length per chunk and is far larger + /// than the other variant; an unboxed enum would pay that size on every ask. + Descriptor(Box), + /// A hop is RELAYING the capsule for us and has staged this many bytes so far. + RelayPending { staged_bytes: u64 }, + } + + /// How often a waiting requestor re-asks a relaying hop. Each poll is one small round trip, so this + /// is chosen to keep the wait responsive without making a multi-minute relay expensive to observe. + const RELAY_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(10); + + /// How long a relay may report NO forward progress before the requestor gives up on it. + /// + /// Sized well above [`RELAY_POLL_INTERVAL`] so ordinary jitter -- a poll that lands between two of + /// the hop's staging writes -- never reads as a stall. + const RELAY_STALL_WINDOW: std::time::Duration = std::time::Duration::from_secs(60); + + /// The hard ceiling on a single relay wait, however healthy the progress looks. + /// + /// **This is a security bound, not a tuning knob (NC-12).** The progress figure is a HOP'S CLAIM + /// about itself, so a hostile hop can fabricate a counter that rises forever and a stall window alone + /// would never catch it. The ceiling is what makes the worst case finite: a lying hop can waste this + /// much of one pull's time from one peer, and no more. It is generous because an honest hop pulling a + /// large capsule over a slow link is the case this whole path exists to serve, and the cost of that + /// generosity is bounded -- the requestor is waiting on ONE peer, which it chose. + pub(super) const RELAY_MAX_WAIT: std::time::Duration = std::time::Duration::from_secs(30 * 60); + + /// Why a relay wait ended without producing a descriptor. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub(super) enum RelayWaitEnd { + /// The hop kept answering but stopped making progress. + Stalled, + /// The hop kept making progress for longer than any wait may last. + Ceiling, + /// The hop stopped answering, or answered something that is neither a descriptor nor progress. + Abandoned, + } + + impl RelayWaitEnd { + /// This node's own vocabulary for the failure. Composed here rather than from anything the peer + /// said, so a hop can never author what this node logs (#1603). + pub(super) fn reason(self) -> &'static str { + match self { + RelayWaitEnd::Stalled => "the relaying hop stopped making progress", + RelayWaitEnd::Ceiling => "the relay exceeded the maximum a single wait may last", + RelayWaitEnd::Abandoned => "the relaying hop stopped answering", + } + } + } + + /// Wait for a hop that is relaying a capsule for us, re-asking with `ask` until it produces the + /// descriptor or one of [`RelayWaitEnd`]'s three endings. + /// + /// # Why the bound is PROGRESS and not a wall clock + /// + /// The descriptor ladder ([`DESCRIPTOR_ASK_DEADLINES`]) bounds a BLOCKING ask, where the cost of + /// waiting is a held stream on both ends -- so a tight cap is exactly right for it, and dig-node#333 + /// is not a case of that cap being too small. The relay ask is no longer blocking: the hop ACKs and + /// keeps pulling, so each further poll costs one small round trip. That changes which instrument is + /// correct. A wall-clock cap on a transfer whose size the requestor cannot know before the descriptor + /// arrives is either too short for a large capsule or a blanket licence for a slow one; forward + /// PROGRESS needs no size, and a hop that is genuinely moving bytes is exactly the hop worth waiting + /// for. + /// + /// Two bounds keep that honest, and both are required. [`RELAY_STALL_WINDOW`] ends a wait on a hop + /// that has stopped moving. [`RELAY_MAX_WAIT`] ends it regardless, because the progress figure is the + /// hop's own claim and a stall window cannot catch a liar who keeps counting (NC-12). + /// + /// **Progress never authorises a byte.** It decides only how long to keep waiting; the capsule that + /// eventually arrives is verified against the chain-anchored root exactly as a direct holder's would + /// be, and a hop that fabricated its way through this wait still cannot produce content that passes. + async fn wait_for_relayed_descriptor( + staged_at_first_ask: u64, + ceiling: std::time::Duration, + mut ask_once: A, + ) -> Result + where + A: FnMut(bool) -> F, + F: std::future::Future>, + { + let started = tokio::time::Instant::now(); + let mut best = staged_at_first_ask; + let mut last_advance = started; + loop { + // The ceiling is checked FIRST, before any further waiting, so a hop that keeps answering can + // never buy one more poll past it. + if started.elapsed() >= ceiling { + return Err(RelayWaitEnd::Ceiling); + } + tokio::time::sleep(RELAY_POLL_INTERVAL).await; + // Every poll is a full ladder climb, for the same reason the first ask was: a hop that + // is mid-relay is doing real work and may answer slowly. + match ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || ask_once(true)).await { + LadderEnd::Answered(DescriptorAnswer::Descriptor(info)) => return Ok(*info), + LadderEnd::Answered(DescriptorAnswer::RelayPending { staged_bytes }) => { + if staged_bytes > best { + best = staged_bytes; + last_advance = tokio::time::Instant::now(); + } else if last_advance.elapsed() >= RELAY_STALL_WINDOW { + return Err(RelayWaitEnd::Stalled); + } + tracing::debug!( + staged_bytes, + "module pull: the hop is still relaying; waiting on its progress" + ); + } + LadderEnd::Refused | LadderEnd::Exhausted => return Err(RelayWaitEnd::Abandoned), + } + } + } + + #[cfg(test)] + mod tests { + use super::*; + + /// The bound the FIELD behaved as if it had: three asks 2.00 s apart, the last abandoned one + /// millisecond before the holder answered (the three-machine run behind dig_ecosystem#3128). The + /// tests below use it as the BEFORE case, so "the new bound is better" is measured against what + /// was actually observed rather than against a bound this file invented. + const FIELD_DEADLINE: [std::time::Duration; 1] = [std::time::Duration::from_secs(2)]; + + /// The measured cold-describe cost of the 135 MB capsule that failed: ~4.0 s on a host whose + /// whole-file `cat` took 0.01 s, so it is compute, not disk. + const COLD_135MB: std::time::Duration = std::time::Duration::from_millis(4_000); + + /// The same describe throughput (~34 MB/s) applied to a 1 GB capsule: ~30 s. Chosen FROM the + /// measurement rather than picked, because the whole question the ladder answers is what happens + /// when the capsule is larger than the one that was measured. + const COLD_1GB: std::time::Duration = std::time::Duration::from_secs(30); + + /// A holder whose FIRST descriptor ask takes `cold` and whose later asks are answered from its + /// memo in ~0 s — the real serve-side shape, because a cold describe runs under `spawn_blocking` + /// on the holder ([`crate::Node::describe_held_module`]) and a blocking task is not cancelled when + /// the requestor's stream drops, so an ABANDONED ask still warms the memo. + /// + /// It counts asks, so a test can distinguish "the ladder re-asked" from "the first ask was simply + /// given longer". + struct MemoizingHolder { + warm_at: tokio::time::Instant, + asks: std::cell::Cell, + } + + impl MemoizingHolder { + fn new(cold: std::time::Duration) -> Self { + MemoizingHolder { + // The describe starts when the holder is first built, and finishes `cold` later + // whether or not anyone is still waiting — that is the non-cancellable property. + warm_at: tokio::time::Instant::now() + cold, + asks: std::cell::Cell::new(0), + } + } + + async fn ask(&self) -> Option<&'static str> { + self.asks.set(self.asks.get() + 1); + tokio::time::sleep_until(self.warm_at).await; + Some("descriptor") + } + } + + /// **Proves:** the capsule that actually failed in the field is obtained under the new bound. + /// + /// **Catches:** the shipped behaviour — an ask abandoned before a legitimately-slow holder can + /// answer, with nothing asking again. + /// + /// **Non-vacuous:** the companion below runs the SAME fixture under the bound the field behaved as + /// if it had, and must fail. The fixture cannot pass by being fast — `COLD_135MB` is the measured + /// cost and exceeds `FIELD_DEADLINE`. + #[tokio::test(start_paused = true)] + async fn the_capsule_that_failed_in_the_field_is_obtained_under_the_new_bound() { + let holder = MemoizingHolder::new(COLD_135MB); + + let answer = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || holder.ask()).await; + + assert_eq!(answer, LadderEnd::Answered("descriptor")); + } + + /// **Proves:** the same 135 MB describe is LOST under the bound the field behaved as if it had — + /// so the test above is load-bearing on the change, not on a generous fixture. + #[tokio::test(start_paused = true)] + async fn the_same_capsule_is_lost_under_the_field_deadline() { + assert!( + COLD_135MB > FIELD_DEADLINE[0], + "the fixture must exceed the observed bound or it proves nothing" + ); + let holder = MemoizingHolder::new(COLD_135MB); + + let answer = ask_within_deadlines(&FIELD_DEADLINE, || holder.ask()).await; + + assert_eq!( + answer, + LadderEnd::Exhausted, + "the observed bound cannot outlast a 4.0 s describe" + ); + } + + /// **Proves:** the LADDER, not merely a larger first rung, is what makes the bound scale — a + /// capsule whose describe costs ~30 s is obtained, and it takes THREE asks to get it. + /// + /// **Catches:** collapsing `DESCRIPTOR_ASK_DEADLINES` back to one rung of any size. A single rung + /// sized for 135 MB loses a 1 GB capsule (the companion below), and a single rung sized for 1 GB + /// would let one unanswerable holder hold a descriptor slot for 30 s before the next is tried. + /// + /// **Non-vacuous:** the ask COUNT is asserted, so a fixture answered on rung 1 could not pass. + #[tokio::test(start_paused = true)] + async fn a_capsule_an_order_of_magnitude_larger_needs_the_later_rungs() { + let holder = MemoizingHolder::new(COLD_1GB); + + let answer = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || holder.ask()).await; + + assert_eq!(answer, LadderEnd::Answered("descriptor")); + assert_eq!( + holder.asks.get(), + 3, + "rungs 1 and 2 must elapse and rung 3 must re-ask onto the warmed memo" + ); + } + + /// **Proves:** a single rung sized for the measured capsule loses the order-of-magnitude-larger + /// one — the reason the fix is a ladder rather than a bigger number. + #[tokio::test(start_paused = true)] + async fn one_rung_sized_for_the_measured_capsule_loses_the_larger_one() { + let holder = MemoizingHolder::new(COLD_1GB); + + let answer = + ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES[..1], || holder.ask()).await; + + assert_eq!(answer, LadderEnd::Exhausted); + } + + /// **Proves:** a holder that ANSWERS "no" spends exactly one rung. A refusal is not slowness, and + /// re-asking it would make every genuine miss cost the whole ladder before the next holder is + /// tried. + #[tokio::test(start_paused = true)] + async fn a_refusal_does_not_climb_the_ladder() { + let asks = std::cell::Cell::new(0usize); + + let answer: LadderEnd<&str> = + ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || async { + asks.set(asks.get() + 1); + None + }) + .await; + + // REFUSED, not exhausted: the two endings are kept apart because only the first licenses the + // escalated re-ask dig-node#322 adds, and reading a silent peer as a refusal would double + // every unanswerable holder's cost. + assert_eq!(answer, LadderEnd::Refused); + assert_eq!(asks.get(), 1, "a refused ask must not be retried"); + } + + /// **Proves:** the ladder is BOUNDED — an unanswerable holder costs the sum of the rungs and no + /// more, so a slow or hostile peer cannot hold a descriptor slot indefinitely. + #[tokio::test(start_paused = true)] + async fn an_unanswerable_holder_costs_exactly_the_ladder() { + let started = tokio::time::Instant::now(); + + let answer: LadderEnd<&str> = + ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || async { + std::future::pending::<()>().await; + None + }) + .await; + + assert_eq!(answer, LadderEnd::Exhausted); + assert_eq!( + started.elapsed(), + DESCRIPTOR_ASK_DEADLINES.iter().sum::(), + "the total wait must be exactly the ladder, never unbounded" + ); + } + + // -- dig-node#333: waiting on a relaying hop ----------------------------------------------------- + + /// The capsule the field run failed on: 134,968,945 bytes, measured on a three-machine + /// `A -> B -> C` (dig_ecosystem#3128). Every relay fixture below is sized from it rather than + /// from a round number, because the whole question is what happens to a REAL capsule. + const FIELD_CAPSULE_BYTES: u64 = 134_968_945; + + /// A hop pulling [`FIELD_CAPSULE_BYTES`] at 1 MB/s -- an unremarkable rate for a residential + /// uplink, and the one that makes the relay take the "minutes" the field observed. + /// + /// Chosen so the fixture DECISIVELY exceeds the descriptor ladder rather than merely brushing it: + /// at ~135 s it is more than twice the ladder's 65 s total, so a test that passes cannot be + /// explained by the ladder having been slightly generous. + const FIELD_RELAY_BYTES_PER_POLL: u64 = 1_000_000 * 10; + + /// A hop that answers every ask with its staged progress, and finally with a descriptor once it + /// has staged `completes_at` bytes. + /// + /// `advance_per_ask` is what varies between the tests below: a hop that is genuinely moving, one + /// that has frozen, and one that fabricates motion it will never finish. Everything else is held + /// constant, so a passing test cannot be explained by two differently-built fixtures. + struct RelayingHop { + staged: std::cell::Cell, + advance_per_ask: u64, + completes_at: Option, + asks: std::cell::Cell, + } + + impl RelayingHop { + fn new(advance_per_ask: u64, completes_at: Option) -> Self { + RelayingHop { + staged: std::cell::Cell::new(0), + advance_per_ask, + completes_at, + asks: std::cell::Cell::new(0), + } + } + + async fn ask(&self) -> Option { + self.asks.set(self.asks.get() + 1); + self.staged.set(self.staged.get() + self.advance_per_ask); + let staged = self.staged.get(); + match self.completes_at { + Some(target) if staged >= target => { + Some(DescriptorAnswer::Descriptor(Box::new(ModuleInfo { + total_size: FIELD_CAPSULE_BYTES, + module_hash: root(), + chunk_hashes: Vec::new(), + chunk_lens: Vec::new(), + }))) + } + _ => Some(DescriptorAnswer::RelayPending { + staged_bytes: staged, + }), + } + } + } + + /// **Proves (dig-node#333):** a requestor waits out a hop that is genuinely relaying, and gets the + /// descriptor -- across a span the descriptor ladder could never have covered. + /// + /// **Catches:** the shipped behaviour exactly. The field run had the requestor abandon at 65 s + /// while the hop went on to cache 134,968,945 bytes minutes later, so the capability was real and + /// no first attempt could ever use it. + /// + /// **Non-vacuous:** the assertion is on ELAPSED VIRTUAL TIME as well as on the descriptor. A wait + /// that merely returned the descriptor could be satisfied by a hop that answered immediately; the + /// elapsed check requires the wait to have outlasted the whole ladder, which is the thing the old + /// code structurally could not do. + #[tokio::test(start_paused = true)] + async fn a_requestor_waits_out_a_hop_that_is_genuinely_relaying() { + let hop = RelayingHop::new(FIELD_RELAY_BYTES_PER_POLL, Some(FIELD_CAPSULE_BYTES)); + let ladder: std::time::Duration = DESCRIPTOR_ASK_DEADLINES.iter().sum(); + let started = tokio::time::Instant::now(); + + // Driven through `descriptor_via_rounds` -- the function `get_module_info` itself calls -- + // rather than through the wait in isolation. A wait that is correct and no longer reached is + // exactly the shape a revert-proof misses. + let answer = descriptor_for_holder(true, RELAY_MAX_WAIT, |_proxy| hop.ask()) + .await + .descriptor; + + assert!( + answer.is_ok(), + "a hop that is moving bytes must be waited for: {answer:?}" + ); + assert!( + started.elapsed() > ladder, + "the fixture must outlast the {ladder:?} descriptor ladder or it proves nothing -- elapsed {:?}", + started.elapsed() + ); + assert!( + started.elapsed() < RELAY_MAX_WAIT, + "an honest relay must finish well inside the ceiling" + ); + } + + /// **Proves:** a hop that has STOPPED making progress is abandoned after + /// [`RELAY_STALL_WINDOW`], not waited on until the ceiling. + /// + /// **Why this is a separate test on a different hop behaviour:** the wait is a two-branch + /// decision, and a suite proving only the keep-waiting branch is satisfied by a wait that never + /// gives up at all. This hop answers every ask -- so it is not silent, and only the FROZEN counter + /// distinguishes it -- and never advances. + #[tokio::test(start_paused = true)] + async fn a_frozen_relay_is_abandoned_after_the_stall_window() { + let hop = RelayingHop::new(0, None); + let started = tokio::time::Instant::now(); + + let answer = descriptor_for_holder(true, RELAY_MAX_WAIT, |_proxy| hop.ask()) + .await + .descriptor; + + assert_eq!( + answer.unwrap_err(), + DescriptorFailure::RelayWait(RelayWaitEnd::Stalled) + ); + assert!( + started.elapsed() < RELAY_STALL_WINDOW + RELAY_POLL_INTERVAL * 2, + "a frozen hop must be dropped at the stall window, not held to the ceiling -- elapsed {:?}", + started.elapsed() + ); + } + + /// **Proves (NC-12):** a hop that FABRICATES endless progress is bounded by [`RELAY_MAX_WAIT`]. + /// + /// **Catches the nearest wrong implementation of this whole change:** a wait bounded only by + /// forward progress. That version passes both tests above and hangs forever here, because the + /// staged byte count is the hop's claim about itself and a liar who keeps counting never stalls. + /// + /// **Why the fixture cannot pass by accident:** this hop advances by ONE byte per ask and never + /// completes, so it is indistinguishable from an honest, very slow relay by progress alone. Only + /// the ceiling can end it. + #[tokio::test(start_paused = true)] + async fn a_hop_that_fabricates_endless_progress_is_bounded_by_the_ceiling() { + let hop = RelayingHop::new(1, None); + let started = tokio::time::Instant::now(); + + // Bounded at twice the ceiling so a REGRESSION FAILS rather than hangs. Without it the + // absent-ceiling case loops forever burning CPU on virtual time, which reads in CI as a stuck + // job rather than a failed assertion -- and a test that hangs on regression is a landmine for + // whoever trips it. Measured: this test's own revert-proof spun for 659 CPU-seconds before it + // was killed. + let answer = tokio::time::timeout( + RELAY_MAX_WAIT * 2, + descriptor_for_holder(true, RELAY_MAX_WAIT, |_proxy| hop.ask()), + ) + .await + .expect( + "the wait MUST end at the ceiling; an unbounded wait is the NC-12 defect itself", + ) + .descriptor; + + assert_eq!( + answer.unwrap_err(), + DescriptorFailure::RelayWait(RelayWaitEnd::Ceiling) + ); + assert!( + started.elapsed() >= RELAY_MAX_WAIT, + "the wait must run to the ceiling before ending" + ); + assert!( + started.elapsed() < RELAY_MAX_WAIT + RELAY_POLL_INTERVAL * 2, + "and must end AT the ceiling, not merely somewhere after it -- elapsed {:?}", + started.elapsed() + ); + } + + /// **Proves (finding 2):** the ceiling handed in is the ceiling USED — a hop gets only what the + /// pull has left, never a fresh [`RELAY_MAX_WAIT`]. + /// + /// **Catches the shape of a half-applied budget:** a parameter that is accepted, threaded, and + /// then ignored in favour of the constant. That version passes every other test in this module, + /// because every other test hands in exactly `RELAY_MAX_WAIT`. + /// + /// **Fixture design:** the hop is the SAME endless-progress liar as the ceiling test above, so the + /// only thing that varies is the budget it is granted. Its ending is identical; only the elapsed + /// time distinguishes the two, which is precisely the property under test. + #[tokio::test(start_paused = true)] + async fn a_hop_gets_only_what_the_pull_has_left() { + let hop = RelayingHop::new(1, None); + let nearly_spent = RELAY_POLL_INTERVAL * 3; + let started = tokio::time::Instant::now(); + + let answer = descriptor_for_holder(true, nearly_spent, |_proxy| hop.ask()).await; + + assert_eq!( + answer.descriptor.unwrap_err(), + DescriptorFailure::RelayWait(RelayWaitEnd::Ceiling) + ); + assert!( + started.elapsed() < nearly_spent + RELAY_POLL_INTERVAL * 2, + "the wait must end at the budget it was GIVEN, not at the per-hop maximum -- elapsed {:?} against a budget of {nearly_spent:?}", + started.elapsed() + ); + assert!( + started.elapsed() < RELAY_MAX_WAIT, + "the control: a ceiling that was ignored would run to {RELAY_MAX_WAIT:?}" + ); + } + + /// **Proves:** the relay wait is REPORTED, so a caller can charge it — and reported as zero when + /// no hop relayed, so an ordinary ask never consumes a pull's budget. + /// + /// **Why both halves:** a report that is always zero would silently disable the budget, and a + /// report that is never zero would exhaust it on ordinary asks. Neither is visible from the + /// outcome alone. + #[tokio::test(start_paused = true)] + async fn only_a_relaying_hop_costs_the_pull_any_relay_time() { + let relaying = RelayingHop::new(FIELD_RELAY_BYTES_PER_POLL, Some(FIELD_CAPSULE_BYTES)); + let waited = descriptor_for_holder(true, RELAY_MAX_WAIT, |_proxy| relaying.ask()) + .await + .relay_waited; + assert!( + waited > std::time::Duration::ZERO, + "a hop that relayed must report the time it cost" + ); + + let holder = RelayingHop::new(u64::MAX, Some(1)); + let none = descriptor_for_holder(true, RELAY_MAX_WAIT, |_proxy| holder.ask()) + .await + .relay_waited; + assert_eq!( + none, + std::time::Duration::ZERO, + "a holder that simply answered must cost the pull no relay budget at all" + ); + } + + /// **Proves:** a hop that stops answering ends the wait immediately, rather than being polled to + /// the ceiling. A relay that died is not a relay in progress. + #[tokio::test(start_paused = true)] + async fn a_hop_that_stops_answering_ends_the_wait() { + // First round: relaying. Every round after: silence. Only the SECOND round can produce the + // abandonment, so a fixture that was silent from the start could not exhibit it. + let rounds = std::cell::Cell::new(0usize); + let answer = descriptor_for_holder(true, RELAY_MAX_WAIT, |_proxy| { + let ix = rounds.get(); + rounds.set(ix + 1); + async move { + if ix == 0 { + Some(DescriptorAnswer::RelayPending { staged_bytes: 1 }) + } else { + None + } + } + }) + .await + .descriptor; + + assert_eq!( + answer.unwrap_err(), + DescriptorFailure::RelayWait(RelayWaitEnd::Abandoned) + ); + } + + // -- dig-node#322: escalating within one invocation ----------------------------------------------- + + /// Record the `proxy` phase of every round an escalation drives, so the SEQUENCE is the assertion + /// rather than a count that a single escalated round would also satisfy. + /// How a fixture holder ends a ROUND, expressed at the one-shot level the production closure + /// actually supplies. + #[derive(Clone, Copy)] + enum RoundEnding { + /// Answers, and the answer is no. One rung is spent and the ladder stops. + Refuses, + /// Never answers at all, so every rung of the ladder elapses. + Silent, + } + + /// The sequence of ROUNDS a descriptor ask drives, by phase, with each round ending as named. + /// + /// The ending is chosen BY PHASE rather than by call index, because a round is not one ask -- a + /// silent holder is asked once per rung, so an index-keyed fixture serves round two's ending to + /// round one's second rung and manufactures an escalation that never happened. That mistake made + /// this exact test report `[false, false, true]`; keying on the phase is what makes the fixture + /// describe rounds instead of asks. + /// + /// Consecutive asks at the same phase collapse into the one round they belong to, so the result + /// is the round sequence and never a rung count. + async fn phases_of( + already_escalated: bool, + plain: RoundEnding, + escalated: RoundEnding, + ) -> Vec { + let seen = std::cell::RefCell::new(Vec::new()); + let _ = descriptor_for_holder(already_escalated, RELAY_MAX_WAIT, |proxy| { + seen.borrow_mut().push(proxy); + let ending = if proxy { escalated } else { plain }; + async move { + match ending { + RoundEnding::Refuses => None, + // Pending forever: under `start_paused` the runtime advances the virtual clock, + // so every rung elapses and the ladder ends Exhausted -- the real shape of an + // unresponsive peer, rather than a value standing in for one. + RoundEnding::Silent => std::future::pending().await, + } + } + }) + .await; + let mut rounds: Vec = Vec::new(); + for phase in seen.into_inner() { + if rounds.last() != Some(&phase) { + rounds.push(phase); + } + } + rounds + } + + /// **Proves (dig-node#322):** a cold requestor's FIRST invocation escalates, once its plain round + /// has been answered with a no -- so the documented single command can relay. + /// + /// **The assertion is the PHASE SEQUENCE, not a round count.** A count of two is satisfied by two + /// plain rounds, and a count of one by an implementation that escalated immediately -- which would + /// remove the amplification bound this change must preserve. `[false, true]` is the only sequence + /// that is both fixed and correct. + #[tokio::test(start_paused = true)] + async fn a_cold_first_invocation_spends_a_plain_round_then_escalates() { + let phases = phases_of(false, RoundEnding::Refuses, RoundEnding::Refuses).await; + + assert_eq!( + phases, + vec![false, true], + "the plain round must come FIRST and the relay ask must follow it in the same invocation" + ); + } + + /// **Proves:** an EXHAUSTED plain round is not escalated, so an unresponsive holder still costs + /// exactly one ladder. + /// + /// **Catches the obvious over-broad version of the #322 fix** -- escalate whenever the round + /// produced no descriptor -- which doubles `get_module_info`'s wall clock on every silent peer. + /// Its production-path companion is + /// [`the_production_get_module_info_climbs_the_whole_ladder`], which pins the same fact in + /// virtual seconds through the real method. + #[tokio::test(start_paused = true)] + async fn an_unresponsive_holder_is_not_escalated() { + let phases = phases_of(false, RoundEnding::Silent, RoundEnding::Refuses).await; + + assert_eq!( + phases, + vec![false], + "a peer that could not answer a plain ask must not be asked to relay as well" + ); + } + + /// **Proves:** a pair that has ALREADY spent its plain round asks once, escalated. The escalation + /// is a second pass per `(capsule, peer)` pair, never a second pass per invocation. + #[tokio::test(start_paused = true)] + async fn an_already_escalated_pair_asks_once() { + let phases = phases_of(true, RoundEnding::Refuses, RoundEnding::Refuses).await; + + assert_eq!(phases, vec![true]); + } + + /// A canonical 64-hex id, so a fixture descriptor's `module_hash` is the right shape. + fn root() -> String { + [0xCDu8; 32].iter().map(|b| format!("{b:02x}")).collect() } } - None } /// One `dig.getModuleInfo` over `peer`, decoded into a [`ModuleInfo`]. `proxy` declares whether this @@ -446,7 +1396,7 @@ async fn descriptor_over( store_id: &str, root: &str, proxy: bool, -) -> Option { +) -> Option { let params = serde_json::to_value(GetModuleInfoParams { store_id: store_id.to_string(), root: root.to_string(), @@ -462,7 +1412,29 @@ async fn descriptor_over( .await .ok()?; let response = read_response_frame(&mut stream).await.ok()?; - serde_json::from_value(response.get("result")?.clone()).ok() + if let Some(result) = response.get("result") { + return serde_json::from_value(result.clone()) + .ok() + .map(|info: ModuleInfo| DescriptorAnswer::Descriptor(Box::new(info))); + } + relay_progress_in(&response).map(|staged_bytes| DescriptorAnswer::RelayPending { staged_bytes }) +} + +/// The staged byte count a RELAY-IN-PROGRESS answer carries, or `None` for any other answer. +/// +/// Keyed on the `data` FIELD rather than on the error code alone, and deliberately so: the code is +/// the taxonomy's ordinary inconclusive miss, which an honest non-relaying node also returns when its +/// own lookup was unsettled. Only a node that is relaying reports the field, so only the field +/// distinguishes "wait for me" from "I could not find out". +fn relay_progress_in(response: &serde_json::Value) -> Option { + let error = response.get("error")?; + if error.get("code")?.as_i64()? != crate::download::content_miss_inconclusive() { + return None; + } + error + .get("data")? + .get(crate::RELAY_PROGRESS_FIELD)? + .as_u64() } /// Open a `dig.fetchModuleRange` frame stream over `peer`, at the phase `proxy` names. @@ -497,6 +1469,32 @@ async fn window_stream_over( Some(stream) } +impl NatModuleTransport { + /// ONE descriptor ask of `provider_peer_id` at the phase `proxy` names -- a dial, a question, a + /// hang-up, and nothing else. + /// + /// This carries NO policy: no ladder, no escalation, no waiting. Every one of those lives in + /// [`descriptor_ask`], which drives this. Keeping the primitive policy-free is what leaves a call + /// site nothing to get a descriptor "almost" the right way WITH — outside this file there is no + /// partial path to reach at all, and inside it a bypass is caught by the two production tests + /// named on [`descriptor_ask`] rather than by the compiler. + /// + /// Each round re-dials: an abandoned round's connection is gone, and the holder's descriptor memo + /// -- not the connection -- is what makes the re-ask cheap. See [`DESCRIPTOR_ASK_DEADLINES`]. + async fn ask_descriptor_once( + &self, + provider_peer_id: &str, + store_id: &str, + root: &str, + proxy: bool, + ) -> Option { + let mut peer = self.connect(provider_peer_id, store_id, root).await.ok()?; + let answer = descriptor_over(&mut peer, store_id, root, proxy).await; + peer.disconnect().await; + answer + } +} + #[async_trait] impl ModuleTransport for NatModuleTransport { async fn get_module_info( @@ -505,21 +1503,29 @@ impl ModuleTransport for NatModuleTransport { store_id: &str, root: &str, ) -> Result { - // Each rung re-dials: the abandoned rung's connection is gone, and the holder's memo — not the - // connection — is what makes the re-ask cheap. See [`DESCRIPTOR_ASK_DEADLINES`]. - let result = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || async { - let mut peer = self.connect(provider_peer_id, store_id, root).await.ok()?; - let proxy = self - .escalation - .escalate_for(store_id, root, provider_peer_id); - let answer = descriptor_over(&mut peer, store_id, root, proxy).await; - peer.disconnect().await; - answer - }) + // The pair's phase, decided ONCE per invocation rather than re-read inside the ladder, so the + // rounds below say plainly which is which. `escalate_for` records the pair, so this is `false` + // exactly on the pair's first invocation. + let already_escalated = self + .escalation + .escalate_for(store_id, root, provider_peer_id); + + // The ceiling handed down is what is LEFT of this PULL's relay-wait budget, not a fresh + // per-peer allowance. See [`RelayWaitBudget`] for why the difference is the whole point. + let outcome = descriptor_ask::descriptor_for_holder( + already_escalated, + self.relay_budget.remaining(store_id, root), + |proxy| self.ask_descriptor_once(provider_peer_id, store_id, root, proxy), + ) .await; + self.relay_budget + .charge(store_id, root, outcome.relay_waited); + // The reason names the STEP and the sentinelled peer; the peer's own answer text is never - // embedded (#1603) — the crate sanitizes at its Display layer and upstream must not defeat it. - result.ok_or_else(|| DownloadError::transport(provider_peer_id, "getModuleInfo failed")) + // embedded (#1603). + outcome + .descriptor + .map_err(|failure| DownloadError::transport(provider_peer_id, failure.reason())) } async fn fetch_module_range( @@ -629,155 +1635,6 @@ pub(crate) fn pool_of(entries: &[(&str, &str)]) -> ConnectedPool { mod tests { use super::*; - /// The bound the FIELD behaved as if it had: three asks 2.00 s apart, the last abandoned one - /// millisecond before the holder answered (the three-machine run behind dig_ecosystem#3128). The - /// tests below use it as the BEFORE case, so "the new bound is better" is measured against what - /// was actually observed rather than against a bound this file invented. - const FIELD_DEADLINE: [std::time::Duration; 1] = [std::time::Duration::from_secs(2)]; - - /// The measured cold-describe cost of the 135 MB capsule that failed: ~4.0 s on a host whose - /// whole-file `cat` took 0.01 s, so it is compute, not disk. - const COLD_135MB: std::time::Duration = std::time::Duration::from_millis(4_000); - - /// The same describe throughput (~34 MB/s) applied to a 1 GB capsule: ~30 s. Chosen FROM the - /// measurement rather than picked, because the whole question the ladder answers is what happens - /// when the capsule is larger than the one that was measured. - const COLD_1GB: std::time::Duration = std::time::Duration::from_secs(30); - - /// A holder whose FIRST descriptor ask takes `cold` and whose later asks are answered from its - /// memo in ~0 s — the real serve-side shape, because a cold describe runs under `spawn_blocking` - /// on the holder ([`crate::Node::describe_held_module`]) and a blocking task is not cancelled when - /// the requestor's stream drops, so an ABANDONED ask still warms the memo. - /// - /// It counts asks, so a test can distinguish "the ladder re-asked" from "the first ask was simply - /// given longer". - struct MemoizingHolder { - warm_at: tokio::time::Instant, - asks: std::cell::Cell, - } - - impl MemoizingHolder { - fn new(cold: std::time::Duration) -> Self { - MemoizingHolder { - // The describe starts when the holder is first built, and finishes `cold` later - // whether or not anyone is still waiting — that is the non-cancellable property. - warm_at: tokio::time::Instant::now() + cold, - asks: std::cell::Cell::new(0), - } - } - - async fn ask(&self) -> Option<&'static str> { - self.asks.set(self.asks.get() + 1); - tokio::time::sleep_until(self.warm_at).await; - Some("descriptor") - } - } - - /// **Proves:** the capsule that actually failed in the field is obtained under the new bound. - /// - /// **Catches:** the shipped behaviour — an ask abandoned before a legitimately-slow holder can - /// answer, with nothing asking again. - /// - /// **Non-vacuous:** the companion below runs the SAME fixture under the bound the field behaved as - /// if it had, and must fail. The fixture cannot pass by being fast — `COLD_135MB` is the measured - /// cost and exceeds `FIELD_DEADLINE`. - #[tokio::test(start_paused = true)] - async fn the_capsule_that_failed_in_the_field_is_obtained_under_the_new_bound() { - let holder = MemoizingHolder::new(COLD_135MB); - - let answer = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || holder.ask()).await; - - assert_eq!(answer, Some("descriptor")); - } - - /// **Proves:** the same 135 MB describe is LOST under the bound the field behaved as if it had — - /// so the test above is load-bearing on the change, not on a generous fixture. - #[tokio::test(start_paused = true)] - async fn the_same_capsule_is_lost_under_the_field_deadline() { - assert!( - COLD_135MB > FIELD_DEADLINE[0], - "the fixture must exceed the observed bound or it proves nothing" - ); - let holder = MemoizingHolder::new(COLD_135MB); - - let answer = ask_within_deadlines(&FIELD_DEADLINE, || holder.ask()).await; - - assert_eq!( - answer, None, - "the observed bound cannot outlast a 4.0 s describe" - ); - } - - /// **Proves:** the LADDER, not merely a larger first rung, is what makes the bound scale — a - /// capsule whose describe costs ~30 s is obtained, and it takes THREE asks to get it. - /// - /// **Catches:** collapsing `DESCRIPTOR_ASK_DEADLINES` back to one rung of any size. A single rung - /// sized for 135 MB loses a 1 GB capsule (the companion below), and a single rung sized for 1 GB - /// would let one unanswerable holder hold a descriptor slot for 30 s before the next is tried. - /// - /// **Non-vacuous:** the ask COUNT is asserted, so a fixture answered on rung 1 could not pass. - #[tokio::test(start_paused = true)] - async fn a_capsule_an_order_of_magnitude_larger_needs_the_later_rungs() { - let holder = MemoizingHolder::new(COLD_1GB); - - let answer = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || holder.ask()).await; - - assert_eq!(answer, Some("descriptor")); - assert_eq!( - holder.asks.get(), - 3, - "rungs 1 and 2 must elapse and rung 3 must re-ask onto the warmed memo" - ); - } - - /// **Proves:** a single rung sized for the measured capsule loses the order-of-magnitude-larger - /// one — the reason the fix is a ladder rather than a bigger number. - #[tokio::test(start_paused = true)] - async fn one_rung_sized_for_the_measured_capsule_loses_the_larger_one() { - let holder = MemoizingHolder::new(COLD_1GB); - - let answer = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES[..1], || holder.ask()).await; - - assert_eq!(answer, None); - } - - /// **Proves:** a holder that ANSWERS "no" spends exactly one rung. A refusal is not slowness, and - /// re-asking it would make every genuine miss cost the whole ladder before the next holder is - /// tried. - #[tokio::test(start_paused = true)] - async fn a_refusal_does_not_climb_the_ladder() { - let asks = std::cell::Cell::new(0usize); - - let answer: Option<&str> = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || async { - asks.set(asks.get() + 1); - None - }) - .await; - - assert_eq!(answer, None); - assert_eq!(asks.get(), 1, "a refused ask must not be retried"); - } - - /// **Proves:** the ladder is BOUNDED — an unanswerable holder costs the sum of the rungs and no - /// more, so a slow or hostile peer cannot hold a descriptor slot indefinitely. - #[tokio::test(start_paused = true)] - async fn an_unanswerable_holder_costs_exactly_the_ladder() { - let started = tokio::time::Instant::now(); - - let answer: Option<&str> = ask_within_deadlines(&DESCRIPTOR_ASK_DEADLINES, || async { - std::future::pending::<()>().await; - None - }) - .await; - - assert_eq!(answer, None); - assert_eq!( - started.elapsed(), - DESCRIPTOR_ASK_DEADLINES.iter().sum::(), - "the total wait must be exactly the ladder, never unbounded" - ); - } - /// A canonical 64-hex id built from a repeated byte, so a test id can never be the wrong length. fn id_of(byte: u8) -> String { [byte; 32].iter().map(|b| format!("{b:02x}")).collect() @@ -872,7 +1729,16 @@ mod tests { let t = transport(pool_of(&[]), Arc::new(HangingLocator)); let started = tokio::time::Instant::now(); - let result = t.get_module_info(&peer, &store(), &root()).await; + // Bounded at twice the ladder so a REGRESSION FAILS rather than hangs. A call site that + // stops going through [`descriptor_ask`] has no ladder at all, so it waits on the pending + // locator forever -- measured, when the gate's bypass was reproduced here. A test that hangs + // on regression reads in CI as a stuck job rather than a failed assertion. + let ladder: std::time::Duration = DESCRIPTOR_ASK_DEADLINES.iter().sum(); + let result = tokio::time::timeout(ladder * 2, t.get_module_info(&peer, &store(), &root())) + .await + .expect( + "`get_module_info` must be BOUNDED by the ladder; an unbounded ask is the defect", + ); assert!( result.is_err(), @@ -885,6 +1751,171 @@ mod tests { ); } + /// Counts how many times the production path asked discovery for this capsule's providers. + /// + /// The count is the ROUND count seen from outside: every descriptor round dials, every dial + /// resolves candidates, and every resolution consults the locator exactly once. So a locator that + /// counts is an observer of the production method's round structure that needs no peer, no + /// socket and no injected seam. + #[derive(Default)] + struct CountingLocator { + calls: std::sync::atomic::AtomicUsize, + } + + #[async_trait] + impl ProviderLocator for CountingLocator { + async fn find_providers( + &self, + _content: &ContentId, + ) -> Result, DownloadError> { + self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + // No providers and no error: an ANSWER of "nobody", so the round is refused rather than + // exhausted -- which is exactly the condition dig-node#322 says must escalate. + Ok(Vec::new()) + } + } + + /// **Proves (dig-node#322, at the PRODUCTION entry point):** one call to the real + /// `ModuleTransport::get_module_info` drives TWO descriptor rounds when the first is refused — + /// the plain round and the escalated one — so the documented single command can relay. + /// + /// **Catches what the unit-level escalation tests structurally cannot:** a call site that stops + /// driving [`descriptor_ask`] at all. The gate proved that was possible — replacing this method's + /// body with a direct one-shot ask deleted both the escalation and the relay wait from the + /// shipped path with every test still green, because every test drove the helper rather than the + /// method. This one drives the method. + /// + /// **Non-vacuous:** the assertion is TWO, not "at least one". A bypass that keeps a single round + /// yields one; a fixture that never reached the transport at all yields zero. Only the escalation + /// produces two, and the companion above pins that the rounds are a plain one then a relay one. + #[tokio::test(start_paused = true)] + async fn a_cold_first_invocation_escalates_through_the_production_transport() { + let peer = peer_hex(1); + let locator = Arc::new(CountingLocator::default()); + let t = transport( + pool_of(&[]), + Arc::clone(&locator) as Arc, + ); + + let result = t.get_module_info(&peer, &store(), &root()).await; + + assert!( + result.is_err(), + "no holder answered, so there is no descriptor" + ); + assert_eq!( + locator.calls.load(std::sync::atomic::Ordering::SeqCst), + 2, + "a refused plain round must be followed by an ESCALATED round inside the same call; one round means the call site is no longer driving the escalation" + ); + } + + /// **Proves (finding 2):** the relay-wait budget is spent per CAPSULE, drains, and saturates — + /// so many hops on one pull share one allowance while a different pull is untouched. + /// + /// **Catches:** a budget keyed by peer, or one that resets, either of which restores the + /// `holders × ceiling` multiplication the budget exists to remove. + #[test] + fn the_relay_budget_is_spent_per_capsule_and_saturates() { + let budget = RelayWaitBudget::default(); + let (store, root) = (store(), root()); + let full = RELAY_WAIT_BUDGET_PER_PULL; + + assert_eq!( + budget.remaining(&store, &root), + full, + "a fresh pull has all of it" + ); + + // Two different HOPS of the same pull, charged separately, draw down the same allowance. + budget.charge(&store, &root, full / 4); + budget.charge(&store, &root, full / 4); + assert_eq!( + budget.remaining(&store, &root), + full / 2, + "hops of one pull SHARE the budget; a per-peer allowance would still read as full here" + ); + + // A different capsule is a different pull. + assert_eq!( + budget.remaining(&store, &id_of(0x77)), + full, + "the control: another capsule must be unaffected, or the ledger is global rather than per pull" + ); + + // Overrun saturates instead of wrapping into a fresh allowance. + budget.charge(&store, &root, full); + assert_eq!( + budget.remaining(&store, &root), + std::time::Duration::ZERO, + "an overrun must leave nothing, never wrap" + ); + } + + /// **Proves (dig-node#333 re-gate):** the ledger's lifetime is the PULL's, not the daemon's — a + /// SECOND pull of the same capsule starts with its whole budget. + /// + /// **Catches the shipped shape of the previous revision**, in which the transport was built once + /// at node wiring and the ledger only ever grew. A capsule whose first honest pull spent the + /// budget could then never take the relay path again until the process restarted — and the + /// refusal was composed as a transport error naming a PEER, blaming that peer for this node's own + /// earlier spend. That is the mis-attribution class this repo has already paid for twice. + /// + /// **Why the fixture must be TWO pulls:** the sibling test above exercises many hops of ONE pull + /// and passes identically whether or not the entry is ever released, because within a pull the + /// entry is supposed to persist. Only a second pull can tell a per-pull ledger from a permanent + /// one, and the assertion is that the second gets the FULL budget rather than merely a non-zero + /// one — a partial release would satisfy "more than nothing" while still charging pull two for + /// pull one. + #[test] + fn a_second_pull_of_the_same_capsule_gets_its_own_budget() { + let budget = RelayWaitBudget::default(); + let (store, root) = (store(), root()); + let full = RELAY_WAIT_BUDGET_PER_PULL; + + // PULL ONE spends every second of it — an honest hop relaying a large capsule. + budget.charge(&store, &root, full); + assert_eq!( + budget.remaining(&store, &root), + std::time::Duration::ZERO, + "the control: pull one must genuinely exhaust the budget, or pull two proves nothing" + ); + + // The pull ends. Whoever drove `download` says so. + budget.release(&store, &root); + + assert_eq!( + budget.remaining(&store, &root), + full, + "pull two must start with the WHOLE budget; a ledger keyed to the capsule for the daemon's lifetime would still read zero here and would refuse the relay path forever, while naming a peer as the cause" + ); + } + + /// **Proves:** releasing a pull leaves OTHER pulls' budgets alone, so the release is scoped and + /// not a global reset. + /// + /// Without this, `release` could be implemented as "clear everything" and still pass the test + /// above — which would hand every concurrent pull a fresh budget whenever any one of them ended, + /// restoring the multiplication under a different name. + #[test] + fn releasing_one_pull_does_not_refund_another() { + let budget = RelayWaitBudget::default(); + let (store, root) = (store(), root()); + let other_root = id_of(0x77); + let full = RELAY_WAIT_BUDGET_PER_PULL; + + budget.charge(&store, &root, full / 2); + budget.charge(&store, &other_root, full / 2); + + budget.release(&store, &root); + + assert_eq!( + budget.remaining(&store, &other_root), + full / 2, + "the OTHER pull keeps what it has spent; a release that refunds it is a global reset" + ); + } + /// **Proves:** an IPv6 candidate becomes a dialable target with its address BRACKETED — the exact /// bug that blocked the entire read leg, where `format!("{host}:{port}")` + `parse::()` /// failed with "invalid socket address syntax" before a socket was ever opened (#1593). diff --git a/crates/dig-wallet/Cargo.toml b/crates/dig-wallet/Cargo.toml index 0319dece..2665b6b0 100644 --- a/crates/dig-wallet/Cargo.toml +++ b/crates/dig-wallet/Cargo.toml @@ -83,8 +83,13 @@ sqlx = { version = "0.8", default-features = false, features = ["sqlite", "runti # `ChainSourceProvider` traits will not unify (`dig-node-core/src/seams/chia_peer/light_client.rs`). # So the lock carries TWO chia-query lines and two dci lines, on purpose. That is sound here only # because no `chia_query` type crosses the dig-wallet -> dig-node-core boundary; do NOT "align them" -# on sight -- doing so breaks dig-node-core. Collapsing to one version is dig_ecosystem#2227, which -# needs a chia-peer release against dci 0.2 first. +# on sight -- doing so breaks dig-node-core. Collapsing to one version is dig_ecosystem#2227. +# +# ITS BLOCKER IS NOT WHAT THIS COMMENT SAID UNTIL 2026-08-23. It named "a chia-peer release +# against dci 0.2", which was never the path: chia-peer 0.2.0 skipped dci 0.2 and went straight +# to dci ^0.3 + chia-protocol ^0.36.1, where chia-query 0.15.0 also sits. They unify again +# already -- at CHIA 0.36. So collapsing the two lines means moving dig-node-core off the 0.26 +# line, i.e. the chia-0.36 cascade (dig_ecosystem#3152). chia-query = "0.6.2" # Diagnostics MUST go through `tracing`, never stderr. dig-node installs the `dig-logging` # subscriber process-globally (dig-node-service::logging), and a Windows service has no stderr