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); +}