diff --git a/Cargo.lock b/Cargo.lock index 543697c..4bf8118 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -499,6 +499,7 @@ dependencies = [ "c0mpute-core", "c0mpute-doctor", "c0mpute-gateway", + "c0mpute-placement", "c0mpute-proto", "c0mpute-secure-chat", "c0mpute-store", @@ -508,7 +509,9 @@ dependencies = [ "dialoguer", "hex", "libc", + "reqwest", "rpassword", + "serde", "serde_json", "tokio", "toml", @@ -599,6 +602,27 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "c0mpute-placement" +version = "0.2.26" +dependencies = [ + "anyhow", + "async-trait", + "axum", + "blake3", + "c0mpute-gateway", + "c0mpute-placement", + "c0mpute-proto", + "c0mpute-store", + "futures", + "reqwest", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", +] + [[package]] name = "c0mpute-proto" version = "0.2.26" diff --git a/Cargo.toml b/Cargo.toml index 7beb051..4f697b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ members = [ "node/crates/c0mpute-core", "node/crates/c0mpute-net", "node/crates/c0mpute-store", + "node/crates/c0mpute-placement", "node/crates/c0mpute-gateway", "node/crates/c0mpute-verify", "node/crates/c0mpute-update", @@ -65,6 +66,7 @@ rpassword = "7" c0mpute-proto = { path = "node/crates/c0mpute-proto" } c0mpute-store = { path = "node/crates/c0mpute-store" } +c0mpute-placement = { path = "node/crates/c0mpute-placement" } c0mpute-net = { path = "node/crates/c0mpute-net" } c0mpute-gateway = { path = "node/crates/c0mpute-gateway" } c0mpute-verify = { path = "node/crates/c0mpute-verify" } diff --git a/docs/prds/003-shard-placement-transport.md b/docs/prds/003-shard-placement-transport.md index 8a5f3eb..ae4ac74 100644 --- a/docs/prds/003-shard-placement-transport.md +++ b/docs/prds/003-shard-placement-transport.md @@ -1,7 +1,7 @@ --- cip: 003 title: "Cross-node shard placement and streaming transport" -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 3 depends-on: 002 blocks: 005, 006 -implementation: +implementation: PR #23 (c0mpute-placement crate; HTTP transport over the CIP-002 shard endpoints; `c0mpute storage peer`) estimate: "3–4 weeks" --- @@ -57,6 +57,20 @@ first thing this CIP fixes. ## Design +### What shipped first: HTTP, not libp2p + +The plan below rewrites the libp2p protocol before placement can work. That +ordering turned out to be unnecessary. CIP-002 already ships shard `PUT`/`GET`/ +`HEAD` endpoints that verify what they are given, so placement was built +against a `ShardTransport` trait with an HTTP implementation on top of those — +and cross-node placement works today, on a real multi-node testnet, with no +libp2p changes at all. + +The streaming libp2p protocol below is still worth doing (it removes an HTTP +hop between peers that are already connected, and gives repair a batched +`Have` probe). It is now a *second implementation of an existing trait* rather +than a prerequisite, which makes it independently schedulable. + ### Fix the transport first `request_response::cbor::Behaviour` buffers an @@ -93,11 +107,27 @@ Given a block needing `n` hosts, score each candidate peer: ``` score = reputation # c0mpute-verify::reputation, >= 0.9 required * uptime_30d # >= 0.99 required (CIP-001) - * free_disk_factor # committed - used, normalised - * (1 / (1 + rtt_ms / 100)) # prefer near peers, weakly + * (0.9 + 0.1 / (1 + rtt_ms / 100)) # prefer near peers, weakly ``` -Then select greedily under **diversity constraints**, in priority order: +Free disk is a hard filter rather than a score term — a peer either has room +for the shard or it does not. + +**The latency weighting is deliberately narrower than this CIP first +specified.** A bare `1 / (1 + rtt/100)` factor makes a 400 ms peer score 20% +below a 1 ms one, which is enough for a fast flaky node to outrank a slow +reliable one. CIP-001 is explicit that availability drives durability and +latency does not, so the term is scaled into a band where it separates +otherwise-equal peers but cannot overturn a reputation gap. A unit test pins +this. + +Greedy selection under a per-domain cap is **optimal, not heuristic**: "at +most `max_per_domain` from each domain" is a partition matroid, and greedy is +optimal over a matroid. So a `DiversityUnsatisfiable` result means no other +assignment would have worked either — no backtracking, and no better answer +being missed. + +Select greedily under **diversity constraints**, in priority order: 1. No two shards of the same block on the same peer. (Hard.) 2. At most `floor(parity / 2)` shards per ASN — 2 of 14 for `standard`. (Hard.) 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 5609d5d..ea67804 100644 --- a/docs/prds/README.md +++ b/docs/prds/README.md @@ -66,9 +66,9 @@ Delivering read/write network storage for c0mpute, implementing |-----|-------|-----------|--------| | [001](001-storage-program.md) | Storage program: durability model, tiers, and economics | — | In progress | | [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 | Draft | +| [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/Cargo.toml b/node/crates/c0mpute-cli/Cargo.toml index bd22353..34ef7cc 100644 --- a/node/crates/c0mpute-cli/Cargo.toml +++ b/node/crates/c0mpute-cli/Cargo.toml @@ -16,12 +16,15 @@ c0mpute-proto = { workspace = true } c0mpute-update = { workspace = true } c0mpute-secure-chat = { workspace = true } c0mpute-store = { workspace = true } +c0mpute-placement = { workspace = true } +reqwest = { workspace = true } c0mpute-gateway = { workspace = true } axum = { workspace = true } anyhow = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } clap = { workspace = true } +serde = { workspace = true } serde_json = { workspace = true } toml = { workspace = true } rpassword = { workspace = true } diff --git a/node/crates/c0mpute-cli/src/main.rs b/node/crates/c0mpute-cli/src/main.rs index bbaffd5..e89a401 100644 --- a/node/crates/c0mpute-cli/src/main.rs +++ b/node/crates/c0mpute-cli/src/main.rs @@ -19,6 +19,7 @@ //! The plugin form mirrors the URL namespace: c0mpute.com/transcode, //! c0mpute.com/coinpay, c0mpute.com/infernet. +mod peers; mod storage; use std::path::PathBuf; @@ -356,6 +357,15 @@ fn maybe_self_update(cli: &Cli) { fn maybe_self_update(_cli: &Cli) {} fn main() -> Result<()> { + // Rust ignores SIGPIPE, so writing to a closed pipe returns EPIPE and the + // stdlib panics on it. For a CLI that prints lists that is a crash on + // `c0mpute storage ls | head`, which is ordinary shell usage. Restore the + // default disposition so the process exits quietly instead. + #[cfg(unix)] + unsafe { + libc::signal(libc::SIGPIPE, libc::SIG_DFL); + } + let cli = Cli::parse(); // Opportunistic self-update on any command (throttled), so c0mpute stays diff --git a/node/crates/c0mpute-cli/src/peers.rs b/node/crates/c0mpute-cli/src/peers.rs new file mode 100644 index 0000000..a5fa48c --- /dev/null +++ b/node/crates/c0mpute-cli/src/peers.rs @@ -0,0 +1,160 @@ +//! The node's storage-peer registry (CIP-003). +//! +//! Peers are read from `/peers.json`. Gossipsub capability ads +//! will populate this automatically once the storage role advertises itself; +//! until then an operator adds peers explicitly, which is also what makes a +//! deliberate small testnet possible. + +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use c0mpute_placement::{PeerCatalog, PeerInfo}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct PeerFile { + #[serde(default)] + pub peers: Vec, +} + +pub fn peers_path(storage_root: &Path) -> PathBuf { + storage_root.join("peers.json") +} + +pub fn load(storage_root: &Path) -> Result { + let path = peers_path(storage_root); + if !path.exists() { + return Ok(PeerCatalog::default()); + } + let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?; + let file: PeerFile = + serde_json::from_slice(&bytes).with_context(|| format!("parse {}", path.display()))?; + Ok(PeerCatalog::new(file.peers)) +} + +pub fn save(storage_root: &Path, catalog: &PeerCatalog) -> Result<()> { + let path = peers_path(storage_root); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + let file = PeerFile { + peers: catalog.peers().to_vec(), + }; + let json = serde_json::to_vec_pretty(&file)?; + // Write-then-rename: a half-written peer list would be parsed as a + // smaller network on the next read, and placement decisions follow from + // exactly that number. + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, &json)?; + std::fs::rename(&tmp, &path)?; + Ok(()) +} + +/// Build a peer record, filling in what can be inferred. +/// +/// A peer with no ASN and a DNS endpoint has an `Unknown` failure domain and +/// is excluded from placement by default — deliberately, since CIP-001's +/// durability figures assume independent hosts. `prefix_from_endpoint` +/// recovers a weak-but-real domain when the endpoint is a literal IP. +pub fn build( + peer_id: String, + endpoint: String, + asn: Option, + region: Option, +) -> PeerInfo { + let ip_prefix = PeerInfo::prefix_from_endpoint(&endpoint); + PeerInfo { + peer_id, + endpoint, + // Until CIP-006's challenges produce real numbers, a manually added + // peer is taken at its word. Recorded here rather than hidden so the + // assumption is visible when reputation starts being measured. + reputation: 1.0, + uptime_30d: 1.0, + free_bytes: u64::MAX, + rtt_ms: 50, + asn, + region, + ip_prefix, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmpdir() -> PathBuf { + let d = std::env::temp_dir().join(format!( + "c0mpute-peers-test-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&d).unwrap(); + d + } + + #[test] + fn missing_file_is_an_empty_catalog_not_an_error() { + let c = load(&tmpdir()).unwrap(); + assert!(c.is_empty()); + } + + #[test] + fn round_trips_through_disk() { + let dir = tmpdir(); + let mut c = PeerCatalog::default(); + c.upsert(build( + "a".into(), + "http://10.0.0.1:7780".into(), + Some(7), + None, + )); + c.upsert(build( + "b".into(), + "http://10.1.0.1:7780".into(), + Some(8), + None, + )); + save(&dir, &c).unwrap(); + + let back = load(&dir).unwrap(); + assert_eq!(back.len(), 2); + assert_eq!(back.get("a").unwrap().asn, Some(7)); + assert_eq!(back.domain_count(), 2); + } + + #[test] + fn infers_an_ip_prefix_when_the_asn_is_unknown() { + let p = build("a".into(), "http://203.0.113.9:7780".into(), None, None); + assert_eq!(p.ip_prefix.as_deref(), Some("203.0.113")); + assert_eq!( + p.domain(), + c0mpute_placement::FailureDomain::IpPrefix("203.0.113".into()) + ); + } + + #[test] + fn a_dns_endpoint_without_an_asn_has_no_domain() { + let p = build( + "a".into(), + "http://node.example.com:7780".into(), + None, + None, + ); + assert_eq!(p.ip_prefix, None); + assert_eq!(p.domain(), c0mpute_placement::FailureDomain::Unknown); + } + + #[test] + fn an_explicit_asn_wins_over_the_inferred_prefix() { + let p = build( + "a".into(), + "http://203.0.113.9:7780".into(), + Some(64512), + None, + ); + assert_eq!(p.domain(), c0mpute_placement::FailureDomain::Asn(64512)); + } +} diff --git a/node/crates/c0mpute-cli/src/storage.rs b/node/crates/c0mpute-cli/src/storage.rs index 8680fe6..5ab0449 100644 --- a/node/crates/c0mpute-cli/src/storage.rs +++ b/node/crates/c0mpute-cli/src/storage.rs @@ -16,18 +16,41 @@ use anyhow::{Context, Result, bail}; use c0mpute_core::config; use c0mpute_gateway::auth::AllowAll; use c0mpute_gateway::storage_api::{self, Limits, StorageApiState}; +use c0mpute_placement::{DistributedStorage, HttpTransport, PlacementPolicy}; use c0mpute_proto::Hash; use c0mpute_store::{ChunkStore, Storage, Tier}; use clap::Subcommand; +use tokio::sync::RwLock; + +use crate::peers; #[derive(Subcommand, Debug)] pub enum StorageCmd { /// Store a file and print its object hash. + /// + /// Placed across the network when storage peers are configured + /// (`c0mpute storage peer add`), otherwise stored on this node alone. Put { file: PathBuf, /// Redundancy tier: hot, standard (default) or critical. #[arg(long, default_value = "standard")] tier: String, + /// Keep every shard on this node even when peers are configured. + #[arg(long)] + local: bool, + /// Place without requiring failure-domain diversity. + /// + /// CIP-001's durability figures assume shard hosts fail + /// independently. Fourteen shards behind one ISP are one host wearing + /// fourteen hats, and nothing downstream can tell the difference. + /// Testnets only. + #[arg(long)] + insecure_ignore_diversity: bool, + }, + /// Storage peers this node knows about. + Peer { + #[command(subcommand)] + cmd: PeerCmd, }, /// Fetch an object by hash. Get { @@ -49,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, @@ -70,6 +106,29 @@ pub enum StorageCmd { }, } +#[derive(Subcommand, Debug)] +pub enum PeerCmd { + /// Register a storage peer. + Add { + peer_id: String, + /// Base URL of the peer's storage API, e.g. http://10.0.0.2:7780 + endpoint: String, + /// Autonomous system number. Without it the failure domain falls back + /// to the peer's IP prefix, or is unknown for a DNS name — and + /// unknown-domain peers are excluded from placement. + #[arg(long)] + asn: Option, + #[arg(long)] + region: Option, + }, + /// List known peers and the failure domains they span. + Ls, + /// Forget a peer. + Rm { peer_id: String }, + /// Check which peers are reachable right now. + Ping, +} + /// Where the local shard store lives. fn storage_root(config_path: &std::path::Path) -> Result { let cfg = config::Config::load_or_default(config_path)?; @@ -106,11 +165,20 @@ fn human(bytes: u64) -> String { pub async fn run(cmd: StorageCmd, config_path: &std::path::Path) -> Result<()> { match cmd { - StorageCmd::Put { file, tier } => put(config_path, file, &tier).await, + StorageCmd::Put { + file, + tier, + local, + insecure_ignore_diversity, + } => put(config_path, file, &tier, local, insecure_ignore_diversity).await, + StorageCmd::Peer { cmd } => peer(config_path, cmd).await, StorageCmd::Get { hash, out, range } => get(config_path, &hash, out, range).await, 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 => { @@ -124,16 +192,71 @@ pub async fn run(cmd: StorageCmd, config_path: &std::path::Path) -> Result<()> { } } -async fn put(config_path: &std::path::Path, file: PathBuf, tier: &str) -> Result<()> { - let tier: Tier = tier.parse()?; - let storage = open(config_path).await?; +/// Open a distributed view of this node's storage, if peers are configured. +async fn open_distributed( + config_path: &std::path::Path, + ignore_diversity: bool, +) -> Result> { + let root = storage_root(config_path)?; + let catalog = peers::load(&root)?; + if catalog.is_empty() { + return Ok(None); + } + let peer_count = catalog.len(); + let domains = catalog.domain_count(); + + let mut config = c0mpute_placement::DistributedConfig::default(); + if ignore_diversity { + config.policy = Some(PlacementPolicy { + max_per_domain: usize::MAX, + allow_unknown_domain: true, + ..PlacementPolicy::for_parity(4) + }); + } + + let storage = DistributedStorage::new( + open(config_path).await?, + Arc::new(HttpTransport::default()), + Arc::new(RwLock::new(catalog)), + ) + .with_config(config); + Ok(Some((storage, peer_count, domains))) +} +async fn put( + config_path: &std::path::Path, + file: PathBuf, + tier: &str, + force_local: bool, + ignore_diversity: bool, +) -> Result<()> { + let tier: Tier = tier.parse()?; let bytes = tokio::fs::read(&file) .await .with_context(|| format!("read {}", file.display()))?; let len = bytes.len() as u64; - let manifest = storage.put_tiered(&bytes, tier).await?; + if ignore_diversity { + eprintln!( + "warning: --insecure-ignore-diversity — shards may all land in one\n\ + failure domain, which makes the durability claim meaningless." + ); + } + + let distributed = if force_local { + None + } else { + open_distributed(config_path, ignore_diversity).await? + }; + + let manifest = match &distributed { + Some((storage, peer_count, domains)) => { + eprintln!(" placing across {peer_count} peer(s) in {domains} failure domain(s)"); + storage.put(&bytes, tier).await? + } + None => open(config_path).await?.put_tiered(&bytes, tier).await?, + }; + println!("blake3:{}", manifest.object_hash.to_hex()); eprintln!( " {} in {} block(s), {} shards, tier {} ({:.1}x expansion, {} raw)", @@ -144,9 +267,121 @@ async fn put(config_path: &std::path::Path, file: PathBuf, tier: &str) -> Result tier.expansion(), human((len as f64 * tier.expansion()).ceil() as u64), ); + if distributed.is_none() { + eprintln!( + " single-node: every shard is on this disk, so the erasure coding is\n \ + overhead without durability. Add peers with `c0mpute storage peer add`." + ); + } Ok(()) } +async fn peer(config_path: &std::path::Path, cmd: PeerCmd) -> Result<()> { + let root = storage_root(config_path)?; + let mut catalog = peers::load(&root)?; + + match cmd { + PeerCmd::Add { + peer_id, + endpoint, + asn, + region, + } => { + let info = peers::build(peer_id.clone(), endpoint.clone(), asn, region); + let domain = info.domain(); + catalog.upsert(info); + peers::save(&root, &catalog)?; + println!("added {peer_id} at {endpoint}"); + println!(" failure domain: {domain:?}"); + if matches!(domain, c0mpute_placement::FailureDomain::Unknown) { + eprintln!( + " warning: unknown failure domain — this peer will be skipped by\n \ + placement. Pass --asn, or use an endpoint with a literal IP." + ); + } + report_capacity(&catalog); + } + PeerCmd::Ls => { + if catalog.is_empty() { + println!("no storage peers configured"); + return Ok(()); + } + println!( + "{:<20} {:<32} {:<24} {}", + "PEER", "ENDPOINT", "DOMAIN", "FREE" + ); + for p in catalog.peers() { + println!( + "{:<20} {:<32} {:<24} {}", + p.peer_id, + p.endpoint, + format!("{:?}", p.domain()), + if p.free_bytes == u64::MAX { + "unknown".to_string() + } else { + human(p.free_bytes) + } + ); + } + report_capacity(&catalog); + } + PeerCmd::Rm { peer_id } => { + catalog.remove(&peer_id); + peers::save(&root, &catalog)?; + println!("removed {peer_id}"); + report_capacity(&catalog); + } + PeerCmd::Ping => { + if catalog.is_empty() { + println!("no storage peers configured"); + return Ok(()); + } + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build()?; + for p in catalog.peers() { + let url = format!("{}/storage/v1/status", p.endpoint.trim_end_matches('/')); + let started = std::time::Instant::now(); + match client.get(&url).send().await { + Ok(r) if r.status().is_success() => { + println!( + "{:<20} up {:>5} ms", + p.peer_id, + started.elapsed().as_millis() + ) + } + Ok(r) => println!("{:<20} HTTP {}", p.peer_id, r.status()), + Err(e) => println!("{:<20} down {e}", p.peer_id), + } + } + } + } + Ok(()) +} + +/// Say plainly whether this network can place at each tier. +/// +/// The peer count is the number people look at; the domain count is the one +/// that decides whether a placement is durable, so print both together. +fn report_capacity(catalog: &c0mpute_placement::PeerCatalog) { + let domains = catalog.domain_count(); + println!( + "\n{} peer(s) across {domains} failure domain(s)", + catalog.len() + ); + for tier in [Tier::Hot, Tier::Standard, Tier::Critical] { + let policy = PlacementPolicy::for_parity(tier.parity()); + let needed = policy.domains_required(tier.n()); + let ok = catalog.len() >= tier.n() && domains >= needed; + println!( + " {:<9} {} — needs {} peers across {needed} domains", + tier.to_string(), + if ok { "ready" } else { "NOT ready" }, + tier.n() + ); + } +} + async fn get( config_path: &std::path::Path, hash: &str, @@ -169,9 +404,22 @@ async fn get( if end < start { bail!("range end {end} is before start {start}"); } + // Range reads are served locally. An object placed across the + // network has no local shards, so this only works for a + // single-node object today; distributed range reads land with the + // filesystem layer (CIP-007), which is what needs them. storage.get_range(&hash, start, end - start + 1).await? } - None => storage.get(&hash).await?, + None => { + // Whether the shards are local or on peers is a property of the + // object, not of the command. Prefer the distributed path when + // peers are configured; it falls back to local shards per block, + // so a single-node object still reads. + match open_distributed(config_path, false).await? { + Some((distributed, _, _)) => distributed.get(&hash).await?, + None => storage.get(&hash).await?, + } + } }; match out { @@ -246,28 +494,64 @@ async fn info(config_path: &std::path::Path, hash: &str) -> Result<()> { println!("shards {}", m.shard_count()); println!("manifest v{}", m.version); - let mut healthy = 0usize; - let mut missing = 0usize; - for block in &m.blocks { - for shard in &block.shards { - if storage.chunk_store().has(&shard.hash).await { - healthy += 1; + // Where the shards live decides how to check them: probe the peers when + // the object was placed across the network, the local disk otherwise. + match open_distributed(config_path, false).await? { + Some((distributed, _, _)) => { + let health = distributed.health(&m).await?; + let placed: Vec<&str> = m.blocks[0] + .shards + .iter() + .filter_map(|s| s.host_hint.as_deref()) + .collect(); + if !placed.is_empty() { + println!("hosts {}", placed.join(", ")); + } + for h in &health { + println!( + "block {:<3} {} of {} shards present — {:?}{}", + h.index, + h.healthy, + h.total, + h.state, + if h.missing.is_empty() { + String::new() + } else { + format!(" (missing shards {:?})", h.missing) + } + ); + } + if health.iter().any(|h| h.state.needs_repair()) { + println!(" repair is CIP-005; not implemented yet"); + } + } + None => { + let mut healthy = 0usize; + let mut missing = 0usize; + for block in &m.blocks { + for shard in &block.shards { + if storage.chunk_store().has(&shard.hash).await { + healthy += 1; + } else { + missing += 1; + } + } + } + let state = if missing == 0 { + "healthy" + } else if missing <= m.parity as usize { + "degraded (readable)" } else { - missing += 1; + "LOST" + }; + println!("health {healthy} present, {missing} missing — {state}"); + if missing > 0 { + println!( + " repair is CIP-005; on a single node there is nowhere to repair from" + ); } } } - let state = if missing == 0 { - "healthy" - } else if missing <= m.parity as usize { - "degraded (readable)" - } else { - "LOST" - }; - println!("health {healthy} present, {missing} missing — {state}"); - if missing > 0 { - println!(" repair is CIP-005; on a single node there is nowhere to repair from"); - } Ok(()) } @@ -468,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-gateway/src/storage_api.rs b/node/crates/c0mpute-gateway/src/storage_api.rs index 34970c8..41f534c 100644 --- a/node/crates/c0mpute-gateway/src/storage_api.rs +++ b/node/crates/c0mpute-gateway/src/storage_api.rs @@ -353,7 +353,7 @@ async fn put_object( .storage .read_manifest(&object_hash) .await - .map_err(|e| ApiError::Internal(e.to_string()))?; + .map_err(|e| ApiError::Internal(format!("{e:#}")))?; return Ok((StatusCode::OK, axum::Json(manifest)).into_response()); } @@ -379,7 +379,7 @@ async fn put_object( .put_stream(stream, Some(object_hash), tier, Some(len)) .await .map_err(|e| { - let msg = e.to_string(); + let msg = format!("{e:#}"); if msg.contains("integrity failure") { ApiError::Unprocessable(msg) } else { @@ -429,7 +429,7 @@ async fn get_object( .storage .get_range_with(&manifest, offset, len) .await - .map_err(|e| ApiError::Internal(e.to_string()))?; + .map_err(|e| ApiError::Internal(format!("{e:#}")))?; let end = offset + bytes.len() as u64 - 1; return Ok(( StatusCode::PARTIAL_CONTENT, @@ -496,7 +496,7 @@ async fn delete_object( .storage .delete(&object_hash) .await - .map_err(|e| ApiError::Internal(e.to_string()))?; + .map_err(|e| ApiError::Internal(format!("{e:#}")))?; state.budget.release(cost); Ok(StatusCode::NO_CONTENT.into_response()) } @@ -518,7 +518,7 @@ async fn load_manifest(state: &StorageApiState, hash: &Hash) -> ApiResult, + /// Also keep a local copy of every shard. + /// + /// Off by default: it adds a full extra copy of the object to this node's + /// disk for no durability the placement does not already provide. + pub keep_local_copy: bool, +} + +pub struct DistributedStorage { + local: Storage, + transport: Arc, + catalog: Arc>, + config: DistributedConfig, +} + +impl DistributedStorage { + pub fn new( + local: Storage, + transport: Arc, + catalog: Arc>, + ) -> Self { + Self { + local, + transport, + catalog, + config: DistributedConfig::default(), + } + } + + pub fn with_config(mut self, config: DistributedConfig) -> Self { + self.config = config; + self + } + + pub fn local(&self) -> &Storage { + &self.local + } + + fn policy_for(&self, tier: Tier) -> PlacementPolicy { + self.config + .policy + .clone() + .unwrap_or_else(|| PlacementPolicy::for_parity(tier.parity())) + } + + /// Shards confirmed before a write is acknowledged. + /// + /// `k + ceil(parity/2)` — 12 of 14 for `standard`. Bounds write latency by + /// the twelfth-fastest peer rather than the slowest, while still leaving + /// the object readable if the two stragglers never land. The remaining + /// placements continue in the background. + fn write_quorum(tier: Tier) -> usize { + tier.k() + tier.parity().div_ceil(2) + } + + /// Store an object across the network. + pub async fn put(&self, data: &[u8], tier: Tier) -> Result { + let block_size = block_size_for(data.len() as u64); + let (k, parity) = (tier.k(), tier.parity()); + let policy = self.policy_for(tier); + let quorum = Self::write_quorum(tier); + + let chunks: Vec<&[u8]> = if data.is_empty() { + vec![&[]] + } else { + data.chunks(block_size as usize).collect() + }; + + let mut blocks = Vec::with_capacity(chunks.len()); + for (index, plaintext) in chunks.iter().enumerate() { + let entry = self + .place_block(plaintext, index as u32, k, parity, &policy, quorum) + .await + .with_context(|| format!("placing block {index}"))?; + blocks.push(entry); + } + + let manifest = ObjectManifest { + version: MANIFEST_VERSION, + object_hash: Hash::of(data), + original_len: data.len() as u64, + block_size, + k: k as u8, + parity: parity as u8, + tier, + blocks, + }; + self.local.write_manifest(&manifest).await?; + info!( + object_hash = %manifest.object_hash, + blocks = manifest.blocks.len(), + %tier, + "placed object across the network" + ); + Ok(manifest) + } + + /// Encode one block and place its shards. + /// + /// Peers are selected per block rather than per object, so a large object + /// spreads across the whole network instead of pinning every one of its + /// blocks to the same fourteen nodes. + async fn place_block( + &self, + plaintext: &[u8], + index: u32, + k: usize, + parity: usize, + policy: &PlacementPolicy, + quorum: usize, + ) -> Result { + let n = k + parity; + let (shards, _) = erasure::encode(plaintext, k, parity)?; + let shard_bytes = shards.first().map(|s| s.bytes.len()).unwrap_or(0) as u64; + + let assignments = { + let catalog = self.catalog.read().await; + select(catalog.peers(), n, shard_bytes, policy)? + }; + + // All n concurrently; acknowledge at quorum. + let mut inflight = FuturesUnordered::new(); + for (a, shard) in assignments.iter().zip(shards.iter()) { + let transport = Arc::clone(&self.transport); + let peer = a.peer.clone(); + let bytes = shard.bytes.clone(); + let hash = Hash::of(&bytes); + let shard_index = shard.index; + inflight.push(async move { + let res = transport.put_shard(&peer, &hash, &bytes).await; + (shard_index, hash, peer, res) + }); + } + + let mut entries: Vec = Vec::with_capacity(n); + let mut failures: Vec = Vec::new(); + while let Some((shard_index, hash, peer, res)) = inflight.next().await { + match res { + Ok(()) => entries.push(ShardEntry { + index: shard_index, + hash, + host_hint: Some(peer.peer_id), + }), + Err(e) => { + warn!(block = index, shard = shard_index, peer = %peer.peer_id, err = %e, + "shard placement failed"); + failures.push(format!("{}: {e}", peer.peer_id)); + } + } + } + + if entries.len() < quorum { + bail!( + "block {index}: only {} of {n} shards placed, need {quorum} for write quorum ({} failed: {})", + entries.len(), + failures.len(), + failures.join("; ") + ); + } + if entries.len() < n { + // Readable but under-replicated. CIP-005's repair loop is what + // restores full parity; recorded loudly so it is not invisible. + warn!( + block = index, + placed = entries.len(), + n, + "block placed below full redundancy; needs repair" + ); + } + + if self.config.keep_local_copy { + for s in &shards { + self.local.chunk_store().put(&s.bytes).await?; + } + } + + entries.sort_by_key(|e| e.index); + Ok(BlockEntry { + index, + len: plaintext.len() as u32, + hash: Hash::of(plaintext), + shards: entries, + }) + } + + /// Read an object back from the network. + pub async fn get(&self, object_hash: &Hash) -> Result> { + let manifest = self.local.read_manifest(object_hash).await?; + let mut out = Vec::with_capacity(manifest.original_len as usize); + for i in 0..manifest.blocks.len() { + out.extend_from_slice(&self.read_block(&manifest, i).await?); + } + let actual = Hash::of(&out); + if actual != manifest.object_hash { + bail!( + "object integrity failure: manifest says {} but decoded bytes hash to {actual}", + manifest.object_hash + ); + } + Ok(out) + } + + /// Reconstruct one block from whichever `k` peers answer first. + /// + /// All `n` are requested rather than a chosen `k`. That costs `n/k` (1.4x + /// for `standard`) in read bandwidth and buys the difference between + /// waiting for the k-th fastest peer and waiting for the slowest of a + /// chosen k — on a network of consumer nodes at 200–500 ms, an easy trade. + pub async fn read_block(&self, manifest: &ObjectManifest, index: usize) -> Result> { + let entry = manifest + .blocks + .get(index) + .ok_or_else(|| anyhow!("block {index} out of range"))?; + let k = manifest.k as usize; + let n = manifest.n(); + + let catalog = self.catalog.read().await; + let mut inflight = FuturesUnordered::new(); + let mut unreachable = Vec::new(); + + for shard in &entry.shards { + let Some(hint) = &shard.host_hint else { + // No hint: a locally-stored shard from a CIP-002 write. + continue; + }; + let Some(peer) = catalog.get(hint).cloned() else { + // The manifest is a cache, not the source of truth. Resolving + // a stale hint via Kad provider records is the CIP-003 design; + // until that lands, an unknown peer is simply skipped. + unreachable.push(hint.clone()); + continue; + }; + let transport = Arc::clone(&self.transport); + let hash = shard.hash; + let shard_index = shard.index; + inflight.push(async move { + let res = transport.get_shard(&peer, &hash).await; + (shard_index, hash, peer.peer_id, res) + }); + } + drop(catalog); + + let mut received: Vec> = vec![None; n]; + let mut found = 0usize; + let mut errors = Vec::new(); + + while let Some((shard_index, hash, peer_id, res)) = inflight.next().await { + match res { + Ok(bytes) => { + // Verify even though the transport does: a second + // implementation of ShardTransport might not, and a + // substituted shard silently corrupts the decode. + if Hash::of(&bytes) != hash { + errors.push(format!("{peer_id}: served the wrong bytes")); + continue; + } + received[shard_index as usize] = Some(Shard { + index: shard_index, + bytes, + }); + found += 1; + if found == k { + // Enough to reconstruct; drop the stragglers. + break; + } + } + Err(e) => errors.push(format!("{peer_id}: {e}")), + } + } + + // Fall back to any locally-held shards — how a CIP-002 object, or one + // written with keep_local_copy, still reads. + if found < k { + for shard in &entry.shards { + if received[shard.index as usize].is_some() { + continue; + } + if let Ok(bytes) = self.local.chunk_store().get(&shard.hash).await { + received[shard.index as usize] = Some(Shard { + index: shard.index, + bytes, + }); + found += 1; + if found == k { + break; + } + } + } + } + + if found < k { + bail!( + "block {index} of object {}: need {k} shards, got {found} \ + ({} peers unreachable, {} errors: {})", + manifest.object_hash, + unreachable.len(), + errors.len(), + errors.join("; ") + ); + } + + let mut plaintext = + erasure::decode(received, k, manifest.parity as usize, entry.len as usize)?; + plaintext.truncate(entry.len as usize); + + let actual = Hash::of(&plaintext); + if actual != entry.hash { + bail!( + "block {index} of object {}: integrity failure, manifest says {} but bytes hash to {actual}", + manifest.object_hash, + entry.hash + ); + } + debug!(block = index, found, k, "reconstructed block"); + Ok(plaintext) + } + + /// Per-block health, for `c0mpute storage info` and CIP-005's repair scan. + pub async fn health(&self, manifest: &ObjectManifest) -> Result> { + let catalog = self.catalog.read().await; + let mut out = Vec::with_capacity(manifest.blocks.len()); + + for block in &manifest.blocks { + let mut healthy = 0usize; + let mut missing = Vec::new(); + for shard in &block.shards { + let held = match shard.host_hint.as_ref().and_then(|h| catalog.get(h)) { + Some(peer) => self + .transport + .has_shards(peer, &[shard.hash]) + .await + .map(|v| v.first().copied().unwrap_or(false)) + .unwrap_or(false), + None => self.local.chunk_store().has(&shard.hash).await, + }; + if held { + healthy += 1; + } else { + missing.push(shard.index); + } + } + out.push(BlockHealth { + index: block.index, + healthy, + total: block.shards.len(), + missing, + state: BlockState::classify(healthy, manifest.k as usize, manifest.parity as usize), + }); + } + Ok(out) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BlockHealth { + pub index: u32, + pub healthy: usize, + pub total: usize, + pub missing: Vec, + pub state: BlockState, +} + +/// CIP-005's classification, thresholds relative to `(k, parity)` rather than +/// absolute. Repair triggers as soon as half the parity budget is spent — +/// waiting until `k+1` would be cheaper, and is what Storj's much wider code +/// affords them, but CIP-001 spent that margin on the cost advantage. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BlockState { + Healthy, + Degraded, + Urgent, + Critical, + Lost, +} + +impl BlockState { + pub fn classify(healthy: usize, k: usize, parity: usize) -> Self { + let n = k + parity; + if healthy >= n { + BlockState::Healthy + } else if healthy < k { + BlockState::Lost + } else if healthy == k { + // One more loss and the block is gone. + BlockState::Critical + } else if healthy <= k + parity / 4 { + BlockState::Urgent + } else { + // Any shard missing at all is a repair trigger: CIP-001 bought the + // cost advantage by spending the durability margin Storj keeps, so + // we cannot wait until the parity budget is nearly gone. + BlockState::Degraded + } + } + + pub fn needs_repair(self) -> bool { + !matches!(self, BlockState::Healthy) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn write_quorum_is_k_plus_half_the_parity() { + assert_eq!(DistributedStorage::write_quorum(Tier::Standard), 12); // 10 + 2 + assert_eq!(DistributedStorage::write_quorum(Tier::Critical), 26); // 20 + 6 + assert_eq!(DistributedStorage::write_quorum(Tier::Hot), 2); // 1 + 1 + } + + #[test] + fn block_state_classifies_against_k_and_parity() { + // RS 10/14 + assert_eq!(BlockState::classify(14, 10, 4), BlockState::Healthy); + assert_eq!(BlockState::classify(13, 10, 4), BlockState::Degraded); + assert_eq!(BlockState::classify(12, 10, 4), BlockState::Degraded); + assert_eq!(BlockState::classify(11, 10, 4), BlockState::Urgent); + assert_eq!(BlockState::classify(10, 10, 4), BlockState::Critical); + assert_eq!(BlockState::classify(9, 10, 4), BlockState::Lost); + } + + #[test] + fn only_healthy_blocks_skip_repair() { + assert!(!BlockState::Healthy.needs_repair()); + for s in [ + BlockState::Degraded, + BlockState::Urgent, + BlockState::Critical, + BlockState::Lost, + ] { + assert!(s.needs_repair(), "{s:?} should need repair"); + } + } +} diff --git a/node/crates/c0mpute-placement/src/lib.rs b/node/crates/c0mpute-placement/src/lib.rs new file mode 100644 index 0000000..59c6d40 --- /dev/null +++ b/node/crates/c0mpute-placement/src/lib.rs @@ -0,0 +1,44 @@ +//! Cross-node shard placement for c0mpute storage (CIP-003). +//! +//! CIP-002 made the storage engine reachable over HTTP, but every shard still +//! landed on one disk — which means the erasure coding was pure overhead with +//! no durability behind it. This crate spreads a block's `n` shards across `n` +//! peers chosen for reputation and failure-domain diversity, and reads them +//! back from whichever `k` answer first. +//! +//! Three pieces: +//! +//! - [`peer`] — who the storage peers are, and which failure domain each +//! belongs to. +//! - [`select`] — choosing `n` of them under CIP-001's durability rules. +//! Pure; no network I/O, because a slow peer lookup must never become a +//! slow write. +//! - [`transport`] — moving shard bytes. HTTP against the CIP-002 endpoints +//! today; the libp2p `/c0mpute/shard/1.0.0` protocol becomes a second +//! implementation of the same trait. +//! +//! and [`distributed::DistributedStorage`], which composes them. +//! +//! The load-bearing decision is that placement **fails loudly** when the +//! network cannot satisfy the diversity policy. CIP-001's durability figures +//! assume shard hosts fail independently; fourteen shards behind one ISP are +//! one sample, not fourteen, and nothing downstream can detect that the +//! assumption was broken. A write that cannot be made durable is an error, +//! not a warning. + +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 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/peer.rs b/node/crates/c0mpute-placement/src/peer.rs new file mode 100644 index 0000000..2d73c41 --- /dev/null +++ b/node/crates/c0mpute-placement/src/peer.rs @@ -0,0 +1,224 @@ +//! Storage peers and the failure domains they belong to (CIP-003). + +use serde::{Deserialize, Serialize}; + +/// The unit of correlated failure. +/// +/// CIP-001's durability arithmetic treats shard hosts as independent samples. +/// Two peers in the same autonomous system share an operator, a transit +/// provider and often a building, so they are one sample wearing two hats. +/// Grouping by domain is what keeps the arithmetic honest. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub enum FailureDomain { + /// Best signal: the peer's autonomous system. + Asn(u32), + /// Fallback when the ASN is unknown. Weaker — two ASNs can share a + /// prefix's neighbourhood and one ASN can span many prefixes — but it is + /// never *wrong*, only coarse. + IpPrefix(String), + /// Nothing is known about where this peer sits. + Unknown, +} + +/// A peer that might hold shards. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct PeerInfo { + pub peer_id: String, + /// Base URL of the peer's storage API, e.g. `http://1.2.3.4:7780`. + pub endpoint: String, + /// From `c0mpute_verify::reputation`. + pub reputation: f32, + /// Fraction of the last 30 days the peer was reachable. + pub uptime_30d: f32, + /// Bytes the peer will still accept. + pub free_bytes: u64, + pub rtt_ms: u32, + pub asn: Option, + pub region: Option, + /// Network prefix (e.g. `"203.0.113"`), used when `asn` is unknown. + pub ip_prefix: Option, +} + +impl PeerInfo { + /// Which failure domain this peer counts against. + /// + /// ASN first, IP prefix second, `Unknown` last. Region is deliberately not + /// part of the identity: it is far coarser than an ASN, so folding it in + /// would let two peers in one datacenter look like two domains simply + /// because their operators labelled them differently. + pub fn domain(&self) -> FailureDomain { + match (self.asn, &self.ip_prefix) { + (Some(asn), _) => FailureDomain::Asn(asn), + (None, Some(prefix)) => FailureDomain::IpPrefix(prefix.clone()), + (None, None) => FailureDomain::Unknown, + } + } + + /// Derive an IP prefix from a hostname or `host:port`, when it is a + /// literal IPv4 address. A DNS name tells us nothing without resolving it, + /// and resolving here would make selection do network I/O. + pub fn prefix_from_endpoint(endpoint: &str) -> Option { + let after_scheme = match endpoint.split_once("://") { + Some((_, rest)) => rest, + None => endpoint, + }; + let host_port = after_scheme.split('/').next()?; + // Only strip a trailing `:port`, never an IPv6 colon. + let host = match host_port.rsplit_once(':') { + Some((h, port)) if !port.is_empty() && port.bytes().all(|b| b.is_ascii_digit()) => h, + _ => host_port, + }; + let host = host.trim_start_matches('[').trim_end_matches(']'); + + let octets: Vec<&str> = host.split('.').collect(); + if octets.len() == 4 && octets.iter().copied().all(|o| o.parse::().is_ok()) { + // /24-equivalent. Coarse on purpose: the point is to catch "these + // are obviously the same rack", not to model routing. + return Some(format!("{}.{}.{}", octets[0], octets[1], octets[2])); + } + None + } +} + +/// What this node currently believes about its storage peers. +/// +/// Populated from gossipsub capability ads today; CIP-006's challenge results +/// will feed `reputation` and `uptime_30d` once those exist. Deliberately a +/// plain snapshot — selection must not do network I/O, or a slow peer lookup +/// becomes a slow write. +#[derive(Clone, Debug, Default)] +pub struct PeerCatalog { + peers: Vec, +} + +impl PeerCatalog { + pub fn new(peers: Vec) -> Self { + Self { peers } + } + + pub fn peers(&self) -> &[PeerInfo] { + &self.peers + } + + pub fn len(&self) -> usize { + self.peers.len() + } + + pub fn is_empty(&self) -> bool { + self.peers.is_empty() + } + + pub fn upsert(&mut self, peer: PeerInfo) { + match self.peers.iter_mut().find(|p| p.peer_id == peer.peer_id) { + Some(existing) => *existing = peer, + None => self.peers.push(peer), + } + } + + pub fn remove(&mut self, peer_id: &str) { + self.peers.retain(|p| p.peer_id != peer_id); + } + + pub fn get(&self, peer_id: &str) -> Option<&PeerInfo> { + self.peers.iter().find(|p| p.peer_id == peer_id) + } + + /// How many distinct failure domains the catalog spans. The headline + /// number for whether this network can store anything durably at all. + pub fn domain_count(&self) -> usize { + self.peers + .iter() + .map(|p| p.domain()) + .collect::>() + .len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn peer(id: &str) -> PeerInfo { + PeerInfo { + peer_id: id.into(), + endpoint: "http://10.0.0.1:7780".into(), + reputation: 0.95, + uptime_30d: 0.995, + free_bytes: 1 << 30, + rtt_ms: 20, + asn: None, + region: None, + ip_prefix: None, + } + } + + #[test] + fn domain_prefers_asn_then_prefix_then_unknown() { + let mut p = peer("a"); + assert_eq!(p.domain(), FailureDomain::Unknown); + + p.ip_prefix = Some("203.0.113".into()); + assert_eq!(p.domain(), FailureDomain::IpPrefix("203.0.113".into())); + + p.asn = Some(64512); + assert_eq!(p.domain(), FailureDomain::Asn(64512), "ASN should win"); + } + + /// Region is intentionally not part of domain identity — it is coarser + /// than an ASN and would make one datacenter look like several domains. + #[test] + fn region_does_not_affect_domain_identity() { + let mut a = peer("a"); + let mut b = peer("b"); + a.asn = Some(7); + b.asn = Some(7); + a.region = Some("us-east".into()); + b.region = Some("eu-west".into()); + assert_eq!(a.domain(), b.domain()); + } + + #[test] + fn prefix_from_endpoint_handles_the_shapes_we_see() { + assert_eq!( + PeerInfo::prefix_from_endpoint("http://203.0.113.42:7780"), + Some("203.0.113".into()) + ); + assert_eq!( + PeerInfo::prefix_from_endpoint("https://198.51.100.7/storage"), + Some("198.51.100".into()) + ); + // A DNS name tells us nothing without resolving, and selection must + // not do network I/O. + assert_eq!( + PeerInfo::prefix_from_endpoint("http://node.example.com:7780"), + None + ); + assert_eq!(PeerInfo::prefix_from_endpoint("http://999.1.1.1"), None); + } + + #[test] + fn catalog_upsert_replaces_rather_than_duplicating() { + let mut c = PeerCatalog::default(); + c.upsert(peer("a")); + c.upsert(peer("a")); + assert_eq!(c.len(), 1); + + let mut updated = peer("a"); + updated.free_bytes = 42; + c.upsert(updated); + assert_eq!(c.len(), 1); + assert_eq!(c.get("a").unwrap().free_bytes, 42); + } + + #[test] + fn domain_count_is_the_networks_real_capacity_for_durability() { + let mut c = PeerCatalog::default(); + for i in 0..10 { + let mut p = peer(&format!("p{i}")); + p.asn = Some(if i < 6 { 100 } else { 200 }); + c.upsert(p); + } + assert_eq!(c.len(), 10); + assert_eq!(c.domain_count(), 2, "ten peers, two places they can fail"); + } +} 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 new file mode 100644 index 0000000..ca038ac --- /dev/null +++ b/node/crates/c0mpute-placement/src/select.rs @@ -0,0 +1,527 @@ +//! Peer selection for shard placement (CIP-003). +//! +//! Choosing `n` peers for a block is where the durability model in CIP-001 +//! actually lives. Those availability figures assume shard hosts fail +//! *independently*; fourteen shards behind one ISP are not fourteen +//! independent samples, and nothing downstream can detect that the assumption +//! was violated. So this module fails loudly rather than placing badly. + +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +use crate::peer::{FailureDomain, PeerInfo}; + +/// Rules a placement must satisfy. Defaults come from CIP-001. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct PlacementPolicy { + /// Minimum `c0mpute-verify::reputation`. + pub min_reputation: f32, + /// Minimum 30-day uptime. CIP-001's durability table collapses from ~6.7 + /// nines to ~3.4 between 0.99 and 0.95, so this gate is doing more work + /// than the parity count is. + pub min_uptime_30d: f32, + /// Most shards of one block allowed in a single failure domain. + /// `floor(parity / 2)` by default: half the parity budget can be lost to + /// one ISP or region going dark, and the object still reads. + pub max_per_domain: usize, + /// Whether peers whose failure domain could not be determined may be used. + /// + /// When `false` (the default) they are excluded. When `true` they are all + /// treated as members of a *single* shared domain, which is the + /// conservative reading — the alternative, giving each unknown peer its + /// own domain, would let fourteen unlocatable peers satisfy every + /// constraint while providing no real diversity at all. + pub allow_unknown_domain: bool, +} + +impl PlacementPolicy { + /// The policy for a tier's `(k, parity)`. + pub fn for_parity(parity: usize) -> Self { + Self { + min_reputation: 0.9, + min_uptime_30d: 0.99, + max_per_domain: (parity / 2).max(1), + allow_unknown_domain: false, + } + } + + /// How many distinct failure domains a placement of `n` shards needs. + pub fn domains_required(&self, n: usize) -> usize { + n.div_ceil(self.max_per_domain) + } +} + +#[derive(Debug, thiserror::Error, PartialEq)] +pub enum PlacementError { + #[error( + "not enough eligible peers: need {needed}, found {eligible} \ + (of {total} known; {below_bar} below reputation {min_reputation} or uptime {min_uptime}, \ + {too_full} without room for a {shard_bytes}-byte shard, {unknown_domain} with an unknown failure domain)" + )] + InsufficientPeers { + needed: usize, + eligible: usize, + total: usize, + below_bar: usize, + too_full: usize, + unknown_domain: usize, + min_reputation: f32, + min_uptime: f32, + shard_bytes: u64, + }, + #[error( + "failure-domain diversity unsatisfiable: {needed} shards at most {max_per_domain} \ + per domain needs {domains_required} distinct domains, but only {domains_available} \ + are available ({placed} shards could be placed)" + )] + DiversityUnsatisfiable { + needed: usize, + placed: usize, + max_per_domain: usize, + domains_required: usize, + domains_available: usize, + }, +} + +/// One shard assigned to one peer. +#[derive(Clone, Debug, PartialEq)] +pub struct Assignment { + pub shard_index: u8, + pub peer: PeerInfo, +} + +/// Rank a peer. Higher is better. +/// +/// Reputation and uptime dominate because they are what the durability model +/// is sensitive to; latency is a weak tiebreak, deliberately. Preferring fast +/// peers too strongly would concentrate placement on whichever few nodes are +/// nearest, which is the opposite of what diversity is for. +/// +/// CIP-003 sketched this as `reputation * uptime * (1 / (1 + rtt/100))`. That +/// weighting does not match the intent: it makes a 400 ms peer score 20% below +/// a 1 ms one, so a fast flaky node outranks a slow reliable one — exactly the +/// trade CIP-001 says not to make, since availability drives durability and +/// latency does not. The latency term is therefore scaled into a narrow band: +/// it separates otherwise-equal peers and cannot overturn a reputation gap. +pub fn score(peer: &PeerInfo) -> f32 { + let latency_factor = 1.0 / (1.0 + peer.rtt_ms as f32 / 100.0); + 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 +/// heuristic: "at most `max_per_domain` from each domain" is a partition +/// matroid, and greedy is optimal over a matroid. So if this returns +/// `DiversityUnsatisfiable`, no other assignment would have worked either — +/// there is no need to backtrack, and no better answer being missed. +pub fn select( + candidates: &[PeerInfo], + n: usize, + 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; + let mut unknown_domain = 0usize; + + 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; + } + if p.free_bytes < shard_bytes { + too_full += 1; + continue; + } + if matches!(p.domain(), FailureDomain::Unknown) && !policy.allow_unknown_domain { + unknown_domain += 1; + continue; + } + eligible.push(p); + } + + if eligible.len() < n { + return Err(PlacementError::InsufficientPeers { + needed: n, + eligible: eligible.len(), + total, + below_bar, + too_full, + unknown_domain, + min_reputation: policy.min_reputation, + min_uptime: policy.min_uptime_30d, + shard_bytes, + }); + } + + // Best first. `total_cmp` rather than `partial_cmp().unwrap()`: a NaN + // score from a malformed peer record should sort, not panic. + eligible.sort_by(|a, b| score(b).total_cmp(&score(a))); + + // 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 { + break; + } + let domain = p.domain(); + let used = per_domain.entry(domain).or_insert(0); + if *used >= policy.max_per_domain { + continue; + } + *used += 1; + chosen.push(p); + } + + if chosen.len() < n { + let domains_available = eligible + .iter() + .map(|p| p.domain()) + .collect::>() + .len(); + return Err(PlacementError::DiversityUnsatisfiable { + needed: n, + placed: chosen.len(), + max_per_domain: policy.max_per_domain, + domains_required: policy.domains_required(n), + domains_available, + }); + } + + Ok(chosen.into_iter().cloned().collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn peer(id: &str, asn: Option, rep: f32, uptime: f32) -> PeerInfo { + PeerInfo { + peer_id: id.to_string(), + endpoint: format!("http://{id}.test"), + reputation: rep, + uptime_30d: uptime, + free_bytes: 1 << 30, + rtt_ms: 50, + asn, + region: None, + ip_prefix: None, + } + } + + /// n peers, each in its own ASN — the easy case. + fn diverse(n: usize) -> Vec { + (0..n) + .map(|i| peer(&format!("p{i}"), Some(1000 + i as u32), 0.95, 0.995)) + .collect() + } + + #[test] + fn selects_n_distinct_peers() { + let policy = PlacementPolicy::for_parity(4); + let got = select(&diverse(20), 14, 1024, &policy).unwrap(); + assert_eq!(got.len(), 14); + + let ids: std::collections::HashSet<_> = got.iter().map(|a| &a.peer.peer_id).collect(); + assert_eq!(ids.len(), 14, "same peer used twice for one block"); + + let idx: Vec = got.iter().map(|a| a.shard_index).collect(); + assert_eq!(idx, (0..14).collect::>()); + } + + #[test] + fn rejects_peers_below_the_reputation_bar() { + let policy = PlacementPolicy::for_parity(4); + let mut peers = diverse(14); + for p in peers.iter_mut().take(3) { + p.reputation = 0.5; + } + let err = select(&peers, 14, 1024, &policy).unwrap_err(); + match err { + PlacementError::InsufficientPeers { + eligible, + below_bar, + .. + } => { + assert_eq!(eligible, 11); + assert_eq!(below_bar, 3); + } + other => panic!("unexpected: {other}"), + } + } + + /// CIP-001: RS 10/14 drops from ~6.7 nines to ~3.4 between 0.99 and 0.95 + /// per-node availability, so the uptime gate is load-bearing. + #[test] + fn rejects_peers_below_the_uptime_bar() { + let policy = PlacementPolicy::for_parity(4); + let mut peers = diverse(16); + for p in peers.iter_mut().take(5) { + p.uptime_30d = 0.95; + } + assert!(select(&peers, 14, 1024, &policy).is_err()); + } + + #[test] + fn rejects_peers_without_room() { + let policy = PlacementPolicy::for_parity(4); + let mut peers = diverse(15); + for p in peers.iter_mut().take(4) { + p.free_bytes = 10; + } + let err = select(&peers, 14, 1_000_000, &policy).unwrap_err(); + assert!(matches!( + err, + PlacementError::InsufficientPeers { too_full: 4, .. } + )); + } + + /// The constraint that actually matters. Plenty of healthy peers, but they + /// are all behind two ASNs, so a placement would be 14 correlated samples + /// wearing the costume of 14 independent ones. + #[test] + fn refuses_to_place_without_failure_domain_diversity() { + let policy = PlacementPolicy::for_parity(4); + let peers: Vec = (0..20) + .map(|i| { + peer( + &format!("p{i}"), + Some(if i < 10 { 100 } else { 200 }), + 0.95, + 0.995, + ) + }) + .collect(); + + let err = select(&peers, 14, 1024, &policy).unwrap_err(); + match err { + PlacementError::DiversityUnsatisfiable { + placed, + domains_required, + domains_available, + max_per_domain, + .. + } => { + assert_eq!(max_per_domain, 2); + assert_eq!(placed, 4, "2 domains x 2 per domain"); + assert_eq!(domains_required, 7); + assert_eq!(domains_available, 2); + } + other => panic!("unexpected: {other}"), + } + } + + #[test] + fn caps_shards_per_domain() { + let policy = PlacementPolicy::for_parity(4); + // 7 ASNs, 4 peers each: enough peers and exactly enough domains. + let peers: Vec = (0..28) + .map(|i| peer(&format!("p{i}"), Some(100 + (i as u32 % 7)), 0.95, 0.995)) + .collect(); + let got = select(&peers, 14, 1024, &policy).unwrap(); + + let mut per_asn: HashMap = HashMap::new(); + for a in &got { + *per_asn.entry(a.peer.asn.unwrap()).or_default() += 1; + } + assert_eq!(per_asn.len(), 7); + for (asn, count) in per_asn { + assert!(count <= 2, "asn {asn} got {count} shards, cap is 2"); + } + } + + /// Unknown-domain peers share one domain rather than each getting their + /// own — otherwise a network of unlocatable peers would satisfy every + /// constraint while providing no diversity. + #[test] + fn unknown_domains_are_one_domain_not_many() { + let mut policy = PlacementPolicy::for_parity(4); + policy.allow_unknown_domain = true; + let peers: Vec = (0..20) + .map(|i| peer(&format!("p{i}"), None, 0.95, 0.995)) + .collect(); + + let err = select(&peers, 14, 1024, &policy).unwrap_err(); + assert!(matches!( + err, + PlacementError::DiversityUnsatisfiable { + placed: 2, + domains_available: 1, + .. + } + )); + } + + #[test] + fn unknown_domain_peers_are_excluded_by_default() { + let policy = PlacementPolicy::for_parity(4); + assert!(!policy.allow_unknown_domain); + let peers: Vec = (0..20) + .map(|i| peer(&format!("p{i}"), None, 0.95, 0.995)) + .collect(); + let err = select(&peers, 14, 1024, &policy).unwrap_err(); + assert!(matches!( + err, + PlacementError::InsufficientPeers { + unknown_domain: 20, + .. + } + )); + } + + /// Falls back to IP-prefix diversity when the ASN is unknown, which is + /// weaker than ASN but never wrong. + #[test] + fn ip_prefix_substitutes_for_an_unknown_asn() { + let policy = PlacementPolicy::for_parity(4); + let peers: Vec = (0..14) + .map(|i| { + let mut p = peer(&format!("p{i}"), None, 0.95, 0.995); + p.ip_prefix = Some(format!("10.{i}")); + p + }) + .collect(); + assert_eq!(select(&peers, 14, 1024, &policy).unwrap().len(), 14); + } + + #[test] + fn prefers_higher_scoring_peers() { + let policy = PlacementPolicy::for_parity(2); // hot: max 1 per domain + let mut peers = diverse(6); + peers[3].reputation = 1.0; + peers[3].rtt_ms = 5; + let got = select(&peers, 3, 1024, &policy).unwrap(); + assert!( + got.iter().any(|a| a.peer.peer_id == "p3"), + "the best peer should have been chosen" + ); + } + + /// Latency is a tiebreak, not a driver: a fast but unreliable peer must + /// not outrank a slower, more available one. + #[test] + fn reputation_outweighs_latency() { + let fast_flaky = PeerInfo { + rtt_ms: 1, + reputation: 0.90, + ..peer("fast", Some(1), 0.90, 0.99) + }; + let slow_solid = PeerInfo { + rtt_ms: 400, + reputation: 1.0, + ..peer("slow", Some(2), 1.0, 1.0) + }; + assert!(score(&slow_solid) > score(&fast_flaky)); + } + + #[test] + fn hot_tier_needs_three_domains() { + let policy = PlacementPolicy::for_parity(2); + assert_eq!(policy.max_per_domain, 1); + assert_eq!(policy.domains_required(3), 3); + + let two_domains: Vec = (0..6) + .map(|i| { + peer( + &format!("p{i}"), + Some(if i < 3 { 1 } else { 2 }), + 0.95, + 0.995, + ) + }) + .collect(); + assert!(select(&two_domains, 3, 1024, &policy).is_err()); + assert!(select(&diverse(3), 3, 1024, &policy).is_ok()); + } + + #[test] + fn nan_scores_do_not_panic() { + let policy = PlacementPolicy { + min_reputation: 0.0, + min_uptime_30d: 0.0, + ..PlacementPolicy::for_parity(4) + }; + let mut peers = diverse(14); + peers[2].reputation = f32::NAN; + // Whatever the ordering, it must not panic. + let _ = select(&peers, 14, 1024, &policy); + } + + #[test] + fn error_message_names_what_is_missing() { + let policy = PlacementPolicy::for_parity(4); + let peers: Vec = (0..20) + .map(|i| peer(&format!("p{i}"), Some(100 + (i as u32 % 3)), 0.95, 0.995)) + .collect(); + let msg = select(&peers, 14, 1024, &policy).unwrap_err().to_string(); + assert!(msg.contains("7 distinct domains"), "unhelpful: {msg}"); + assert!(msg.contains("only 3"), "unhelpful: {msg}"); + } +} diff --git a/node/crates/c0mpute-placement/src/transport.rs b/node/crates/c0mpute-placement/src/transport.rs new file mode 100644 index 0000000..3800da9 --- /dev/null +++ b/node/crates/c0mpute-placement/src/transport.rs @@ -0,0 +1,339 @@ +//! Moving shards between nodes (CIP-003). +//! +//! Placement is written against this trait rather than against libp2p so that +//! the two can land independently. The HTTP implementation talks to the +//! CIP-002 `/storage/v1/shards/...` endpoints, which already exist and +//! already verify what they are given — so cross-node placement works today, +//! and the libp2p `/c0mpute/shard/1.0.0` protocol becomes a second +//! implementation of the same three methods rather than a prerequisite. + +use anyhow::{Result, anyhow, bail}; +use async_trait::async_trait; +use c0mpute_proto::Hash; + +use crate::peer::PeerInfo; + +#[async_trait] +pub trait ShardTransport: Send + Sync { + /// Store one shard on `peer`. The peer re-hashes and rejects a mismatch, + /// so a corrupted transfer fails at the receiver rather than silently + /// becoming a bad shard. + async fn put_shard(&self, peer: &PeerInfo, hash: &Hash, bytes: &[u8]) -> Result<()>; + + /// Fetch one shard from `peer`. The caller re-hashes; never trust a peer. + async fn get_shard(&self, peer: &PeerInfo, hash: &Hash) -> Result>; + + /// Does `peer` still hold these shards? Used by repair (CIP-005) to spot + /// degraded blocks without transferring anything. + async fn has_shards(&self, peer: &PeerInfo, hashes: &[Hash]) -> Result>; +} + +/// Talks to a peer's CIP-002 storage API. +pub struct HttpTransport { + client: reqwest::Client, +} + +impl HttpTransport { + pub fn new(timeout: std::time::Duration) -> Result { + Ok(Self { + client: reqwest::Client::builder().timeout(timeout).build()?, + }) + } + + fn shard_url(peer: &PeerInfo, hash: &Hash) -> String { + format!( + "{}/storage/v1/shards/{}", + peer.endpoint.trim_end_matches('/'), + hash.to_hex() + ) + } +} + +impl Default for HttpTransport { + fn default() -> Self { + Self::new(std::time::Duration::from_secs(30)).expect("default reqwest client builds") + } +} + +#[async_trait] +impl ShardTransport for HttpTransport { + async fn put_shard(&self, peer: &PeerInfo, hash: &Hash, bytes: &[u8]) -> Result<()> { + let resp = self + .client + .put(Self::shard_url(peer, hash)) + .body(bytes.to_vec()) + .send() + .await?; + let status = resp.status(); + // 200 means the peer already held it — content-addressed dedup, not a + // failure. + if status.is_success() { + return Ok(()); + } + let body = resp.text().await.unwrap_or_default(); + bail!( + "peer {} rejected shard {hash}: {status} {body}", + peer.peer_id + ) + } + + async fn get_shard(&self, peer: &PeerInfo, hash: &Hash) -> Result> { + let resp = self.client.get(Self::shard_url(peer, hash)).send().await?; + if !resp.status().is_success() { + bail!( + "peer {} has no shard {hash}: {}", + peer.peer_id, + resp.status() + ); + } + let bytes = resp.bytes().await?.to_vec(); + + // Verify before returning. A peer that serves the wrong bytes must not + // be able to poison a reconstruction — with k of n shards there is no + // downstream check that would catch a substituted shard until the + // whole block fails its hash, and then we would not know which peer + // did it. + let actual = Hash::of(&bytes); + if actual != *hash { + return Err(anyhow!( + "peer {} served bytes hashing to {actual}, not {hash}", + peer.peer_id + )); + } + Ok(bytes) + } + + async fn has_shards(&self, peer: &PeerInfo, hashes: &[Hash]) -> Result> { + // One HEAD per shard. CIP-003 specifies a batched `Have` probe; over + // HTTP that would need a new endpoint, so this is the honest version + // until the libp2p transport lands with real batching. Fine at CIP-003 + // scale, too chatty for CIP-005's hourly scan of millions of blocks. + let mut out = Vec::with_capacity(hashes.len()); + for h in hashes { + let held = match self.client.head(Self::shard_url(peer, h)).send().await { + Ok(r) => r.status().is_success(), + Err(_) => false, + }; + out.push(held); + } + Ok(out) + } +} + +#[cfg(any(test, feature = "testing"))] +pub mod memory { + //! In-process transport with fault injection, for tests. + + use std::collections::{HashMap, HashSet}; + use std::sync::{Arc, Mutex}; + + use super::*; + + #[derive(Default)] + struct Inner { + /// peer_id -> shard hash -> bytes + held: HashMap>>, + /// Peers that reject every operation, simulating an unreachable node. + offline: HashSet, + /// Peers that serve corrupted bytes, simulating a dishonest node. + corrupt: HashSet, + put_calls: usize, + get_calls: usize, + } + + #[derive(Clone, Default)] + pub struct MemoryTransport { + inner: Arc>, + } + + impl MemoryTransport { + pub fn new() -> Self { + Self::default() + } + + pub fn take_offline(&self, peer_id: &str) { + self.inner.lock().unwrap().offline.insert(peer_id.into()); + } + + pub fn bring_online(&self, peer_id: &str) { + self.inner.lock().unwrap().offline.remove(peer_id); + } + + pub fn make_corrupt(&self, peer_id: &str) { + self.inner.lock().unwrap().corrupt.insert(peer_id.into()); + } + + /// How many shards a peer is holding. + pub fn shard_count(&self, peer_id: &str) -> usize { + self.inner + .lock() + .unwrap() + .held + .get(peer_id) + .map(|m| m.len()) + .unwrap_or(0) + } + + pub fn peers_holding(&self, hash: &Hash) -> Vec { + let inner = self.inner.lock().unwrap(); + let key = hash.to_hex(); + let mut out: Vec = inner + .held + .iter() + .filter(|(_, m)| m.contains_key(&key)) + .map(|(p, _)| p.clone()) + .collect(); + out.sort(); + out + } + + pub fn get_calls(&self) -> usize { + self.inner.lock().unwrap().get_calls + } + + pub fn put_calls(&self) -> usize { + self.inner.lock().unwrap().put_calls + } + } + + #[async_trait] + impl ShardTransport for MemoryTransport { + async fn put_shard(&self, peer: &PeerInfo, hash: &Hash, bytes: &[u8]) -> Result<()> { + let mut inner = self.inner.lock().unwrap(); + inner.put_calls += 1; + if inner.offline.contains(&peer.peer_id) { + bail!("peer {} is offline", peer.peer_id); + } + inner + .held + .entry(peer.peer_id.clone()) + .or_default() + .insert(hash.to_hex(), bytes.to_vec()); + Ok(()) + } + + async fn get_shard(&self, peer: &PeerInfo, hash: &Hash) -> Result> { + let mut inner = self.inner.lock().unwrap(); + inner.get_calls += 1; + if inner.offline.contains(&peer.peer_id) { + bail!("peer {} is offline", peer.peer_id); + } + let corrupt = inner.corrupt.contains(&peer.peer_id); + let bytes = inner + .held + .get(&peer.peer_id) + .and_then(|m| m.get(&hash.to_hex())) + .cloned() + .ok_or_else(|| anyhow!("peer {} has no shard {hash}", peer.peer_id))?; + + if corrupt { + let mut bad = bytes.clone(); + if let Some(b) = bad.first_mut() { + *b = b.wrapping_add(1); + } + let actual = Hash::of(&bad); + return Err(anyhow!( + "peer {} served bytes hashing to {actual}, not {hash}", + peer.peer_id + )); + } + Ok(bytes) + } + + async fn has_shards(&self, peer: &PeerInfo, hashes: &[Hash]) -> Result> { + let inner = self.inner.lock().unwrap(); + if inner.offline.contains(&peer.peer_id) { + return Ok(vec![false; hashes.len()]); + } + let held = inner.held.get(&peer.peer_id); + Ok(hashes + .iter() + .map(|h| held.is_some_and(|m| m.contains_key(&h.to_hex()))) + .collect()) + } + } +} + +#[cfg(test)] +mod tests { + use super::memory::MemoryTransport; + use super::*; + + fn peer(id: &str) -> PeerInfo { + PeerInfo { + peer_id: id.into(), + endpoint: format!("http://{id}:7780"), + reputation: 1.0, + uptime_30d: 1.0, + free_bytes: 1 << 30, + rtt_ms: 10, + asn: Some(1), + region: None, + ip_prefix: None, + } + } + + #[test] + fn shard_url_is_the_cip_002_endpoint() { + let p = peer("a"); + let h = Hash::of(b"x"); + assert_eq!( + HttpTransport::shard_url(&p, &h), + format!("http://a:7780/storage/v1/shards/{}", h.to_hex()) + ); + } + + #[test] + fn shard_url_tolerates_a_trailing_slash() { + let mut p = peer("a"); + p.endpoint = "http://a:7780/".into(); + let h = Hash::of(b"x"); + assert!(!HttpTransport::shard_url(&p, &h).contains("//storage")); + } + + #[tokio::test] + async fn memory_transport_round_trips() { + let t = MemoryTransport::new(); + let p = peer("a"); + let bytes = b"shard bytes".to_vec(); + let h = Hash::of(&bytes); + + t.put_shard(&p, &h, &bytes).await.unwrap(); + assert_eq!(t.get_shard(&p, &h).await.unwrap(), bytes); + assert_eq!(t.has_shards(&p, &[h]).await.unwrap(), vec![true]); + assert_eq!(t.peers_holding(&h), vec!["a".to_string()]); + } + + #[tokio::test] + async fn offline_peers_fail_both_ways() { + let t = MemoryTransport::new(); + let p = peer("a"); + let bytes = b"x".to_vec(); + let h = Hash::of(&bytes); + t.put_shard(&p, &h, &bytes).await.unwrap(); + + t.take_offline("a"); + assert!(t.get_shard(&p, &h).await.is_err()); + assert!(t.put_shard(&p, &h, &bytes).await.is_err()); + assert_eq!(t.has_shards(&p, &[h]).await.unwrap(), vec![false]); + + t.bring_online("a"); + assert_eq!(t.get_shard(&p, &h).await.unwrap(), bytes); + } + + /// A dishonest peer must be caught at the transport, not at block decode — + /// by then we would know the block is wrong but not who broke it. + #[tokio::test] + async fn corrupt_peers_are_detected_at_the_transport() { + let t = MemoryTransport::new(); + let p = peer("a"); + let bytes = b"honest bytes".to_vec(); + let h = Hash::of(&bytes); + t.put_shard(&p, &h, &bytes).await.unwrap(); + + t.make_corrupt("a"); + let err = t.get_shard(&p, &h).await.unwrap_err().to_string(); + assert!(err.contains("served bytes hashing to"), "unexpected: {err}"); + assert!(err.contains('a'), "the error should name the peer: {err}"); + } +} diff --git a/node/crates/c0mpute-placement/tests/distributed.rs b/node/crates/c0mpute-placement/tests/distributed.rs new file mode 100644 index 0000000..0df97fc --- /dev/null +++ b/node/crates/c0mpute-placement/tests/distributed.rs @@ -0,0 +1,490 @@ +//! Cross-node placement and retrieval (CIP-003 acceptance criteria). +//! +//! Two levels: a fast in-memory transport for the failure matrix, and a real +//! multi-node HTTP test that stands up actual gateway servers and talks to +//! them over the CIP-002 endpoints. + +use std::sync::Arc; + +use c0mpute_placement::transport::memory::MemoryTransport; +use c0mpute_placement::{ + BlockState, DistributedConfig, DistributedStorage, PeerCatalog, PeerInfo, PlacementError, + PlacementPolicy, ShardTransport, +}; +use c0mpute_proto::Hash; +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-placement-{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()) +} + +/// `count` healthy peers, each in its own failure domain. +fn healthy_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 + i as u32, + 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 = 0xfeed_face_dead_beef; + for _ in 0..len { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + out.push((state & 0xff) as u8); + } + out +} + +struct Net { + storage: DistributedStorage, + transport: MemoryTransport, +} + +async fn net(tag: &str, peer_count: usize) -> Net { + let transport = MemoryTransport::new(); + let catalog = Arc::new(RwLock::new(PeerCatalog::new(healthy_peers(peer_count)))); + let storage = DistributedStorage::new( + local_storage(tag).await, + Arc::new(transport.clone()), + catalog, + ); + Net { storage, transport } +} + +// ------------------------------------------------------------------ placement + +#[tokio::test] +async fn shards_land_on_distinct_peers() { + let n = net("distinct", 20).await; + let data = varied(100_000); + + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + assert_eq!(manifest.blocks.len(), 1); + assert_eq!(manifest.blocks[0].shards.len(), 14); + + let hosts: std::collections::HashSet<_> = manifest.blocks[0] + .shards + .iter() + .map(|s| s.host_hint.clone().unwrap()) + .collect(); + assert_eq!(hosts.len(), 14, "a block's shards must not share a peer"); + + // And no peer is holding more than one shard of it. + for host in &hosts { + assert_eq!(n.transport.shard_count(host), 1); + } +} + +#[tokio::test] +async fn host_hints_are_recorded_and_reads_go_to_peers() { + let n = net("hints", 20).await; + let data = varied(50_000); + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + assert!( + manifest.blocks[0] + .shards + .iter() + .all(|s| s.host_hint.is_some()), + "every placed shard should record where it went" + ); + + let before = n.transport.get_calls(); + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); + assert!( + n.transport.get_calls() > before, + "the read should have gone to peers, not to local disk" + ); +} + +#[tokio::test] +async fn multi_block_objects_spread_across_the_network() { + let n = net("spread", 40).await; + // Three blocks at the 4 MiB default. + let data = varied(4 * 1024 * 1024 * 3); + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + assert_eq!(manifest.blocks.len(), 3); + + // Peers are chosen per block, so more than 14 distinct hosts are involved. + let hosts: std::collections::HashSet<_> = manifest + .blocks + .iter() + .flat_map(|b| b.shards.iter()) + .map(|s| s.host_hint.clone().unwrap()) + .collect(); + assert!(hosts.len() >= 14, "only {} hosts used", hosts.len()); + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +#[tokio::test] +async fn hot_tier_places_three_replicas_on_three_peers() { + let n = net("hot", 10).await; + let data = varied(10_000); + let manifest = n.storage.put(&data, Tier::Hot).await.unwrap(); + + assert_eq!(manifest.blocks[0].shards.len(), 3); + let hosts: std::collections::HashSet<_> = manifest.blocks[0] + .shards + .iter() + .map(|s| s.host_hint.clone().unwrap()) + .collect(); + assert_eq!(hosts.len(), 3); + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +// ------------------------------------------------------------------ durability + +#[tokio::test] +async fn survives_losing_the_parity_budget() { + let n = net("parity", 20).await; + let data = varied(200_000); + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + for shard in manifest.blocks[0].shards.iter().take(4) { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + assert_eq!( + n.storage.get(&manifest.object_hash).await.unwrap(), + data, + "RS 10/14 must tolerate 4 lost hosts" + ); +} + +#[tokio::test] +async fn fails_clearly_past_the_parity_budget() { + let n = net("lost", 20).await; + let data = varied(200_000); + let 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 err = format!( + "{:#}", + n.storage.get(&manifest.object_hash).await.unwrap_err() + ); + assert!(err.contains("need 10 shards"), "unhelpful error: {err}"); + assert!(err.contains("got 9"), "unhelpful error: {err}"); +} + +/// A peer serving wrong bytes must not corrupt the reconstruction — parity +/// covers it and the read still succeeds. +#[tokio::test] +async fn a_dishonest_peer_cannot_poison_a_read() { + let n = net("dishonest", 20).await; + let data = varied(120_000); + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + for shard in manifest.blocks[0].shards.iter().take(3) { + n.transport.make_corrupt(shard.host_hint.as_ref().unwrap()); + } + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +// ----------------------------------------------------------------- write path + +/// Write acknowledges at k + ceil(parity/2) = 12 of 14, so two dead peers do +/// not fail the write — they leave it under-replicated for repair. +#[tokio::test] +async fn write_succeeds_at_quorum_with_two_peers_down() { + let n = net("quorum", 20).await; + n.transport.take_offline("peer0"); + n.transport.take_offline("peer1"); + + let data = varied(80_000); + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + let placed = manifest.blocks[0].shards.len(); + assert!( + (12..=14).contains(&placed), + "expected quorum placement, got {placed}" + ); + assert_eq!(n.storage.get(&manifest.object_hash).await.unwrap(), data); +} + +#[tokio::test] +async fn write_fails_below_quorum() { + let n = net("noquorum", 16).await; + // Only 16 peers, and 5 of them are dead: at most 11 placements, under the + // quorum of 12. + for i in 0..5 { + n.transport.take_offline(&format!("peer{i}")); + } + let err = format!( + "{:#}", + n.storage + .put(&varied(50_000), Tier::Standard) + .await + .unwrap_err() + ); + assert!(err.contains("write quorum"), "unhelpful error: {err}"); +} + +// ------------------------------------------------------------------ diversity + +/// The property the whole design rests on. Twenty healthy peers, but they all +/// sit behind two ASNs, so placement would be two correlated samples dressed +/// as fourteen independent ones. It must refuse. +#[tokio::test] +async fn refuses_to_place_on_a_network_without_diversity() { + let transport = MemoryTransport::new(); + let mut peers = healthy_peers(20); + for (i, p) in peers.iter_mut().enumerate() { + p.asn = Some(if i < 10 { 100 } else { 200 }); + } + let catalog = Arc::new(RwLock::new(PeerCatalog::new(peers))); + let storage = DistributedStorage::new( + local_storage("nodiversity").await, + Arc::new(transport.clone()), + catalog, + ); + + let err = format!( + "{:#}", + storage + .put(&varied(50_000), Tier::Standard) + .await + .unwrap_err() + ); + assert!(err.contains("diversity unsatisfiable"), "unhelpful: {err}"); + assert!(err.contains("7 distinct domains"), "unhelpful: {err}"); + + // And nothing was written — a refused placement must not leave shards + // scattered across the peers it did reach. + assert_eq!(transport.put_calls(), 0); +} + +#[tokio::test] +async fn a_too_small_network_is_an_error_not_a_silent_downgrade() { + let n = net("tiny", 6).await; + let err = format!( + "{:#}", + n.storage + .put(&varied(10_000), Tier::Standard) + .await + .unwrap_err() + ); + assert!( + err.contains("not enough eligible peers"), + "unhelpful: {err}" + ); + assert!(err.contains("need 14"), "unhelpful: {err}"); +} + +/// An operator who knowingly runs a small network can relax the policy, but it +/// has to be deliberate. +#[tokio::test] +async fn policy_can_be_relaxed_explicitly() { + let transport = MemoryTransport::new(); + let mut peers = healthy_peers(14); + for p in peers.iter_mut() { + p.asn = Some(1); // all one domain + } + let catalog = Arc::new(RwLock::new(PeerCatalog::new(peers))); + let storage = DistributedStorage::new( + local_storage("relaxed").await, + Arc::new(transport.clone()), + catalog, + ) + .with_config(DistributedConfig { + policy: Some(PlacementPolicy { + max_per_domain: 14, + ..PlacementPolicy::for_parity(4) + }), + keep_local_copy: false, + }); + + let data = varied(30_000); + let manifest = storage.put(&data, Tier::Standard).await.unwrap(); + assert_eq!(storage.get(&manifest.object_hash).await.unwrap(), data); +} + +// -------------------------------------------------------------------- health + +#[tokio::test] +async fn health_reports_degradation_per_block() { + let n = net("health", 20).await; + let data = varied(60_000); + let manifest = n.storage.put(&data, Tier::Standard).await.unwrap(); + + let health = n.storage.health(&manifest).await.unwrap(); + assert_eq!(health.len(), 1); + assert_eq!(health[0].healthy, 14); + assert_eq!(health[0].state, BlockState::Healthy); + + for shard in manifest.blocks[0].shards.iter().take(3) { + n.transport.take_offline(shard.host_hint.as_ref().unwrap()); + } + let health = n.storage.health(&manifest).await.unwrap(); + assert_eq!(health[0].healthy, 11); + assert_eq!(health[0].missing.len(), 3); + assert_eq!(health[0].state, BlockState::Urgent); + assert!(health[0].state.needs_repair()); +} + +// ------------------------------------------------------ real multi-node HTTP + +/// The end-to-end case: five real gateway servers, shards pushed over the +/// CIP-002 HTTP endpoints, and an object reconstructed from them. +/// +/// Uses `hot` (n=3) so a five-node testnet is enough; the failure matrix above +/// covers RS 10/14 on the in-memory transport. +#[tokio::test] +async fn places_and_reads_across_real_http_nodes() { + use axum::Router; + use c0mpute_gateway::storage_api::{self, StorageApiState}; + use c0mpute_placement::HttpTransport; + + async fn spawn_node(tag: &str) -> (String, Storage) { + let storage = local_storage(tag).await; + let state = StorageApiState::local(storage.clone()); + let app: Router = storage_api::router(state); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{addr}"), storage) + } + + let mut peers = Vec::new(); + let mut stores = Vec::new(); + for i in 0..5 { + let (endpoint, store) = spawn_node(&format!("http-node{i}")).await; + peers.push(PeerInfo { + peer_id: format!("node{i}"), + endpoint, + reputation: 1.0, + uptime_30d: 1.0, + free_bytes: 1 << 30, + rtt_ms: 1, + asn: Some(64500 + i), + region: None, + ip_prefix: None, + }); + stores.push(store); + } + + let catalog = Arc::new(RwLock::new(PeerCatalog::new(peers.clone()))); + let client = DistributedStorage::new( + local_storage("http-client").await, + Arc::new(HttpTransport::default()), + catalog, + ); + + let data = varied(250_000); + let manifest = client.put(&data, Tier::Hot).await.unwrap(); + assert_eq!(manifest.blocks[0].shards.len(), 3); + + // The shards really are on three different servers' disks. + let mut holders = 0; + for store in &stores { + for shard in &manifest.blocks[0].shards { + if store.chunk_store().has(&shard.hash).await { + holders += 1; + break; + } + } + } + assert_eq!(holders, 3, "shards should be spread over three real nodes"); + + assert_eq!(client.get(&manifest.object_hash).await.unwrap(), data); + + // The client itself holds no shard bytes — only the manifest. + for shard in &manifest.blocks[0].shards { + assert!( + !client.local().chunk_store().has(&shard.hash).await, + "the writer should not keep a redundant local copy by default" + ); + } +} + +/// A shard PUT to a real node under the wrong hash is rejected by the +/// receiver, so a corrupted transfer can never become a stored bad shard. +#[tokio::test] +async fn real_nodes_reject_shards_that_do_not_match_their_hash() { + use axum::Router; + use c0mpute_gateway::storage_api::{self, StorageApiState}; + use c0mpute_placement::HttpTransport; + + let storage = local_storage("http-reject").await; + let app: Router = storage_api::router(StorageApiState::local(storage.clone())); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let peer = PeerInfo { + peer_id: "n".into(), + endpoint: format!("http://{addr}"), + reputation: 1.0, + uptime_30d: 1.0, + free_bytes: 1 << 30, + rtt_ms: 1, + asn: Some(1), + region: None, + ip_prefix: None, + }; + + let transport = HttpTransport::default(); + let bytes = b"the real bytes".to_vec(); + let wrong = Hash::of(b"a different shard"); + + let err = transport + .put_shard(&peer, &wrong, &bytes) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("rejected shard"), "unexpected: {err}"); + assert!(!storage.chunk_store().has(&wrong).await); + + // The honest write works. + let right = Hash::of(&bytes); + transport.put_shard(&peer, &right, &bytes).await.unwrap(); + assert!(storage.chunk_store().has(&right).await); + assert_eq!(transport.get_shard(&peer, &right).await.unwrap(), bytes); +} + +#[tokio::test] +async fn placement_error_types_are_distinguishable() { + // Callers (the CLI, and CIP-005's repair loop) need to tell "grow the + // network" apart from "this network can never be diverse enough". + let policy = PlacementPolicy::for_parity(4); + let few = healthy_peers(3); + assert!(matches!( + c0mpute_placement::select(&few, 14, 1024, &policy).unwrap_err(), + PlacementError::InsufficientPeers { .. } + )); + + let mut same_domain = healthy_peers(20); + for p in same_domain.iter_mut() { + p.asn = Some(42); + } + assert!(matches!( + c0mpute_placement::select(&same_domain, 14, 1024, &policy).unwrap_err(), + PlacementError::DiversityUnsatisfiable { .. } + )); +} 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); +}