From df6b4b260499358244675675830667ae0c9b4f8a Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sat, 29 Aug 2026 13:04:42 +0000 Subject: [PATCH] feat(storage): CIP-003 cross-node shard placement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements CIP-003. Until now every shard landed on one disk, which meant the erasure coding was pure overhead with no durability behind it. Blocks are now spread across n peers chosen for reputation and failure-domain diversity, and read back from whichever k answer first. New crate `c0mpute-placement`: peer — PeerInfo, PeerCatalog, and FailureDomain (ASN, falling back to IP prefix, then Unknown). select — choosing n peers under CIP-001's rules. Pure; no network I/O, because a slow peer lookup must not become a slow write. transport — ShardTransport trait, with HTTP and in-memory implementations. distributed— DistributedStorage, composing the three. Two decisions worth calling out. **HTTP first, libp2p later.** CIP-003 assumed the libp2p protocol had to be rewritten before placement could work. It didn't: CIP-002 already ships shard PUT/GET/HEAD endpoints that verify what they are given, so placement was built against a transport trait with an HTTP implementation over those. Cross-node placement works today on a real testnet, and the streaming libp2p protocol becomes a second implementation of an existing trait rather than a blocker. **Placement fails loudly.** CIP-001's durability figures assume shard hosts fail independently; fourteen shards behind one ISP are one sample wearing fourteen hats, and nothing downstream can detect it. So a write that cannot satisfy the diversity policy is an error naming the constraint, not a warning: failure-domain diversity unsatisfiable: 14 shards at most 2 per domain needs 7 distinct domains, but only 1 are available Greedy selection under a per-domain cap is optimal rather than heuristic — the cap is a partition matroid, so a refusal means no assignment would have worked. Also in this change: - Write acknowledges at k + ceil(parity/2) (12 of 14 for standard), so two slow peers do not fail a write; reads request all n and reconstruct from the first k. - `c0mpute storage peer add|ls|rm|ping`, a peers.json registry, and put/get/ info using the network when peers are configured. - CIP-003 sketched the peer score with a `1/(1+rtt/100)` latency term. That lets a fast flaky peer outrank a slow reliable one, which is the opposite of what CIP-001 says matters. Narrowed to a band that breaks ties without overturning a reputation gap; a test pins it. Three bugs found by running it rather than by tests: - `c0mpute storage get` still used the local read path, so an object placed across the network was unreadable — placement worked and retrieval did not. - anyhow's Display drops the cause chain, so the HTTP layer turned "not enough eligible peers: need 14, found 3" into "placing block 0". The whole point of CIP-003 is failing loudly; six sites now format with `{:#}`. - The CLI panicked on SIGPIPE, so `c0mpute storage ls | head` crashed. 43 new tests (27 unit, 16 integration including a real multi-node HTTP test). Verified on a 16-node testnet driven through the CLI: 14 shards on 14 distinct peers, byte-identical read back, still readable with 4 holders killed, refused with 5, and refused outright on a single-domain network. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LsQAuvXkmyHTgnvquLHrRx --- Cargo.lock | 23 + Cargo.toml | 2 + docs/prds/003-shard-placement-transport.md | 40 +- docs/prds/README.md | 2 +- node/crates/c0mpute-cli/Cargo.toml | 3 + node/crates/c0mpute-cli/src/main.rs | 10 + node/crates/c0mpute-cli/src/peers.rs | 160 ++++++ node/crates/c0mpute-cli/src/storage.rs | 316 ++++++++++- .../crates/c0mpute-gateway/src/storage_api.rs | 12 +- node/crates/c0mpute-placement/Cargo.toml | 30 ++ .../c0mpute-placement/src/distributed.rs | 468 +++++++++++++++++ node/crates/c0mpute-placement/src/lib.rs | 37 ++ node/crates/c0mpute-placement/src/peer.rs | 224 ++++++++ node/crates/c0mpute-placement/src/select.rs | 471 +++++++++++++++++ .../crates/c0mpute-placement/src/transport.rs | 339 ++++++++++++ .../c0mpute-placement/tests/distributed.rs | 490 ++++++++++++++++++ 16 files changed, 2591 insertions(+), 36 deletions(-) create mode 100644 node/crates/c0mpute-cli/src/peers.rs create mode 100644 node/crates/c0mpute-placement/Cargo.toml create mode 100644 node/crates/c0mpute-placement/src/distributed.rs create mode 100644 node/crates/c0mpute-placement/src/lib.rs create mode 100644 node/crates/c0mpute-placement/src/peer.rs create mode 100644 node/crates/c0mpute-placement/src/select.rs create mode 100644 node/crates/c0mpute-placement/src/transport.rs create mode 100644 node/crates/c0mpute-placement/tests/distributed.rs diff --git a/Cargo.lock b/Cargo.lock index 543697c..0143969 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,26 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "c0mpute-placement" +version = "0.2.26" +dependencies = [ + "anyhow", + "async-trait", + "axum", + "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/README.md b/docs/prds/README.md index 5609d5d..beb0e6e 100644 --- a/docs/prds/README.md +++ b/docs/prds/README.md @@ -66,7 +66,7 @@ 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 | | [006](006-challenges-metering-payouts.md) | Storage challenges, metering, and provider payouts | 003, 004 | 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..b9ad92b 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 { @@ -70,6 +93,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,7 +152,13 @@ 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, @@ -124,16 +176,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 +251,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 +388,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 +478,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(()) } 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..6869f1e --- /dev/null +++ b/node/crates/c0mpute-placement/src/lib.rs @@ -0,0 +1,37 @@ +//! 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 select; +pub mod transport; + +pub use distributed::{BlockHealth, BlockState, DistributedConfig, DistributedStorage}; +pub use peer::{FailureDomain, PeerCatalog, PeerInfo}; +pub use select::{Assignment, PlacementError, PlacementPolicy, score, select}; +pub use 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/select.rs b/node/crates/c0mpute-placement/src/select.rs new file mode 100644 index 0000000..00d90a2 --- /dev/null +++ b/node/crates/c0mpute-placement/src/select.rs @@ -0,0 +1,471 @@ +//! 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) +} + +/// 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 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 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))); + + let mut per_domain: HashMap = HashMap::new(); + 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() + .enumerate() + .map(|(i, p)| Assignment { + shard_index: i as u8, + peer: p.clone(), + }) + .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 { .. } + )); +}