dig_download/

progress.rs

//! Progress reporting + resume state.
//!
//! Two things a caller / a crash-restart needs to see:
//!
//! - **Progress** — a live [`DownloadEvent`] stream (bytes done, per-range completions, source
//!   health, pause/resume, terminal outcome) plus a coalesced [`DownloadProgress`] snapshot, so a UI
//!   or an agent can watch a download without polling.
//! - **Resume state** — a durable [`DownloadState`] (which ranges are complete + verified, and the
//!   resource commitment) written to a [`StateStore`] as the download makes progress, so
//!   [`resume`](crate::DownloadHandle) — after a pause OR a crash — re-fetches only the still-missing
//!   ranges and NEVER a completed+verified one.

use std::collections::BTreeSet;
use std::time::Duration;

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::error::DownloadError;

/// A coalesced snapshot of a download's progress — the "how far along" view.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct DownloadProgress {
    /// Verified bytes written to the sink so far.
    pub bytes_done: u64,
    /// Total resource ciphertext length (0 until the commitment is established).
    pub total_length: u64,
    /// Ranges completed + verified.
    pub ranges_done: usize,
    /// Total ranges in the plan (0 until planned).
    pub ranges_total: usize,
    /// Distinct providers with a range currently in flight.
    pub active_sources: usize,
}

impl DownloadProgress {
    /// Fraction complete in `[0.0, 1.0]` by bytes (0 until the total length is known).
    pub fn fraction(&self) -> f64 {
        if self.total_length == 0 {
            0.0
        } else {
            self.bytes_done as f64 / self.total_length as f64
        }
    }

    /// Whether every planned range is done (and the plan is non-trivial).
    pub fn is_complete(&self) -> bool {
        self.ranges_total > 0 && self.ranges_done == self.ranges_total
    }
}

/// A live event emitted as a download progresses. Delivered on the handle's event stream.
#[derive(Debug, Clone)]
pub enum DownloadEvent {
    /// The resource was located + planned: this many ranges over this total length.
    Planned {
        /// Total ranges in the plan.
        ranges_total: usize,
        /// Total resource ciphertext length.
        total_length: u64,
    },
    /// A range was fetched, verified, and written. Carries the updated coalesced snapshot.
    RangeCompleted {
        /// The range index that completed.
        range: usize,
        /// The provider (64-hex `peer_id`) that served it.
        provider: String,
        /// The progress snapshot after this completion.
        progress: DownloadProgress,
    },
    /// A range fetch from a provider failed (transport or verify) and will be retried elsewhere.
    RangeFailed {
        /// The range index that failed.
        range: usize,
        /// The provider that failed to serve it.
        provider: String,
        /// A short reason (stable text).
        reason: String,
    },
    /// The provider set was refreshed (a `find_providers` re-run) because ranges were running out of
    /// live sources.
    ProvidersRefreshed {
        /// The number of providers now known.
        providers: usize,
    },
    /// The download was paused (no new range fetches will be issued until resumed).
    Paused,
    /// The download was resumed after a pause.
    Resumed,
    /// The download finished successfully — every range verified + written + finalized.
    Completed {
        /// The total verified bytes written.
        total_length: u64,
    },
    /// The download ended in failure (terminal). Carries the reason text.
    Failed {
        /// The terminal failure reason.
        reason: String,
    },
}

/// Durable resume state for one download: the resource commitment metadata + the set of ranges
/// already completed + verified. Serialized to a [`StateStore`] so a paused OR crashed download
/// resumes without re-fetching a verified range.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DownloadState {
    /// The download key (a stable id for this content target — see
    /// [`crate::orchestrator::download_key`]).
    pub key: String,
    /// Total resource ciphertext length (once the commitment is established; 0 before).
    pub total_length: u64,
    /// The per-chunk lengths (the commitment's `chunk_lens`), so a resume re-plans identically.
    pub chunk_lens: Vec<u64>,
    /// The chain-anchored generation root (64-hex), if known.
    pub root: Option<String>,
    /// The whole-resource inclusion proof (base64), if known.
    pub inclusion_proof: Option<String>,
    /// Range indices already completed + verified (never re-fetched on resume).
    pub done_ranges: BTreeSet<usize>,
}

impl DownloadState {
    /// A fresh, empty state for `key` (nothing planned or done yet).
    pub fn new(key: impl Into<String>) -> Self {
        DownloadState {
            key: key.into(),
            total_length: 0,
            chunk_lens: Vec::new(),
            root: None,
            inclusion_proof: None,
            done_ranges: BTreeSet::new(),
        }
    }

    /// Whether the resource commitment has been established (chunk layout known).
    pub fn has_commitment(&self) -> bool {
        !self.chunk_lens.is_empty()
    }

    /// Mark range `index` complete.
    pub fn mark_done(&mut self, index: usize) {
        self.done_ranges.insert(index);
    }

    /// Whether range `index` is already complete (and must not be re-fetched).
    pub fn is_done(&self, index: usize) -> bool {
        self.done_ranges.contains(&index)
    }
}

/// How long a persisted bad-descriptor verdict keeps a holder out of the DESCRIPTOR role.
///
/// Reputation decays because a verdict is evidence about a moment, not a permanent label: a holder can
/// be reinstalled, fixed, or have served a stale generation. A verdict that never expired would turn
/// one bad answer into permanent exclusion — and, aggregated, into a denial primitive.
pub const BAD_DESCRIPTOR_TTL: Duration = Duration::from_secs(24 * 60 * 60);

/// The most bad-descriptor verdicts retained per target.
///
/// Reputation is written in response to peer behaviour, so it must not itself be a growth vector: the
/// oldest verdict is evicted once the cap is reached. A capsule with more than this many distinct
/// lying holders is not a case reputation can help with.
pub const MAX_BAD_DESCRIPTOR_PEERS: usize = 32;

/// One persisted "this peer served a bad module descriptor" verdict, with when it was recorded (unix
/// seconds) so it can decay.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BadDescriptorVerdict {
    /// The 64-hex `peer_id` of the holder whose descriptor failed a final gate.
    pub peer_id: String,
    /// When the verdict was recorded, in unix seconds.
    pub recorded_at_unix: u64,
}

/// The current unix time in seconds (0 if the clock is before the epoch — a clock that absurd only
/// costs the caller a prematurely-expired verdict).
pub fn unix_now() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Whether `peer_id` is a well-formed 64-hex holder id.
///
/// A `peer_id` arrives off the wire as free-form text, and reputation is the one place where such a
/// string would be PERSISTED and later matched against. Only ids of the canonical shape are stored, so
/// no peer-supplied text can shape a stored key (#1603/#1609).
fn is_hex_peer_id(peer_id: &str) -> bool {
    peer_id.len() == 64 && peer_id.bytes().all(|b| b.is_ascii_hexdigit())
}

/// Record `peer_id`'s verdict in `verdicts`, dropping expired ones, de-duplicating (a repeat verdict
/// refreshes the timestamp), and evicting the oldest once [`MAX_BAD_DESCRIPTOR_PEERS`] is reached.
///
/// A malformed (non-64-hex) id is IGNORED rather than stored — see [`is_hex_peer_id`].
fn record_verdict(verdicts: &mut Vec<BadDescriptorVerdict>, peer_id: &str, now: u64) {
    if !is_hex_peer_id(peer_id) {
        return;
    }
    verdicts.retain(|v| !is_expired(v, now) && v.peer_id != peer_id);
    verdicts.push(BadDescriptorVerdict {
        peer_id: peer_id.to_string(),
        recorded_at_unix: now,
    });
    while verdicts.len() > MAX_BAD_DESCRIPTOR_PEERS {
        verdicts.remove(0); // oldest first — the vec stays in insertion order
    }
}

/// The still-live verdicts' peer ids, expired ones dropped.
fn live_peers(verdicts: &[BadDescriptorVerdict], now: u64) -> Vec<String> {
    verdicts
        .iter()
        .filter(|v| !is_expired(v, now))
        .map(|v| v.peer_id.clone())
        .collect()
}

/// Whether a verdict has aged past [`BAD_DESCRIPTOR_TTL`] (a verdict timestamped in the future — a
/// clock step — is treated as current, never as eternally valid).
fn is_expired(verdict: &BadDescriptorVerdict, now: u64) -> bool {
    now.saturating_sub(verdict.recorded_at_unix) > BAD_DESCRIPTOR_TTL.as_secs()
}

/// Persists [`DownloadState`] so a download resumes across pause + process restart.
///
/// The orchestrator checkpoints after each range completes and on pause. A resume loads the state and
/// re-plans; only the ranges NOT in `done_ranges` are fetched. The trait abstracts the medium so
/// tests use an [`InMemoryStateStore`] and a node uses [`FileStateStore`] (or a store-backed one).
#[async_trait]
pub trait StateStore: Send + Sync {
    /// Load the persisted state for `key`, or `None` if there is no checkpoint yet.
    async fn load(&self, key: &str) -> Result<Option<DownloadState>, DownloadError>;

    /// Persist `state` (overwriting any prior checkpoint for its key).
    async fn save(&self, state: &DownloadState) -> Result<(), DownloadError>;

    /// Delete the checkpoint for `key` (called after a successful, finalized download).
    async fn clear(&self, key: &str) -> Result<(), DownloadError>;

    /// Remember that `peer_id` supplied a descriptor for `target_key` that failed a final integrity
    /// gate, so a later call — or a later PROCESS — does not pay for the same lie again.
    ///
    /// Demotion within one pull is not enough: the holder order is deterministic, so a fresh call, a
    /// retry, or a restart re-asks the same liars from scratch, each paying up to
    /// `MAX_DESCRIPTOR_ATTEMPTS` full pull attempts in bandwidth and staging disk (#1611).
    ///
    /// A verdict is advisory and decays ([`BAD_DESCRIPTOR_TTL`]); it must never become a denial
    /// primitive, so a caller consults it to ORDER/FILTER descriptor sources and falls back to the
    /// full holder set rather than giving up. Demoted holders stay fully usable for CHUNK fetches —
    /// chunk bytes are independently hash-attributed, so excluding them would cost availability for no
    /// integrity gain.
    ///
    /// The default is a no-op, so an existing [`StateStore`] keeps compiling and simply forgets
    /// reputation between calls (the pre-#1611 behaviour — an efficiency loss, never a correctness one).
    async fn record_bad_descriptor(
        &self,
        target_key: &str,
        peer_id: &str,
    ) -> Result<(), DownloadError> {
        let _ = (target_key, peer_id);
        Ok(())
    }

    /// The peers with a still-live [`record_bad_descriptor`](Self::record_bad_descriptor) verdict for
    /// `target_key`. The default returns none (this store keeps no reputation).
    async fn bad_descriptor_peers(&self, target_key: &str) -> Result<Vec<String>, DownloadError> {
        let _ = target_key;
        Ok(Vec::new())
    }
}

/// An in-memory [`StateStore`] — the test store, and the default when no persistence is wanted (a
/// pause+resume within one process still works; a crash loses it). Thread-safe.
#[derive(Debug, Default)]
pub struct InMemoryStateStore {
    inner: tokio::sync::Mutex<std::collections::HashMap<String, DownloadState>>,
    /// Bad-descriptor verdicts per target key — reputation that survives a repeat CALL, though not (by
    /// construction) a process restart. [`FileStateStore`] is the durable one.
    reputation: tokio::sync::Mutex<std::collections::HashMap<String, Vec<BadDescriptorVerdict>>>,
}

impl InMemoryStateStore {
    /// A new, empty in-memory state store.
    pub fn new() -> Self {
        InMemoryStateStore::default()
    }
}

#[async_trait]
impl StateStore for InMemoryStateStore {
    async fn load(&self, key: &str) -> Result<Option<DownloadState>, DownloadError> {
        Ok(self.inner.lock().await.get(key).cloned())
    }

    async fn save(&self, state: &DownloadState) -> Result<(), DownloadError> {
        self.inner
            .lock()
            .await
            .insert(state.key.clone(), state.clone());
        Ok(())
    }

    async fn clear(&self, key: &str) -> Result<(), DownloadError> {
        self.inner.lock().await.remove(key);
        Ok(())
    }

    async fn record_bad_descriptor(
        &self,
        target_key: &str,
        peer_id: &str,
    ) -> Result<(), DownloadError> {
        let mut reputation = self.reputation.lock().await;
        let verdicts = reputation.entry(target_key.to_string()).or_default();
        record_verdict(verdicts, peer_id, unix_now());
        Ok(())
    }

    async fn bad_descriptor_peers(&self, target_key: &str) -> Result<Vec<String>, DownloadError> {
        let reputation = self.reputation.lock().await;
        Ok(reputation
            .get(target_key)
            .map(|v| live_peers(v, unix_now()))
            .unwrap_or_default())
    }
}

/// The fixed-width, path-safe file stem a download key maps to: `SHA-256(key)` in lower hex.
///
/// Always exactly [`CHECKPOINT_STEM_LEN`] characters, for every key. See [`FileStateStore::file_for`]
/// for why this is a digest and not the key's own bytes.
fn checkpoint_file_stem(key: &str) -> String {
    use sha2::Digest;
    crate::module::hex_of(sha2::Sha256::digest(key.as_bytes()))
}

/// Length in characters of every [`checkpoint_file_stem`] — a SHA-256 digest in lower hex.
const CHECKPOINT_STEM_LEN: usize = 64;

/// A file-backed [`StateStore`]: one JSON checkpoint file per download key, under a directory. A
/// crashed download resumes by re-reading its checkpoint. The filename is a fixed-width SHA-256
/// DIGEST of the key, so it is both filesystem-safe and bounded in length whatever the key is.
#[derive(Debug, Clone)]
pub struct FileStateStore {
    dir: std::path::PathBuf,
}

impl FileStateStore {
    /// A file state store writing checkpoints under `dir` (created on first save if missing).
    pub fn new(dir: impl Into<std::path::PathBuf>) -> Self {
        FileStateStore { dir: dir.into() }
    }

    fn path_for(&self, key: &str) -> std::path::PathBuf {
        self.file_for(key, ".json")
    }

    /// The reputation sidecar beside a target's checkpoint. Kept SEPARATE from the checkpoint because
    /// the two have different lifetimes: a checkpoint is cleared the moment a download completes, while
    /// what a holder did must outlive that success.
    fn reputation_path_for(&self, key: &str) -> std::path::PathBuf {
        self.file_for(key, ".holders.json")
    }

    /// `<sha256_hex(key)><suffix>` under this store's directory.
    ///
    /// The key is DIGESTED rather than hex-encoded (#38). Hex encoding gave path-safety — no key text
    /// can shape a path — but its output grows with the key, and the real keys are long: a module
    /// checkpoint key is `module:<64hex>:<64hex>` = 136 bytes, which hex-encodes to 272 characters and
    /// with `.json` reaches 277 — past Linux's `NAME_MAX` of 255. Every capsule checkpoint write on
    /// Linux therefore failed with `File name too long (os error 36)`, deterministically.
    ///
    /// A digest keeps BOTH properties the hex encoding was chosen for and adds the missing one:
    /// the output alphabet is still `[0-9a-f]` (path-safe by construction, no key text reaches the
    /// path), distinct keys still get distinct names (collision resistance, unlike a truncation, which
    /// would silently alias two capsules onto one checkpoint and corrupt resume state), and the name is
    /// now FIXED-WIDTH at 64 characters however long the key is. The longest name this can produce is
    /// 64 + `".holders.json".len()` = 77 characters, asserted by [`digest_name_is_bounded`].
    ///
    /// [`digest_name_is_bounded`]: tests::digest_name_is_bounded
    fn file_for(&self, key: &str, suffix: &str) -> std::path::PathBuf {
        let mut name = checkpoint_file_stem(key);
        debug_assert_eq!(name.len(), CHECKPOINT_STEM_LEN);
        name.reserve_exact(suffix.len());
        name.push_str(suffix);
        self.dir.join(name)
    }

    /// The verdicts persisted for `key` (an absent or unreadable sidecar reads as none — reputation is
    /// advisory, so it must never fail a download).
    fn read_verdicts(&self, key: &str) -> Vec<BadDescriptorVerdict> {
        std::fs::read(self.reputation_path_for(key))
            .ok()
            .and_then(|bytes| serde_json::from_slice(&bytes).ok())
            .unwrap_or_default()
    }
}

#[async_trait]
impl StateStore for FileStateStore {
    async fn load(&self, key: &str) -> Result<Option<DownloadState>, DownloadError> {
        let path = self.path_for(key);
        match std::fs::read(&path) {
            Ok(bytes) => {
                let state = serde_json::from_slice(&bytes).map_err(DownloadError::state)?;
                Ok(Some(state))
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(DownloadError::state(e)),
        }
    }

    async fn save(&self, state: &DownloadState) -> Result<(), DownloadError> {
        std::fs::create_dir_all(&self.dir).map_err(DownloadError::state)?;
        let bytes = serde_json::to_vec(state).map_err(DownloadError::state)?;
        std::fs::write(self.path_for(&state.key), bytes).map_err(DownloadError::state)
    }

    async fn clear(&self, key: &str) -> Result<(), DownloadError> {
        match std::fs::remove_file(self.path_for(key)) {
            Ok(()) => Ok(()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(e) => Err(DownloadError::state(e)),
        }
    }

    async fn record_bad_descriptor(
        &self,
        target_key: &str,
        peer_id: &str,
    ) -> Result<(), DownloadError> {
        let mut verdicts = self.read_verdicts(target_key);
        record_verdict(&mut verdicts, peer_id, unix_now());
        std::fs::create_dir_all(&self.dir).map_err(DownloadError::state)?;
        let bytes = serde_json::to_vec(&verdicts).map_err(DownloadError::state)?;
        std::fs::write(self.reputation_path_for(target_key), bytes).map_err(DownloadError::state)
    }

    async fn bad_descriptor_peers(&self, target_key: &str) -> Result<Vec<String>, DownloadError> {
        Ok(live_peers(&self.read_verdicts(target_key), unix_now()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn progress_fraction_and_complete() {
        let mut p = DownloadProgress {
            total_length: 100,
            bytes_done: 25,
            ranges_total: 4,
            ranges_done: 1,
            active_sources: 2,
        };
        assert!((p.fraction() - 0.25).abs() < 1e-9);
        assert!(!p.is_complete());
        p.ranges_done = 4;
        p.bytes_done = 100;
        assert!(p.is_complete());
        assert_eq!(DownloadProgress::default().fraction(), 0.0);
    }

    #[test]
    fn state_marks_and_queries_done() {
        let mut s = DownloadState::new("k");
        assert!(!s.is_done(2));
        s.mark_done(2);
        assert!(s.is_done(2));
        assert_eq!(s.done_ranges.len(), 1);
    }

    #[tokio::test]
    async fn in_memory_store_round_trips() {
        let store = InMemoryStateStore::new();
        assert!(store.load("k").await.unwrap().is_none());
        let mut s = DownloadState::new("k");
        s.mark_done(1);
        s.total_length = 42;
        store.save(&s).await.unwrap();
        assert_eq!(store.load("k").await.unwrap().unwrap(), s);
        store.clear("k").await.unwrap();
        assert!(store.load("k").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn file_store_round_trips_and_survives_reload() {
        let dir = std::env::temp_dir().join(format!(
            "dig-download-test-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let store = FileStateStore::new(&dir);
        assert!(store.load("abc").await.unwrap().is_none());

        let mut s = DownloadState::new("abc");
        s.total_length = 100;
        s.chunk_lens = vec![10, 20];
        s.root = Some("aa".repeat(32));
        s.mark_done(0);
        store.save(&s).await.unwrap();

        // A brand-new store instance (simulating a process restart) reads the same checkpoint.
        let reloaded = FileStateStore::new(&dir);
        assert_eq!(reloaded.load("abc").await.unwrap().unwrap(), s);

        store.clear("abc").await.unwrap();
        assert!(store.load("abc").await.unwrap().is_none());
        // clear on a missing key is a no-op.
        store.clear("abc").await.unwrap();
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// The longest name `FileStateStore` can produce must fit in a filename, on every platform.
    ///
    /// `NAME_MAX` is 255 on Linux (and 255 UTF-16 units on Windows/NTFS), so the bound is checked
    /// against 255 rather than a `cfg`-gated constant. This is the ARITHMETIC half of #38: it holds
    /// for ANY key the key constructors can emit, because the stem is a fixed-width digest and the
    /// only variable left is the suffix.
    #[test]
    fn digest_name_is_bounded() {
        const NAME_MAX: usize = 255;
        let long_key = crate::module::module_download_key(&"ab".repeat(32), &"cd".repeat(32));
        assert_eq!(
            long_key.len(),
            136,
            "the production key really is this long"
        );

        // Measured on what `FileStateStore` ACTUALLY produces — not on the stem helper — so a
        // regression in `file_for` itself is visible here and not only in the round-trip test. The
        // reputation sidecar carries the longest suffix any call site passes: the worst case.
        let store = FileStateStore::new("dir");
        for produced in [
            store.path_for(&long_key),
            store.reputation_path_for(&long_key),
        ] {
            let name = produced.file_name().unwrap().to_string_lossy().into_owned();
            assert!(
                name.chars().count() < NAME_MAX,
                "{} chars exceeds NAME_MAX: {name}",
                name.chars().count()
            );
        }
        assert_eq!(
            store
                .reputation_path_for(&long_key)
                .file_name()
                .unwrap()
                .to_string_lossy()
                .chars()
                .count(),
            CHECKPOINT_STEM_LEN + ".holders.json".len(),
        );

        // The bound holds for ANY key because the stem is fixed-width: the 136-byte production key
        // and a 3-byte key produce the same-length stem.
        assert_eq!(checkpoint_file_stem(&long_key).len(), CHECKPOINT_STEM_LEN);
        assert_eq!(checkpoint_file_stem("abc").len(), CHECKPOINT_STEM_LEN);
        // Distinct keys stay distinct — the property a truncation would have destroyed.
        assert_ne!(
            checkpoint_file_stem(&long_key),
            checkpoint_file_stem(&crate::module::module_download_key(
                &"ab".repeat(32),
                &"ce".repeat(32)
            ))
        );
    }

    /// #38: a checkpoint under a REAL `module:<64hex>:<64hex>` key must round-trip through
    /// `FileStateStore`.
    ///
    /// The suite could not see this defect before: every `module.rs` test uses `InMemoryStateStore`,
    /// which has no filename at all, and the one `FileStateStore` test used a 3-character key. The
    /// old hex-encoding scheme turned this 136-byte key into a 277-character name and every write
    /// failed on Linux with `File name too long (os error 36)`.
    #[tokio::test]
    async fn file_store_round_trips_a_real_module_download_key() {
        let dir = std::env::temp_dir().join(format!(
            "dig-download-modkey-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let key = crate::module::module_download_key(&"ab".repeat(32), &"cd".repeat(32));
        let store = FileStateStore::new(&dir);

        let mut s = DownloadState::new(&key);
        s.total_length = 100;
        s.chunk_lens = vec![10, 20];
        s.mark_done(0);
        // The write is where `os error 36` struck.
        store.save(&s).await.unwrap();

        // A restarted process reads the same checkpoint back under the same long key.
        assert_eq!(
            FileStateStore::new(&dir).load(&key).await.unwrap().unwrap(),
            s
        );

        // The reputation sidecar carries the longest suffix, so it is the worst case — exercise it too.
        let peer = "ef".repeat(32);
        store.record_bad_descriptor(&key, &peer).await.unwrap();
        assert_eq!(
            FileStateStore::new(&dir)
                .bad_descriptor_peers(&key)
                .await
                .unwrap(),
            vec![peer]
        );

        // Nothing on disk exceeds the filename limit.
        for entry in std::fs::read_dir(&dir).unwrap() {
            let name = entry.unwrap().file_name();
            assert!(
                name.to_string_lossy().chars().count() < 255,
                "checkpoint filename must fit NAME_MAX: {name:?}"
            );
        }

        store.clear(&key).await.unwrap();
        assert!(store.load(&key).await.unwrap().is_none());
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn download_event_variants_construct() {
        // Smoke: the event shapes build (exercised richly in the orchestrator tests).
        let _ = DownloadEvent::Planned {
            ranges_total: 3,
            total_length: 30,
        };
        let _ = DownloadEvent::Paused;
        let _ = DownloadEvent::Resumed;
        let _ = DownloadEvent::ProvidersRefreshed { providers: 2 };
        let _ = DownloadEvent::Completed { total_length: 30 };
        let _ = DownloadEvent::Failed { reason: "x".into() };
    }

    #[tokio::test]
    async fn in_memory_store_remembers_a_bad_descriptor_verdict() {
        let store = InMemoryStateStore::new();
        let peer = "ab".repeat(32);
        assert!(store.bad_descriptor_peers("k").await.unwrap().is_empty());
        store.record_bad_descriptor("k", &peer).await.unwrap();
        assert_eq!(store.bad_descriptor_peers("k").await.unwrap(), vec![peer]);
        // Reputation is per TARGET: another capsule's holders are unaffected.
        assert!(store
            .bad_descriptor_peers("other")
            .await
            .unwrap()
            .is_empty());
    }

    #[tokio::test]
    async fn file_store_reputation_survives_a_process_restart_and_outlives_the_checkpoint() {
        let dir = std::env::temp_dir().join(format!(
            "dig-download-rep-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let store = FileStateStore::new(&dir);
        let peer = "cd".repeat(32);
        store
            .record_bad_descriptor("module:x", &peer)
            .await
            .unwrap();

        // A brand-new instance (a restarted process) still sees the verdict — the durability the
        // in-call `demoted` vec never had.
        let restarted = FileStateStore::new(&dir);
        assert_eq!(
            restarted.bad_descriptor_peers("module:x").await.unwrap(),
            vec![peer.clone()]
        );

        // Clearing the CHECKPOINT must not forget what a holder did: the two have different lifetimes.
        restarted.clear("module:x").await.unwrap();
        assert_eq!(
            restarted.bad_descriptor_peers("module:x").await.unwrap(),
            vec![peer]
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A verdict DECAYS: one bad answer must not exclude a holder forever (holders get fixed, and a
    /// verdict can be the record of a stale generation).
    #[test]
    fn a_verdict_expires_after_its_ttl() {
        let now = 10_000_000u64;
        let fresh = BadDescriptorVerdict {
            peer_id: "ab".repeat(32),
            recorded_at_unix: now - 60,
        };
        let stale = BadDescriptorVerdict {
            peer_id: "cd".repeat(32),
            recorded_at_unix: now - BAD_DESCRIPTOR_TTL.as_secs() - 1,
        };
        assert_eq!(
            live_peers(&[fresh.clone(), stale], now),
            vec![fresh.peer_id],
            "only the un-expired verdict is live"
        );
    }

    /// Reputation is written in response to peer behaviour, so it must not itself grow without bound:
    /// the record is capped and a repeat verdict refreshes rather than duplicates.
    #[test]
    fn the_verdict_record_is_bounded_and_deduplicated() {
        let now = 10_000_000u64;
        let mut verdicts = Vec::new();
        for i in 0..(MAX_BAD_DESCRIPTOR_PEERS + 10) {
            record_verdict(&mut verdicts, &format!("{i:064x}"), now);
        }
        assert_eq!(verdicts.len(), MAX_BAD_DESCRIPTOR_PEERS, "capped");
        assert!(
            !verdicts.iter().any(|v| v.peer_id == format!("{:064x}", 0)),
            "the oldest verdicts were evicted first"
        );

        let repeat = format!("{:064x}", MAX_BAD_DESCRIPTOR_PEERS + 9);
        record_verdict(&mut verdicts, &repeat, now + 5);
        assert_eq!(
            verdicts.iter().filter(|v| v.peer_id == repeat).count(),
            1,
            "a repeat verdict refreshes the entry instead of duplicating it"
        );
    }

    /// A `peer_id` is free-form text off the wire, and reputation is the one place it would be
    /// PERSISTED and later matched: a malformed id is dropped, never stored.
    #[test]
    fn a_malformed_peer_id_is_never_recorded() {
        let mut verdicts = Vec::new();
        record_verdict(&mut verdicts, "../../etc/passwd", 1);
        record_verdict(&mut verdicts, "not-hex", 1);
        record_verdict(&mut verdicts, &"ab".repeat(31), 1); // too short
        assert!(
            verdicts.is_empty(),
            "only 64-hex ids are stored: {verdicts:?}"
        );
    }
}