Skip to main content

dig_download/
module.rs

1//! [`ModuleDownloader`] — the whole-`.dig`-module peer pull (the reshare leg, #1576).
2//!
3//! Where the resource [`Downloader`](crate::Downloader) fetches ONE resource within a capsule, the
4//! `ModuleDownloader` pulls the ENTIRE `.dig` module blob for a `(store_id, root)` generation — the
5//! complete, content-addressed, on-chain-anchored container — so a node that read one resource from a
6//! peer can become a COMPLETE resharer of the whole capsule (it re-serves every retrieval key,
7//! including private/encrypted resources, with valid proofs). This is the "whole-module semantics"
8//! decision on #1576; it is delivered over the EXISTING ranged-fetch transport (multi-source,
9//! resumable, per-source attributable) rather than a bespoke one-shot stream.
10//!
11//! ## The flow (decider plan #1576)
12//!
13//! 1. **Locate** the module's holders via the injected [`ProviderLocator`] (`find_providers` on the
14//!    capsule [`ContentId`]).
15//! 2. **Handshake** `dig.getModuleInfo` against a holder ([`ModuleTransport::get_module_info`]) →
16//!    [`ModuleInfo`] (`total_size`, `module_hash`, per-chunk `chunk_hashes` + `chunk_lens`). This is
17//!    the transfer descriptor: it defines the chunk plan AND the per-chunk + whole-blob checks.
18//! 3. **Spread** the chunks across the located holders ([`ModuleTransport::fetch_module_range`],
19//!    `dig.fetchModuleRange`), each chunk one range, round-robin from a per-chunk starting holder so a
20//!    multi-holder set is genuinely pulled from multiple sources. Chunks are pulled in ascending order
21//!    (one in flight); parallel in-flight chunks are a later optimization, not a contract.
22//! 4. **Attribute** each returned range against `chunk_hashes[i]` the instant it arrives — a tampered
23//!    or short range is REJECTED, its reason recorded against the serving holder, and the chunk
24//!    re-fetched from the next holder (per-source attribution, fail-closed before assembly). A frame
25//!    that OVERSHOOTS the requested window is clipped, not rejected: answering at chunk granularity is
26//!    legitimate (the §2.2 clip contract, #836).
27//! 5. **Resume** across pause / crash via the injected [`StateStore`]: a checkpointed chunk is read
28//!    back from staging and RE-ATTRIBUTED against `chunk_hashes` rather than trusted, so it is skipped
29//!    when still intact and re-fetched when the staging file has been corrupted since (#1605). A
30//!    resumed pull always ends in the same two final gates below — resume can never bypass them.
31//! 6. **Assemble** verified chunks in order into the [`Sink`]'s staging area, hashing each into a
32//!    running whole-module SHA-256 as it lands, then run the two fail-closed final gates BEFORE
33//!    finalize — (a) that running hash equals `module_hash` (whole-blob integrity), and (b) the staged
34//!    module verifies against its chain-anchored `root` via the injected [`ModuleAnchorVerifier`],
35//!    which reads it through the bounded [`ModuleReader`] seam (NC-9 —
36//!    verified-content-is-not-safe-until-chain-bound; a right-shaped-but-forged module a lying
37//!    holder-set could otherwise agree on is caught here). Only if BOTH pass is the sink finalized +
38//!    the resume checkpoint cleared. A failure leaves the staging file unfinalized (never written
39//!    through) and is terminal for the pull.
40//!
41//!    **Peak memory is ONE CHUNK, not one module** (#1610). Nothing sized by the declared `total_size`
42//!    is ever allocated, so a small host can reshare a capsule far larger than its RAM. Streaming the
43//!    hash opens no window on partially-verified bytes: a chunk is absorbed only after it matches
44//!    `chunk_hashes[i]`, the bytes live in the STAGING area — never the artifact — and promotion is a
45//!    single atomic step strictly after both gates pass.
46//!
47//! ## Trust model
48//!
49//! The [`ModuleInfo`] is obtained from ONE holder and used to plan + attribute ranges. It is NOT
50//! trusted for safety: the per-chunk `chunk_hashes` only give cheap early rejection + attribution,
51//! and a holder-set that consistently lies about BOTH the descriptor and the bytes still fails the
52//! whole-blob `module_hash` check (if it disagrees with the served bytes) or — decisively — the
53//! chain-anchor gate, which binds the assembled `.dig` to the on-chain `(store_id, root)`. The
54//! chain-anchor gate is the sole root of trust; everything before it is optimization.
55//!
56//! ## Injection seams
57//!
58//! Like the rest of the crate, the network + store-format are INJECTABLE so the engine is tested over
59//! an in-memory harness:
60//! - [`ModuleTransport`] — the two `dig.getModuleInfo` / `dig.fetchModuleRange` peer calls. The real
61//!   dig-nat/dig-peer adapter is wired by dig-node's serve/client legs (the module client methods do
62//!   not yet exist on the shared peer client); this crate ships the seam + the in-memory
63//!   [`testkit::MockModuleTransport`](crate::testkit) used by the tests.
64//! - [`ModuleAnchorVerifier`] + [`ModuleReader`] — bind the STAGED module to the chain root, read
65//!   through a bounded window rather than handed over as one slice. dig-node injects the
66//!   digstore verifier (which parses the `.dig`, extracts its committed root, and checks it equals the
67//!   `getAnchoredRoot` value). There is no fail-open production default: the no-op
68//!   `AcceptAnyModuleAnchor` exists ONLY under `cfg(test)` / the `testkit` feature, so a default
69//!   consumer build cannot even name it.
70
71use std::sync::Arc;
72
73use async_trait::async_trait;
74use dig_dht::ContentId;
75use dig_rpc_protocol::types::ModuleInfo;
76use sha2::{Digest, Sha256};
77
78use crate::error::{
79    hex64_or_sentinel, sanitize_untrusted_text, DownloadError, VerifyError, MAX_ERROR_REASON_CHARS,
80};
81use crate::locate::ProviderLocator;
82use crate::progress::{DownloadState, StateStore};
83use crate::sink::{promote_verified, Sink};
84
85/// The two peer calls the module pull needs, abstracted for testability (in-memory
86/// [`MockModuleTransport`](crate::testkit) in tests; the real dig-nat/dig-peer adapter is wired by
87/// dig-node's serve/client legs, #1576 sub-family 4).
88#[async_trait]
89pub trait ModuleTransport: Send + Sync {
90    /// `dig.getModuleInfo` — the transfer descriptor for the `(store_id, root)` module from
91    /// `provider_peer_id`. `store_id` / `root` are the 64-hex generation ids.
92    ///
93    /// # Errors
94    /// A recoverable [`DownloadError::Transport`] on connect/stream failure — the caller tries
95    /// another holder.
96    async fn get_module_info(
97        &self,
98        provider_peer_id: &str,
99        store_id: &str,
100        root: &str,
101    ) -> Result<ModuleInfo, DownloadError>;
102
103    /// `dig.fetchModuleRange` — the `[offset, offset+length)` window of the module blob from
104    /// `provider_peer_id`.
105    ///
106    /// # Errors
107    /// A recoverable [`DownloadError::Transport`]/[`DownloadError::Timeout`] — the caller re-fetches
108    /// the chunk from another holder.
109    async fn fetch_module_range(
110        &self,
111        provider_peer_id: &str,
112        store_id: &str,
113        root: &str,
114        offset: u64,
115        length: u64,
116    ) -> Result<Vec<u8>, DownloadError>;
117}
118
119/// Random-access read over the module bytes a pull has staged — the seam the anchor gate sees
120/// INSTEAD of a `&[u8]` of the whole module (#1610).
121///
122/// A `&[u8]` parameter forced the puller to hold the entire module in RAM before it could ask the one
123/// question that decides the pull, so peak RSS was the module size and a small host simply could not
124/// reshare a large capsule. Behind this trait the same gate reads the staging area on demand, so peak
125/// RSS is one chunk.
126///
127/// ## What an implementation MUST guarantee
128///
129/// Every byte this reader returns is already **chunk-hash-verified against the descriptor**, and the
130/// readable window is exactly the `total_size` the whole-module-hash gate has already accepted — the
131/// gate never sees an unverified or out-of-window byte (`StagedModuleReader` is this crate's
132/// implementation and enforces both).
133#[async_trait]
134pub trait ModuleReader: Send + Sync {
135    /// The module's verified length in bytes. Reads are clamped to `[0, len())`.
136    fn len(&self) -> u64;
137
138    /// Whether the module is empty — present because clippy requires it beside [`len`](Self::len);
139    /// a real `.dig` module is never empty.
140    fn is_empty(&self) -> bool {
141        self.len() == 0
142    }
143
144    /// Read the `[offset, offset + len)` window of the module.
145    ///
146    /// # Errors
147    /// [`DownloadError::Sink`] when the window falls outside the module or the staged bytes cannot be
148    /// read back / no longer match the descriptor's chunk hashes. A read error is never "zeroes": the
149    /// caller MUST treat it as a failure to verify, never as absent content.
150    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, DownloadError>;
151}
152
153/// Binds a fully-staged `.dig` module to its chain-anchored `(store_id, root)` — the sole root of
154/// trust of the module pull (NC-9). dig-node injects the digstore verifier; this crate ships only the
155/// explicitly-opt-in, fail-OPEN [`AcceptAnyModuleAnchor`] for tests.
156#[async_trait]
157pub trait ModuleAnchorVerifier: Send + Sync {
158    /// Whether the module behind `module` is the genuine `.dig` container committed on-chain under
159    /// `(store_id, root)` (i.e. its embedded generation root equals the `getAnchoredRoot` value).
160    ///
161    /// `module` is a **borrowed, read-only** view of the staged bytes and is valid only for the
162    /// duration of this call: it cannot be retained, and it cannot promote or mutate anything. The
163    /// bytes it yields are chunk-hash-verified and bounded to the already-hash-gated module length,
164    /// so reading them incrementally is not a weaker check than being handed the whole slice was —
165    /// it is the same bytes, materialized one window at a time.
166    ///
167    /// An implementation that consults the chain MUST report [`ModuleAnchor::Unavailable`] when it
168    /// could not reach an answer, NEVER [`ModuleAnchor::NotAnchored`]. The two are acted on very
169    /// differently: `NotAnchored` is EVIDENCE against the holder that supplied the descriptor and earns
170    /// it a durable demotion, while `Unavailable` is this node's own failure and is terminal for the
171    /// pull. Collapsing them lets a chain-source blip brand every honest holder tried (see
172    /// [`ModuleAnchor`]). A read error from `module` is likewise the LOCAL node failing to read its own
173    /// staging area ⇒ `Unavailable`, not `NotAnchored`.
174    async fn verify_module_anchor(
175        &self,
176        module: &dyn ModuleReader,
177        store_id: &str,
178        root: &str,
179    ) -> ModuleAnchor;
180}
181
182/// The three answers a [`ModuleAnchorVerifier`] can give — the reason this is not a `bool`.
183///
184/// A two-valued answer forces an implementation that cannot reach the chain to say "not anchored",
185/// which is a claim about the HOLDER. That mislabels an honest holder serving a correct blob during a
186/// chain-source outage, and the resulting durable verdict then INVERTS the node's descriptor preference
187/// for the whole reputation TTL: remembered honest holders are skipped and unremembered peers — which
188/// is what a sybil is — are asked first.
189#[derive(Debug, Clone, PartialEq, Eq)]
190pub enum ModuleAnchor {
191    /// The blob IS the module committed on-chain under `(store_id, root)`.
192    Anchored,
193    /// The blob is definitively NOT that module — evidence against the descriptor's source.
194    NotAnchored,
195    /// The check could not be completed (chain source unreachable, timeout, malformed local state).
196    /// Says NOTHING about the holder: terminal for the pull, never a verdict. Carries a short reason
197    /// for the terminal error.
198    Unavailable(String),
199}
200
201/// A **fail-OPEN** [`ModuleAnchorVerifier`] that accepts any blob without checking the chain — for
202/// tests ONLY. Provides NO chain-anchored integrity; a production caller MUST inject the real digstore
203/// anchor verifier.
204///
205/// COMPILED OUT of a default consumer build (`cfg(any(test, feature = "testkit"))`): `#[doc(hidden)]`
206/// hides a type, it does not gate access, and the reshare path's only root of trust must not be
207/// bypassable by a `use`. A consumer that genuinely wants a no-op verifier opts in by enabling the
208/// `testkit` feature — a visible, reviewable choice in its `Cargo.toml`.
209#[cfg(any(test, feature = "testkit"))]
210#[doc(hidden)]
211#[derive(Debug, Clone, Copy, Default)]
212pub struct AcceptAnyModuleAnchor;
213
214#[cfg(any(test, feature = "testkit"))]
215#[async_trait]
216impl ModuleAnchorVerifier for AcceptAnyModuleAnchor {
217    async fn verify_module_anchor(
218        &self,
219        _module: &dyn ModuleReader,
220        _store_id: &str,
221        _root: &str,
222    ) -> ModuleAnchor {
223        ModuleAnchor::Anchored
224    }
225}
226
227/// The default [`ModuleDownloadConfig::max_module_size`] — 512 MiB.
228///
229/// **This is a DISK policy knob, no longer a memory bound (#1610).** The puller used to assemble the
230/// whole module in RAM, so the declared size was what one lying `getModuleInfo` could make a node
231/// allocate — a ceiling above host memory was then an out-of-memory primitive costing the attacker one
232/// message. Chunks are now hashed as they land and the anchor gate reads the staging area, so peak RSS
233/// is ONE CHUNK regardless of the declared size and no RAM ceiling is being defended.
234///
235/// What the bound still limits is the STAGING BYTES an unproven descriptor can make this node write to
236/// disk before either gate can reject it, so it is kept rather than removed. A deployment that
237/// reshares larger capsules now raises it on the disk budget alone; it no longer has to size host
238/// memory to the largest capsule it wants to serve.
239pub const DEFAULT_MAX_MODULE_SIZE: u64 = 512 * 1024 * 1024;
240
241/// The hard upper bound on the number of chunks a [`ModuleInfo`] may declare.
242///
243/// The declared chunk COUNT sizes the puller's `offsets` + `done` vectors, so an absurd count is the
244/// same one-message allocation attack as an absurd `total_size` — bounded here, before any allocation.
245/// 1 Mi chunks covers any real capsule at any sane chunk size.
246pub const MAX_MODULE_CHUNK_COUNT: usize = 1024 * 1024;
247
248/// How many DIFFERENT holders' descriptors a single pull will try before giving up.
249///
250/// The descriptor defines the whole plan, so a holder that answers `getModuleInfo` first with a
251/// well-formed but WRONG descriptor would otherwise deny the capsule's reshare permanently (holder
252/// order is deterministic, so every retry re-asks the same liar). A descriptor whose blob fails the
253/// whole-blob or chain-anchor gate is DEMOTED and the next holder's descriptor tried instead.
254pub const MAX_DESCRIPTOR_ATTEMPTS: usize = 3;
255
256/// Tunables for a module pull.
257#[derive(Debug, Clone)]
258pub struct ModuleDownloadConfig {
259    /// Per-ASK timeout — the window any single transport call to a holder gets, whether it is a
260    /// `fetchModuleRange` or the `getModuleInfo` handshake. A holder that does not answer within it is
261    /// treated as a failed source for that ask and the next holder is tried.
262    ///
263    /// It covers the descriptor ask as well as ranges because the descriptor phase's worst-case wait
264    /// (`MAX_DESCRIPTOR_ATTEMPTS` × holders × this window) has to be enforceable by this crate; a
265    /// transport whose `get_module_info` never resolves otherwise holds a pull open forever. The name
266    /// is kept for compatibility — renaming a public field is a breaking change, and a truthful bound
267    /// is worth more than a tidier name.
268    pub range_timeout: std::time::Duration,
269
270    /// Upper bound on the `total_size` a [`ModuleInfo`] may declare. The descriptor comes from an
271    /// UNTRUSTED holder and sizes the bytes this node will STAGE on disk before either final gate can
272    /// reject it, so it is refused above this bound before a single range is fetched. It is a disk
273    /// policy knob, not a memory bound — see [`DEFAULT_MAX_MODULE_SIZE`]. Defaults to
274    /// [`DEFAULT_MAX_MODULE_SIZE`].
275    pub max_module_size: u64,
276}
277
278impl Default for ModuleDownloadConfig {
279    fn default() -> Self {
280        ModuleDownloadConfig {
281            range_timeout: std::time::Duration::from_secs(30),
282            max_module_size: DEFAULT_MAX_MODULE_SIZE,
283        }
284    }
285}
286
287/// The multi-source, resumable, fail-closed whole-`.dig`-module puller. Built once from injected
288/// dependencies, then [`download`](ModuleDownloader::download)ed against many `(store_id, root)`
289/// generations.
290pub struct ModuleDownloader {
291    locator: Arc<dyn ProviderLocator>,
292    transport: Arc<dyn ModuleTransport>,
293    anchor: Arc<dyn ModuleAnchorVerifier>,
294    state_store: Arc<dyn StateStore>,
295    config: ModuleDownloadConfig,
296}
297
298impl ModuleDownloader {
299    /// Build a downloader from its injected seams.
300    pub fn new(
301        locator: Arc<dyn ProviderLocator>,
302        transport: Arc<dyn ModuleTransport>,
303        anchor: Arc<dyn ModuleAnchorVerifier>,
304        state_store: Arc<dyn StateStore>,
305        config: ModuleDownloadConfig,
306    ) -> Self {
307        ModuleDownloader {
308            locator,
309            transport,
310            anchor,
311            state_store,
312            config,
313        }
314    }
315
316    /// Pull the whole `.dig` module for `(store_id, root)`, writing the verified blob into `sink`.
317    ///
318    /// Returns the verified module byte length on success. See the module docs for the full flow +
319    /// trust model.
320    ///
321    /// # Errors
322    /// - [`DownloadError::NotFound`] — no holders located, or no located holder ANSWERED
323    ///   `getModuleInfo` within the attempt budget. Nothing was proven false in either case, so no
324    ///   source is attributable and the failure is not recast as a gate `Verify` (SPEC §17.5a).
325    /// - [`DownloadError::NoProviders`] — holders exhausted with chunks still missing.
326    /// - [`DownloadError::Verify`] — the whole-blob `module_hash` or the chain-anchor gate failed
327    ///   (fail-closed; the sink is NOT finalized).
328    pub async fn download(
329        &self,
330        store_id: &str,
331        root: &str,
332        sink: &dyn Sink,
333    ) -> Result<u64, DownloadError> {
334        let content = module_content_id(store_id, root).ok_or(DownloadError::NotDownloadable)?;
335
336        // 1. LOCATE the holders.
337        let mut providers = self.locator.find_providers(&content).await?;
338        if providers.is_empty() {
339            return Err(DownloadError::NotFound {
340                content: module_download_key(store_id, root),
341            });
342        }
343
344        // 2. Pull against ONE holder's descriptor at a time. A descriptor that is well-formed but
345        //    WRONG survives every per-chunk check and only dies at the final gates — so its SOURCE is
346        //    demoted and the next holder's descriptor tried, rather than the pull going terminal. One
347        //    holder winning the `getModuleInfo` race must not be able to deny a capsule's reshare.
348        let key = module_download_key(store_id, root);
349        let mut exclusions = DescriptorExclusions::new(self.remembered_verdicts(&key).await);
350        let mut attempts = 0usize;
351        loop {
352            // Reputation is consulted, never obeyed to the point of denial: the moment excluding the
353            // remembered holders would leave NOBODY to ask, the memory is dropped for the rest of this
354            // call and every located holder becomes askable again (#1611). This covers the case where
355            // some honest holders are remembered and the liar is not — excluding them, demoting the
356            // liar, and then finding "no usable holder" would deny a pull the network can serve.
357            if exclusions.usable_holders(&providers) == 0 && exclusions.stop_trusting_memory() {
358                tracing::warn!(
359                    holders = providers.len(),
360                    "no holder is left to ask for a descriptor once past verdicts are honoured; \
361                     re-asking every holder rather than letting reputation deny the pull"
362                );
363                continue;
364            }
365            // A descriptor that never ARRIVES spends an attempt too (#37). The budget used to be
366            // charged only for descriptors successfully obtained whose pull then failed, so a holder
367            // set that was merely slow or transiently unreachable ended the whole pull on the first
368            // ask — the one failure mode a retry budget exists for. `fetch_module_info` already asks
369            // every un-demoted holder within one round; what was missing is the ACROSS-round re-ask.
370            //
371            // Worst case is unchanged in shape and bounded by the same constant: at most
372            // `MAX_DESCRIPTOR_ATTEMPTS` rounds, each asking each un-demoted holder once, so the wait
373            // is `MAX_DESCRIPTOR_ATTEMPTS × holders × the transport's own per-ask timeout`. No holder
374            // is re-asked inside a round and no round is added beyond the budget, so an unanswerable
375            // holder set cannot hold the pull open indefinitely.
376            let (source, info) = match self
377                .fetch_module_info(&providers, &exclusions.excluded(), store_id, root)
378                .await
379            {
380                Ok(obtained) => obtained,
381                Err(e) => {
382                    attempts += 1;
383                    if attempts >= MAX_DESCRIPTOR_ATTEMPTS {
384                        return Err(e);
385                    }
386                    tracing::warn!(
387                        error = %e,
388                        attempts,
389                        "module pull: no holder answered getModuleInfo; re-asking the holder set"
390                    );
391                    continue;
392                }
393            };
394            attempts += 1;
395            let failure = match self
396                .pull_with_descriptor(&info, store_id, root, sink, &mut providers)
397                .await
398            {
399                Ok(len) => return Ok(len),
400                // A LOCAL failure — this node's memory, sink, state store, or an anchor check it could
401                // not COMPLETE — ends the pull and attributes nothing to any holder.
402                Err(PullFailure::Terminal(e)) => return Err(e),
403                Err(failure) => failure,
404            };
405            let proven_false = failure.is_proven_false();
406            tracing::warn!(
407                peer = %hex64_or_sentinel(&source, "peer-id"),
408                error = %failure.error(),
409                proven_false,
410                "module pull: descriptor attempt failed; demoting this source and re-handshaking \
411                 with another holder"
412            );
413            // Only a PROVEN-false descriptor earns a durable verdict. Chunk exhaustion demotes for
414            // this call (that is #1613's point) but carries no evidence the descriptor was false —
415            // the bytes may simply be unavailable, and the peers refusing them need not be the peer
416            // that supplied the descriptor. Persisting it would let sybils that refuse their assigned
417            // chunks brand an HONEST holder for 24 h, per capsule, until only attacker-supplied
418            // descriptors are ever asked for (#1611 security finding).
419            if proven_false {
420                if let Err(store_err) = self.state_store.record_bad_descriptor(&key, &source).await
421                {
422                    tracing::debug!(error = %store_err, "could not persist a bad-descriptor verdict");
423                }
424            }
425            exclusions.demote(source);
426            // Give up when THIS call's attempt budget is spent OR no askable holder is left (the
427            // memory has already been dropped above if it was what exhausted them). The budget counts
428            // attempts made here, not the exclusion set — which also carries remembered verdicts that
429            // cost this call nothing. The returned error is the DESCRIPTOR failure, never a "not
430            // found": blaming discovery for a descriptor lie is exactly the ambiguity that cost four
431            // #1586 rounds.
432            if attempts >= MAX_DESCRIPTOR_ATTEMPTS
433                || (exclusions.usable_holders(&providers) == 0 && !exclusions.trusts_memory())
434            {
435                return Err(match failure {
436                    PullFailure::BadDescriptor(e) | PullFailure::UnsatisfiableDescriptor(e) => e,
437                    PullFailure::Terminal(e) => e,
438                });
439            }
440            // The checkpoint and the bytes it describes are KEPT across the demotion (dig-node#328).
441            // Demoting a source says nothing about the bytes already verified under the old
442            // descriptor, and three guards already make carrying them into the next attempt safe:
443            //
444            //   1. a resumed chunk is re-read and re-hashed against the CURRENT descriptor's
445            //      `chunk_hashes[i]` (`read_back_verified_chunk`), so a byte staged under a lying
446            //      descriptor can never count toward an honest one — it simply re-fetches;
447            //   2. a descriptor of a DIFFERENT shape does not resume at all: `load_or_fresh_state`
448            //      returns `resumes_staging: false` on any `chunk_lens` mismatch and
449            //      `pull_with_descriptor` then truncates the staging area itself;
450            //   3. a longer tail cannot survive into the artifact: `promote_verified` shortens the
451            //      staging area to the verified length and REFUSES to promote if a byte is still
452            //      readable past it.
453            //
454            // Wiping here was therefore redundant, and it cost a full re-download on EVERY demotion —
455            // including a pure transport abort, where the descriptor was never shown to be false. On
456            // real hardware that turned a 135 MB capsule with ~20 MB already staged into 182 MB of
457            // inbound traffic, 0.12% MORE than starting from scratch.
458            //
459            // This holds for a PROVEN-false descriptor too, and deliberately: guard (1) re-attributes
460            // every resumed chunk against the next descriptor, so a liar's bytes are re-fetched rather
461            // than trusted, and a local read-back is orders of magnitude cheaper than the refetch a
462            // fail-closed wipe would force in the rare proven-liar case.
463        }
464    }
465
466    /// Run one whole pull — plan, resume, fetch, and the two final gates — against ONE holder's
467    /// descriptor. Returns [`PullFailure::BadDescriptor`] exactly when the assembled blob fails a
468    /// final gate (the descriptor's source lied and another holder should be asked), and
469    /// [`PullFailure::Terminal`] for every other failure.
470    async fn pull_with_descriptor(
471        &self,
472        info: &ModuleInfo,
473        store_id: &str,
474        root: &str,
475        sink: &dyn Sink,
476        providers: &mut Vec<dig_dht::ProviderRecord>,
477    ) -> Result<u64, PullFailure> {
478        let layout = ChunkPlan::from_info(info, self.config.max_module_size)?;
479
480        // Load resume state; a checkpoint for a DIFFERENT generation shape is discarded (never mixed)
481        // so a resume re-plans identically to the original.
482        let key = module_download_key(store_id, root);
483        let Resume {
484            mut state,
485            resumes_staging,
486        } = self.load_or_fresh_state(&key, &layout).await?;
487        if !resumes_staging {
488            // No checkpoint resumes THIS plan, so anything already in the staging area belongs to a
489            // different shape (an earlier abandoned attempt). It is discarded with the checkpoint —
490            // otherwise a longer stale tail rides out inside this plan's promotion.
491            sink.truncate(0).await?;
492        }
493
494        // STAGE + HASH the module CHUNK BY CHUNK, in ascending chunk order (#1610). There is no
495        // whole-module buffer: peak RSS is ONE CHUNK, and the attacker-declared `total_size` sizes no
496        // allocation at all — it now only bounds the staged bytes on disk.
497        //
498        // The fail-closed property is unchanged, and rests on the SAME two facts as before:
499        //   1. a chunk is absorbed into the running whole-module hash only AFTER it has matched the
500        //      descriptor's `chunk_hashes[i]`, so no unattributed byte ever reaches the hash gate; and
501        //   2. bytes land in the STAGING area, which is never the artifact — promotion is the atomic
502        //      `promote_verified` below, strictly after BOTH gates pass. Nothing partially verified is
503        //      observable at the final path, exactly as when a RAM blob held the same bytes.
504        let checkpointed = std::mem::take(&mut state.done_ranges);
505        let mut hasher = Sha256::new();
506        let mut any_chunk_verified = false;
507        for index in 0..layout.chunk_count() {
508            let (offset, len) = layout.chunk_span(index);
509            let staged = if checkpointed.contains(&index) {
510                self.read_back_verified_chunk(sink, info, index, offset, len)
511                    .await
512            } else {
513                None
514            };
515            let bytes = match staged {
516                Some(bytes) => bytes,
517                None => {
518                    // FETCH + ATTRIBUTE the chunk, fanned round-robin across the holders.
519                    let bytes = match self
520                        .fetch_verified_chunk(providers, info, &layout, index, store_id, root)
521                        .await
522                    {
523                        Ok(bytes) => bytes,
524                        // Exhaustion is attributed to the DESCRIPTOR whether or not a chunk has
525                        // verified: a liar can buy credibility for one byte, so only the attempt
526                        // budget may bound the retry (#1613). A non-recoverable failure (a sink/state
527                        // fault) stays terminal — it is the local node failing, not a holder lying.
528                        Err(e)
529                            if e.is_recoverable()
530                                || matches!(e, DownloadError::NotFound { .. }) =>
531                        {
532                            return Err(PullFailure::UnsatisfiableDescriptor(
533                                describe_chunk_exhaustion(e, any_chunk_verified),
534                            ))
535                        }
536                        Err(e) => return Err(PullFailure::Terminal(e)),
537                    };
538                    sink.write_at(offset, &bytes).await?;
539                    state.mark_done(index);
540                    self.state_store.save(&state).await?;
541                    bytes
542                }
543            };
544            any_chunk_verified = true;
545            hasher.update(&bytes);
546            state.mark_done(index);
547        }
548
549        // The two FAIL-CLOSED final gates, BEFORE finalize. Neither pass ⇒ the staging file is never
550        // promoted (the module is rejected, not written through — NC-9).
551        let assembled_hash = hex_of(hasher.finalize());
552        if assembled_hash != info.module_hash {
553            return Err(PullFailure::BadDescriptor(DownloadError::Verify(
554                VerifyError::Metadata(format!(
555                    "assembled module_hash {assembled_hash} != declared {}",
556                    hex64_or_sentinel(&info.module_hash, "module-hash")
557                )),
558            )));
559        }
560        // The anchor gate reads the staging area through a bounded, read-only, chunk-re-verifying
561        // window instead of being handed the whole module. It runs AFTER the whole-module hash gate,
562        // so every byte it can see belongs to a blob this node has already hashed end to end, and it
563        // runs BEFORE `promote_verified`, so its verdict still gates the whole artifact.
564        //
565        // The gate reads through the SINK, so a sink that cannot expose its staged bytes is refused
566        // here — explicitly, and named for what it is. Such a sink could never be promoted either
567        // (`promote_verified` refuses an unprovable staged length), so this is the same refusal one
568        // step earlier; naming it "the chain anchor could not be verified" would blame the chain for a
569        // local capability the sink simply does not have.
570        if !sink.supports_read_back() {
571            return Err(PullFailure::Terminal(DownloadError::sink(
572                "this sink cannot read back its staged bytes, so the chain-anchor gate has nothing \
573                 to read and the module could never be promoted; implement Sink::read_at + \
574                 Sink::supports_read_back",
575            )));
576        }
577        let reader = StagedModuleReader::new(sink, &layout, &info.chunk_hashes);
578        match self
579            .anchor
580            .verify_module_anchor(&reader, store_id, root)
581            .await
582        {
583            ModuleAnchor::Anchored => {}
584            ModuleAnchor::NotAnchored => {
585                return Err(PullFailure::BadDescriptor(DownloadError::Verify(
586                    VerifyError::Metadata(format!(
587                        "assembled module is not chain-anchored under ({store_id}, {root})"
588                    )),
589                )))
590            }
591            // The gate could not reach an answer. That is THIS node's failure, so it is terminal and
592            // earns the holder nothing: branding an honest holder for a chain-source blip would invert
593            // the node's descriptor preference toward unremembered (i.e. sybil) peers for the whole
594            // reputation TTL.
595            ModuleAnchor::Unavailable(reason) => {
596                return Err(PullFailure::Terminal(DownloadError::state(format!(
597                    "cannot verify the chain anchor for ({store_id}, {root}): {}",
598                    sanitize_untrusted_text(&reason, MAX_ERROR_REASON_CHARS)
599                ))))
600            }
601        }
602
603        // A promotion refusal is fail-closed AND recoverable: the checkpoint that led here is dropped
604        // with the bytes it describes, so a later pull re-fetches instead of failing identically forever
605        // (a checkpoint can outlive its staging file — GC reaps the `.download.tmp` while the
606        // `StateStore` keeps its record elsewhere). Best-effort: the promotion error is what the caller
607        // must see.
608        if let Err(e) = promote_verified(sink, layout.total_size).await {
609            let _ = self.state_store.clear(&key).await;
610            let _ = sink.truncate(0).await;
611            return Err(PullFailure::Terminal(e));
612        }
613        self.state_store.clear(&key).await?;
614        Ok(layout.total_size)
615    }
616
617    /// The holders this node has already caught supplying a PROVEN-false descriptor for `key` — the
618    /// [`StateStore`]'s remembered reputation, used to start a pull with the known liars already
619    /// demoted instead of re-discovering them (#1611). An unreadable store simply yields none:
620    /// reputation is advisory and must never fail a pull.
621    async fn remembered_verdicts(&self, key: &str) -> Vec<String> {
622        match self.state_store.bad_descriptor_peers(key).await {
623            Ok(peers) => peers,
624            Err(e) => {
625                tracing::debug!(error = %e, "could not read remembered descriptor verdicts");
626                Vec::new()
627            }
628        }
629    }
630
631    /// Try each not-yet-demoted holder's `dig.getModuleInfo` until one answers, returning the
632    /// answering holder's `peer_id` alongside its descriptor so a lying source can be attributed +
633    /// demoted.
634    ///
635    /// If every holder fails, the terminal error names the STEP (`getModuleInfo`) and carries each
636    /// holder's own reason — a swallowed reason resurfacing as an unrelated message cost six blind
637    /// diagnosis rounds on the read leg (#836).
638    async fn fetch_module_info(
639        &self,
640        providers: &[dig_dht::ProviderRecord],
641        excluded: &[String],
642        store_id: &str,
643        root: &str,
644    ) -> Result<(String, ModuleInfo), DownloadError> {
645        let mut reasons = HolderReasons::default();
646        let mut tried = 0usize;
647        for provider in providers {
648            let peer = &provider.provider_peer_id;
649            if excluded.iter().any(|d| d == peer) {
650                continue; // demoted in this call, or carrying a remembered verdict
651            }
652            tried += 1;
653            // BOUND THE ASK HERE, not in the transport. `SPEC.md` states the descriptor phase's
654            // worst-case wait as `MAX_DESCRIPTOR_ATTEMPTS x holders x the per-ask timeout`; before
655            // this, nothing in the crate enforced the last factor, so the promise rested on an
656            // INJECTED transport happening to have a timeout. A `ModuleTransport` whose
657            // `get_module_info` never resolves satisfies the trait and held a pull open forever, and
658            // #37's across-round re-ask multiplied the asks that exposure applies to.
659            //
660            // `range_timeout` is deliberately reused rather than a new knob added: the sibling
661            // orchestrator already bounds its holder METADATA probe by the same field
662            // (`orchestrator.rs`), so one per-ask window covering every per-holder transport call is
663            // this crate's established contract, not an invention. A new public field would also be a
664            // semver-incompatible bump on a `0.x` config struct with public fields, forcing every
665            // consumer to re-adopt for a knob nobody asked for.
666            //
667            // A timeout is the holder's failure, not the descriptor's: it is recorded like any other
668            // recoverable per-holder reason, so the next holder is asked and the budget still governs.
669            let asked = tokio::time::timeout(
670                self.config.range_timeout,
671                self.transport.get_module_info(peer, store_id, root),
672            )
673            .await;
674            match asked {
675                Ok(Ok(info)) => return Ok((peer.clone(), info)),
676                Ok(Err(e)) if e.is_recoverable() => reasons.record(peer, e),
677                Ok(Err(e)) => return Err(e),
678                Err(_) => reasons.record(
679                    peer,
680                    format!(
681                        "getModuleInfo timed out after {:?}",
682                        self.config.range_timeout
683                    ),
684                ),
685            }
686        }
687        Err(DownloadError::NotFound {
688            content: format!(
689                "getModuleInfo failed on all {tried} usable holder(s) ({} demoted) for module {} — \
690                 {reasons}",
691                excluded.len(),
692                module_download_key(store_id, root),
693            ),
694        })
695    }
696
697    /// Fetch chunk `index` from the holders, verifying each returned range against
698    /// `chunk_hashes[index]` for per-source attribution: a tampered range is rejected and the next
699    /// holder tried. Fetching cycles the holders starting at `index` (round-robin spread), so a
700    /// multi-holder set is pulled from multiple sources; one re-locate is attempted before giving up.
701    ///
702    /// Every rejection reason is recorded per holder and reported in the terminal error (#836).
703    async fn fetch_verified_chunk(
704        &self,
705        providers: &mut Vec<dig_dht::ProviderRecord>,
706        info: &ModuleInfo,
707        layout: &ChunkPlan,
708        index: usize,
709        store_id: &str,
710        root: &str,
711    ) -> Result<Vec<u8>, DownloadError> {
712        let (offset, len) = layout.chunk_span(index);
713        let expected_hash = &info.chunk_hashes[index];
714        let mut reasons = HolderReasons::default();
715
716        let mut relocated = false;
717        loop {
718            let count = providers.len();
719            for step in 0..count {
720                let peer = providers[(index + step) % count].provider_peer_id.clone();
721                match self
722                    .fetch_chunk_from(&peer, store_id, root, offset, len, expected_hash)
723                    .await
724                {
725                    Ok(bytes) => return Ok(bytes),
726                    Err(reason) => reasons.record(&peer, reason),
727                }
728            }
729            if relocated {
730                return Err(DownloadError::NotFound {
731                    content: format!(
732                        "fetchModuleRange failed for chunk {index} ([{offset}, {}) of module {}) on \
733                         all {count} known holder(s) — {reasons}",
734                        offset + len,
735                        module_download_key(store_id, root),
736                    ),
737                });
738            }
739            // Every known holder failed this chunk — ask the DHT for more before giving up.
740            let content =
741                module_content_id(store_id, root).ok_or(DownloadError::NotDownloadable)?;
742            let refreshed = self.locator.find_providers(&content).await?;
743            merge_new_providers(providers, refreshed);
744            relocated = true;
745        }
746    }
747
748    /// Fetch and attribute ONE chunk from ONE holder, returning either the verified bytes or a named
749    /// reason this holder could not serve it.
750    ///
751    /// A frame that overshoots the requested window is CLIPPED to it, never rejected: a holder
752    /// legitimately answers at its own chunk granularity (the §2.2 clip contract,
753    /// [`assemble_range_stream`](crate::source::assemble_range_stream), #836). Only bytes that both
754    /// fill the window and hash to `expected_hash` are accepted.
755    async fn fetch_chunk_from(
756        &self,
757        peer: &str,
758        store_id: &str,
759        root: &str,
760        offset: u64,
761        len: u64,
762        expected_hash: &str,
763    ) -> Result<Vec<u8>, String> {
764        let fetched = tokio::time::timeout(
765            self.config.range_timeout,
766            self.transport
767                .fetch_module_range(peer, store_id, root, offset, len),
768        )
769        .await;
770
771        let mut bytes = match fetched {
772            Ok(Ok(bytes)) => bytes,
773            Ok(Err(e)) => return Err(format!("transport: {e}")),
774            Err(_) => return Err(format!("timed out after {:?}", self.config.range_timeout)),
775        };
776
777        if bytes.len() as u64 > len {
778            bytes.truncate(len as usize); // CLIP — a chunk-granular holder is legitimate.
779        }
780        if bytes.len() as u64 != len {
781            return Err(format!(
782                "short range: wanted {len} bytes, got {}",
783                bytes.len()
784            ));
785        }
786        if sha256_hex(&bytes) != expected_hash {
787            return Err("chunk hash mismatch".to_string());
788        }
789        Ok(bytes)
790    }
791
792    /// Load the resume checkpoint for `key`, or a fresh one if none exists / the persisted generation
793    /// shape does not match the current [`ModuleInfo`] (a stale checkpoint is never partially reused).
794    async fn load_or_fresh_state(
795        &self,
796        key: &str,
797        layout: &ChunkPlan,
798    ) -> Result<Resume, DownloadError> {
799        let fresh = || {
800            let mut s = DownloadState::new(key);
801            s.total_length = layout.total_size;
802            s.chunk_lens = layout.chunk_lens.clone();
803            Resume {
804                state: s,
805                resumes_staging: false,
806            }
807        };
808        match self.state_store.load(key).await? {
809            Some(prev) if prev.chunk_lens == layout.chunk_lens => Ok(Resume {
810                state: prev,
811                resumes_staging: true,
812            }),
813            _ => Ok(fresh()),
814        }
815    }
816
817    /// Read one already-checkpointed chunk back from the sink's staging area, returning it only if it
818    /// still passes the SAME attribution a freshly-fetched chunk gets.
819    ///
820    /// The staging file is not a trusted input (it survives a crash, another process, and bit-rot), so
821    /// a resumed pull must not inherit corruption it can no longer localize. A chunk that cannot be
822    /// read back, reads short, or fails its hash yields `None` and is simply re-fetched — resume is an
823    /// optimization, never a correctness dependency (#1605).
824    async fn read_back_verified_chunk(
825        &self,
826        sink: &dyn Sink,
827        info: &ModuleInfo,
828        index: usize,
829        offset: u64,
830        len: u64,
831    ) -> Option<Vec<u8>> {
832        let bytes = sink.read_at(offset, len).await.ok()?;
833        if bytes.len() as u64 != len || sha256_hex(&bytes) != info.chunk_hashes[index] {
834            tracing::warn!(
835                chunk = index,
836                offset,
837                "staged chunk failed re-attribution on resume; re-fetching"
838            );
839            return None;
840        }
841        Some(bytes)
842    }
843}
844
845/// The [`ModuleReader`] the puller hands the anchor gate: a bounded, read-only, chunk-re-verifying
846/// window onto the sink's staging area (#1610).
847///
848/// It exists so the gate never needs the whole module in RAM. Two properties make that safe to
849/// substitute for the `&[u8]` it replaced:
850///
851/// - **Bounded.** Reads outside `[0, total_size)` are refused, so the gate cannot see a byte outside
852///   the blob the whole-module hash gate accepted — including any longer tail a demoted descriptor
853///   may have left staged (a staging area is never shortened by writing).
854/// - **Re-verified.** Every chunk is re-read from the artifact and re-hashed against the descriptor's
855///   `chunk_hashes` on each read, so a staging area mutated between the hash gate and the anchor gate
856///   fails closed instead of feeding the gate bytes nothing has attributed. Being handed a RAM blob
857///   gave weaker cover than this: it verified a COPY while promotion promoted the file.
858///
859/// Peak memory for one read is the caller's requested span plus one chunk. The anchor verifier is an
860/// injected, trusted component of the node (never a peer), so the span is not attacker-controlled.
861struct StagedModuleReader<'a> {
862    sink: &'a dyn Sink,
863    layout: &'a ChunkPlan,
864    chunk_hashes: &'a [String],
865}
866
867impl<'a> StagedModuleReader<'a> {
868    fn new(sink: &'a dyn Sink, layout: &'a ChunkPlan, chunk_hashes: &'a [String]) -> Self {
869        StagedModuleReader {
870            sink,
871            layout,
872            chunk_hashes,
873        }
874    }
875
876    /// Read chunk `index` back from staging and re-attribute it against the descriptor.
877    async fn verified_chunk(&self, index: usize) -> Result<Vec<u8>, DownloadError> {
878        let (offset, len) = self.layout.chunk_span(index);
879        let bytes = self.sink.read_at(offset, len).await?;
880        if bytes.len() as u64 != len {
881            return Err(DownloadError::sink(format!(
882                "staged chunk {index} reads {} bytes, expected {len}",
883                bytes.len()
884            )));
885        }
886        if sha256_hex(&bytes) != self.chunk_hashes[index] {
887            return Err(DownloadError::sink(format!(
888                "staged chunk {index} no longer matches its verified hash"
889            )));
890        }
891        Ok(bytes)
892    }
893}
894
895#[async_trait]
896impl ModuleReader for StagedModuleReader<'_> {
897    fn len(&self) -> u64 {
898        self.layout.total_size
899    }
900
901    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, DownloadError> {
902        if len == 0 {
903            return Ok(Vec::new());
904        }
905        let end = offset.checked_add(len).filter(|e| *e <= self.len());
906        let Some(end) = end else {
907            return Err(DownloadError::sink(format!(
908                "read [{offset}, {offset}+{len}) falls outside the {}-byte module",
909                self.len()
910            )));
911        };
912        let mut out = Vec::with_capacity(usize::try_from(len).map_err(|_| {
913            DownloadError::sink(format!(
914                "read of {len} bytes exceeds this platform's address space"
915            ))
916        })?);
917        // Start at the last chunk beginning at or before `offset`; zero-length chunks in between are
918        // stepped over by the loop rather than special-cased.
919        let mut index = self
920            .layout
921            .offsets
922            .partition_point(|&start| start <= offset)
923            .saturating_sub(1);
924        while (out.len() as u64) < len {
925            if index >= self.layout.chunk_count() {
926                // Unreachable while `end <= total_size` holds, but fail CLOSED rather than return a
927                // short read that a caller could mistake for the whole window.
928                return Err(DownloadError::sink(format!(
929                    "the staged chunk plan does not cover [{offset}, {end})"
930                )));
931            }
932            let (chunk_offset, chunk_len) = self.layout.chunk_span(index);
933            if chunk_len == 0 {
934                index += 1;
935                continue;
936            }
937            let chunk = self.verified_chunk(index).await?;
938            let want_from = offset + out.len() as u64;
939            let start = usize::try_from(want_from - chunk_offset).unwrap_or(usize::MAX);
940            let take = chunk
941                .len()
942                .saturating_sub(start)
943                .min(usize::try_from(len - out.len() as u64).unwrap_or(usize::MAX));
944            out.extend_from_slice(&chunk[start..start + take]);
945            index += 1;
946        }
947        Ok(out)
948    }
949}
950
951/// The resume checkpoint a pull starts from, and whether it belongs to THIS descriptor's plan.
952///
953/// `resumes_staging` is the licence to inherit what is already staged. A discarded (shape-mismatched
954/// or absent) checkpoint means the staging area — which no write ever shortens — may still hold a
955/// different plan's bytes, so it is reset rather than resumed.
956struct Resume {
957    state: DownloadState,
958    resumes_staging: bool,
959}
960
961/// Why one descriptor's pull attempt failed — and therefore whether ANOTHER holder's descriptor is
962/// worth trying.
963///
964/// A final-gate failure means the DESCRIPTOR was a lie and an honest holder may still serve the
965/// capsule. Chunk exhaustion is AMBIGUOUS and classified by [`classify_chunk_exhaustion`]: unavailable
966/// bytes and an unsatisfiable descriptor look identical from inside one attempt.
967enum PullFailure {
968    /// The descriptor was PROVEN false: the assembled blob failed the whole-blob-hash or chain-anchor
969    /// gate, or the descriptor was structurally unusable. Attributable to the holder that supplied it,
970    /// which is demoted for this call AND recorded durably.
971    BadDescriptor(DownloadError),
972    /// The descriptor could not be SATISFIED — the chunks it declares could not be fetched from any
973    /// holder. Its source is demoted for this call so another descriptor is tried (#1613), but nothing
974    /// here proves the descriptor was false: the bytes may be genuinely unavailable, and the holders
975    /// refusing them need not be the holder that supplied the descriptor. So it earns NO durable
976    /// verdict — see [`DescriptorEvidence`].
977    UnsatisfiableDescriptor(DownloadError),
978    /// Any other failure (a local sink/state fault) — terminal for the pull.
979    Terminal(DownloadError),
980}
981
982/// Which holders a pull will not ask for a descriptor, and whether it is still honouring the
983/// [`StateStore`]'s remembered verdicts.
984///
985/// Two sources of exclusion with DIFFERENT authority: holders demoted in THIS call (a failed attempt
986/// happened here — always excluded) and holders remembered from a past call (advisory, and droppable).
987/// Keeping them apart is what lets reputation be dropped without also forgiving a liar caught seconds
988/// ago — the bug of a single flat list.
989struct DescriptorExclusions {
990    demoted_here: Vec<String>,
991    remembered: Vec<String>,
992    trusts_memory: bool,
993}
994
995impl DescriptorExclusions {
996    fn new(remembered: Vec<String>) -> Self {
997        let trusts_memory = !remembered.is_empty();
998        DescriptorExclusions {
999            demoted_here: Vec::new(),
1000            remembered,
1001            trusts_memory,
1002        }
1003    }
1004
1005    /// The peers not to ask right now.
1006    fn excluded(&self) -> Vec<String> {
1007        let mut excluded = self.demoted_here.clone();
1008        if self.trusts_memory {
1009            excluded.extend(self.remembered.iter().cloned());
1010        }
1011        excluded
1012    }
1013
1014    /// How many located holders are still askable for a descriptor.
1015    fn usable_holders(&self, providers: &[dig_dht::ProviderRecord]) -> usize {
1016        let excluded = self.excluded();
1017        providers
1018            .iter()
1019            .filter(|p| !excluded.contains(&p.provider_peer_id))
1020            .count()
1021    }
1022
1023    /// Record that `peer` failed an attempt in THIS call (never droppable).
1024    fn demote(&mut self, peer: String) {
1025        self.demoted_here.push(peer);
1026    }
1027
1028    /// Stop honouring the remembered verdicts, returning whether that actually changed anything (i.e.
1029    /// whether the memory was what left nobody to ask).
1030    fn stop_trusting_memory(&mut self) -> bool {
1031        let was_trusting = self.trusts_memory;
1032        self.trusts_memory = false;
1033        was_trusting && !self.remembered.is_empty()
1034    }
1035
1036    fn trusts_memory(&self) -> bool {
1037        self.trusts_memory
1038    }
1039}
1040
1041/// Explain a chunk-level exhaustion: were the BYTES unavailable under a credible descriptor, or was
1042/// the DESCRIPTOR itself unsatisfiable?
1043///
1044/// This is DIAGNOSIS, not control flow. Exhaustion always demotes the descriptor source and re-tries
1045/// another holder's descriptor (bounded by [`MAX_DESCRIPTOR_ATTEMPTS`] and the un-demoted holder set),
1046/// because "did any chunk verify?" is not a bound an attacker respects: a holder declaring
1047/// `chunk_lens = [1, rest]` serves that ONE byte — matching its own fabricated first hash — and then
1048/// refuses everything, so a retry gated on the flag never happens and one liar denies the capsule's
1049/// reshare for the price of a single byte (#1613). The attempt budget alone guarantees termination.
1050///
1051/// Whether a chunk verified is still worth SAYING: exhaustion after real progress is more likely
1052/// genuine unavailability, exhaustion with none more likely a fabricated descriptor, and an operator
1053/// reading the log should not have to guess which.
1054fn describe_chunk_exhaustion(e: DownloadError, any_chunk_verified: bool) -> DownloadError {
1055    let diagnosis = if any_chunk_verified {
1056        "some chunk(s) had already verified under this descriptor, so the missing bytes are more \
1057         likely genuinely unavailable than fabricated"
1058    } else {
1059        "no chunk ever verified under this descriptor, so it is more likely fabricated than the \
1060         bytes unavailable"
1061    };
1062    match e {
1063        DownloadError::NotFound { content } => DownloadError::NotFound {
1064            content: format!("{content} — {diagnosis}"),
1065        },
1066        other => other,
1067    }
1068}
1069
1070impl PullFailure {
1071    /// The underlying error, whatever the attribution.
1072    fn error(&self) -> &DownloadError {
1073        match self {
1074            PullFailure::BadDescriptor(e)
1075            | PullFailure::UnsatisfiableDescriptor(e)
1076            | PullFailure::Terminal(e) => e,
1077        }
1078    }
1079
1080    /// Whether this failure PROVES the descriptor false, and so may be remembered against its source.
1081    fn is_proven_false(&self) -> bool {
1082        matches!(self, PullFailure::BadDescriptor(_))
1083    }
1084}
1085
1086impl From<DownloadError> for PullFailure {
1087    fn from(e: DownloadError) -> Self {
1088        PullFailure::Terminal(e)
1089    }
1090}
1091
1092/// Why each holder could not serve a step, accumulated so the terminal error explains the failure
1093/// instead of swallowing it (#836). Holder ids are sentinelled — a `provider_peer_id` is free-form
1094/// text off the wire, and a log an attacker can write is not evidence (#1603).
1095#[derive(Debug, Default)]
1096struct HolderReasons(Vec<String>);
1097
1098impl HolderReasons {
1099    /// Record `reason` against `peer`, and trace it as it happens (per-holder visibility even when a
1100    /// later holder succeeds and no error is ever returned).
1101    fn record(&mut self, peer: &str, reason: impl std::fmt::Display) {
1102        let peer = hex64_or_sentinel(peer, "peer-id");
1103        // The REASON is as untrusted as the peer id: a foreign error's `Display` plausibly carries a
1104        // remote message or status line, and an un-escaped newline in it forges a whole log line just
1105        // as effectively as a hostile peer id would (#1603). Escape + bound it here, at the one place
1106        // every holder reason funnels through.
1107        let reason = sanitize_untrusted_text(&reason.to_string(), MAX_ERROR_REASON_CHARS);
1108        tracing::debug!(%peer, %reason, "module pull: holder rejected");
1109        self.0.push(format!("{peer}: {reason}"));
1110    }
1111}
1112
1113impl std::fmt::Display for HolderReasons {
1114    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1115        if self.0.is_empty() {
1116            return f.write_str("no holder reasons recorded");
1117        }
1118        write!(f, "reasons: [{}]", self.0.join("; "))
1119    }
1120}
1121
1122/// The chunk layout derived from a [`ModuleInfo`]: per-chunk lengths + their cumulative offsets, with
1123/// the descriptor's self-consistency checked once up front.
1124#[derive(Debug)]
1125struct ChunkPlan {
1126    total_size: u64,
1127    chunk_lens: Vec<u64>,
1128    offsets: Vec<u64>,
1129}
1130
1131impl ChunkPlan {
1132    /// Validate a [`ModuleInfo`] and derive its chunk plan. The descriptor MUST carry `chunk_lens`
1133    /// (required for the byte→chunk mapping), have one length per `chunk_hashes` entry, have the
1134    /// lengths sum to `total_size` — otherwise the per-chunk fail-closed check is unimplementable —
1135    /// declare no more than `max_module_size` bytes, and declare no more than
1136    /// [`MAX_MODULE_CHUNK_COUNT`] chunks. All of its arithmetic is CHECKED — a wrapping sum must be a
1137    /// typed rejection, not a panic (see the inline note below).
1138    ///
1139    /// The size bound is checked FIRST and before any allocation: the descriptor comes from an
1140    /// untrusted holder and `total_size` bounds the bytes staged on disk before either final gate can
1141    /// reject them.
1142    fn from_info(info: &ModuleInfo, max_module_size: u64) -> Result<Self, PullFailure> {
1143        // Every rejection below is a statement about the DESCRIPTOR, so it is attributable to the holder
1144        // that supplied it. The one exception is the allocation further down, which is a local outcome —
1145        // see its comment.
1146        let false_descriptor = |reason: String| {
1147            PullFailure::BadDescriptor(DownloadError::Verify(VerifyError::Metadata(reason)))
1148        };
1149        if info.total_size > max_module_size {
1150            return Err(false_descriptor(format!(
1151                "declared module total_size {} exceeds the maximum {max_module_size}",
1152                info.total_size
1153            )));
1154        }
1155        if info.chunk_lens.is_empty() {
1156            return Err(false_descriptor(
1157                "ModuleInfo carries no chunk_lens (cannot map ranges to chunk hashes)".into(),
1158            ));
1159        }
1160        // Bound the declared COUNT before cloning it: the count sizes the plan's own vectors, so an
1161        // absurd one is the same one-message allocation attack as an absurd `total_size`.
1162        if info.chunk_lens.len() > MAX_MODULE_CHUNK_COUNT {
1163            return Err(false_descriptor(format!(
1164                "declared chunk_lens count {} exceeds the maximum {MAX_MODULE_CHUNK_COUNT}",
1165                info.chunk_lens.len()
1166            )));
1167        }
1168        if info.chunk_lens.len() != info.chunk_hashes.len() {
1169            return Err(false_descriptor(format!(
1170                "chunk_lens ({}) != chunk_hashes ({})",
1171                info.chunk_lens.len(),
1172                info.chunk_hashes.len()
1173            )));
1174        }
1175        // CHECKED arithmetic, both here and for the offsets below. Unchecked, a descriptor of
1176        // `{ total_size: 0, chunk_lens: [1, u64::MAX] }` WRAPS to a sum of 0, matches its declared
1177        // total, and passes every check above — then either aborts the node inside `sum()` (with
1178        // overflow checks on, as dig-node's release profile has them) or yields spans that index far
1179        // past the assembled blob. This validator's whole job is to be TOTAL over a hostile
1180        // descriptor, so no arithmetic in it may wrap or panic.
1181        let sum = info
1182            .chunk_lens
1183            .iter()
1184            .try_fold(0u64, |acc, &len| acc.checked_add(len))
1185            .ok_or_else(|| {
1186                false_descriptor("chunk_lens sum overflows u64 (hostile descriptor)".into())
1187            })?;
1188        if sum != info.total_size {
1189            return Err(false_descriptor(format!(
1190                "chunk_lens sum {sum} != total_size {}",
1191                info.total_size
1192            )));
1193        }
1194        let chunk_lens = info.chunk_lens.clone();
1195        let mut offsets = Vec::new();
1196        // Not descriptor evidence — the count is already bounded above, so a refusal here is this host
1197        // running out of memory and must brand nobody. But it is still UNSATISFIABLE rather than
1198        // terminal: another holder's descriptor deserves a try (blame and next-step are two
1199        // INDEPENDENT axes, chosen separately).
1200        offsets.try_reserve_exact(chunk_lens.len()).map_err(|e| {
1201            PullFailure::UnsatisfiableDescriptor(DownloadError::sink(format!(
1202                "this host cannot allocate the {}-entry chunk plan this descriptor declares: {e}",
1203                chunk_lens.len()
1204            )))
1205        })?;
1206        let mut acc = 0u64;
1207        for &len in &chunk_lens {
1208            offsets.push(acc);
1209            acc = acc.checked_add(len).ok_or_else(|| {
1210                false_descriptor("chunk offsets overflow u64 (hostile descriptor)".into())
1211            })?;
1212        }
1213        Ok(ChunkPlan {
1214            total_size: info.total_size,
1215            chunk_lens,
1216            offsets,
1217        })
1218    }
1219
1220    fn chunk_count(&self) -> usize {
1221        self.chunk_lens.len()
1222    }
1223
1224    /// The `(offset, length)` byte span of chunk `index`.
1225    fn chunk_span(&self, index: usize) -> (u64, u64) {
1226        (self.offsets[index], self.chunk_lens[index])
1227    }
1228}
1229
1230/// Append any newly-discovered holders (by `peer_id`) not already known.
1231fn merge_new_providers(
1232    known: &mut Vec<dig_dht::ProviderRecord>,
1233    fresh: Vec<dig_dht::ProviderRecord>,
1234) {
1235    for p in fresh {
1236        if !known
1237            .iter()
1238            .any(|k| k.provider_peer_id == p.provider_peer_id)
1239        {
1240            known.push(p);
1241        }
1242    }
1243}
1244
1245/// The 64-hex SHA-256 of `bytes` — the per-chunk content-id derivation.
1246fn sha256_hex(bytes: &[u8]) -> String {
1247    hex_of(Sha256::digest(bytes))
1248}
1249
1250/// Lower-hex encode bytes. The single shared hex encoder for the crate.
1251///
1252/// Split out from [`sha256_hex`] because the whole-module hash is now accumulated INCREMENTALLY
1253/// over the chunks (#1610) and so finalizes a digest that never had a contiguous `&[u8]` behind it.
1254pub(crate) fn hex_of(digest: impl AsRef<[u8]>) -> String {
1255    let mut out = String::with_capacity(64);
1256    for b in digest.as_ref() {
1257        out.push(char::from_digit((b >> 4) as u32, 16).unwrap());
1258        out.push(char::from_digit((b & 0x0f) as u32, 16).unwrap());
1259    }
1260    out
1261}
1262
1263/// The stable resume key for a module pull: `module:<store_id>:<root>`. Distinct from the resource
1264/// [`download_key`](crate::orchestrator::download_key) keyspace so a module checkpoint never collides
1265/// with a resource one.
1266pub fn module_download_key(store_id: &str, root: &str) -> String {
1267    format!("module:{store_id}:{root}")
1268}
1269
1270/// The capsule [`ContentId`] a module pull locates holders by — the `(store_id, root)` generation.
1271/// `store_id` / `root` must be 64-hex; a malformed id yields `None`.
1272pub fn module_content_id(store_id: &str, root: &str) -> Option<ContentId> {
1273    Some(ContentId::root(hex32(store_id)?, hex32(root)?))
1274}
1275
1276/// Decode a 64-hex string into a 32-byte array, or `None` if malformed.
1277fn hex32(s: &str) -> Option<[u8; 32]> {
1278    if s.len() != 64 {
1279        return None;
1280    }
1281    let mut out = [0u8; 32];
1282    for (i, byte) in out.iter_mut().enumerate() {
1283        *byte = u8::from_str_radix(&s[i * 2..i * 2 + 2], 16).ok()?;
1284    }
1285    Some(out)
1286}
1287
1288#[cfg(test)]
1289mod tests {
1290    use super::*;
1291    use crate::progress::InMemoryStateStore;
1292    use crate::sink::InMemorySink;
1293    use crate::testkit::{
1294        mock_providers, MockModuleTransport, MockProviderLocator, RejectAllModuleAnchor,
1295    };
1296    use std::collections::BTreeSet;
1297
1298    /// A 64-hex id whose every byte is `byte`.
1299    fn hex_id(byte: u8) -> String {
1300        format!("{byte:02x}").repeat(32)
1301    }
1302
1303    fn locator_with(n: u8, store_id: &str, root: &str) -> Arc<MockProviderLocator> {
1304        let content = module_content_id(store_id, root).unwrap();
1305        Arc::new(MockProviderLocator::fixed(mock_providers(n, &content)))
1306    }
1307
1308    #[tokio::test]
1309    async fn happy_path_assembles_verified_module_from_multiple_sources() {
1310        let store_id = hex_id(0x11);
1311        let root = hex_id(0x22);
1312        // 26 bytes over 8-byte chunks = 4 chunks, spread round-robin across 3 holders.
1313        let module = b"the whole .dig module blob".to_vec();
1314
1315        let transport = Arc::new(MockModuleTransport::serving(
1316            &store_id,
1317            &root,
1318            module.clone(),
1319            8,
1320        ));
1321        let downloader = ModuleDownloader::new(
1322            locator_with(3, &store_id, &root),
1323            transport.clone(),
1324            Arc::new(AcceptAnyModuleAnchor),
1325            Arc::new(InMemoryStateStore::new()),
1326            ModuleDownloadConfig::default(),
1327        );
1328        let sink = InMemorySink::new();
1329
1330        let len = downloader
1331            .download(&store_id, &root, &sink)
1332            .await
1333            .expect("pull succeeds");
1334
1335        assert_eq!(len, module.len() as u64);
1336        assert_eq!(
1337            sink.contents().await,
1338            module,
1339            "reassembled blob is byte-exact"
1340        );
1341        assert!(sink.is_finalized().await, "verified module is finalized");
1342
1343        // MULTI-SOURCE: the 4 chunks were pulled from more than one holder.
1344        let distinct: BTreeSet<String> = transport
1345            .fetches()
1346            .await
1347            .into_iter()
1348            .map(|(p, _)| p)
1349            .collect();
1350        assert!(
1351            distinct.len() > 1,
1352            "chunks came from multiple holders: {distinct:?}"
1353        );
1354    }
1355
1356    #[tokio::test]
1357    async fn resume_after_interrupt_refetches_only_missing_chunks() {
1358        let store_id = hex_id(0x33);
1359        let root = hex_id(0x44);
1360        let module = (0u8..40).collect::<Vec<u8>>(); // 40 bytes / 8 = 5 chunks
1361        let state_store = Arc::new(InMemoryStateStore::new());
1362        let sink = InMemorySink::new();
1363
1364        // First pass: only 2 fetches succeed, then the source starves → the pull fails partway.
1365        let interrupted = Arc::new(
1366            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
1367                .with_success_budget(2),
1368        );
1369        let first = ModuleDownloader::new(
1370            locator_with(1, &store_id, &root),
1371            interrupted.clone(),
1372            Arc::new(AcceptAnyModuleAnchor),
1373            state_store.clone(),
1374            ModuleDownloadConfig::default(),
1375        );
1376        let err = first
1377            .download(&store_id, &root, &sink)
1378            .await
1379            .expect_err("interrupted pull fails before finalize");
1380        assert!(
1381            matches!(err, DownloadError::NotFound { .. }),
1382            "exhaustion is terminal and names its step: {err}"
1383        );
1384        assert!(
1385            !sink.is_finalized().await,
1386            "an incomplete pull is never finalized"
1387        );
1388
1389        // Second pass: a healthy source resumes against the SAME state + sink.
1390        let healthy = Arc::new(MockModuleTransport::serving(
1391            &store_id,
1392            &root,
1393            module.clone(),
1394            8,
1395        ));
1396        let second = ModuleDownloader::new(
1397            locator_with(1, &store_id, &root),
1398            healthy.clone(),
1399            Arc::new(AcceptAnyModuleAnchor),
1400            state_store,
1401            ModuleDownloadConfig::default(),
1402        );
1403        let len = second
1404            .download(&store_id, &root, &sink)
1405            .await
1406            .expect("resumed pull succeeds");
1407        assert_eq!(len, module.len() as u64);
1408        assert_eq!(sink.contents().await, module);
1409        assert!(sink.is_finalized().await);
1410
1411        // RESUME optimization: the 2 already-verified chunks (offsets 0, 8) are NOT re-fetched.
1412        let resumed_offsets: BTreeSet<u64> = healthy
1413            .fetches()
1414            .await
1415            .into_iter()
1416            .map(|(_, o)| o)
1417            .collect();
1418        assert!(!resumed_offsets.contains(&0), "chunk 0 not re-fetched");
1419        assert!(!resumed_offsets.contains(&8), "chunk 1 not re-fetched");
1420        assert_eq!(
1421            resumed_offsets,
1422            BTreeSet::from([16, 24, 32]),
1423            "only missing chunks fetched"
1424        );
1425    }
1426
1427    #[tokio::test]
1428    async fn tampered_range_is_rejected_and_routed_around() {
1429        let store_id = hex_id(0x55);
1430        let root = hex_id(0x66);
1431        let module = b"honest bytes across several chunks here".to_vec();
1432
1433        // Holder 1 tampers every chunk; holders 2 + 3 are honest → the pull recovers.
1434        let peer1 = crate::testkit::mock_peer_hex(1);
1435        let transport = Arc::new(
1436            MockModuleTransport::serving(&store_id, &root, module.clone(), 8).tampering(&peer1),
1437        );
1438        let downloader = ModuleDownloader::new(
1439            locator_with(3, &store_id, &root),
1440            transport,
1441            Arc::new(AcceptAnyModuleAnchor),
1442            Arc::new(InMemoryStateStore::new()),
1443            ModuleDownloadConfig::default(),
1444        );
1445        let sink = InMemorySink::new();
1446
1447        let len = downloader
1448            .download(&store_id, &root, &sink)
1449            .await
1450            .expect("pull recovers around the tampering holder");
1451        assert_eq!(
1452            sink.contents().await,
1453            module,
1454            "only honest bytes were accepted"
1455        );
1456        assert_eq!(len, module.len() as u64);
1457    }
1458
1459    #[tokio::test]
1460    async fn all_sources_tampering_fails_closed_without_finalize() {
1461        let store_id = hex_id(0x77);
1462        let root = hex_id(0x88);
1463        let module = b"content nobody serves honestly".to_vec();
1464        let peer1 = crate::testkit::mock_peer_hex(1);
1465
1466        // The ONLY holder tampers → every chunk fails its hash → nowhere to get honest bytes.
1467        let transport =
1468            Arc::new(MockModuleTransport::serving(&store_id, &root, module, 8).tampering(&peer1));
1469        let downloader = ModuleDownloader::new(
1470            locator_with(1, &store_id, &root),
1471            transport,
1472            Arc::new(AcceptAnyModuleAnchor),
1473            Arc::new(InMemoryStateStore::new()),
1474            ModuleDownloadConfig::default(),
1475        );
1476        let sink = InMemorySink::new();
1477
1478        let err = downloader
1479            .download(&store_id, &root, &sink)
1480            .await
1481            .unwrap_err();
1482        assert!(
1483            matches!(err, DownloadError::NotFound { .. }),
1484            "no honest source left is terminal: {err}"
1485        );
1486        assert!(
1487            !sink.is_finalized().await,
1488            "tampered content is never written through"
1489        );
1490    }
1491
1492    #[tokio::test]
1493    async fn anchor_rejection_fails_closed_without_finalize() {
1494        let store_id = hex_id(0x99);
1495        let root = hex_id(0xAA);
1496        let module = b"assembles cleanly but is not chain-anchored".to_vec();
1497
1498        // Every per-chunk + whole-blob check passes, but the chain-anchor gate rejects the blob.
1499        let transport = Arc::new(MockModuleTransport::serving(&store_id, &root, module, 8));
1500        let downloader = ModuleDownloader::new(
1501            locator_with(2, &store_id, &root),
1502            transport,
1503            Arc::new(RejectAllModuleAnchor),
1504            Arc::new(InMemoryStateStore::new()),
1505            ModuleDownloadConfig::default(),
1506        );
1507        let sink = InMemorySink::new();
1508
1509        let err = downloader
1510            .download(&store_id, &root, &sink)
1511            .await
1512            .unwrap_err();
1513        assert!(
1514            matches!(err, DownloadError::Verify(_)),
1515            "anchor rejection is a verify failure"
1516        );
1517        assert!(
1518            !sink.is_finalized().await,
1519            "an unanchored module is never finalized"
1520        );
1521    }
1522
1523    #[tokio::test]
1524    async fn wrong_whole_module_hash_fails_closed() {
1525        let store_id = hex_id(0xBB);
1526        let root = hex_id(0xCC);
1527        let module = b"chunks are honest, module_hash lies".to_vec();
1528
1529        // Per-chunk hashes are honest (every range verifies) but the declared module_hash is wrong.
1530        let transport = Arc::new(
1531            MockModuleTransport::serving(&store_id, &root, module, 8).with_corrupt_module_hash(),
1532        );
1533        let downloader = ModuleDownloader::new(
1534            locator_with(2, &store_id, &root),
1535            transport,
1536            Arc::new(AcceptAnyModuleAnchor),
1537            Arc::new(InMemoryStateStore::new()),
1538            ModuleDownloadConfig::default(),
1539        );
1540        let sink = InMemorySink::new();
1541
1542        let err = downloader
1543            .download(&store_id, &root, &sink)
1544            .await
1545            .unwrap_err();
1546        assert!(matches!(err, DownloadError::Verify(_)));
1547        assert!(!sink.is_finalized().await);
1548    }
1549
1550    #[tokio::test]
1551    async fn no_holders_located_is_not_found() {
1552        let store_id = hex_id(0x01);
1553        let root = hex_id(0x02);
1554        let transport = Arc::new(MockModuleTransport::serving(
1555            &store_id,
1556            &root,
1557            vec![1, 2, 3],
1558            8,
1559        ));
1560        let downloader = ModuleDownloader::new(
1561            Arc::new(MockProviderLocator::fixed(vec![])),
1562            transport,
1563            Arc::new(AcceptAnyModuleAnchor),
1564            Arc::new(InMemoryStateStore::new()),
1565            ModuleDownloadConfig::default(),
1566        );
1567        let sink = InMemorySink::new();
1568        let err = downloader
1569            .download(&store_id, &root, &sink)
1570            .await
1571            .unwrap_err();
1572        assert!(matches!(err, DownloadError::NotFound { .. }));
1573    }
1574
1575    /// §2.2 CLIP CONTRACT — a holder that answers at CHUNK granularity returns MORE bytes than the
1576    /// requested window. That is legitimate (dig-download 0.7.4, #836), so the puller must clip the
1577    /// frame to the window and keep using the holder — NOT reject it as a length liar. Rejecting here
1578    /// would make every chunk-granular server unusable and starve the pull.
1579    #[tokio::test]
1580    async fn an_over_long_range_is_clipped_not_rejected() {
1581        let store_id = hex_id(0xD1);
1582        let root = hex_id(0xD2);
1583        let module = b"a chunk-granular holder overserves every window".to_vec();
1584
1585        let transport = Arc::new(
1586            MockModuleTransport::serving(&store_id, &root, module.clone(), 8).overserving(),
1587        );
1588        let downloader = ModuleDownloader::new(
1589            locator_with(1, &store_id, &root),
1590            transport,
1591            Arc::new(AcceptAnyModuleAnchor),
1592            Arc::new(InMemoryStateStore::new()),
1593            ModuleDownloadConfig::default(),
1594        );
1595        let sink = InMemorySink::new();
1596
1597        let len = downloader
1598            .download(&store_id, &root, &sink)
1599            .await
1600            .expect("an over-long frame is clipped, so the pull completes");
1601        assert_eq!(len, module.len() as u64);
1602        assert_eq!(
1603            sink.contents().await,
1604            module,
1605            "clipped to exactly the requested window — no bleed-through of the extra bytes"
1606        );
1607        assert!(sink.is_finalized().await);
1608    }
1609
1610    /// #836 — a swallowed transport reason re-surfacing as an unrelated message cost six blind
1611    /// diagnosis iterations. When the holder set is exhausted the terminal error MUST name the failing
1612    /// STEP and carry the per-holder reasons, so the log alone explains the failure.
1613    #[tokio::test]
1614    async fn exhausted_holders_name_the_failing_step_and_the_reasons() {
1615        let store_id = hex_id(0xE1);
1616        let root = hex_id(0xE2);
1617        let peer1 = crate::testkit::mock_peer_hex(1);
1618        let transport = Arc::new(
1619            MockModuleTransport::serving(
1620                &store_id,
1621                &root,
1622                b"nobody serves this honestly".to_vec(),
1623                8,
1624            )
1625            .tampering(&peer1),
1626        );
1627        let downloader = ModuleDownloader::new(
1628            locator_with(1, &store_id, &root),
1629            transport,
1630            Arc::new(AcceptAnyModuleAnchor),
1631            Arc::new(InMemoryStateStore::new()),
1632            ModuleDownloadConfig::default(),
1633        );
1634        let sink = InMemorySink::new();
1635
1636        let message = downloader
1637            .download(&store_id, &root, &sink)
1638            .await
1639            .unwrap_err()
1640            .to_string();
1641
1642        assert!(
1643            message.contains("fetchModuleRange"),
1644            "names the step that failed: {message}"
1645        );
1646        assert!(
1647            message.contains("chunk 0"),
1648            "names the chunk that could not be fetched: {message}"
1649        );
1650        assert!(
1651            message.contains("chunk hash mismatch"),
1652            "carries the per-holder reason instead of swallowing it: {message}"
1653        );
1654        assert!(
1655            message.contains(&peer1),
1656            "attributes the reason to a holder"
1657        );
1658    }
1659
1660    /// A hostile `getModuleInfo` descriptor declares a `total_size` the puller would STAGE on disk
1661    /// before either final gate could reject it. The declared size is refused against the configured
1662    /// cap before a single range is fetched.
1663    #[tokio::test]
1664    async fn an_oversized_declared_module_is_refused_before_staging() {
1665        let store_id = hex_id(0xF1);
1666        let root = hex_id(0xF2);
1667        let transport = Arc::new(
1668            MockModuleTransport::serving(&store_id, &root, b"small blob, huge lie".to_vec(), 8)
1669                .declaring_total_size(64 * 1024 * 1024 * 1024),
1670        );
1671        let downloader = ModuleDownloader::new(
1672            locator_with(1, &store_id, &root),
1673            transport,
1674            Arc::new(AcceptAnyModuleAnchor),
1675            Arc::new(InMemoryStateStore::new()),
1676            ModuleDownloadConfig {
1677                max_module_size: 1024,
1678                ..ModuleDownloadConfig::default()
1679            },
1680        );
1681        let sink = InMemorySink::new();
1682
1683        let err = downloader
1684            .download(&store_id, &root, &sink)
1685            .await
1686            .unwrap_err();
1687        assert!(
1688            matches!(err, DownloadError::Verify(_)),
1689            "an over-cap descriptor is a verify failure: {err}"
1690        );
1691        assert!(
1692            err.to_string().contains("exceeds the maximum"),
1693            "names the bound it broke: {err}"
1694        );
1695        assert!(!sink.is_finalized().await);
1696    }
1697
1698    /// #1605 — a crash-RESUMED pull must not trust its own staging bytes. A chunk read back from
1699    /// staging is re-attributed against `chunk_hashes` exactly like a freshly-fetched one, so a
1700    /// staging file corrupted between runs is RE-FETCHED and the pull still completes correctly
1701    /// (rather than assembling corruption and dying at the whole-blob gate with no way forward).
1702    #[tokio::test]
1703    async fn a_corrupted_staged_chunk_is_re_fetched_on_resume() {
1704        let store_id = hex_id(0xA1);
1705        let root = hex_id(0xA2);
1706        let module = (0u8..40).collect::<Vec<u8>>(); // 40 bytes / 8 = 5 chunks
1707        let state_store = Arc::new(InMemoryStateStore::new());
1708        let sink = InMemorySink::new();
1709
1710        // First pass: 2 chunks land, then the source starves.
1711        let interrupted = Arc::new(
1712            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
1713                .with_success_budget(2),
1714        );
1715        ModuleDownloader::new(
1716            locator_with(1, &store_id, &root),
1717            interrupted,
1718            Arc::new(AcceptAnyModuleAnchor),
1719            state_store.clone(),
1720            ModuleDownloadConfig::default(),
1721        )
1722        .download(&store_id, &root, &sink)
1723        .await
1724        .expect_err("interrupted pull fails before finalize");
1725
1726        // Corrupt the staged bytes of chunk 0 behind the puller's back (bit-rot / tampering with the
1727        // staging file between runs).
1728        sink.write_at(0, &[0xFF; 8])
1729            .await
1730            .expect("staging is writable");
1731
1732        let healthy = Arc::new(MockModuleTransport::serving(
1733            &store_id,
1734            &root,
1735            module.clone(),
1736            8,
1737        ));
1738        let len = ModuleDownloader::new(
1739            locator_with(1, &store_id, &root),
1740            healthy.clone(),
1741            Arc::new(AcceptAnyModuleAnchor),
1742            state_store,
1743            ModuleDownloadConfig::default(),
1744        )
1745        .download(&store_id, &root, &sink)
1746        .await
1747        .expect("resume detects the corrupt staged chunk and re-fetches it");
1748
1749        assert_eq!(len, module.len() as u64);
1750        assert_eq!(
1751            sink.contents().await,
1752            module,
1753            "the corrupted staged chunk was replaced with honest bytes"
1754        );
1755        let refetched: BTreeSet<u64> = healthy
1756            .fetches()
1757            .await
1758            .into_iter()
1759            .map(|(_, o)| o)
1760            .collect();
1761        assert!(
1762            refetched.contains(&0),
1763            "the corrupt chunk was re-fetched: {refetched:?}"
1764        );
1765        assert!(
1766            !refetched.contains(&8),
1767            "the still-valid staged chunk was NOT re-fetched: {refetched:?}"
1768        );
1769    }
1770
1771    /// #1603 — `ProviderRecord::provider_peer_id` is free-form text off the wire, so a hostile holder
1772    /// can publish arbitrary content there. It must never reach an error/log verbatim; a non-canonical
1773    /// id is replaced by a sentinel. A log an attacker can write is not evidence.
1774    #[tokio::test]
1775    async fn a_non_canonical_peer_id_is_sentinelled_not_echoed() {
1776        let store_id = hex_id(0xB1);
1777        let root = hex_id(0xB2);
1778        let hostile = "not-hex <script>alert(1)</script>\n[FATAL] forged log line";
1779        let content = module_content_id(&store_id, &root).unwrap();
1780        let locator = Arc::new(MockProviderLocator::fixed(vec![
1781            crate::testkit::mock_provider_with_peer_id(hostile, &content),
1782        ]));
1783
1784        // The transport rejects everything, so every failure reason mentions the holder.
1785        let transport = Arc::new(MockModuleTransport::serving(
1786            "unrelated-store",
1787            &root,
1788            vec![1, 2, 3],
1789            8,
1790        ));
1791        let downloader = ModuleDownloader::new(
1792            locator,
1793            transport,
1794            Arc::new(AcceptAnyModuleAnchor),
1795            Arc::new(InMemoryStateStore::new()),
1796            ModuleDownloadConfig::default(),
1797        );
1798        let sink = InMemorySink::new();
1799
1800        let message = downloader
1801            .download(&store_id, &root, &sink)
1802            .await
1803            .unwrap_err()
1804            .to_string();
1805        assert!(
1806            !message.contains("<script>") && !message.contains("[FATAL]"),
1807            "peer-supplied text is never echoed: {message}"
1808        );
1809        assert!(
1810            message.contains("non-canonical-peer-id"),
1811            "a sentinel stands in for it: {message}"
1812        );
1813        assert!(
1814            !message.contains('\n'),
1815            "the whole record is ONE line — a forged log line cannot ride in on the reason: {message}"
1816        );
1817    }
1818
1819    /// A peer-supplied hash from the descriptor is equally untrusted text and equally sentinelled when
1820    /// it is reported in the whole-blob mismatch message.
1821    #[test]
1822    fn untrusted_hex_is_sentinelled() {
1823        let canonical = "ab".repeat(32);
1824        assert_eq!(hex64_or_sentinel(&canonical, "peer-id"), canonical);
1825        assert_eq!(
1826            hex64_or_sentinel("AB".repeat(32).as_str(), "peer-id"),
1827            "ab".repeat(32),
1828            "canonical form is lowercase"
1829        );
1830        assert_eq!(
1831            hex64_or_sentinel("short", "peer-id"),
1832            "<non-canonical-peer-id>"
1833        );
1834        assert_eq!(
1835            hex64_or_sentinel("zz".repeat(32).as_str(), "hash"),
1836            "<non-canonical-hash>"
1837        );
1838    }
1839
1840    #[test]
1841    fn malformed_ids_are_not_downloadable() {
1842        assert!(module_content_id("too-short", &hex_id(1)).is_none());
1843        assert!(module_content_id(&hex_id(1), "zz").is_none());
1844        assert!(module_content_id(&hex_id(1), &hex_id(2)).is_some());
1845    }
1846
1847    #[test]
1848    fn download_key_is_module_scoped() {
1849        let k = module_download_key(&hex_id(1), &hex_id(2));
1850        assert!(k.starts_with("module:"));
1851    }
1852
1853    /// RESHARE-DENIAL — the descriptor defines the WHOLE plan, and holder order is deterministic, so
1854    /// one holder that wins the `getModuleInfo` race with a well-formed-but-WRONG descriptor would
1855    /// otherwise deny a capsule's reshare forever: the pull assembles honest bytes, dies at the
1856    /// whole-blob gate, and every retry re-asks the same liar. The lying SOURCE must be demoted and
1857    /// the next holder's descriptor tried — an honest holder in the set means the pull SUCCEEDS.
1858    #[tokio::test]
1859    async fn a_lying_descriptor_source_is_demoted_and_an_honest_holder_completes_the_pull() {
1860        let store_id = hex_id(0xC1);
1861        let root = hex_id(0xC2);
1862        let module = b"honest bytes, one lying descriptor source".to_vec();
1863        let liar = crate::testkit::mock_peer_hex(1);
1864
1865        let transport = Arc::new(
1866            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
1867                .lying_descriptor_from(&liar),
1868        );
1869        let downloader = ModuleDownloader::new(
1870            locator_with(2, &store_id, &root),
1871            transport.clone(),
1872            Arc::new(AcceptAnyModuleAnchor),
1873            Arc::new(InMemoryStateStore::new()),
1874            ModuleDownloadConfig::default(),
1875        );
1876        let sink = InMemorySink::new();
1877
1878        let len = downloader
1879            .download(&store_id, &root, &sink)
1880            .await
1881            .expect("the honest holder's descriptor completes the pull");
1882        assert_eq!(len, module.len() as u64);
1883        assert_eq!(sink.contents().await, module);
1884        assert!(sink.is_finalized().await);
1885
1886        let handshakes = transport.module_info_calls().await;
1887        assert_eq!(
1888            handshakes.len(),
1889            2,
1890            "the liar's descriptor was demoted and another holder re-handshaked: {handshakes:?}"
1891        );
1892        assert_eq!(handshakes[0], liar, "the liar answered first");
1893        assert_ne!(
1894            handshakes[1], liar,
1895            "the demoted source is never re-asked: {handshakes:?}"
1896        );
1897    }
1898
1899    /// A pull that resumes over an earlier run's staging + checkpoint must not re-adopt a descriptor
1900    /// THIS call already proved wrong: the checkpoint the liar's plan produced is dropped on demotion
1901    /// and the pull re-handshakes with another holder.
1902    ///
1903    /// The demotion set is per-CALL (a local `Vec`), so this covers within-call resume only — a fresh
1904    /// process re-asks the same liar first and demotes it again. Holder reputation that outlives a call
1905    /// would have to live in the [`StateStore`]; it is deliberately not claimed here.
1906    #[tokio::test]
1907    async fn a_pull_does_not_re_adopt_a_demoted_descriptor_within_the_same_call() {
1908        let store_id = hex_id(0xC5);
1909        let root = hex_id(0xC6);
1910        let module = (0u8..40).collect::<Vec<u8>>(); // 40 bytes / 8 = 5 chunks
1911        let liar = crate::testkit::mock_peer_hex(1);
1912        let state_store = Arc::new(InMemoryStateStore::new());
1913        let sink = InMemorySink::new();
1914
1915        // First run: 2 chunks land, then the source starves — the pull ends without finalize.
1916        let interrupted = Arc::new(
1917            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
1918                .lying_descriptor_from(&liar)
1919                .with_success_budget(2),
1920        );
1921        ModuleDownloader::new(
1922            locator_with(2, &store_id, &root),
1923            interrupted,
1924            Arc::new(AcceptAnyModuleAnchor),
1925            state_store.clone(),
1926            ModuleDownloadConfig::default(),
1927        )
1928        .download(&store_id, &root, &sink)
1929        .await
1930        .expect_err("the interrupted pull fails before finalize");
1931
1932        // Resumed run against the SAME staging + checkpoint, with the liar still answering first.
1933        let healthy = Arc::new(
1934            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
1935                .lying_descriptor_from(&liar),
1936        );
1937        let len = ModuleDownloader::new(
1938            locator_with(2, &store_id, &root),
1939            healthy,
1940            Arc::new(AcceptAnyModuleAnchor),
1941            state_store,
1942            ModuleDownloadConfig::default(),
1943        )
1944        .download(&store_id, &root, &sink)
1945        .await
1946        .expect("the resumed pull completes via the honest holder");
1947        assert_eq!(len, module.len() as u64);
1948        assert_eq!(sink.contents().await, module);
1949        assert!(sink.is_finalized().await);
1950    }
1951
1952    /// Demotion is BOUNDED: when every holder lies about the descriptor the pull still ends — with a
1953    /// fail-closed verify error and no finalize — rather than looping over holders forever.
1954    #[tokio::test]
1955    async fn every_descriptor_source_lying_is_terminal_and_never_finalizes() {
1956        let store_id = hex_id(0xC3);
1957        let root = hex_id(0xC4);
1958        let liar = crate::testkit::mock_peer_hex(1);
1959        let transport = Arc::new(
1960            MockModuleTransport::serving(&store_id, &root, b"only a liar holds this".to_vec(), 8)
1961                .lying_descriptor_from(&liar),
1962        );
1963        let downloader = ModuleDownloader::new(
1964            locator_with(1, &store_id, &root),
1965            transport.clone(),
1966            Arc::new(AcceptAnyModuleAnchor),
1967            Arc::new(InMemoryStateStore::new()),
1968            ModuleDownloadConfig::default(),
1969        );
1970        let sink = InMemorySink::new();
1971
1972        let err = downloader
1973            .download(&store_id, &root, &sink)
1974            .await
1975            .unwrap_err();
1976        assert!(
1977            matches!(err, DownloadError::Verify(_)),
1978            "fail-closed: {err}"
1979        );
1980        assert!(!sink.is_finalized().await);
1981        assert!(
1982            transport.module_info_calls().await.len() <= MAX_DESCRIPTOR_ATTEMPTS,
1983            "descriptor attempts are bounded"
1984        );
1985    }
1986
1987    /// A hostile descriptor whose `chunk_lens` SUM WRAPS: `1 + u64::MAX == 0`, which equals the
1988    /// declared `total_size`, so every self-consistency check passes on unchecked arithmetic. The
1989    /// descriptor validator's whole job is to be TOTAL over a hostile descriptor before any
1990    /// allocation, so the wrap must be a typed error — never a panic (with overflow checks on, an
1991    /// unchecked `sum()` aborts the node from ONE `getModuleInfo` response, zero bytes fetched) and
1992    /// never an accepted plan (with them off, the derived spans index past a zero-length blob).
1993    #[test]
1994    fn a_wrapping_chunk_len_sum_is_rejected_not_panicked() {
1995        let hostile = ModuleInfo {
1996            total_size: 0,
1997            module_hash: "ab".repeat(32),
1998            chunk_hashes: vec!["cd".repeat(32), "ef".repeat(32)],
1999            chunk_lens: vec![1, u64::MAX],
2000        };
2001        let err = ChunkPlan::from_info(&hostile, DEFAULT_MAX_MODULE_SIZE)
2002            .expect_err("a wrapping descriptor is refused");
2003        assert!(
2004            err.is_proven_false(),
2005            "a hostile descriptor is attributable to the holder that supplied it"
2006        );
2007        let err = err.error();
2008        assert!(
2009            matches!(err, DownloadError::Verify(_)),
2010            "a hostile descriptor is a verify failure: {err}"
2011        );
2012        assert!(
2013            err.to_string().contains("overflow"),
2014            "names the arithmetic it broke: {err}"
2015        );
2016    }
2017
2018    /// The declared chunk COUNT is as unbounded as the declared size: a descriptor claiming billions
2019    /// of zero-length chunks costs the puller two big allocations (`offsets` + the `done` bitmap)
2020    /// before a byte is fetched. It is refused against a fixed cap.
2021    #[test]
2022    fn an_absurd_chunk_count_is_refused() {
2023        let hostile = ModuleInfo {
2024            total_size: 0,
2025            module_hash: "ab".repeat(32),
2026            chunk_hashes: Vec::new(),
2027            chunk_lens: vec![0; MAX_MODULE_CHUNK_COUNT + 1],
2028        };
2029        let err = ChunkPlan::from_info(&hostile, DEFAULT_MAX_MODULE_SIZE)
2030            .expect_err("an over-count descriptor is refused");
2031        assert!(err.is_proven_false(), "attributable to its source");
2032        let err = err.error();
2033        assert!(
2034            err.to_string().contains("chunk_lens"),
2035            "names the bound it broke: {err}"
2036        );
2037    }
2038
2039    #[test]
2040    fn chunk_plan_rejects_inconsistent_descriptor() {
2041        // chunk_lens sum (5) != total_size (99)
2042        let bad = ModuleInfo {
2043            total_size: 99,
2044            module_hash: "ab".repeat(32),
2045            chunk_hashes: vec!["cd".repeat(32)],
2046            chunk_lens: vec![5],
2047        };
2048        assert!(ChunkPlan::from_info(&bad, DEFAULT_MAX_MODULE_SIZE).is_err());
2049
2050        // missing chunk_lens
2051        let no_lens = ModuleInfo {
2052            total_size: 5,
2053            module_hash: "ab".repeat(32),
2054            chunk_hashes: vec!["cd".repeat(32)],
2055            chunk_lens: vec![],
2056        };
2057        assert!(ChunkPlan::from_info(&no_lens, DEFAULT_MAX_MODULE_SIZE).is_err());
2058
2059        // chunk_hashes / chunk_lens length disagree
2060        let mismatched = ModuleInfo {
2061            total_size: 5,
2062            module_hash: "ab".repeat(32),
2063            chunk_hashes: vec!["cd".repeat(32), "ef".repeat(32)],
2064            chunk_lens: vec![5],
2065        };
2066        assert!(ChunkPlan::from_info(&mismatched, DEFAULT_MAX_MODULE_SIZE).is_err());
2067    }
2068
2069    /// A throwaway directory for the file-backed promotion tests.
2070    fn temp_dir(tag: &str) -> std::path::PathBuf {
2071        let d = std::env::temp_dir().join(format!(
2072            "dig-download-module-{tag}-{}-{}",
2073            std::process::id(),
2074            std::time::SystemTime::now()
2075                .duration_since(std::time::UNIX_EPOCH)
2076                .unwrap()
2077                .as_nanos()
2078        ));
2079        std::fs::create_dir_all(&d).unwrap();
2080        d
2081    }
2082
2083    /// CACHE POISONING — the artifact VERIFIED must be the artifact PROMOTED.
2084    ///
2085    /// A holder that wins the `getModuleInfo` race with a self-consistent fabrication LARGER than the
2086    /// real module gets its long bytes staged (they pass every per-chunk check and the whole-blob hash
2087    /// gate) and only dies at the chain-anchor gate. The pull then re-handshakes and completes against
2088    /// the honest, SHORTER module — so unless the staging area is provably reduced to the verified
2089    /// length, the promoted `.dig` is honest bytes followed by the attacker's tail: a file whose
2090    /// SHA-256 is not `module_hash`, cached and re-announced as a holder by the reshare leg.
2091    #[tokio::test]
2092    async fn the_promoted_artifact_is_byte_equal_to_the_verified_one_after_a_shorter_retry() {
2093        let dir = temp_dir("shrinking-lie");
2094        let final_path = dir.join("module.dig");
2095        let store_id = hex_id(0xE1);
2096        let root = hex_id(0xE2);
2097        let honest = b"honest!!".to_vec(); // 8 bytes, one 8-byte chunk
2098        let fabricated = vec![0xAA; 32]; // 32 bytes, four self-consistent chunks
2099        let liar = crate::testkit::mock_peer_hex(1); // providers[0] — wins the handshake race
2100
2101        let transport = Arc::new(
2102            MockModuleTransport::serving(&store_id, &root, honest.clone(), 8)
2103                .serving_alternate_module_from(&liar, fabricated.clone()),
2104        );
2105        let downloader = ModuleDownloader::new(
2106            locator_with(3, &store_id, &root),
2107            transport,
2108            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(honest.clone())),
2109            Arc::new(InMemoryStateStore::new()),
2110            ModuleDownloadConfig::default(),
2111        );
2112        let sink = crate::sink::FileSink::new(&final_path);
2113
2114        let len = downloader
2115            .download(&store_id, &root, &sink)
2116            .await
2117            .expect("the honest holder's descriptor completes the pull");
2118        assert_eq!(
2119            len,
2120            honest.len() as u64,
2121            "the VERIFIED length is the honest one"
2122        );
2123
2124        let promoted = std::fs::read(&final_path).expect("the module was promoted");
2125        assert_eq!(
2126            promoted,
2127            honest,
2128            "the promoted artifact carries the attacker's tail: {} promoted bytes vs {} verified",
2129            promoted.len(),
2130            honest.len()
2131        );
2132
2133        let _ = std::fs::remove_dir_all(&dir);
2134    }
2135
2136    /// The SAME divergence with NO attacker: a staging file left by an earlier attempt at a DIFFERENT
2137    /// module shape (its checkpoint is discarded as mismatched, but the FILE is not) must not survive
2138    /// into the promotion of a shorter, freshly-verified module.
2139    #[tokio::test]
2140    async fn leftover_staging_of_another_shape_never_survives_into_the_promotion() {
2141        let dir = temp_dir("stale-staging");
2142        let final_path = dir.join("module.dig");
2143        let store_id = hex_id(0xE3);
2144        let root = hex_id(0xE4);
2145        let honest = b"honest!!".to_vec();
2146
2147        // An earlier, differently-shaped attempt left a LONGER staging file plus its checkpoint.
2148        let staging = crate::sink::staging_path_for(&final_path);
2149        std::fs::write(&staging, vec![0xAA; 32]).unwrap();
2150        let state_store = Arc::new(InMemoryStateStore::new());
2151        let key = module_download_key(&store_id, &root);
2152        let mut stale = DownloadState::new(&key);
2153        stale.total_length = 32;
2154        stale.chunk_lens = vec![8, 8, 8, 8];
2155        stale.mark_done(0);
2156        state_store.save(&stale).await.unwrap();
2157
2158        let downloader = ModuleDownloader::new(
2159            locator_with(2, &store_id, &root),
2160            Arc::new(MockModuleTransport::serving(
2161                &store_id,
2162                &root,
2163                honest.clone(),
2164                8,
2165            )),
2166            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(honest.clone())),
2167            state_store,
2168            ModuleDownloadConfig::default(),
2169        );
2170        let sink = crate::sink::FileSink::new(&final_path);
2171
2172        let len = downloader.download(&store_id, &root, &sink).await.unwrap();
2173        assert_eq!(len, honest.len() as u64);
2174        assert_eq!(
2175            std::fs::read(&final_path).unwrap(),
2176            honest,
2177            "the stale longer staging tail was promoted with the verified bytes"
2178        );
2179
2180        let _ = std::fs::remove_dir_all(&dir);
2181    }
2182
2183    /// A sink that implements `read_at` (delegating to an inner [`InMemorySink`]) but IGNORES
2184    /// `truncate` — the ONE-DEFAULT case: a staging area that simply cannot shrink, paired with a
2185    /// working read-back. `truncate`'s own default is now fail-closed (it must be OPTED IN to model
2186    /// "ignores truncate" rather than "unsupported"), so this claims success without ever shortening
2187    /// the inner buffer — the exact contract the trait doc's opt-in example describes. Everything else
2188    /// delegates to the inner sink.
2189    struct UnshrinkableSink(InMemorySink);
2190
2191    #[async_trait]
2192    impl Sink for UnshrinkableSink {
2193        async fn write_at(&self, offset: u64, bytes: &[u8]) -> Result<(), DownloadError> {
2194            self.0.write_at(offset, bytes).await
2195        }
2196        async fn truncate(&self, _len: u64) -> Result<(), DownloadError> {
2197            Ok(()) // models a staging area that cannot shrink: claims success but never truncates
2198        }
2199        // Read-back WORKS here, so the refusal below can only come from the length proof itself — not
2200        // from the "cannot observe my staging area" refusal that guards an unproven sink.
2201        fn supports_read_back(&self) -> bool {
2202            true
2203        }
2204        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, DownloadError> {
2205            self.0.read_at(offset, len).await
2206        }
2207        async fn finalize(&self) -> Result<(), DownloadError> {
2208            self.0.finalize().await
2209        }
2210    }
2211
2212    /// The promotion check is a CONFIRMATION, not just a shortening: a sink that cannot shrink must
2213    /// FAIL CLOSED rather than promote an artifact longer than the verified one.
2214    #[tokio::test]
2215    async fn a_staging_area_that_cannot_shrink_is_never_promoted() {
2216        let store_id = hex_id(0xE7);
2217        let root = hex_id(0xE8);
2218        let honest = b"honest!!".to_vec();
2219        let fabricated = vec![0xAA; 32];
2220        let liar = crate::testkit::mock_peer_hex(1);
2221
2222        let downloader = ModuleDownloader::new(
2223            locator_with(3, &store_id, &root),
2224            Arc::new(
2225                MockModuleTransport::serving(&store_id, &root, honest.clone(), 8)
2226                    .serving_alternate_module_from(&liar, fabricated),
2227            ),
2228            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(honest.clone())),
2229            Arc::new(InMemoryStateStore::new()),
2230            ModuleDownloadConfig::default(),
2231        );
2232        let sink = UnshrinkableSink(InMemorySink::new());
2233
2234        let err = downloader
2235            .download(&store_id, &root, &sink)
2236            .await
2237            .expect_err("a staging area that still holds the demoted tail is not promoted");
2238        assert!(
2239            err.to_string().contains("past the verified length"),
2240            "names the promotion invariant it refused: {err}"
2241        );
2242        assert!(!sink.0.is_finalized().await, "and it never finalized");
2243    }
2244
2245    /// A sink implementing ONLY `write_at` + `finalize` — BOTH `truncate` and `read_at` left on the
2246    /// trait's defaults. This is the TWO-DEFAULT combination that used to fail OPEN: the old
2247    /// `truncate` default silently no-op'd (nothing ever shortened) while `read_at`'s default already
2248    /// failed closed, so [`promote_verified`]'s "bytes past the verified end" probe read that
2249    /// failure as "nothing past the end" and promoted a longer, un-truncated, poisoned artifact.
2250    /// Delegates storage + finalized-tracking to an inner [`InMemorySink`]; everything else is the
2251    /// plain trait default — this is exactly the shape of a pre-existing external `Sink` that never
2252    /// anticipated a promotable staging area.
2253    struct TwoDefaultSink(InMemorySink);
2254
2255    #[async_trait]
2256    impl Sink for TwoDefaultSink {
2257        async fn write_at(&self, offset: u64, bytes: &[u8]) -> Result<(), DownloadError> {
2258            self.0.write_at(offset, bytes).await
2259        }
2260        async fn finalize(&self) -> Result<(), DownloadError> {
2261            self.0.finalize().await
2262        }
2263    }
2264
2265    /// A sink on BOTH the `truncate` and `read_at` defaults fails closed — at the FIRST place it is
2266    /// asked to shorten anything, which is the abandoned-plan reset, long before promotion.
2267    ///
2268    /// Named for what it actually proves. It was previously named for the promotion tail and asserted
2269    /// only `DownloadError::Sink(_)`, which this pull produces without reaching `promote_verified` at
2270    /// all: `pull_with_descriptor` resets a non-resuming staging area first, the fail-closed `truncate`
2271    /// default rejects that, and the pull dies before any liar tail is staged. It therefore passed with
2272    /// the entire promotion proof deleted. The promotion path for a defaulted `read_at` is covered by
2273    /// `the_documented_whole_commit_sink_recipe_cannot_promote_unproven_bytes`; this cell covers the
2274    /// reset, and asserts the MESSAGE so it cannot silently start proving something else.
2275    #[tokio::test]
2276    async fn a_sink_on_both_truncate_and_read_at_defaults_fails_closed_at_the_first_reset() {
2277        let store_id = hex_id(0xE9);
2278        let root = hex_id(0xEA);
2279        let honest = b"honest!!".to_vec();
2280        let fabricated = vec![0xAA; 32];
2281        let liar = crate::testkit::mock_peer_hex(1);
2282
2283        let downloader = ModuleDownloader::new(
2284            locator_with(3, &store_id, &root),
2285            Arc::new(
2286                MockModuleTransport::serving(&store_id, &root, honest.clone(), 8)
2287                    .serving_alternate_module_from(&liar, fabricated),
2288            ),
2289            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(honest.clone())),
2290            Arc::new(InMemoryStateStore::new()),
2291            ModuleDownloadConfig::default(),
2292        );
2293        let sink = TwoDefaultSink(InMemorySink::new());
2294
2295        let err = downloader
2296            .download(&store_id, &root, &sink)
2297            .await
2298            .expect_err("a sink on both defaults must fail closed, never promote blind");
2299        // The MESSAGE, not just the variant: this pull dies at the abandoned-plan reset, and asserting
2300        // only `Sink(_)` passed even with the whole promotion proof deleted.
2301        assert!(
2302            err.to_string().contains("truncation unsupported"),
2303            "it is the fail-closed truncate DEFAULT that refuses, at the plan reset: {err}"
2304        );
2305        assert!(!sink.0.is_finalized().await, "and it never finalized");
2306    }
2307
2308    /// RESHARE-DENIAL, the CHEAPEST variant — a descriptor whose per-chunk hashes are FABRICATED.
2309    ///
2310    /// Nobody can satisfy those hashes, so the pull exhausts every holder on chunk 0 and never reaches
2311    /// a final gate. Treating that exhaustion as terminal lets a holder that serves ZERO bytes deny a
2312    /// capsule's reshare forever: while no chunk has verified, the descriptor itself is the suspect, so
2313    /// its source is demoted and an honest holder's descriptor completes the pull.
2314    #[tokio::test]
2315    async fn a_fabricated_chunk_hash_descriptor_source_is_demoted_and_the_pull_completes() {
2316        let store_id = hex_id(0xF1);
2317        let root = hex_id(0xF2);
2318        let module = b"honest bytes behind a zero-byte liar".to_vec();
2319        let liar = crate::testkit::mock_peer_hex(1);
2320
2321        let transport = Arc::new(
2322            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
2323                .fabricating_chunk_hashes_from(&liar),
2324        );
2325        let downloader = ModuleDownloader::new(
2326            locator_with(3, &store_id, &root),
2327            transport.clone(),
2328            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(module.clone())),
2329            Arc::new(InMemoryStateStore::new()),
2330            ModuleDownloadConfig::default(),
2331        );
2332        let sink = InMemorySink::new();
2333
2334        let len = downloader
2335            .download(&store_id, &root, &sink)
2336            .await
2337            .expect("an honest holder's descriptor completes the pull");
2338        assert_eq!(len, module.len() as u64);
2339        assert_eq!(sink.contents().await, module);
2340
2341        let handshakes = transport.module_info_calls().await;
2342        assert_eq!(handshakes[0], liar, "the liar answered first");
2343        assert!(
2344            handshakes.len() >= 2 && handshakes[1] != liar,
2345            "the fabricating source was demoted and another holder re-handshaked: {handshakes:?}"
2346        );
2347    }
2348
2349    /// #1613 — a descriptor retry gated on "did ANY chunk verify?" is bypassable for ONE BYTE.
2350    ///
2351    /// The liar declares `chunk_lens = [1, rest]` with an honest hash for the first byte and a
2352    /// fabricated one for the rest, serves that single byte, then refuses everything. Under the old
2353    /// bound the first verified chunk flipped the pull into "the descriptor is credible", so the
2354    /// inevitable exhaustion on chunk 1 was Terminal — no demotion, no re-handshake — and the pull died
2355    /// with two honest holders standing right there. Cost to the attacker: one byte.
2356    ///
2357    /// The attempt budget, not the flag, is what bounds termination: exhaustion always demotes the
2358    /// descriptor source and tries the next holder's descriptor while the budget allows.
2359    #[tokio::test]
2360    async fn a_liar_that_serves_one_byte_then_refuses_is_still_demoted() {
2361        let store_id = hex_id(0x5A);
2362        let root = hex_id(0x5B);
2363        let module = b"an honest module blob of some length".to_vec();
2364        let liar = crate::testkit::mock_peer_hex(1);
2365
2366        let transport = Arc::new(
2367            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
2368                .serving_one_byte_then_refusing_from(&liar),
2369        );
2370        let downloader = ModuleDownloader::new(
2371            locator_with(3, &store_id, &root),
2372            transport.clone(),
2373            Arc::new(AcceptAnyModuleAnchor),
2374            Arc::new(InMemoryStateStore::new()),
2375            ModuleDownloadConfig::default(),
2376        );
2377        let sink = InMemorySink::new();
2378
2379        let len = downloader
2380            .download(&store_id, &root, &sink)
2381            .await
2382            .expect("an honest holder's descriptor completes the pull");
2383
2384        assert_eq!(len, module.len() as u64);
2385        assert_eq!(sink.contents().await, module, "only honest bytes promoted");
2386        let handshakes = transport.module_info_calls().await;
2387        assert!(
2388            handshakes.len() > 1,
2389            "the one-byte liar was demoted and another holder's descriptor tried: {handshakes:?}"
2390        );
2391        assert!(
2392            handshakes.len() <= MAX_DESCRIPTOR_ATTEMPTS + 1,
2393            "descriptor retries stay bounded by the attempt budget: {handshakes:?}"
2394        );
2395    }
2396
2397    /// The exhaustion DIAGNOSIS still distinguishes the two cases even though the control flow no
2398    /// longer does: exhaustion after some chunk verified is more likely genuine unavailability, and
2399    /// exhaustion with none verified more likely a fabricated descriptor. That is a message, not a gate.
2400    #[test]
2401    fn exhaustion_diagnosis_names_whether_any_chunk_had_verified() {
2402        let base = DownloadError::NotFound {
2403            content: "fetchModuleRange failed for chunk 3".into(),
2404        };
2405        let with_progress = describe_chunk_exhaustion(base, true).to_string();
2406        let without_progress = describe_chunk_exhaustion(
2407            DownloadError::NotFound {
2408                content: "fetchModuleRange failed for chunk 3".into(),
2409            },
2410            false,
2411        )
2412        .to_string();
2413        assert!(
2414            with_progress.contains("chunk 3"),
2415            "keeps the original reason: {with_progress}"
2416        );
2417        assert_ne!(
2418            with_progress, without_progress,
2419            "the two exhaustion cases read differently"
2420        );
2421    }
2422
2423    /// #1611 — a liar caught in ONE call must not be re-asked as a descriptor source in the NEXT.
2424    ///
2425    /// Demotion used to live in a local `Vec` inside `download()`, so a fresh call re-asked the same
2426    /// liars from scratch and paid up to `MAX_DESCRIPTOR_ATTEMPTS` full pull attempts again. The verdict
2427    /// is now persisted in the `StateStore`, so the second call skips the known liar's handshake
2428    /// entirely — while still fetching CHUNKS from it (chunk bytes are hash-attributed, so excluding it
2429    /// there would cost availability for no integrity gain).
2430    #[tokio::test]
2431    async fn a_liar_demoted_in_one_call_is_not_re_asked_in_the_next() {
2432        let store_id = hex_id(0x6A);
2433        let root = hex_id(0x6B);
2434        let module = b"an honest module across chunks".to_vec();
2435        let liar = crate::testkit::mock_peer_hex(1);
2436        let state_store = Arc::new(InMemoryStateStore::new());
2437
2438        // Call 1: the liar wins the handshake race, fails the whole-blob gate, and is demoted.
2439        let first_transport = Arc::new(
2440            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
2441                .lying_descriptor_from(&liar),
2442        );
2443        let first = ModuleDownloader::new(
2444            locator_with(3, &store_id, &root),
2445            first_transport.clone(),
2446            Arc::new(AcceptAnyModuleAnchor),
2447            state_store.clone(),
2448            ModuleDownloadConfig::default(),
2449        );
2450        first
2451            .download(&store_id, &root, &InMemorySink::new())
2452            .await
2453            .expect("an honest holder completes call 1");
2454        assert!(
2455            first_transport.module_info_calls().await.contains(&liar),
2456            "call 1 did ask the liar (that is how it learned)"
2457        );
2458
2459        // Call 2: the SAME state store, so the verdict is remembered.
2460        let second_transport = Arc::new(
2461            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
2462                .lying_descriptor_from(&liar),
2463        );
2464        let second = ModuleDownloader::new(
2465            locator_with(3, &store_id, &root),
2466            second_transport.clone(),
2467            Arc::new(AcceptAnyModuleAnchor),
2468            state_store.clone(),
2469            ModuleDownloadConfig::default(),
2470        );
2471        let len = second
2472            .download(&store_id, &root, &InMemorySink::new())
2473            .await
2474            .expect("call 2 completes");
2475
2476        assert_eq!(len, module.len() as u64);
2477        assert!(
2478            !second_transport.module_info_calls().await.contains(&liar),
2479            "the remembered liar is never asked for a descriptor again: {:?}",
2480            second_transport.module_info_calls().await
2481        );
2482        assert_eq!(
2483            second_transport.module_info_calls().await.len(),
2484            1,
2485            "and exactly one honest handshake was needed"
2486        );
2487        assert!(
2488            second_transport
2489                .fetches()
2490                .await
2491                .iter()
2492                .any(|(peer, _)| peer == &liar),
2493            "a demoted descriptor source is still used for CHUNK fetches"
2494        );
2495        assert_eq!(
2496            state_store
2497                .bad_descriptor_peers(&module_download_key(&store_id, &root))
2498                .await
2499                .unwrap(),
2500            vec![liar],
2501            "the verdict is what the store persisted"
2502        );
2503    }
2504
2505    /// Reputation must not be able to deny a pull: when EVERY located holder carries a past verdict the
2506    /// memory is ignored for that attempt (a verdict is evidence about a moment, and holders get fixed).
2507    #[tokio::test]
2508    async fn reputation_never_denies_a_pull_when_every_holder_is_remembered() {
2509        let store_id = hex_id(0x6C);
2510        let root = hex_id(0x6D);
2511        let module = b"honest bytes from a once-bad holder".to_vec();
2512        let key = module_download_key(&store_id, &root);
2513        let state_store = Arc::new(InMemoryStateStore::new());
2514        // The only holder is remembered as a past liar — but it serves honestly now.
2515        state_store
2516            .record_bad_descriptor(&key, &crate::testkit::mock_peer_hex(1))
2517            .await
2518            .unwrap();
2519
2520        let downloader = ModuleDownloader::new(
2521            locator_with(1, &store_id, &root),
2522            Arc::new(MockModuleTransport::serving(
2523                &store_id,
2524                &root,
2525                module.clone(),
2526                8,
2527            )),
2528            Arc::new(AcceptAnyModuleAnchor),
2529            state_store,
2530            ModuleDownloadConfig::default(),
2531        );
2532        let sink = InMemorySink::new();
2533        let len = downloader
2534            .download(&store_id, &root, &sink)
2535            .await
2536            .expect("a remembered holder is still asked when it is the only one");
2537        assert_eq!(len, module.len() as u64);
2538        assert_eq!(sink.contents().await, module);
2539    }
2540
2541    /// GATE #1, the PARTIAL case — reputation must not deny a pull the network can serve.
2542    ///
2543    /// Verdicts on the two HONEST holders and none on the liar: honouring the memory excludes the honest
2544    /// holders from the descriptor role, the liar wins the handshake, fails a final gate, is demoted —
2545    /// and `usable == 0`, so the pull returned the descriptor error with honest holders sitting right
2546    /// there. `demoted` started empty before #1611, so that was a regression, and the total-case test
2547    /// (`reputation_never_denies_a_pull_when_every_holder_is_remembered`) never reached it. The escape
2548    /// now triggers on "no usable holder remains", not on "all holders remembered".
2549    #[tokio::test]
2550    async fn reputation_never_denies_a_pull_when_only_the_honest_holders_are_remembered() {
2551        let store_id = hex_id(0x7A);
2552        let root = hex_id(0x7B);
2553        let module = b"honest bytes the network can still serve".to_vec();
2554        let key = module_download_key(&store_id, &root);
2555        let liar = crate::testkit::mock_peer_hex(1);
2556        let state_store = Arc::new(InMemoryStateStore::new());
2557        // The HONEST holders carry the verdicts; the liar does not.
2558        for honest in [
2559            crate::testkit::mock_peer_hex(2),
2560            crate::testkit::mock_peer_hex(3),
2561        ] {
2562            state_store
2563                .record_bad_descriptor(&key, &honest)
2564                .await
2565                .unwrap();
2566        }
2567
2568        let downloader = ModuleDownloader::new(
2569            locator_with(3, &store_id, &root),
2570            Arc::new(
2571                MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
2572                    .lying_descriptor_from(&liar),
2573            ),
2574            Arc::new(AcceptAnyModuleAnchor),
2575            state_store,
2576            ModuleDownloadConfig::default(),
2577        );
2578        let sink = InMemorySink::new();
2579
2580        let len = downloader
2581            .download(&store_id, &root, &sink)
2582            .await
2583            .expect("remembered HONEST holders are re-asked rather than denying the pull");
2584        assert_eq!(len, module.len() as u64);
2585        assert_eq!(sink.contents().await, module);
2586    }
2587
2588    /// GATE #1(b) — chunk exhaustion must NOT leave a durable verdict against the descriptor source.
2589    ///
2590    /// DHT provider announcement is unauthenticated, so if unsatisfied chunks were durable evidence,
2591    /// sybil holders could refuse their assigned chunks and get an HONEST descriptor source blacklisted
2592    /// on the victim for 24 h — per capsule, repeatably — until only attacker-supplied descriptors were
2593    /// ever asked for. Exhaustion still demotes for THIS call (#1613); it just earns no memory.
2594    #[tokio::test]
2595    async fn chunk_exhaustion_demotes_for_this_call_but_records_no_durable_verdict() {
2596        let store_id = hex_id(0x7C);
2597        let root = hex_id(0x7D);
2598        let key = module_download_key(&store_id, &root);
2599        let state_store = Arc::new(InMemoryStateStore::new());
2600
2601        // The only holder answers `getModuleInfo` honestly and then serves nothing: the chunks cannot be
2602        // fetched, so the pull exhausts and fails — with no evidence its descriptor was false.
2603        let downloader = ModuleDownloader::new(
2604            locator_with(1, &store_id, &root),
2605            Arc::new(
2606                MockModuleTransport::serving(&store_id, &root, b"unavailable bytes".to_vec(), 8)
2607                    .with_success_budget(0),
2608            ),
2609            Arc::new(AcceptAnyModuleAnchor),
2610            state_store.clone(),
2611            ModuleDownloadConfig::default(),
2612        );
2613        let sink = InMemorySink::new();
2614
2615        downloader
2616            .download(&store_id, &root, &sink)
2617            .await
2618            .expect_err("unavailable chunks fail the pull");
2619        assert!(
2620            state_store
2621                .bad_descriptor_peers(&key)
2622                .await
2623                .unwrap()
2624                .is_empty(),
2625            "no durable verdict from mere unavailability — else sybils can brand an honest holder"
2626        );
2627        assert!(!sink.is_finalized().await);
2628    }
2629
2630    /// GATE #4 — the sink recipe this crate's own docs give MUST NOT fail open.
2631    ///
2632    /// `truncate` overridden to `Ok(())` (the blessed "I commit whole" opt-in) with `read_at` left on its
2633    /// default was the one untested combination: nothing is shortened, the past-the-end probe's
2634    /// "read-back unsupported" reads as "nothing there", and an unproven artifact promotes. Promotion now
2635    /// requires `supports_read_back`, so a sink that cannot show its staged bytes is refused.
2636    #[tokio::test]
2637    async fn the_documented_whole_commit_sink_recipe_cannot_promote_unproven_bytes() {
2638        /// `truncate` → `Ok(())` exactly as the [`Sink::truncate`] doc's opt-in shows, `read_at` left on
2639        /// the trait default. The shape a real store-write sink following that recipe would have.
2640        struct RecipeSink(InMemorySink);
2641
2642        #[async_trait]
2643        impl Sink for RecipeSink {
2644            async fn write_at(&self, offset: u64, bytes: &[u8]) -> Result<(), DownloadError> {
2645                self.0.write_at(offset, bytes).await
2646            }
2647            async fn truncate(&self, _len: u64) -> Result<(), DownloadError> {
2648                Ok(()) // "this sink commits whole, so there is never a tail to shrink"
2649            }
2650            async fn finalize(&self) -> Result<(), DownloadError> {
2651                self.0.finalize().await
2652            }
2653        }
2654
2655        let store_id = hex_id(0x7E);
2656        let root = hex_id(0x7F);
2657        let honest = b"honest!!".to_vec();
2658        let liar = crate::testkit::mock_peer_hex(1);
2659
2660        // The liar stages a LONGER fabrication first, then the honest, shorter module is pulled — so the
2661        // staging area holds a tail the verified bytes do not contain.
2662        let downloader = ModuleDownloader::new(
2663            locator_with(3, &store_id, &root),
2664            Arc::new(
2665                MockModuleTransport::serving(&store_id, &root, honest.clone(), 8)
2666                    .serving_alternate_module_from(&liar, vec![0xAA; 32]),
2667            ),
2668            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(honest)),
2669            Arc::new(InMemoryStateStore::new()),
2670            ModuleDownloadConfig::default(),
2671        );
2672        let sink = RecipeSink(InMemorySink::new());
2673
2674        let err = downloader
2675            .download(&store_id, &root, &sink)
2676            .await
2677            .expect_err("a sink that cannot prove its staged length is never promoted");
2678        assert!(
2679            err.to_string().contains("cannot read back"),
2680            "names WHY it refused — an unprovable promotion, not a length verdict: {err}"
2681        );
2682        assert!(!sink.0.is_finalized().await, "and it never finalized");
2683    }
2684
2685    /// A declared size far past any real module costs no allocation at all (#1610).
2686    ///
2687    /// Before #1610 this descriptor made the puller try to reserve ~18 EiB — the failure the deleted
2688    /// `try_zeroed_blob` had to classify. The plan now derives from the descriptor without allocating
2689    /// anything proportional to `total_size`, so the same hostile claim is simply *planned* and then
2690    /// dies as unfetchable chunks (see the two end-to-end tests below). Pinning it here keeps the
2691    /// no-allocation property from silently regressing back into a reservation.
2692    #[test]
2693    fn an_18_exbibyte_declaration_costs_no_allocation() {
2694        let hostile = ModuleInfo {
2695            total_size: u64::MAX,
2696            module_hash: hex_id(0x01),
2697            chunk_hashes: vec![hex_id(0x02)],
2698            chunk_lens: vec![u64::MAX],
2699        };
2700        let Ok(plan) = ChunkPlan::from_info(&hostile, u64::MAX) else {
2701            panic!("the plan is derived, not allocated — an 18 EiB claim is now cheap to hold")
2702        };
2703        assert_eq!(plan.total_size, u64::MAX);
2704        assert_eq!(plan.chunk_count(), 1);
2705    }
2706
2707    /// GATE, end to end — a ~100-byte inflated descriptor must not deny the capsule.
2708    ///
2709    /// The attacker announces as a provider, wins the `getModuleInfo` race, and answers a SELF-CONSISTENT
2710    /// descriptor whose `total_size` (and matching final `chunk_len`) this host cannot allocate. It must
2711    /// cost the attacker the descriptor role and nothing else: no durable verdict against anyone, and the
2712    /// honest holder standing right there completes the pull.
2713    ///
2714    /// Only the LIAR inflates. With every holder inflating — as this test first did — a puller that dies
2715    /// instead of retrying is indistinguishable from one that recovers, so the regression was invisible.
2716    #[tokio::test]
2717    async fn an_honest_holder_completes_the_pull_after_an_impossible_descriptor() {
2718        let store_id = hex_id(0x8A);
2719        let root = hex_id(0x8B);
2720        let key = module_download_key(&store_id, &root);
2721        let module = b"a real module served honestly".to_vec();
2722        let liar = crate::testkit::mock_peer_hex(1);
2723        let state_store = Arc::new(InMemoryStateStore::new());
2724
2725        let transport = Arc::new(
2726            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
2727                .inflating_total_size_from(&liar, u64::MAX),
2728        );
2729        let downloader = ModuleDownloader::new(
2730            locator_with(3, &store_id, &root),
2731            transport.clone(),
2732            Arc::new(AcceptAnyModuleAnchor),
2733            state_store.clone(),
2734            // A ceiling that admits the declared size, so the SIZE guard is not what rejects it and the
2735            // allocation is genuinely what fails.
2736            ModuleDownloadConfig {
2737                max_module_size: u64::MAX,
2738                ..ModuleDownloadConfig::default()
2739            },
2740        );
2741        let sink = InMemorySink::new();
2742
2743        let len = downloader
2744            .download(&store_id, &root, &sink)
2745            .await
2746            .expect("an honest holder's descriptor completes the pull");
2747        assert_eq!(len, module.len() as u64);
2748        assert_eq!(sink.contents().await, module);
2749        assert!(
2750            transport.module_info_calls().await.len() > 1,
2751            "the unallocatable descriptor's source was demoted and another holder asked: {:?}",
2752            transport.module_info_calls().await
2753        );
2754        assert!(
2755            state_store
2756                .bad_descriptor_peers(&key)
2757                .await
2758                .unwrap()
2759                .is_empty(),
2760            "and NOBODY is branded: an unsatisfiable descriptor is not PROOF the holder lied — the \
2761             bytes it declares may simply be unavailable"
2762        );
2763        let _ = &liar;
2764    }
2765
2766    /// The same failure when EVERY holder declares an impossible module: the pull fails closed
2767    /// (bounded by the attempt budget) and promotes nothing.
2768    #[tokio::test]
2769    async fn every_holder_declaring_an_impossible_module_fails_closed() {
2770        let store_id = hex_id(0x8E);
2771        let root = hex_id(0x8F);
2772        let key = module_download_key(&store_id, &root);
2773        let state_store = Arc::new(InMemoryStateStore::new());
2774
2775        let downloader = ModuleDownloader::new(
2776            locator_with(2, &store_id, &root),
2777            Arc::new(
2778                MockModuleTransport::serving(&store_id, &root, b"a real module".to_vec(), 8)
2779                    .declaring_total_size(u64::MAX),
2780            ),
2781            Arc::new(AcceptAnyModuleAnchor),
2782            state_store.clone(),
2783            ModuleDownloadConfig {
2784                max_module_size: u64::MAX,
2785                ..ModuleDownloadConfig::default()
2786            },
2787        );
2788        let sink = InMemorySink::new();
2789
2790        downloader
2791            .download(&store_id, &root, &sink)
2792            .await
2793            .expect_err("no holder offers a module that could exist");
2794        assert!(
2795            state_store
2796                .bad_descriptor_peers(&key)
2797                .await
2798                .unwrap()
2799                .is_empty(),
2800            "no holder is branded for a descriptor merely unsatisfiable"
2801        );
2802        assert!(!sink.is_finalized().await, "and nothing is promoted");
2803    }
2804
2805    /// GATE — an anchor gate that cannot REACH an answer must not brand the holder.
2806    ///
2807    /// With a `bool` return an implementation that consults the chain had to answer `false` during an
2808    /// outage, which was read as "proven not anchored" and persisted. An honest holder, a correct blob
2809    /// and a chain-source blip then branded every holder tried — and for the whole TTL the node's
2810    /// descriptor preference INVERTED, skipping remembered honest holders and asking unremembered
2811    /// (i.e. sybil) peers first. `ModuleAnchor::Unavailable` is the third answer that fixes it.
2812    #[tokio::test]
2813    async fn an_unreachable_chain_anchor_is_terminal_and_brands_nobody() {
2814        let store_id = hex_id(0x8C);
2815        let root = hex_id(0x8D);
2816        let key = module_download_key(&store_id, &root);
2817        let state_store = Arc::new(InMemoryStateStore::new());
2818
2819        let downloader = ModuleDownloader::new(
2820            locator_with(3, &store_id, &root),
2821            Arc::new(MockModuleTransport::serving(
2822                &store_id,
2823                &root,
2824                b"a correct, genuinely anchored module".to_vec(),
2825                8,
2826            )),
2827            Arc::new(crate::testkit::UnreachableChainAnchor),
2828            state_store.clone(),
2829            ModuleDownloadConfig::default(),
2830        );
2831        let sink = InMemorySink::new();
2832
2833        let err = downloader
2834            .download(&store_id, &root, &sink)
2835            .await
2836            .expect_err("an unverifiable anchor is fail-closed");
2837        assert!(
2838            err.to_string().contains("cannot verify the chain anchor"),
2839            "it reports an unfinished CHECK, not a verdict on the module: {err}"
2840        );
2841        assert!(
2842            state_store
2843                .bad_descriptor_peers(&key)
2844                .await
2845                .unwrap()
2846                .is_empty(),
2847            "and no honest holder is branded by this node's own outage"
2848        );
2849        assert!(
2850            !sink.is_finalized().await,
2851            "fail-closed: nothing is promoted while the anchor is unproven"
2852        );
2853    }
2854
2855    // ---------------------------------------------------------------------------------------------
2856    // #1610 — the streaming whole-module hash + the reader-based anchor gate.
2857    // ---------------------------------------------------------------------------------------------
2858
2859    /// The whole-module hash is taken in CHUNK order, not in the order chunks became available.
2860    ///
2861    /// The nearest wrong implementation absorbs each chunk as it lands — which is what the previous
2862    /// structure did in effect, rehydrating the checkpointed chunks first and only then fetching the
2863    /// rest. Every existing resume test survives that bug, because their checkpoints hold a PREFIX
2864    /// (chunks 0,1): prefix-first arrival order and ascending chunk order are the same sequence, so
2865    /// the fixture cannot tell them apart.
2866    ///
2867    /// The distinguishing fixture is a checkpoint holding exactly the MIDDLE chunk of five, over
2868    /// content where every chunk differs. Arrival-order hashing then computes `c2‖c0‖c1‖c3‖c4`, which
2869    /// fails the whole-module gate; chunk-order hashing completes the pull.
2870    #[tokio::test]
2871    async fn the_whole_module_hash_is_taken_in_chunk_order_not_arrival_order() {
2872        let store_id = hex_id(0x90);
2873        let root = hex_id(0x91);
2874        let key = module_download_key(&store_id, &root);
2875        // 40 bytes / 8 = 5 chunks, each with distinct content.
2876        let module = (0u8..40).collect::<Vec<u8>>();
2877
2878        // Stage ONLY the middle chunk (index 2, bytes [16, 24)) and checkpoint exactly it.
2879        let sink = InMemorySink::new();
2880        sink.write_at(16, &module[16..24]).await.unwrap();
2881        let state_store = Arc::new(InMemoryStateStore::new());
2882        let mut state = DownloadState::new(&key);
2883        state.total_length = module.len() as u64;
2884        state.chunk_lens = vec![8; 5];
2885        state.mark_done(2);
2886        state_store.save(&state).await.unwrap();
2887
2888        let downloader = ModuleDownloader::new(
2889            locator_with(1, &store_id, &root),
2890            Arc::new(MockModuleTransport::serving(
2891                &store_id,
2892                &root,
2893                module.clone(),
2894                8,
2895            )),
2896            // Anchored on the exact bytes — so this test also proves the READER reassembles the module
2897            // the gate sees in chunk order, not merely that the hash does.
2898            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(module.clone())),
2899            state_store,
2900            ModuleDownloadConfig::default(),
2901        );
2902
2903        let len = downloader
2904            .download(&store_id, &root, &sink)
2905            .await
2906            .expect("a mid-module checkpoint resumes and still passes both gates");
2907        assert_eq!(len, module.len() as u64);
2908        assert_eq!(sink.contents().await, module);
2909        assert!(sink.is_finalized().await);
2910    }
2911
2912    /// The anchor gate cannot read outside the module the whole-module hash gate accepted.
2913    ///
2914    /// A staging area is never SHORTENED by writing, so bytes past `total_size` can genuinely be
2915    /// there — a longer earlier attempt's tail. An unbounded reader would hand them to the gate as if
2916    /// they were part of the verified artifact.
2917    #[tokio::test]
2918    async fn the_anchor_gate_cannot_read_past_the_verified_module() {
2919        /// An anchor gate that tries to read one byte beyond the module's end.
2920        struct ReadsPastTheEnd;
2921
2922        #[async_trait]
2923        impl ModuleAnchorVerifier for ReadsPastTheEnd {
2924            async fn verify_module_anchor(
2925                &self,
2926                module: &dyn ModuleReader,
2927                _store_id: &str,
2928                _root: &str,
2929            ) -> ModuleAnchor {
2930                match module.read_at(module.len().saturating_sub(1), 2).await {
2931                    Ok(_) => ModuleAnchor::Anchored,
2932                    Err(e) => ModuleAnchor::Unavailable(e.to_string()),
2933                }
2934            }
2935        }
2936
2937        let store_id = hex_id(0x92);
2938        let root = hex_id(0x93);
2939        let module = (0u8..40).collect::<Vec<u8>>();
2940
2941        let downloader = ModuleDownloader::new(
2942            locator_with(1, &store_id, &root),
2943            Arc::new(MockModuleTransport::serving(
2944                &store_id,
2945                &root,
2946                module.clone(),
2947                8,
2948            )),
2949            Arc::new(ReadsPastTheEnd),
2950            Arc::new(InMemoryStateStore::new()),
2951            ModuleDownloadConfig::default(),
2952        );
2953        let sink = InMemorySink::new();
2954
2955        let err = downloader
2956            .download(&store_id, &root, &sink)
2957            .await
2958            .expect_err("a read past the verified end is refused, so the gate reaches no answer");
2959        assert!(
2960            err.to_string().contains("falls outside"),
2961            "the refusal names the out-of-range window: {err}"
2962        );
2963        assert!(!sink.is_finalized().await, "and nothing is promoted");
2964    }
2965
2966    /// Staging bytes that change between the hash gate and the anchor read fail CLOSED, and brand
2967    /// nobody.
2968    ///
2969    /// The two gates now read the staging area at two different moments (they used to share one RAM
2970    /// copy), so the window between them has to be closed by re-attributing every chunk the reader
2971    /// serves. Asserting only "the pull fails" would not distinguish a reader that re-verifies from
2972    /// one that does not: without the check the anchor gate simply sees the corrupted bytes and
2973    /// answers `NotAnchored`, which also fails the pull — but as a durable verdict against a holder
2974    /// that did nothing wrong. So the load-bearing assertion is WHO gets blamed.
2975    #[tokio::test]
2976    async fn staging_corrupted_between_the_two_gates_fails_closed_and_brands_nobody() {
2977        /// A sink that stages honestly but returns a flipped byte on read-back — a staging area
2978        /// mutated (by bit-rot, another process, or a local attacker) after the hash gate passed.
2979        struct CorruptingReadBack(InMemorySink);
2980
2981        #[async_trait]
2982        impl Sink for CorruptingReadBack {
2983            async fn write_at(&self, offset: u64, bytes: &[u8]) -> Result<(), DownloadError> {
2984                self.0.write_at(offset, bytes).await
2985            }
2986            async fn truncate(&self, len: u64) -> Result<(), DownloadError> {
2987                self.0.truncate(len).await
2988            }
2989            fn supports_read_back(&self) -> bool {
2990                true
2991            }
2992            async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, DownloadError> {
2993                let mut bytes = self.0.read_at(offset, len).await?;
2994                if let Some(first) = bytes.first_mut() {
2995                    *first ^= 0xFF;
2996                }
2997                Ok(bytes)
2998            }
2999            async fn finalize(&self) -> Result<(), DownloadError> {
3000                self.0.finalize().await
3001            }
3002        }
3003
3004        let store_id = hex_id(0x94);
3005        let root = hex_id(0x95);
3006        let key = module_download_key(&store_id, &root);
3007        let module = (0u8..40).collect::<Vec<u8>>();
3008        let state_store = Arc::new(InMemoryStateStore::new());
3009
3010        let downloader = ModuleDownloader::new(
3011            locator_with(2, &store_id, &root),
3012            Arc::new(MockModuleTransport::serving(
3013                &store_id,
3014                &root,
3015                module.clone(),
3016                8,
3017            )),
3018            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(module.clone())),
3019            state_store.clone(),
3020            ModuleDownloadConfig::default(),
3021        );
3022        let sink = CorruptingReadBack(InMemorySink::new());
3023
3024        let err = downloader
3025            .download(&store_id, &root, &sink)
3026            .await
3027            .expect_err("the gate must not run on bytes nothing has attributed");
3028        assert!(
3029            err.to_string()
3030                .contains("no longer matches its verified hash"),
3031            "the failure names the staging area, not the chain or a holder: {err}"
3032        );
3033        assert!(
3034            state_store
3035                .bad_descriptor_peers(&key)
3036                .await
3037                .unwrap()
3038                .is_empty(),
3039            "local corruption is never evidence against a holder that served correct bytes"
3040        );
3041        assert!(!sink.0.is_finalized().await, "and nothing is promoted");
3042    }
3043
3044    /// A source demoted by a TRANSPORT failure keeps the bytes it already staged: the next holder's
3045    /// identical descriptor resumes from the checkpoint and fetches only the chunks that are missing.
3046    ///
3047    /// The fixture is built so that the ONLY way to pass is to preserve staging. Every holder serves
3048    /// the same honest descriptor and the same honest bytes; the transport is merely SEVERED beyond
3049    /// chunk 5 for the duration of the first descriptor attempt, so nothing here is a lie and no gate
3050    /// can be reached by a fail-closed path. The assertion counts SERVED RANGES rather than a final
3051    /// length: a re-download produces the identical artifact, so only the ranges can tell the two
3052    /// apart. With the wipe restored, offsets 0..40 are served twice and the duplicate assertion
3053    /// fires (dig-node#328).
3054    #[tokio::test]
3055    async fn a_transport_demotion_resumes_from_its_partial_instead_of_refetching_the_module() {
3056        let store_id = hex_id(0x5A);
3057        let root = hex_id(0x5B);
3058        // 80 bytes over 8-byte chunks = 10 chunks; the sever bites at chunk 5.
3059        let module: Vec<u8> = (0..80u8).collect();
3060        let severed_at = 40u64;
3061
3062        let transport = Arc::new(
3063            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
3064                .severing_beyond(severed_at),
3065        );
3066        let downloader = ModuleDownloader::new(
3067            locator_with(2, &store_id, &root),
3068            transport.clone(),
3069            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(module.clone())),
3070            Arc::new(InMemoryStateStore::new()),
3071            ModuleDownloadConfig::default(),
3072        );
3073        let sink = InMemorySink::new();
3074
3075        let len = downloader.download(&store_id, &root, &sink).await.unwrap();
3076        assert_eq!(len, module.len() as u64);
3077        assert_eq!(
3078            sink.contents().await,
3079            module,
3080            "and the artifact is the module"
3081        );
3082
3083        let mut served: Vec<u64> = transport
3084            .fetches()
3085            .await
3086            .into_iter()
3087            .map(|(_, o)| o)
3088            .collect();
3089        served.sort_unstable();
3090        let deduped = {
3091            let mut d = served.clone();
3092            d.dedup();
3093            d
3094        };
3095        assert_eq!(
3096            served, deduped,
3097            "a chunk verified before the demotion was re-fetched after it: served offsets {served:?}"
3098        );
3099        assert_eq!(
3100            served,
3101            (0..10).map(|i| i * 8).collect::<Vec<u64>>(),
3102            "every chunk is served exactly once across both descriptor attempts"
3103        );
3104        assert!(
3105            transport.module_info_calls().await.len() >= 2,
3106            "the fixture must actually drive the descriptor-demotion loop"
3107        );
3108    }
3109
3110    /// A descriptor that never ARRIVES spends an attempt and the holder set is re-asked (#37).
3111    ///
3112    /// Both holders time out on the first round of `getModuleInfo`, then answer. Before the fix the
3113    /// `?` on `fetch_module_info` returned before `attempts += 1`, so the pull ended on the first
3114    /// round with holders that would have answered a second ask standing right there.
3115    #[tokio::test]
3116    async fn a_holder_set_that_times_out_on_the_first_descriptor_ask_is_asked_again() {
3117        let store_id = hex_id(0x5C);
3118        let root = hex_id(0x5D);
3119        let module = b"a whole module".to_vec();
3120
3121        let transport = Arc::new(
3122            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
3123                .failing_the_first_info_asks(2), // exactly one full round of asks, both holders
3124        );
3125        let downloader = ModuleDownloader::new(
3126            locator_with(2, &store_id, &root),
3127            transport.clone(),
3128            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(module.clone())),
3129            Arc::new(InMemoryStateStore::new()),
3130            ModuleDownloadConfig::default(),
3131        );
3132        let sink = InMemorySink::new();
3133
3134        let len = downloader
3135            .download(&store_id, &root, &sink)
3136            .await
3137            .expect("a transiently unreachable holder set is re-asked, not surrendered to");
3138        assert_eq!(len, module.len() as u64);
3139        assert_eq!(
3140            transport.module_info_calls().await.len(),
3141            3,
3142            "one failed round over both holders, then one answered ask"
3143        );
3144    }
3145
3146    /// The re-ask is BOUNDED: a holder set that never answers cannot hold the pull open. The budget
3147    /// is the same `MAX_DESCRIPTOR_ATTEMPTS` a failed pull spends, and the error names the descriptor
3148    /// step rather than blaming discovery.
3149    #[tokio::test]
3150    async fn a_holder_set_that_never_answers_gives_up_within_the_attempt_budget() {
3151        let store_id = hex_id(0x5E);
3152        let root = hex_id(0x5F);
3153        let module = b"a whole module".to_vec();
3154
3155        let transport = Arc::new(
3156            MockModuleTransport::serving(&store_id, &root, module.clone(), 8)
3157                .failing_the_first_info_asks(usize::MAX),
3158        );
3159        let downloader = ModuleDownloader::new(
3160            locator_with(2, &store_id, &root),
3161            transport.clone(),
3162            Arc::new(crate::testkit::OnlyThisModuleAnchor::new(module.clone())),
3163            Arc::new(InMemoryStateStore::new()),
3164            ModuleDownloadConfig::default(),
3165        );
3166
3167        let err = downloader
3168            .download(&store_id, &root, &InMemorySink::new())
3169            .await
3170            .expect_err("an unanswerable holder set must end the pull, not retry forever");
3171        assert!(
3172            err.to_string().contains("getModuleInfo"),
3173            "the error names the step that failed: {err}"
3174        );
3175        // Nobody answered, so nothing was proven false and no source is attributable: the honest
3176        // report is a `NotFound` carrying each holder's own reason, NOT a gate `Verify` that would
3177        // manufacture blame against holders that merely did not answer (SPEC 17.5a).
3178        assert!(
3179            matches!(err, DownloadError::NotFound { .. }),
3180            "an unanswered descriptor ask attributes nothing to any holder: {err:?}"
3181        );
3182        assert_eq!(
3183            transport.module_info_calls().await.len(),
3184            2 * MAX_DESCRIPTOR_ATTEMPTS,
3185            "at most MAX_DESCRIPTOR_ATTEMPTS rounds, each asking each holder once"
3186        );
3187    }
3188
3189    /// **Guard (2) of the three named at the demotion site, made falsifiable.**
3190    ///
3191    /// `load_or_fresh_state` refuses to resume a checkpoint whose `chunk_lens` do not match the
3192    /// CURRENT descriptor's, and `pull_with_descriptor` then wipes the staging area. The comment at
3193    /// the demotion site and `SPEC.md` both cite that refusal as one of the three reasons staged bytes
3194    /// may safely survive a descriptor demotion — but deleting the `prev.chunk_lens ==
3195    /// layout.chunk_lens` condition left the entire suite green, including the two tests named for it.
3196    /// A load-bearing guarantee that no test can falsify is how the next refactor deletes it.
3197    ///
3198    /// **What it does NOT do, which is why the earlier tests were blind and why a first attempt at
3199    /// this one was blind too:** it does not keep a longer stale tail out of the artifact, because
3200    /// `promote_verified` truncates the staging area to the verified length before it probes. And it
3201    /// does not keep stale BYTES out, because guard (1) re-hashes every resumed chunk against the
3202    /// current descriptor. Any test asserting the artifact's contents therefore passes with the guard
3203    /// deleted — the artifact is correct either way.
3204    ///
3205    /// What the guard uniquely delivers is the claim `load_or_fresh_state`'s own doc makes: **a stale
3206    /// checkpoint is never PARTIALLY reused.** Without it, the abandoned plan's `done_ranges` are
3207    /// carried into a differently-shaped plan and the puller reads back chunks it never staged, on a
3208    /// staging area belonging to a plan it has abandoned — safe only because a lower layer catches it.
3209    /// That is observable directly, so this test observes it directly.
3210    ///
3211    /// The fixture varies exactly one thing and keeps a truthful control: the SAME staged bytes and
3212    /// the SAME checkpointed indices are replayed against a matching descriptor, where resuming is
3213    /// correct and expected. Without the control, an implementation that simply never resumed anything
3214    /// would satisfy the mismatch half and this test would be pinning a coincidence.
3215    #[tokio::test]
3216    async fn a_stale_shaped_checkpoint_is_discarded_whole_not_partially_reused() {
3217        /// Records every staging-area operation the puller performs, so "did it try to resume?" is a
3218        /// direct observation rather than an inference from the artifact.
3219        #[derive(Default)]
3220        struct SpyingSink {
3221            inner: InMemorySink,
3222            truncates: tokio::sync::Mutex<Vec<u64>>,
3223            reads: tokio::sync::Mutex<Vec<(u64, u64)>>,
3224        }
3225
3226        #[async_trait]
3227        impl Sink for SpyingSink {
3228            async fn write_at(&self, offset: u64, bytes: &[u8]) -> Result<(), DownloadError> {
3229                self.inner.write_at(offset, bytes).await
3230            }
3231            async fn truncate(&self, len: u64) -> Result<(), DownloadError> {
3232                self.truncates.lock().await.push(len);
3233                self.inner.truncate(len).await
3234            }
3235            fn supports_read_back(&self) -> bool {
3236                true
3237            }
3238            async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, DownloadError> {
3239                self.reads.lock().await.push((offset, len));
3240                self.inner.read_at(offset, len).await
3241            }
3242            async fn finalize(&self) -> Result<(), DownloadError> {
3243                self.inner.finalize().await
3244            }
3245        }
3246
3247        /// Runs one pull against a staging area holding 100 bytes of an abandoned two-chunk plan and a
3248        /// checkpoint declaring `checkpoint_lens` with both of its chunks done. Returns the
3249        /// staging-area operations the puller performed.
3250        async fn pull_over_a_checkpoint_of(
3251            tag: u8,
3252            checkpoint_lens: Vec<u64>,
3253            module: Vec<u8>,
3254            chunk_size: usize,
3255        ) -> (Vec<u64>, Vec<(u64, u64)>, Vec<u8>) {
3256            let store_id = hex_id(tag);
3257            let root = hex_id(tag ^ 0xFF);
3258            let transport = Arc::new(MockModuleTransport::serving(
3259                &store_id,
3260                &root,
3261                module.clone(),
3262                chunk_size,
3263            ));
3264
3265            let sink = Arc::new(SpyingSink::default());
3266            sink.write_at(0, &vec![0xAA; module.len()])
3267                .await
3268                .expect("seed the staging area of the abandoned attempt");
3269
3270            let state_store = Arc::new(InMemoryStateStore::new());
3271            let mut stale = DownloadState::new(module_download_key(&store_id, &root));
3272            stale.total_length = checkpoint_lens.iter().sum();
3273            stale.chunk_lens = checkpoint_lens;
3274            stale.done_ranges = BTreeSet::from([0usize, 1usize]);
3275            state_store.save(&stale).await.expect("seed the checkpoint");
3276
3277            ModuleDownloader::new(
3278                locator_with(1, &store_id, &root),
3279                transport,
3280                Arc::new(AcceptAnyModuleAnchor),
3281                state_store,
3282                ModuleDownloadConfig::default(),
3283            )
3284            .download(&store_id, &root, sink.as_ref())
3285            .await
3286            .expect("the pull completes either way — the artifact is not what distinguishes them");
3287
3288            let truncates = sink.truncates.lock().await.clone();
3289            let reads = sink.reads.lock().await.clone();
3290            (truncates, reads, sink.inner.contents().await)
3291        }
3292
3293        // The plan actually being pulled: two 50-byte chunks.
3294        let module: Vec<u8> = (0..100u8).collect();
3295
3296        // MISMATCHED — the checkpoint describes four 25-byte chunks, a shape this descriptor does not
3297        // have. `[25; 4]` sums to the same 100 bytes as the real plan on purpose: the guard keys on
3298        // the SHAPE, so an equal total is the case a length-only check would wave through.
3299        let (truncates, reads, contents) =
3300            pull_over_a_checkpoint_of(0xD1, vec![25; 4], module.clone(), 50).await;
3301
3302        assert_eq!(
3303            contents, module,
3304            "the artifact is correct (it is either way)"
3305        );
3306        assert!(
3307            truncates.contains(&0),
3308            "the abandoned plan's staging area is discarded WHOLE before the new plan stages a byte; \
3309             truncates were {truncates:?}",
3310        );
3311        assert!(
3312            !reads.iter().any(|&(_, len)| len == 50),
3313            "no chunk of the NEW plan is read back, because the stale checkpoint granted no licence \
3314             to resume one; reads were {reads:?}",
3315        );
3316
3317        // CONTROL — the same staged bytes and the same checkpointed indices, against a descriptor of
3318        // the MATCHING shape. Here resuming is correct, and it must actually happen: an
3319        // implementation that had simply stopped resuming would satisfy the assertions above while
3320        // breaking the feature they are meant to bound.
3321        let (truncates, reads, contents) =
3322            pull_over_a_checkpoint_of(0xD2, vec![50, 50], module.clone(), 50).await;
3323
3324        assert_eq!(contents, module, "the artifact is correct here too");
3325        assert!(
3326            !truncates.contains(&0),
3327            "a MATCHING checkpoint keeps its staging area — it is not discarded; truncates were \
3328             {truncates:?}",
3329        );
3330        assert!(
3331            reads.iter().any(|&(_, len)| len == 50),
3332            "and its checkpointed chunks ARE read back for resume; reads were {reads:?}",
3333        );
3334    }
3335
3336    /// **F2 — `SPEC.md` promised a descriptor-phase bound that this crate did not enforce.**
3337    ///
3338    /// The clause reads: the worst-case wait is `MAX_DESCRIPTOR_ATTEMPTS × holders × the transport's
3339    /// per-ask timeout`, and *"an unanswerable holder set cannot hold a pull open indefinitely"*.
3340    /// `tokio::time::timeout` appeared exactly once in this module — around `fetch_module_range` — so
3341    /// the descriptor half of that bound rested entirely on an **injected** transport choosing to have
3342    /// a timeout. A `ModuleTransport` whose `get_module_info` simply never resolves satisfied the
3343    /// trait and hung the pull forever, and #37's across-round re-ask tripled the number of asks that
3344    /// exposure applies to.
3345    ///
3346    /// The fixture is a holder whose `getModuleInfo` never returns — the nearest thing to a real
3347    /// half-open connection, and the one case a per-ask timeout exists for. Time is virtual
3348    /// (`start_paused`), so the assertion is about the BOUND existing, not about wall-clock duration.
3349    ///
3350    /// The outer `timeout` is what makes this test fail rather than HANG when the bound is removed:
3351    /// under paused time an unbounded inner await leaves the runtime idle, the outer timer
3352    /// auto-advances, and the `expect` below fires with a message naming the defect. A test whose
3353    /// revert-proof is "the suite stops responding" is not a usable revert-proof.
3354    #[tokio::test(start_paused = true)]
3355    async fn a_holder_that_never_answers_getmoduleinfo_cannot_hold_the_pull_open() {
3356        /// Answers `getModuleInfo` with a future that never resolves; ranges are never reached.
3357        struct NeverAnswersInfo;
3358
3359        #[async_trait]
3360        impl ModuleTransport for NeverAnswersInfo {
3361            async fn get_module_info(
3362                &self,
3363                _peer: &str,
3364                _store_id: &str,
3365                _root: &str,
3366            ) -> Result<ModuleInfo, DownloadError> {
3367                std::future::pending().await
3368            }
3369            async fn fetch_module_range(
3370                &self,
3371                _peer: &str,
3372                _store_id: &str,
3373                _root: &str,
3374                _offset: u64,
3375                _len: u64,
3376            ) -> Result<Vec<u8>, DownloadError> {
3377                unreachable!("the pull never gets a descriptor, so no range is ever asked for")
3378            }
3379        }
3380
3381        let store_id = hex_id(0xE1);
3382        let root = hex_id(0xE2);
3383        let config = ModuleDownloadConfig::default();
3384
3385        // Generous next to the 30s per-ask default and the 3-attempt budget, so this can only elapse
3386        // if NO per-ask bound exists at all — it cannot mask a bound that is merely slow.
3387        let outer = config.range_timeout * (MAX_DESCRIPTOR_ATTEMPTS as u32) * 100;
3388
3389        let error = tokio::time::timeout(
3390            outer,
3391            ModuleDownloader::new(
3392                locator_with(1, &store_id, &root),
3393                Arc::new(NeverAnswersInfo),
3394                Arc::new(AcceptAnyModuleAnchor),
3395                Arc::new(InMemoryStateStore::new()),
3396                config,
3397            )
3398            .download(&store_id, &root, &InMemorySink::new()),
3399        )
3400        .await
3401        .expect("the descriptor ask is bounded by the crate, not by the transport's good manners")
3402        .expect_err("a holder that never answers cannot produce a module");
3403
3404        let message = error.to_string();
3405        assert!(
3406            message.contains("getModuleInfo"),
3407            "the failure names the step that timed out: {message}",
3408        );
3409        assert!(
3410            message.contains("timed out") || message.contains("Timeout"),
3411            "and reports it as a timeout, not as a fabricated not-found: {message}",
3412        );
3413    }
3414}