From 5ac36c7bdfa206b4adf19dc670e616c396069d1b Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sat, 29 Aug 2026 12:38:34 +0000 Subject: [PATCH] feat(storage): CIP-002 storage HTTP API + `c0mpute storage` CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements CIP-002: turns the c0mpute-store engine into a usable service and puts it behind a CLI. Single node — every shard lands on the local disk and host_hint stays None; cross-node placement is CIP-003. c0mpute-store - Tier (hot / standard / critical) carrying (k, parity) and CIP-001 pricing. hot is RS k=1, which makes parity shards byte-identical copies: real 3-copy replication with 1x repair amplification. - Block layer: objects split into blocks, each independently RS-encoded, so memory is bounded by block size and a range read touches only the blocks it needs. CIP-007's random-access files depend on this. - Manifest v2 (version, block_size, tier, blocks[]) with a shim that still parses v1. - block_size_for() scales 4 MiB -> 256 MiB with object size, keeping a 1 TiB object's manifest under a megabyte instead of ~262k blocks. - put_stream / read_stream / get_range. put is now a wrapper over put_stream so there is one write path. - Storage::list() for enumerating what a node holds. c0mpute-gateway - /storage/v1/{objects,shards,manifests,status} with Range support, tier selection, idempotent PUT, single-flight per hash, and a disk budget. - Commit-then-verify on every write: bytes must hash to the hash the caller committed to, or 422. - ed25519 signed-request auth (DIP-0007 shape) bound to method + path + body hash, with a 5-minute skew window. Writes require it; reads do not, since the hash is the capability. c0mpute-cli - `c0mpute storage put|get|ls|info|verify|rm|status|tiers|serve`, per plugins/storage/module.toml. Commands that need later CIPs (volume, mount, provide) are absent rather than stubbed, and --help says which CIP brings each one. Two bugs found by running it, not by the tests: 1. Data loss on rollback. A write that fails its hash commitment used to delete every shard hash it touched. Shards are content-addressed and shared, so re-uploading an existing object's bytes under a wrong hash produces the same shard hashes — the rollback deleted the intact object's shards. One malformed request destroyed real data. ChunkStore::put_new now reports whether it created a chunk, and rollback only removes what it created. Regression tests at both the store and HTTP level. 2. Tracing wrote to stdout, so `HASH=$(c0mpute storage put f)` captured log lines and every scripted use broke. Diagnostics now go to stderr; daemon mode is unaffected because it points both at the same log file. Also: Config's api/storage/gateway sections get serde defaults, so a partial config.toml that sets only [storage] root loads instead of erroring. 65 new tests (31 store, 12 auth, 22 HTTP integration); workspace is green and clippy-clean on the touched crates. Verified end to end against a running server: 12 MiB round-trips byte-identical, range reads match dd, and an object still reconstructs after deleting 4 of 14 shards. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LsQAuvXkmyHTgnvquLHrRx --- Cargo.lock | 10 + docs/prds/001-storage-program.md | 4 +- docs/prds/002-storage-http-api.md | 50 +- docs/prds/README.md | 4 +- node/crates/c0mpute-cli/Cargo.toml | 3 + node/crates/c0mpute-cli/src/main.rs | 12 + node/crates/c0mpute-cli/src/storage.rs | 470 +++++++++ node/crates/c0mpute-core/src/config.rs | 8 +- node/crates/c0mpute-core/src/lib.rs | 19 +- node/crates/c0mpute-gateway/Cargo.toml | 9 + node/crates/c0mpute-gateway/src/auth.rs | 368 +++++++ node/crates/c0mpute-gateway/src/lib.rs | 21 + .../crates/c0mpute-gateway/src/storage_api.rs | 667 +++++++++++++ .../c0mpute-gateway/tests/storage_api.rs | 783 +++++++++++++++ node/crates/c0mpute-store/Cargo.toml | 3 + node/crates/c0mpute-store/src/lib.rs | 30 +- node/crates/c0mpute-store/src/storage.rs | 894 ++++++++++++++++-- node/crates/c0mpute-store/src/tier.rs | 145 +++ 18 files changed, 3385 insertions(+), 115 deletions(-) create mode 100644 node/crates/c0mpute-cli/src/storage.rs create mode 100644 node/crates/c0mpute-gateway/src/auth.rs create mode 100644 node/crates/c0mpute-gateway/src/storage_api.rs create mode 100644 node/crates/c0mpute-gateway/tests/storage_api.rs create mode 100644 node/crates/c0mpute-store/src/tier.rs diff --git a/Cargo.lock b/Cargo.lock index bf927a2..543697c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -494,11 +494,14 @@ name = "c0mpute-cli" version = "0.2.26" dependencies = [ "anyhow", + "axum", "base64", "c0mpute-core", "c0mpute-doctor", + "c0mpute-gateway", "c0mpute-proto", "c0mpute-secure-chat", + "c0mpute-store", "c0mpute-update", "clap", "daemonize", @@ -563,12 +566,16 @@ version = "0.2.26" dependencies = [ "anyhow", "axum", + "base64", "bytes", "c0mpute-net", "c0mpute-proto", "c0mpute-store", + "ed25519-dalek", + "futures", "serde", "serde_json", + "thiserror 2.0.18", "tokio", "tower", "tower-http", @@ -633,11 +640,14 @@ version = "0.2.26" dependencies = [ "anyhow", "blake3", + "bytes", "c0mpute-proto", + "futures", "hex", "reed-solomon-erasure", "serde", "serde_json", + "thiserror 2.0.18", "tokio", "tracing", ] diff --git a/docs/prds/001-storage-program.md b/docs/prds/001-storage-program.md index 366e950..b3db1bd 100644 --- a/docs/prds/001-storage-program.md +++ b/docs/prds/001-storage-program.md @@ -1,7 +1,7 @@ --- cip: 001 title: "Storage program: durability model, tiers, and economics" -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) depends-on: blocks: 002, 003, 004, 005, 006, 007 -implementation: +implementation: PR #21 (scripts/storage-durability-sim.py; tiers in c0mpute-store::tier) estimate: "1 week (analysis + simulation, no production code)" --- diff --git a/docs/prds/002-storage-http-api.md b/docs/prds/002-storage-http-api.md index efb3b05..7095f2f 100644 --- a/docs/prds/002-storage-http-api.md +++ b/docs/prds/002-storage-http-api.md @@ -1,7 +1,7 @@ --- cip: 002 title: "Storage HTTP API on the gateway" -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 2 depends-on: 001 blocks: 003, 004, 012 -implementation: +implementation: PR #21 (c0mpute-store block layer + manifest v2, c0mpute-gateway storage API, `c0mpute storage` CLI) estimate: "1.5–2 weeks" --- @@ -107,14 +107,15 @@ This CIP adds streaming variants alongside the existing ones: ```rust impl Storage { - /// Consume an AsyncRead, hashing and RS-encoding in fixed blocks. - pub async fn put_stream( - &self, reader: R, expected: Option, tier: Tier, - ) -> Result; + /// Consume a byte stream, hashing and RS-encoding block by block. + pub async fn put_stream( + &self, stream: S, expected: Option, tier: Tier, size_hint: Option, + ) -> Result + where S: Stream>; - /// Produce an AsyncRead that reconstructs lazily, block by block. - pub fn get_stream(&self, object_hash: &Hash) - -> Result; + /// Yield the object's blocks in order, reconstructing lazily. + pub fn read_stream(&self, manifest: ObjectManifest) + -> Pin> + Send>>; /// Byte-range read. Needed by CIP-007 for random-access files. pub async fn get_range( @@ -123,6 +124,14 @@ impl Storage { } ``` +A `Stream>` rather than `AsyncRead`: axum bodies are +already byte streams in both directions (`Body::into_data_stream`, +`Body::from_stream`), so this avoids a bridging dependency on both sides. +`size_hint` carries the HTTP `Content-Length` through to [`block_size_for`]. + +The non-streaming `put` is a thin wrapper over `put_stream`, so there is one +write path rather than two that drift. + Both are built on a **block layer**: an object is split into fixed-size blocks (default 4 MiB, recorded in the manifest) and each block is independently RS-encoded into `n` shards. Consequences, all of which later CIPs depend on: @@ -158,6 +167,29 @@ Version 1 manifests (flat `shards`, single implicit block) still parse — a `#[serde(default)]` shim maps them to a one-block v2. There is no production data to migrate, but the shim keeps the existing tests meaningful. +### Rollback must delete only what the write created + +A write that fails its hash commitment has to undo itself, and the obvious +implementation — remember every shard hash written, then delete them all — is +**wrong in a way that loses data**. + +Shards are content-addressed and therefore shared between objects. Uploading +the bytes of an object that *already exists*, under a wrong committed hash, +produces exactly the same shard hashes. Rolling back everything the write +touched deletes the intact object's shards: one malformed request, from anyone +who can obtain the content, destroys it. + +So `ChunkStore` grows `put_new`, which reports whether a call created the chunk +or found it already present, and rollback removes only newly-created hashes. +Refcounting is still deliberately avoided (CIP-004); this is strictly narrower +and needs no coordination. + +Found by driving the running server with curl, not by the unit tests — those +stored nothing beforehand, so there was nothing for the bad write to destroy. +Both the store and the HTTP suite now carry a regression test that stores an +object first. Worth remembering for CIP-005 and CIP-004, which both delete +content-addressed data and will meet the same trap. + ### Auth Writes (`PUT`, `DELETE`) require the DIP-0007 signed-request envelope in diff --git a/docs/prds/README.md b/docs/prds/README.md index 35b97dd..5609d5d 100644 --- a/docs/prds/README.md +++ b/docs/prds/README.md @@ -64,8 +64,8 @@ Delivering read/write network storage for c0mpute, implementing | # | Title | Depends on | Status | |-----|-------|-----------|--------| -| [001](001-storage-program.md) | Storage program: durability model, tiers, and economics | — | Draft | -| [002](002-storage-http-api.md) | Storage HTTP API on the gateway | 001 | Draft | +| [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 | | [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 | diff --git a/node/crates/c0mpute-cli/Cargo.toml b/node/crates/c0mpute-cli/Cargo.toml index 2c918d2..bd22353 100644 --- a/node/crates/c0mpute-cli/Cargo.toml +++ b/node/crates/c0mpute-cli/Cargo.toml @@ -15,6 +15,9 @@ c0mpute-doctor = { workspace = true } c0mpute-proto = { workspace = true } c0mpute-update = { workspace = true } c0mpute-secure-chat = { workspace = true } +c0mpute-store = { workspace = true } +c0mpute-gateway = { workspace = true } +axum = { workspace = true } anyhow = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } diff --git a/node/crates/c0mpute-cli/src/main.rs b/node/crates/c0mpute-cli/src/main.rs index 1558f87..bbaffd5 100644 --- a/node/crates/c0mpute-cli/src/main.rs +++ b/node/crates/c0mpute-cli/src/main.rs @@ -19,6 +19,8 @@ //! The plugin form mirrors the URL namespace: c0mpute.com/transcode, //! c0mpute.com/coinpay, c0mpute.com/infernet. +mod storage; + use std::path::PathBuf; use std::process::Command; @@ -65,6 +67,15 @@ enum Cmd { #[command(subcommand)] cmd: JobCmd, }, + /// Erasure-coded object storage on this node (DIP-0012). + /// + /// Objects are content-addressed and split into Reed-Solomon shards. Run + /// `c0mpute storage tiers` for the redundancy and price table. + #[command(after_long_help = storage::unimplemented_note())] + Storage { + #[command(subcommand)] + cmd: storage::StorageCmd, + }, /// Plugin management (list / install / enable / disable / uninstall). #[command(alias = "plugins")] Plugin { @@ -436,6 +447,7 @@ async fn run_app(cli: Cli) -> Result<()> { Cmd::StatusAggregator { bind } => c0mpute_core::status_aggregator::run(bind).await, Cmd::Worker { cmd } => run_worker(cmd, &config_path).await, Cmd::Job { cmd } => run_job(cmd).await, + Cmd::Storage { cmd } => storage::run(cmd, &config_path).await, Cmd::Plugin { cmd } => run_plugin(cmd), Cmd::Transcode { cmd } => run_transcode(cmd).await, diff --git a/node/crates/c0mpute-cli/src/storage.rs b/node/crates/c0mpute-cli/src/storage.rs new file mode 100644 index 0000000..8680fe6 --- /dev/null +++ b/node/crates/c0mpute-cli/src/storage.rs @@ -0,0 +1,470 @@ +//! `c0mpute storage` — the storage sub-feature of the umbrella CLI. +//! +//! Declared by `plugins/storage/module.toml` (`cli = "c0mpute storage"`) and +//! specified in `docs/prds/009-mount-cli.md`. This is the CIP-002 slice of +//! that surface: objects, shards and the local node's storage service. +//! +//! Volume, mount and provider commands arrive with their CIPs — see +//! [`unimplemented_note`] for what maps to what. They are deliberately absent +//! rather than stubbed, so `--help` never advertises something that does not +//! work. + +use std::path::PathBuf; +use std::sync::Arc; + +use anyhow::{Context, Result, bail}; +use c0mpute_core::config; +use c0mpute_gateway::auth::AllowAll; +use c0mpute_gateway::storage_api::{self, Limits, StorageApiState}; +use c0mpute_proto::Hash; +use c0mpute_store::{ChunkStore, Storage, Tier}; +use clap::Subcommand; + +#[derive(Subcommand, Debug)] +pub enum StorageCmd { + /// Store a file and print its object hash. + Put { + file: PathBuf, + /// Redundancy tier: hot, standard (default) or critical. + #[arg(long, default_value = "standard")] + tier: String, + }, + /// Fetch an object by hash. + Get { + hash: String, + /// Write to this path instead of stdout. + #[arg(short = 'o', long)] + out: Option, + /// Read only this byte range, e.g. `1000-2047`. + #[arg(long)] + range: Option, + }, + /// List objects held on this node. + Ls { + /// Print one hash per line with no header. + #[arg(long)] + quiet: bool, + }, + /// Show an object's block and shard layout. + Info { hash: String }, + /// Re-read an object and verify every block against its hash. + Verify { hash: String }, + /// Delete an object and its shards. + Rm { + hash: String, + #[arg(long)] + yes: bool, + }, + /// Disk usage, budget, and per-tier redundancy. + Status, + /// Show the tier table: redundancy, durability and price. + Tiers, + /// Run the storage HTTP API on this node. + Serve { + #[arg(long, default_value = "127.0.0.1:7780")] + bind: String, + /// Accept unauthenticated writes. Only for a node that is not + /// reachable from the network. + #[arg(long)] + insecure_allow_anonymous_writes: bool, + }, +} + +/// Where the local shard store lives. +fn storage_root(config_path: &std::path::Path) -> Result { + let cfg = config::Config::load_or_default(config_path)?; + Ok(cfg.storage.root) +} + +async fn open(config_path: &std::path::Path) -> Result { + let root = storage_root(config_path)?; + let store = ChunkStore::open(&root) + .await + .with_context(|| format!("open chunk store at {}", root.display()))?; + Ok(Storage::new(store)) +} + +fn parse_hash(raw: &str) -> Result { + let hex = raw.strip_prefix("blake3:").unwrap_or(raw); + Hash::from_hex(hex).map_err(|_| anyhow::anyhow!("`{raw}` is not a blake3 hash")) +} + +fn human(bytes: u64) -> String { + const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"]; + let mut v = bytes as f64; + let mut i = 0; + while v >= 1024.0 && i < UNITS.len() - 1 { + v /= 1024.0; + i += 1; + } + if i == 0 { + format!("{bytes} B") + } else { + format!("{v:.1} {}", UNITS[i]) + } +} + +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::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::Rm { hash, yes } => rm(config_path, &hash, yes).await, + StorageCmd::Status => status(config_path).await, + StorageCmd::Tiers => { + tiers(); + Ok(()) + } + StorageCmd::Serve { + bind, + insecure_allow_anonymous_writes, + } => serve(config_path, &bind, insecure_allow_anonymous_writes).await, + } +} + +async fn put(config_path: &std::path::Path, file: PathBuf, tier: &str) -> Result<()> { + let tier: Tier = tier.parse()?; + let storage = open(config_path).await?; + + 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?; + println!("blake3:{}", manifest.object_hash.to_hex()); + eprintln!( + " {} in {} block(s), {} shards, tier {} ({:.1}x expansion, {} raw)", + human(len), + manifest.blocks.len(), + manifest.shard_count(), + tier, + tier.expansion(), + human((len as f64 * tier.expansion()).ceil() as u64), + ); + Ok(()) +} + +async fn get( + config_path: &std::path::Path, + hash: &str, + out: Option, + range: Option, +) -> Result<()> { + let hash = parse_hash(hash)?; + let storage = open(config_path).await?; + if !storage.has(&hash).await { + bail!("no object {hash} on this node"); + } + + let bytes = match range { + Some(spec) => { + let (start, end) = spec + .split_once('-') + .context("range must look like `START-END`")?; + let start: u64 = start.parse().context("range start")?; + let end: u64 = end.parse().context("range end")?; + if end < start { + bail!("range end {end} is before start {start}"); + } + storage.get_range(&hash, start, end - start + 1).await? + } + None => storage.get(&hash).await?, + }; + + match out { + Some(path) => { + tokio::fs::write(&path, &bytes).await?; + eprintln!("wrote {} to {}", human(bytes.len() as u64), path.display()); + } + None => { + use std::io::Write; + std::io::stdout().write_all(&bytes)?; + } + } + Ok(()) +} + +async fn ls(config_path: &std::path::Path, quiet: bool) -> Result<()> { + let storage = open(config_path).await?; + let objects = storage.list().await?; + if objects.is_empty() && !quiet { + println!("no objects on this node"); + return Ok(()); + } + if !quiet { + println!( + "{:<66} {:>10} {:>8} {}", + "OBJECT", "SIZE", "TIER", "BLOCKS" + ); + } + for hash in objects { + if quiet { + println!("blake3:{}", hash.to_hex()); + continue; + } + match storage.read_manifest(&hash).await { + Ok(m) => println!( + "blake3:{} {:>10} {:>8} {}", + hash.to_hex(), + human(m.original_len), + m.tier.to_string(), + m.blocks.len() + ), + Err(e) => println!("blake3:{} ", hash.to_hex()), + } + } + Ok(()) +} + +async fn info(config_path: &std::path::Path, hash: &str) -> Result<()> { + let hash = parse_hash(hash)?; + let storage = open(config_path).await?; + let m = storage.read_manifest(&hash).await?; + + println!("object blake3:{}", m.object_hash.to_hex()); + println!( + "size {} ({} bytes)", + human(m.original_len), + m.original_len + ); + println!( + "tier {} — RS {}/{}, {:.1}x expansion, tolerates {} shard losses", + m.tier, + m.k, + m.n(), + m.tier.expansion(), + m.parity + ); + println!( + "blocks {} of {}", + m.blocks.len(), + human(m.block_size as u64) + ); + 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; + } else { + missing += 1; + } + } + } + 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(()) +} + +async fn verify(config_path: &std::path::Path, hash: &str) -> Result<()> { + let hash = parse_hash(hash)?; + let storage = open(config_path).await?; + let m = storage.read_manifest(&hash).await?; + + let mut bad = Vec::new(); + for i in 0..m.blocks.len() { + if let Err(e) = storage.read_block(&m, i).await { + bad.push(format!("block {i}: {e}")); + } + } + if bad.is_empty() { + println!( + "ok — {} block(s) of blake3:{} verified against their hashes", + m.blocks.len(), + m.object_hash.to_hex() + ); + Ok(()) + } else { + for b in &bad { + eprintln!("FAIL {b}"); + } + bail!( + "{} of {} blocks failed verification", + bad.len(), + m.blocks.len() + ); + } +} + +async fn rm(config_path: &std::path::Path, hash: &str, yes: bool) -> Result<()> { + let hash = parse_hash(hash)?; + let storage = open(config_path).await?; + if !storage.has(&hash).await { + bail!("no object {hash} on this node"); + } + if !yes { + let m = storage.read_manifest(&hash).await?; + eprintln!( + "about to delete blake3:{} ({}, {} shards)", + hash.to_hex(), + human(m.original_len), + m.shard_count() + ); + eprintln!("re-run with --yes to confirm"); + return Ok(()); + } + storage.delete(&hash).await?; + println!("deleted blake3:{}", hash.to_hex()); + Ok(()) +} + +async fn status(config_path: &std::path::Path) -> Result<()> { + let root = storage_root(config_path)?; + let storage = open(config_path).await?; + let cfg = config::Config::load_or_default(config_path)?; + let used = storage_api::DiskBudget::scan(&root); + let objects = storage.list().await?; + + println!("root {}", root.display()); + println!("objects {}", objects.len()); + println!( + "used {} of {}", + human(used), + cfg.storage + .cap_bytes + .map(human) + .unwrap_or_else(|| "uncapped".into()) + ); + + let mut logical = 0u64; + let mut degraded = 0usize; + for h in &objects { + if let Ok(m) = storage.read_manifest(h).await { + logical += m.original_len; + for block in &m.blocks { + let mut present = 0; + for s in &block.shards { + if storage.chunk_store().has(&s.hash).await { + present += 1; + } + } + if present < block.shards.len() { + degraded += 1; + } + } + } + } + println!( + "logical {} across {} objects", + human(logical), + objects.len() + ); + if degraded > 0 { + println!("degraded {degraded} block(s) missing at least one shard"); + } + println!(); + println!("single-node: every shard is on this disk, so the erasure coding is"); + println!("overhead without durability until cross-node placement (CIP-003)."); + Ok(()) +} + +fn tiers() { + println!( + "{:<10} {:<10} {:>9} {:>10} {:>9} {}", + "TIER", "SCHEME", "EXPANSION", "TOLERATES", "$/GB-mo", "BEST FOR" + ); + for (tier, scheme, best) in [ + (Tier::Hot, "3-copy", "small hot files, metadata"), + (Tier::Standard, "RS 10/14", "default; bulk data, media"), + (Tier::Critical, "RS 20/32", "irreplaceable, long retention"), + ] { + println!( + "{:<10} {:<10} {:>8.1}x {:>10} {:>9} {}", + tier.to_string(), + scheme, + tier.expansion(), + format!("{} lost", tier.parity()), + format!("${}", tier.price_usd_per_gb_month()), + best + ); + } + println!(); + println!("Expansion is the cost of goods: raw GB paid for per usable GB sold."); + println!("See docs/prds/001-storage-program.md for the durability arithmetic."); +} + +async fn serve(config_path: &std::path::Path, bind: &str, allow_anon: bool) -> Result<()> { + let cfg = config::Config::load_or_default(config_path)?; + let storage = open(config_path).await?; + let addr: std::net::SocketAddr = bind.parse().context("--bind must be HOST:PORT")?; + + if !allow_anon { + bail!( + "refusing to serve without an auth keyring.\n\ + Signed-request verification needs CoinPay DID resolution, which lands\n\ + with CIP-006. For a node that is not reachable from the network, pass\n\ + --insecure-allow-anonymous-writes." + ); + } + + let limits = Limits { + disk_budget_bytes: cfg.storage.cap_bytes, + ..Limits::default() + }; + let state = StorageApiState::new(storage, Arc::new(AllowAll), limits); + let app = storage_api::router(state); + + eprintln!("storage API on http://{addr} (anonymous writes ALLOWED)"); + eprintln!(" PUT /storage/v1/objects/{{hash}}"); + eprintln!(" GET /storage/v1/objects/{{hash}} (supports Range)"); + eprintln!(" GET /storage/v1/status"); + let listener = tokio::net::TcpListener::bind(addr).await?; + axum::serve(listener, app).await?; + Ok(()) +} + +/// Which CIP delivers each command that is not here yet. Referenced from the +/// long help so the gap is discoverable without reading `docs/prds/`. +pub fn unimplemented_note() -> &'static str { + "Not yet available:\n \ + volume create|list|destroy CIP-004\n \ + mount | umount CIP-009 (needs CIP-007 and CIP-008)\n \ + provide | earnings | retire CIP-006\n \ + repair | recover CIP-005\n\ + \nSee docs/prds/README.md for the delivery order." +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn human_sizes_read_sensibly() { + assert_eq!(human(0), "0 B"); + assert_eq!(human(512), "512 B"); + assert_eq!(human(1024), "1.0 KiB"); + assert_eq!(human(1024 * 1024 * 3 / 2), "1.5 MiB"); + assert_eq!(human(1u64 << 40), "1.0 TiB"); + } + + #[test] + fn hashes_parse_with_or_without_the_scheme() { + let h = Hash::of(b"x"); + assert_eq!(parse_hash(&h.to_hex()).unwrap(), h); + assert_eq!(parse_hash(&format!("blake3:{}", h.to_hex())).unwrap(), h); + assert!(parse_hash("nope").is_err()); + } + + #[test] + fn unimplemented_note_points_at_real_cips() { + let note = unimplemented_note(); + for cip in ["CIP-004", "CIP-005", "CIP-006", "CIP-007", "CIP-009"] { + assert!(note.contains(cip), "note should mention {cip}"); + } + } +} diff --git a/node/crates/c0mpute-core/src/config.rs b/node/crates/c0mpute-core/src/config.rs index c1e8e8e..afd2566 100644 --- a/node/crates/c0mpute-core/src/config.rs +++ b/node/crates/c0mpute-core/src/config.rs @@ -3,14 +3,20 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; -use directories::ProjectDirs; use c0mpute_proto::Role; +use directories::ProjectDirs; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct Config { + // Each section defaults, so a partial config.toml that sets only what the + // operator cares about (e.g. just `[storage] root`) loads cleanly instead + // of erroring on the sections they left out. + #[serde(default)] pub api: ApiConfig, + #[serde(default)] pub storage: StorageConfig, + #[serde(default)] pub gateway: GatewayConfig, #[serde(default)] pub roles: Vec, diff --git a/node/crates/c0mpute-core/src/lib.rs b/node/crates/c0mpute-core/src/lib.rs index 7e9b9a0..8313820 100644 --- a/node/crates/c0mpute-core/src/lib.rs +++ b/node/crates/c0mpute-core/src/lib.rs @@ -15,8 +15,8 @@ pub mod supervisor; pub use buyer::{AuctionOutcome, JobAuction, run_auction}; pub use capabilities::{Registry, advertise_loop, tags_from_config}; -pub use dispatch::{run_worker_subscriber, workload_types_from_roles}; pub use config::Config; +pub use dispatch::{run_worker_subscriber, workload_types_from_roles}; pub use register::{Registration, run_register}; pub use runner::TranscodeJobInline; pub use supervisor::Supervisor; @@ -27,9 +27,20 @@ use tracing::info; /// Convenience: install a default `tracing-subscriber` for the binary. pub fn init_tracing() -> Result<()> { use tracing_subscriber::{EnvFilter, fmt}; - let filter = EnvFilter::try_from_default_env() - .unwrap_or_else(|_| EnvFilter::new("info,c0mpute=debug")); - fmt().with_env_filter(filter).try_init().ok(); + let filter = + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info,c0mpute=debug")); + // Diagnostics go to stderr so stdout stays a clean data channel: commands + // like `c0mpute storage put` print only an object hash, which callers + // capture with `$(...)`. With the default stdout writer the logs land in + // that capture and every scripted use breaks. + // + // Daemon mode is unaffected: daemonize_worker points stdout and stderr at + // the same log file. + fmt() + .with_env_filter(filter) + .with_writer(std::io::stderr) + .try_init() + .ok(); info!("tracing initialised"); Ok(()) } diff --git a/node/crates/c0mpute-gateway/Cargo.toml b/node/crates/c0mpute-gateway/Cargo.toml index 75f68ed..3cfed79 100644 --- a/node/crates/c0mpute-gateway/Cargo.toml +++ b/node/crates/c0mpute-gateway/Cargo.toml @@ -18,3 +18,12 @@ tower-http = { workspace = true } bytes = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } +futures = { workspace = true } +thiserror = { workspace = true } +ed25519-dalek = { workspace = true } +base64 = { workspace = true } + +[dev-dependencies] +tower = { workspace = true, features = ["util"] } +c0mpute-store = { workspace = true } +futures = { workspace = true } diff --git a/node/crates/c0mpute-gateway/src/auth.rs b/node/crates/c0mpute-gateway/src/auth.rs new file mode 100644 index 0000000..b2887a9 --- /dev/null +++ b/node/crates/c0mpute-gateway/src/auth.rs @@ -0,0 +1,368 @@ +//! Signed-request auth for storage writes (CIP-002, DIP-0007). +//! +//! Reads of public objects need no auth — knowing a blake3 hash is itself the +//! capability, and `private`-tier objects are ciphertext (CIP-011), so +//! confidentiality comes from the key rather than an ACL. Writes are +//! authenticated so a stranger cannot fill an operator's disk. +//! +//! The envelope is an ed25519 signature over a canonical string. Identity is +//! a CoinPay DID; this module verifies the signature against a keyring and +//! leaves DID *resolution* (fetching a DID's current key from CoinPay) to the +//! caller, which is what lets the same code serve tests, a local dev node, and +//! a production node with a live registry. + +use std::collections::HashMap; +use std::time::{SystemTime, UNIX_EPOCH}; + +use base64::Engine; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use ed25519_dalek::{Signature, Verifier, VerifyingKey}; +use serde::{Deserialize, Serialize}; + +/// Header carrying the signed-request envelope. +pub const AUTH_HEADER: &str = "x-coinpay-auth"; + +/// Requests older than this are rejected, to bound replay. +pub const MAX_CLOCK_SKEW_SECS: u64 = 300; + +/// Domain separator, so a storage envelope can never be replayed against +/// another c0mpute surface that adopts the same scheme. +const SIGNING_DOMAIN: &str = "c0mpute-storage-v1"; + +/// The authenticated caller. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Identity { + pub did: String, +} + +impl Identity { + /// The identity used when authorization is disabled. + pub fn anonymous() -> Self { + Self { + did: "anonymous".to_string(), + } + } +} + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum AuthError { + #[error("missing {AUTH_HEADER} header")] + Missing, + #[error("malformed auth envelope: {0}")] + Malformed(String), + #[error("unknown signer {0}")] + UnknownSigner(String), + #[error("signature does not verify")] + BadSignature, + #[error("request timestamp is {0}s outside the allowed skew")] + Expired(u64), +} + +/// The envelope, base64url-encoded into the header value. +#[derive(Debug, Serialize, Deserialize)] +pub struct Envelope { + pub did: String, + /// Unix seconds. + pub ts: u64, + /// base64url ed25519 signature over [`canonical_string`]. + pub sig: String, +} + +/// The exact bytes a client signs. +/// +/// Binding the method, path and body hash means a captured envelope cannot be +/// replayed against a different object or a different verb. +pub fn canonical_string(method: &str, path: &str, ts: u64, body_hash_hex: &str) -> String { + format!("{SIGNING_DOMAIN}\n{method}\n{path}\n{ts}\n{body_hash_hex}") +} + +/// What a request needs to present to be authorized. +pub struct AuthRequest<'a> { + pub method: &'a str, + pub path: &'a str, + /// Hex blake3 of the body the client committed to. For storage writes this + /// is the object hash in the URL, so it costs nothing to bind. + pub body_hash_hex: &'a str, + pub header: Option<&'a str>, +} + +pub trait Authorizer: Send + Sync + std::fmt::Debug { + fn authorize(&self, req: &AuthRequest<'_>) -> Result; +} + +/// Accepts everything. For local development and single-operator nodes that +/// are not exposed to the network. +#[derive(Debug, Default, Clone, Copy)] +pub struct AllowAll; + +impl Authorizer for AllowAll { + fn authorize(&self, _req: &AuthRequest<'_>) -> Result { + Ok(Identity::anonymous()) + } +} + +/// Verifies ed25519 signed-request envelopes against a keyring of DIDs. +#[derive(Debug, Default)] +pub struct SignedEnvelope { + keys: HashMap, + /// Overridable for tests; `None` means "read the system clock". + now_override: Option, +} + +impl SignedEnvelope { + pub fn new() -> Self { + Self::default() + } + + /// Trust `did` when it signs with `key`. + pub fn with_key(mut self, did: impl Into, key: VerifyingKey) -> Self { + self.keys.insert(did.into(), key); + self + } + + #[cfg(test)] + fn at_time(mut self, now: u64) -> Self { + self.now_override = Some(now); + self + } + + fn now(&self) -> u64 { + self.now_override.unwrap_or_else(|| { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) + }) + } +} + +impl Authorizer for SignedEnvelope { + fn authorize(&self, req: &AuthRequest<'_>) -> Result { + let raw = req.header.ok_or(AuthError::Missing)?; + let decoded = URL_SAFE_NO_PAD + .decode(raw) + .map_err(|e| AuthError::Malformed(e.to_string()))?; + let env: Envelope = + serde_json::from_slice(&decoded).map_err(|e| AuthError::Malformed(e.to_string()))?; + + let now = self.now(); + let skew = now.abs_diff(env.ts); + if skew > MAX_CLOCK_SKEW_SECS { + return Err(AuthError::Expired(skew - MAX_CLOCK_SKEW_SECS)); + } + + let key = self + .keys + .get(&env.did) + .ok_or_else(|| AuthError::UnknownSigner(env.did.clone()))?; + + let sig_bytes = URL_SAFE_NO_PAD + .decode(&env.sig) + .map_err(|e| AuthError::Malformed(e.to_string()))?; + let sig_arr: [u8; 64] = sig_bytes + .try_into() + .map_err(|_| AuthError::Malformed("signature is not 64 bytes".into()))?; + let sig = Signature::from_bytes(&sig_arr); + + let msg = canonical_string(req.method, req.path, env.ts, req.body_hash_hex); + key.verify(msg.as_bytes(), &sig) + .map_err(|_| AuthError::BadSignature)?; + + Ok(Identity { did: env.did }) + } +} + +/// Build a header value for `canonical_string`, for clients and tests. +pub fn sign_envelope( + did: &str, + signing_key: &ed25519_dalek::SigningKey, + method: &str, + path: &str, + ts: u64, + body_hash_hex: &str, +) -> String { + use ed25519_dalek::Signer; + let msg = canonical_string(method, path, ts, body_hash_hex); + let sig = signing_key.sign(msg.as_bytes()); + let env = Envelope { + did: did.to_string(), + ts, + sig: URL_SAFE_NO_PAD.encode(sig.to_bytes()), + }; + URL_SAFE_NO_PAD.encode(serde_json::to_vec(&env).expect("envelope serializes")) +} + +#[cfg(test)] +mod tests { + use super::*; + use ed25519_dalek::SigningKey; + + const DID: &str = "did:coinpay:test"; + const NOW: u64 = 1_756_512_000; + + fn keypair() -> SigningKey { + SigningKey::from_bytes(&[7u8; 32]) + } + + fn verifier(sk: &SigningKey) -> SignedEnvelope { + SignedEnvelope::new() + .with_key(DID, sk.verifying_key()) + .at_time(NOW) + } + + #[test] + fn allow_all_lets_anything_through() { + let req = AuthRequest { + method: "PUT", + path: "/storage/v1/objects/abc", + body_hash_hex: "abc", + header: None, + }; + assert_eq!(AllowAll.authorize(&req).unwrap(), Identity::anonymous()); + } + + #[test] + fn valid_envelope_authorizes() { + let sk = keypair(); + let header = sign_envelope(DID, &sk, "PUT", "/o/abc", NOW, "abc"); + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(&header), + }; + assert_eq!(verifier(&sk).authorize(&req).unwrap().did, DID); + } + + #[test] + fn missing_header_is_rejected() { + let sk = keypair(); + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: None, + }; + assert_eq!( + verifier(&sk).authorize(&req).unwrap_err(), + AuthError::Missing + ); + } + + /// An envelope signed for one object must not work for another — this is + /// the property that stops a captured header being reused. + #[test] + fn envelope_is_bound_to_its_path_and_body() { + let sk = keypair(); + let header = sign_envelope(DID, &sk, "PUT", "/o/abc", NOW, "abc"); + + let wrong_path = AuthRequest { + method: "PUT", + path: "/o/different", + body_hash_hex: "abc", + header: Some(&header), + }; + assert_eq!( + verifier(&sk).authorize(&wrong_path).unwrap_err(), + AuthError::BadSignature + ); + + let wrong_body = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "deadbeef", + header: Some(&header), + }; + assert_eq!( + verifier(&sk).authorize(&wrong_body).unwrap_err(), + AuthError::BadSignature + ); + + let wrong_method = AuthRequest { + method: "DELETE", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(&header), + }; + assert_eq!( + verifier(&sk).authorize(&wrong_method).unwrap_err(), + AuthError::BadSignature + ); + } + + #[test] + fn unknown_signer_is_rejected() { + let sk = keypair(); + let other = SigningKey::from_bytes(&[9u8; 32]); + let header = sign_envelope("did:coinpay:stranger", &other, "PUT", "/o/abc", NOW, "abc"); + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(&header), + }; + assert!(matches!( + verifier(&sk).authorize(&req).unwrap_err(), + AuthError::UnknownSigner(_) + )); + } + + #[test] + fn stale_envelope_is_rejected() { + let sk = keypair(); + let old = NOW - MAX_CLOCK_SKEW_SECS - 60; + let header = sign_envelope(DID, &sk, "PUT", "/o/abc", old, "abc"); + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(&header), + }; + assert!(matches!( + verifier(&sk).authorize(&req).unwrap_err(), + AuthError::Expired(_) + )); + } + + /// A clock ahead of ours by less than the skew budget is fine; well ahead + /// is not. Rejecting only the past would let a client mint far-future + /// envelopes that never expire. + #[test] + fn future_envelope_beyond_skew_is_rejected() { + let sk = keypair(); + let near = sign_envelope(DID, &sk, "PUT", "/o/abc", NOW + 60, "abc"); + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(&near), + }; + assert!(verifier(&sk).authorize(&req).is_ok()); + + let far = sign_envelope(DID, &sk, "PUT", "/o/abc", NOW + 10_000, "abc"); + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(&far), + }; + assert!(matches!( + verifier(&sk).authorize(&req).unwrap_err(), + AuthError::Expired(_) + )); + } + + #[test] + fn garbage_header_is_malformed_not_a_panic() { + let sk = keypair(); + for bad in ["!!!!", "", "aGVsbG8"] { + let req = AuthRequest { + method: "PUT", + path: "/o/abc", + body_hash_hex: "abc", + header: Some(bad), + }; + assert!(verifier(&sk).authorize(&req).is_err()); + } + } +} diff --git a/node/crates/c0mpute-gateway/src/lib.rs b/node/crates/c0mpute-gateway/src/lib.rs index ab4f143..e52e0c3 100644 --- a/node/crates/c0mpute-gateway/src/lib.rs +++ b/node/crates/c0mpute-gateway/src/lib.rs @@ -5,6 +5,9 @@ //! The gateway is deliberately stateless beyond the local cache — every //! request is content-addressed, so any gateway can serve any chunk. +pub mod auth; +pub mod storage_api; + use std::net::SocketAddr; use std::sync::Arc; @@ -32,6 +35,24 @@ pub fn router(state: GatewayState) -> Router { .with_state(state) } +/// The gateway plus the storage API (CIP-002), for nodes running the +/// `storage` role. +pub fn router_with_storage(state: GatewayState, storage: storage_api::StorageApiState) -> Router { + router(state).merge(storage_api::router(storage)) +} + +pub async fn serve_with_storage( + state: GatewayState, + storage: storage_api::StorageApiState, + addr: SocketAddr, +) -> Result<()> { + let app = router_with_storage(state, storage); + info!(%addr, "gateway listening (storage role enabled)"); + let listener = tokio::net::TcpListener::bind(addr).await?; + axum::serve(listener, app).await?; + Ok(()) +} + pub async fn serve(state: GatewayState, addr: SocketAddr) -> Result<()> { let app = router(state); info!(%addr, "gateway listening"); diff --git a/node/crates/c0mpute-gateway/src/storage_api.rs b/node/crates/c0mpute-gateway/src/storage_api.rs new file mode 100644 index 0000000..34970c8 --- /dev/null +++ b/node/crates/c0mpute-gateway/src/storage_api.rs @@ -0,0 +1,667 @@ +//! Storage HTTP API (CIP-002). +//! +//! Turns the `c0mpute-store` engine into a service: +//! +//! ```text +//! PUT /storage/v1/objects/{hash} store an object +//! GET /storage/v1/objects/{hash} reconstruct and stream it back +//! HEAD /storage/v1/objects/{hash} existence + length, no body +//! DELETE /storage/v1/objects/{hash} drop manifest + shards +//! GET /storage/v1/shards/{hash} serve one shard (peer fetch, repair) +//! PUT /storage/v1/shards/{hash} accept one shard (peer placement) +//! HEAD /storage/v1/shards/{hash} do you hold it? +//! GET /storage/v1/manifests/{hash} the manifest as JSON +//! GET /storage/v1/status disk budget and usage +//! ``` +//! +//! Single-node: every shard lands on the local disk and `host_hint` stays +//! `None`. Cross-node placement is CIP-003. + +use std::collections::HashSet; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use axum::Router; +use axum::body::Body; +use axum::extract::{Path, Query, Request, State}; +use axum::http::{HeaderMap, StatusCode, header}; +use axum::response::{IntoResponse, Response}; +use axum::routing::get; +use c0mpute_proto::Hash; +use c0mpute_store::{ObjectManifest, Storage, Tier}; +use futures::TryStreamExt; +use serde::{Deserialize, Serialize}; +use tokio::sync::Mutex; +use tracing::{info, warn}; + +use crate::auth::{AUTH_HEADER, AllowAll, AuthError, AuthRequest, Authorizer}; + +/// Header selecting the redundancy tier on a write. +pub const TIER_HEADER: &str = "x-c0mpute-tier"; +/// Headers a peer sends when placing a single shard. +pub const SHARD_OBJECT_HEADER: &str = "x-c0mpute-object"; +pub const SHARD_INDEX_HEADER: &str = "x-c0mpute-shard-index"; + +/// Default ceiling on a single object. Blocks keep memory bounded, but an +/// unbounded object still means an unbounded manifest. +pub const DEFAULT_MAX_OBJECT_BYTES: u64 = 1 << 40; // 1 TiB + +#[derive(Clone, Debug)] +pub struct Limits { + pub max_object_bytes: u64, + /// Total bytes this node will hold, if capped. + pub disk_budget_bytes: Option, +} + +impl Default for Limits { + fn default() -> Self { + Self { + max_object_bytes: DEFAULT_MAX_OBJECT_BYTES, + disk_budget_bytes: None, + } + } +} + +/// Tracks how much of the operator's committed disk is in use. +/// +/// Seeded by walking the shard directory once at startup — O(files), a few +/// seconds for a large store, and the alternative is trusting a counter that +/// drifts across restarts. +#[derive(Debug)] +pub struct DiskBudget { + limit: Option, + used: AtomicU64, +} + +impl DiskBudget { + pub fn new(limit: Option, used: u64) -> Self { + Self { + limit, + used: AtomicU64::new(used), + } + } + + /// Measure current usage by walking the store's shard directory. + pub fn scan(root: &std::path::Path) -> u64 { + fn walk(dir: &std::path::Path, total: &mut u64) { + let Ok(rd) = std::fs::read_dir(dir) else { + return; + }; + for entry in rd.flatten() { + let path = entry.path(); + match entry.file_type() { + Ok(t) if t.is_dir() => walk(&path, total), + Ok(t) if t.is_file() => { + if let Ok(md) = entry.metadata() { + *total += md.len(); + } + } + _ => {} + } + } + } + let mut total = 0; + walk(&root.join("shards"), &mut total); + total + } + + pub fn used(&self) -> u64 { + self.used.load(Ordering::Relaxed) + } + + pub fn limit(&self) -> Option { + self.limit + } + + /// True if `bytes` more would fit. Advisory: the write is charged after + /// the fact, so concurrent writes can overshoot slightly. + pub fn would_fit(&self, bytes: u64) -> bool { + match self.limit { + None => true, + Some(limit) => self.used().saturating_add(bytes) <= limit, + } + } + + pub fn charge(&self, bytes: u64) { + self.used.fetch_add(bytes, Ordering::Relaxed); + } + + pub fn release(&self, bytes: u64) { + // saturating: a double-release must not wrap to a huge number. + let _ = self + .used + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |u| { + Some(u.saturating_sub(bytes)) + }); + } +} + +#[derive(Clone)] +pub struct StorageApiState { + pub storage: Storage, + pub auth: Arc, + pub limits: Limits, + pub budget: Arc, + /// Object hashes with a PUT in flight, so a duplicate concurrent write + /// gets a 409 instead of two writers racing on the same manifest. + inflight: Arc>>, +} + +impl StorageApiState { + pub fn new(storage: Storage, auth: Arc, limits: Limits) -> Self { + let used = DiskBudget::scan(storage.chunk_store().root()); + let budget = Arc::new(DiskBudget::new(limits.disk_budget_bytes, used)); + Self { + storage, + auth, + limits, + budget, + inflight: Arc::new(Mutex::new(HashSet::new())), + } + } + + /// Local single-operator node: no auth, no disk cap. + pub fn local(storage: Storage) -> Self { + Self::new(storage, Arc::new(AllowAll), Limits::default()) + } +} + +pub fn router(state: StorageApiState) -> Router { + Router::new() + .route( + "/storage/v1/objects/{hash}", + get(get_object) + .head(head_object) + .put(put_object) + .delete(delete_object), + ) + .route( + "/storage/v1/shards/{hash}", + get(get_shard).head(head_shard).put(put_shard), + ) + .route("/storage/v1/manifests/{hash}", get(get_manifest)) + .route("/storage/v1/status", get(status)) + .with_state(state) +} + +// --------------------------------------------------------------------- errors + +#[derive(Debug, thiserror::Error)] +pub enum ApiError { + #[error("{0}")] + BadRequest(String), + #[error("unauthorized: {0}")] + Unauthorized(AuthError), + #[error("not found")] + NotFound, + #[error("a write for this object is already in flight")] + Conflict, + #[error("object exceeds the {0} byte limit")] + TooLarge(u64), + #[error("{0}")] + Unprocessable(String), + #[error("disk budget exhausted")] + OutOfSpace, + #[error("{0}")] + Internal(String), +} + +#[derive(Serialize)] +struct ErrorBody { + error: String, + code: u16, +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + let status = match &self { + ApiError::BadRequest(_) => StatusCode::BAD_REQUEST, + ApiError::Unauthorized(_) => StatusCode::UNAUTHORIZED, + ApiError::NotFound => StatusCode::NOT_FOUND, + ApiError::Conflict => StatusCode::CONFLICT, + ApiError::TooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE, + ApiError::Unprocessable(_) => StatusCode::UNPROCESSABLE_ENTITY, + ApiError::OutOfSpace => StatusCode::INSUFFICIENT_STORAGE, + ApiError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR, + }; + let body = ErrorBody { + error: self.to_string(), + code: status.as_u16(), + }; + (status, axum::Json(body)).into_response() + } +} + +type ApiResult = Result; + +// ---------------------------------------------------------------- helpers + +/// Accept `blake3:` or a bare hex string. +fn parse_hash(raw: &str) -> ApiResult { + let hex = raw.strip_prefix("blake3:").unwrap_or(raw); + Hash::from_hex(hex).map_err(|_| ApiError::BadRequest(format!("invalid hash `{raw}`"))) +} + +fn parse_tier(headers: &HeaderMap) -> ApiResult { + match headers.get(TIER_HEADER) { + None => Ok(Tier::default()), + Some(v) => v + .to_str() + .ok() + .and_then(|s| s.parse::().ok()) + .ok_or_else(|| ApiError::BadRequest("invalid tier".into())), + } +} + +fn header_str<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { + headers.get(name).and_then(|v| v.to_str().ok()) +} + +fn authorize( + state: &StorageApiState, + headers: &HeaderMap, + method: &str, + path: &str, + body_hash: &str, +) -> ApiResult<()> { + let req = AuthRequest { + method, + path, + body_hash_hex: body_hash, + header: header_str(headers, AUTH_HEADER), + }; + state + .auth + .authorize(&req) + .map(|_| ()) + .map_err(ApiError::Unauthorized) +} + +/// A single HTTP byte range. Only the common `bytes=a-b` / `bytes=a-` forms +/// are supported; multipart ranges are not. +fn parse_range(raw: &str, total: u64) -> Option<(u64, u64)> { + let spec = raw.strip_prefix("bytes=")?.trim(); + if spec.contains(',') { + return None; + } + let (start_s, end_s) = spec.split_once('-')?; + if start_s.is_empty() { + // `bytes=-N` — the final N bytes. + let n: u64 = end_s.parse().ok()?; + let n = n.min(total); + return Some((total.saturating_sub(n), n)); + } + let start: u64 = start_s.parse().ok()?; + if start >= total { + return None; + } + let end = if end_s.is_empty() { + total - 1 + } else { + end_s.parse::().ok()?.min(total - 1) + }; + if end < start { + return None; + } + Some((start, end - start + 1)) +} + +// ---------------------------------------------------------------- objects + +#[derive(Deserialize)] +pub struct PutQuery { + /// Optional tier, as an alternative to the header. + pub tier: Option, +} + +async fn put_object( + State(state): State, + Path(raw_hash): Path, + Query(q): Query, + headers: HeaderMap, + req: Request, +) -> ApiResult { + let object_hash = parse_hash(&raw_hash)?; + let path = format!("/storage/v1/objects/{raw_hash}"); + authorize(&state, &headers, "PUT", &path, &object_hash.to_hex())?; + + let tier = match q.tier { + Some(t) => t + .parse::() + .map_err(|e| ApiError::BadRequest(e.to_string()))?, + None => parse_tier(&headers)?, + }; + + let len: u64 = headers + .get(header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse().ok()) + .ok_or_else(|| ApiError::BadRequest("Content-Length is required".into()))?; + + if len > state.limits.max_object_bytes { + return Err(ApiError::TooLarge(state.limits.max_object_bytes)); + } + // Charge the expanded size — the tier decides how much disk this costs. + let cost = (len as f64 * tier.expansion()).ceil() as u64; + if !state.budget.would_fit(cost) { + return Err(ApiError::OutOfSpace); + } + + // Idempotent: an object we already hold is not rewritten. + if state.storage.has(&object_hash).await { + let manifest = state + .storage + .read_manifest(&object_hash) + .await + .map_err(|e| ApiError::Internal(e.to_string()))?; + return Ok((StatusCode::OK, axum::Json(manifest)).into_response()); + } + + // Single-flight per object hash. + { + let mut inflight = state.inflight.lock().await; + if !inflight.insert(object_hash) { + return Err(ApiError::Conflict); + } + } + let _guard = InflightGuard { + state: state.clone(), + hash: object_hash, + }; + + let stream = req + .into_body() + .into_data_stream() + .map_err(|e| anyhow::anyhow!("request body: {e}")); + + let manifest = state + .storage + .put_stream(stream, Some(object_hash), tier, Some(len)) + .await + .map_err(|e| { + let msg = e.to_string(); + if msg.contains("integrity failure") { + ApiError::Unprocessable(msg) + } else { + ApiError::Internal(msg) + } + })?; + + state.budget.charge(cost); + info!(object_hash = %object_hash, %tier, bytes = len, "stored object"); + Ok((StatusCode::CREATED, axum::Json(manifest)).into_response()) +} + +/// Clears the in-flight marker however the handler exits. +struct InflightGuard { + state: StorageApiState, + hash: Hash, +} + +impl Drop for InflightGuard { + fn drop(&mut self) { + let state = self.state.clone(); + let hash = self.hash; + tokio::spawn(async move { + state.inflight.lock().await.remove(&hash); + }); + } +} + +async fn get_object( + State(state): State, + Path(raw_hash): Path, + headers: HeaderMap, +) -> ApiResult { + let object_hash = parse_hash(&raw_hash)?; + let manifest = load_manifest(&state, &object_hash).await?; + let total = manifest.original_len; + + if let Some(range_raw) = header_str(&headers, header::RANGE.as_str()) { + let Some((offset, len)) = parse_range(range_raw, total) else { + return Ok(( + StatusCode::RANGE_NOT_SATISFIABLE, + [(header::CONTENT_RANGE, format!("bytes */{total}"))], + ) + .into_response()); + }; + let bytes = state + .storage + .get_range_with(&manifest, offset, len) + .await + .map_err(|e| ApiError::Internal(e.to_string()))?; + let end = offset + bytes.len() as u64 - 1; + return Ok(( + StatusCode::PARTIAL_CONTENT, + [ + (header::CONTENT_TYPE, "application/octet-stream".to_string()), + ( + header::CONTENT_RANGE, + format!("bytes {offset}-{end}/{total}"), + ), + (header::ACCEPT_RANGES, "bytes".to_string()), + ], + bytes, + ) + .into_response()); + } + + // Whole object: stream block by block so memory stays bounded. + let stream = state.storage.read_stream(manifest); + Ok(( + StatusCode::OK, + [ + (header::CONTENT_TYPE, "application/octet-stream".to_string()), + (header::CONTENT_LENGTH, total.to_string()), + (header::ACCEPT_RANGES, "bytes".to_string()), + ], + Body::from_stream(stream), + ) + .into_response()) +} + +async fn head_object( + State(state): State, + Path(raw_hash): Path, +) -> ApiResult { + let object_hash = parse_hash(&raw_hash)?; + let manifest = load_manifest(&state, &object_hash).await?; + Ok(( + StatusCode::OK, + [ + (header::CONTENT_TYPE, "application/octet-stream".to_string()), + (header::CONTENT_LENGTH, manifest.original_len.to_string()), + (header::ACCEPT_RANGES, "bytes".to_string()), + ( + header::HeaderName::from_static("x-c0mpute-tier"), + manifest.tier.to_string(), + ), + ], + ) + .into_response()) +} + +async fn delete_object( + State(state): State, + Path(raw_hash): Path, + headers: HeaderMap, +) -> ApiResult { + let object_hash = parse_hash(&raw_hash)?; + let path = format!("/storage/v1/objects/{raw_hash}"); + authorize(&state, &headers, "DELETE", &path, &object_hash.to_hex())?; + + let manifest = load_manifest(&state, &object_hash).await?; + let cost = (manifest.original_len as f64 * manifest.tier.expansion()).ceil() as u64; + state + .storage + .delete(&object_hash) + .await + .map_err(|e| ApiError::Internal(e.to_string()))?; + state.budget.release(cost); + Ok(StatusCode::NO_CONTENT.into_response()) +} + +async fn get_manifest( + State(state): State, + Path(raw_hash): Path, +) -> ApiResult { + let object_hash = parse_hash(&raw_hash)?; + let manifest = load_manifest(&state, &object_hash).await?; + Ok(axum::Json(manifest).into_response()) +} + +async fn load_manifest(state: &StorageApiState, hash: &Hash) -> ApiResult { + if !state.storage.has(hash).await { + return Err(ApiError::NotFound); + } + state + .storage + .read_manifest(hash) + .await + .map_err(|e| ApiError::Internal(e.to_string())) +} + +// ----------------------------------------------------------------- shards + +async fn get_shard( + State(state): State, + Path(raw_hash): Path, +) -> ApiResult { + let hash = parse_hash(&raw_hash)?; + match state.storage.chunk_store().get(&hash).await { + Ok(bytes) => Ok(( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/octet-stream")], + bytes, + ) + .into_response()), + Err(e) => { + // A read error here is either "absent" or "corrupt on disk"; both + // mean the peer should look elsewhere, but corruption is a fault + // we want in the logs. + if state.storage.chunk_store().has(&hash).await { + warn!(shard = %hash, err = %e, "held shard failed verification"); + } + Err(ApiError::NotFound) + } + } +} + +async fn head_shard( + State(state): State, + Path(raw_hash): Path, +) -> ApiResult { + let hash = parse_hash(&raw_hash)?; + if state.storage.chunk_store().has(&hash).await { + Ok(StatusCode::OK.into_response()) + } else { + Err(ApiError::NotFound) + } +} + +/// Accept one shard from a peer (placement, CIP-003; repair, CIP-005). +async fn put_shard( + State(state): State, + Path(raw_hash): Path, + headers: HeaderMap, + body: axum::body::Bytes, +) -> ApiResult { + let shard_hash = parse_hash(&raw_hash)?; + let path = format!("/storage/v1/shards/{raw_hash}"); + authorize(&state, &headers, "PUT", &path, &shard_hash.to_hex())?; + + if !state.budget.would_fit(body.len() as u64) { + return Err(ApiError::OutOfSpace); + } + + // Commit-then-verify: a peer cannot make us store bytes under a hash they + // do not hash to. + let actual = Hash::of(&body); + if actual != shard_hash { + return Err(ApiError::Unprocessable(format!( + "shard integrity failure: committed to {shard_hash} but body hashes to {actual}" + ))); + } + + if state.storage.chunk_store().has(&shard_hash).await { + return Ok(StatusCode::OK.into_response()); + } + + let len = body.len() as u64; + state + .storage + .chunk_store() + .put(&body) + .await + .map_err(|e| ApiError::Internal(e.to_string()))?; + state.budget.charge(len); + Ok(StatusCode::CREATED.into_response()) +} + +// ----------------------------------------------------------------- status + +#[derive(Serialize)] +struct StatusBody { + used_bytes: u64, + limit_bytes: Option, + max_object_bytes: u64, + default_tier: String, +} + +async fn status(State(state): State) -> Response { + axum::Json(StatusBody { + used_bytes: state.budget.used(), + limit_bytes: state.budget.limit(), + max_object_bytes: state.limits.max_object_bytes, + default_tier: Tier::default().to_string(), + }) + .into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_both_hash_forms() { + let h = Hash::of(b"x"); + assert_eq!(parse_hash(&h.to_hex()).unwrap(), h); + assert_eq!(parse_hash(&format!("blake3:{}", h.to_hex())).unwrap(), h); + assert!(parse_hash("nonsense").is_err()); + } + + #[test] + fn parses_ranges() { + assert_eq!(parse_range("bytes=0-99", 1000), Some((0, 100))); + assert_eq!(parse_range("bytes=100-", 1000), Some((100, 900))); + assert_eq!(parse_range("bytes=-50", 1000), Some((950, 50))); + // Clamped to the object. + assert_eq!(parse_range("bytes=990-5000", 1000), Some((990, 10))); + // Unsatisfiable or unsupported. + assert_eq!(parse_range("bytes=2000-3000", 1000), None); + assert_eq!(parse_range("bytes=50-10", 1000), None); + assert_eq!(parse_range("bytes=0-10,20-30", 1000), None); + assert_eq!(parse_range("items=0-10", 1000), None); + } + + #[test] + fn budget_tracks_and_never_wraps() { + let b = DiskBudget::new(Some(1000), 0); + assert!(b.would_fit(1000)); + assert!(!b.would_fit(1001)); + b.charge(600); + assert_eq!(b.used(), 600); + assert!(!b.would_fit(500)); + b.release(600); + assert_eq!(b.used(), 0); + // Over-release must not underflow into a huge number. + b.release(999_999); + assert_eq!(b.used(), 0); + } + + #[test] + fn unlimited_budget_always_fits() { + let b = DiskBudget::new(None, u64::MAX / 2); + assert!(b.would_fit(u64::MAX)); + } +} diff --git a/node/crates/c0mpute-gateway/tests/storage_api.rs b/node/crates/c0mpute-gateway/tests/storage_api.rs new file mode 100644 index 0000000..6174741 --- /dev/null +++ b/node/crates/c0mpute-gateway/tests/storage_api.rs @@ -0,0 +1,783 @@ +//! Integration tests for the storage HTTP API (CIP-002 acceptance criteria). +//! +//! Drives the real axum router in-process via `oneshot`, so these exercise +//! routing, headers, status codes and streaming — not just the storage engine +//! underneath. + +use std::sync::Arc; + +use axum::Router; +use axum::body::{Body, to_bytes}; +use axum::http::{Request, StatusCode, header}; +use c0mpute_gateway::auth::{AllowAll, SignedEnvelope, sign_envelope}; +use c0mpute_gateway::storage_api::{self, Limits, StorageApiState}; +use c0mpute_proto::Hash; +use c0mpute_store::{ChunkStore, Storage, Tier}; +use ed25519_dalek::SigningKey; +use tower::ServiceExt; + +const DID: &str = "did:coinpay:test"; + +fn tempdir(tag: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!( + "c0mpute-api-test-{tag}-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +async fn storage_at(dir: &std::path::Path) -> Storage { + Storage::new(ChunkStore::open(dir).await.unwrap()) +} + +async fn app_with(tag: &str, limits: Limits) -> (Router, Storage, std::path::PathBuf) { + let dir = tempdir(tag); + let storage = storage_at(&dir).await; + let state = StorageApiState::new(storage.clone(), Arc::new(AllowAll), limits); + (storage_api::router(state), storage, dir) +} + +async fn app(tag: &str) -> (Router, Storage, std::path::PathBuf) { + app_with(tag, Limits::default()).await +} + +fn varied(len: usize) -> Vec { + let mut out = Vec::with_capacity(len); + let mut state: u64 = 0x1234_5678_9abc_def0; + for _ in 0..len { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + out.push((state & 0xff) as u8); + } + out +} + +fn put_req(hash: &Hash, body: Vec) -> Request { + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::CONTENT_LENGTH, body.len()) + .body(Body::from(body)) + .unwrap() +} + +async fn body_bytes(resp: axum::response::Response) -> Vec { + to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap() + .to_vec() +} + +// ------------------------------------------------------------------ round trip + +#[tokio::test] +async fn put_then_get_round_trips() { + let (app, _s, _d) = app("roundtrip").await; + let data = varied(100_000); + let hash = Hash::of(&data); + + let resp = app + .clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + let resp = app + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_bytes(resp).await, data); +} + +#[tokio::test] +async fn accepts_blake3_prefixed_hashes() { + let (app, _s, _d) = app("prefix").await; + let data = varied(2048); + let hash = Hash::of(&data); + + let resp = app + .clone() + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/blake3:{}", hash.to_hex())) + .header(header::CONTENT_LENGTH, data.len()) + .body(Body::from(data.clone())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); +} + +#[tokio::test] +async fn put_is_idempotent() { + let (app, _s, _d) = app("idempotent").await; + let data = varied(5_000); + let hash = Hash::of(&data); + + let first = app + .clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + assert_eq!(first.status(), StatusCode::CREATED); + + let second = app.oneshot(put_req(&hash, data)).await.unwrap(); + assert_eq!(second.status(), StatusCode::OK, "re-PUT should not rewrite"); +} + +// ------------------------------------------------------------ commit & verify + +/// The property that makes the store trustworthy without trusting the +/// uploader: bytes must hash to the hash the client committed to in the URL. +#[tokio::test] +async fn wrong_committed_hash_is_422_and_stores_nothing() { + let (app, storage, dir) = app("badhash").await; + let data = varied(9_000); + let lie = Hash::of(b"a completely different object"); + + let resp = app.oneshot(put_req(&lie, data.clone())).await.unwrap(); + assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); + + assert!(!storage.has(&lie).await); + assert!(!storage.has(&Hash::of(&data)).await); + let mut shards = 0; + for e in walkdir(&dir.join("shards")) { + if e.is_file() { + shards += 1; + } + } + assert_eq!(shards, 0, "rejected write left {shards} shards behind"); +} + +/// Regression: a rejected PUT must not damage an object that already holds +/// the same bytes. +/// +/// Shards are content-addressed, so re-uploading an existing object's content +/// under a wrong committed hash produces identical shard hashes. Rolling back +/// every hash the failed write touched deleted the good object's shards — +/// one malformed request causing real data loss. Found by driving the server +/// with curl; the unit tests missed it because nothing was stored first. +#[tokio::test] +async fn rejected_put_does_not_destroy_an_existing_object() { + let (app, storage, _d) = app("rollback-safety").await; + let data = varied(300_000); + let hash = Hash::of(&data); + + let resp = app + .clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + // Same bytes, wrong committed hash. + let lie = Hash::of(b"something else entirely"); + let resp = app + .clone() + .oneshot(put_req(&lie, data.clone())) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY); + + // The good object must still be fully readable. + assert!(storage.has(&hash).await); + let resp = app + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!( + body_bytes(resp).await, + data, + "a rejected write destroyed an intact object" + ); +} + +#[tokio::test] +async fn shard_put_verifies_its_hash() { + let (app, _s, _d) = app("shardverify").await; + let bytes = varied(1024); + let real = Hash::of(&bytes); + let lie = Hash::of(b"not it"); + + let bad = app + .clone() + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/shards/{}", lie.to_hex())) + .body(Body::from(bytes.clone())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(bad.status(), StatusCode::UNPROCESSABLE_ENTITY); + + let good = app + .clone() + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/shards/{}", real.to_hex())) + .body(Body::from(bytes)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(good.status(), StatusCode::CREATED); + + let head = app + .oneshot( + Request::builder() + .method("HEAD") + .uri(format!("/storage/v1/shards/{}", real.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(head.status(), StatusCode::OK); +} + +// ----------------------------------------------------------------- durability + +#[tokio::test] +async fn survives_parity_budget_of_shard_loss() { + let (app, storage, _d) = app("parity").await; + let data = varied(200_000); + let hash = Hash::of(&data); + app.clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + + let manifest = storage.read_manifest(&hash).await.unwrap(); + // Standard = RS 10/14; four losses are inside the parity budget. + for shard in manifest.blocks[0].shards.iter().take(4) { + storage.chunk_store().delete(&shard.hash).await.unwrap(); + } + let resp = app + .clone() + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_bytes(resp).await, data); + + // A fifth loss is unrecoverable. + storage + .chunk_store() + .delete(&manifest.blocks[0].shards[4].hash) + .await + .unwrap(); + let resp = app + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + // The body streams, so the failure surfaces as a truncated stream rather + // than a status code — the status is already sent by then. What must not + // happen is silently returning wrong bytes. + let got = to_bytes(resp.into_body(), usize::MAX).await; + match got { + Err(_) => {} + Ok(b) => assert_ne!( + b.as_ref(), + data.as_slice(), + "returned corrupt data as success" + ), + } +} + +// --------------------------------------------------------------------- ranges + +#[tokio::test] +async fn range_requests_return_exact_bytes() { + let (app, _s, _d) = app("range").await; + let data = varied(3_000_000); + let hash = Hash::of(&data); + app.clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + + for (spec, start, len) in [ + ("bytes=1000000-1004095", 1_000_000usize, 4096usize), + ("bytes=0-0", 0, 1), + ("bytes=2999000-", 2_999_000, 1000), + ("bytes=-500", 2_999_500, 500), + ] { + let resp = app + .clone() + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::RANGE, spec) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::PARTIAL_CONTENT, "spec {spec}"); + let cr = resp + .headers() + .get(header::CONTENT_RANGE) + .unwrap() + .to_str() + .unwrap() + .to_string(); + assert!(cr.ends_with("/3000000"), "content-range was {cr}"); + assert_eq!( + body_bytes(resp).await, + &data[start..start + len], + "spec {spec}" + ); + } +} + +#[tokio::test] +async fn unsatisfiable_range_is_416() { + let (app, _s, _d) = app("range416").await; + let data = varied(1000); + let hash = Hash::of(&data); + app.clone().oneshot(put_req(&hash, data)).await.unwrap(); + + let resp = app + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::RANGE, "bytes=5000-6000") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::RANGE_NOT_SATISFIABLE); +} + +// ----------------------------------------------------------------- metadata + +#[tokio::test] +async fn head_reports_length_and_tier_without_a_body() { + let (app, _s, _d) = app("head").await; + let data = varied(4321); + let hash = Hash::of(&data); + app.clone() + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::CONTENT_LENGTH, data.len()) + .header(storage_api::TIER_HEADER, "critical") + .body(Body::from(data.clone())) + .unwrap(), + ) + .await + .unwrap(); + + let resp = app + .oneshot( + Request::builder() + .method("HEAD") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!( + resp.headers().get(header::CONTENT_LENGTH).unwrap(), + &data.len().to_string() + ); + assert_eq!(resp.headers().get("x-c0mpute-tier").unwrap(), "critical"); + assert!(body_bytes(resp).await.is_empty()); +} + +#[tokio::test] +async fn manifest_endpoint_describes_the_layout() { + let (app, _s, _d) = app("manifest").await; + let data = varied(50_000); + let hash = Hash::of(&data); + app.clone().oneshot(put_req(&hash, data)).await.unwrap(); + + let resp = app + .oneshot( + Request::get(format!("/storage/v1/manifests/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let m: serde_json::Value = serde_json::from_slice(&body_bytes(resp).await).unwrap(); + assert_eq!(m["version"], 2); + assert_eq!(m["tier"], "standard"); + assert_eq!(m["k"], 10); + assert_eq!(m["parity"], 4); + assert_eq!(m["blocks"].as_array().unwrap().len(), 1); + assert_eq!(m["blocks"][0]["shards"].as_array().unwrap().len(), 14); +} + +#[tokio::test] +async fn tier_header_selects_redundancy() { + for (tier, shards) in [("hot", 3), ("standard", 14), ("critical", 32)] { + let (app, storage, _d) = app(&format!("tier-{tier}")).await; + let data = varied(10_000); + let hash = Hash::of(&data); + let resp = app + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::CONTENT_LENGTH, data.len()) + .header(storage_api::TIER_HEADER, tier) + .body(Body::from(data)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + let m = storage.read_manifest(&hash).await.unwrap(); + assert_eq!(m.tier, tier.parse::().unwrap()); + assert_eq!(m.shard_count(), shards, "tier {tier}"); + } +} + +#[tokio::test] +async fn unknown_tier_is_400() { + let (app, _s, _d) = app("badtier").await; + let data = varied(100); + let hash = Hash::of(&data); + let resp = app + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::CONTENT_LENGTH, data.len()) + .header(storage_api::TIER_HEADER, "glacier") + .body(Body::from(data)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +// -------------------------------------------------------------------- errors + +#[tokio::test] +async fn missing_content_length_is_400() { + let (app, _s, _d) = app("nolen").await; + let data = varied(100); + let hash = Hash::of(&data); + let resp = app + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::from(data)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn unknown_object_is_404() { + let (app, _s, _d) = app("missing").await; + let hash = Hash::of(b"never stored"); + for path in ["objects", "manifests"] { + let resp = app + .clone() + .oneshot( + Request::get(format!("/storage/v1/{path}/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND, "path {path}"); + } +} + +#[tokio::test] +async fn malformed_hash_is_400() { + let (app, _s, _d) = app("badhex").await; + let resp = app + .oneshot( + Request::get("/storage/v1/objects/not-a-hash") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn object_over_the_limit_is_413() { + let (app, _s, _d) = app_with( + "toolarge", + Limits { + max_object_bytes: 1024, + disk_budget_bytes: None, + }, + ) + .await; + let data = varied(5000); + let hash = Hash::of(&data); + let resp = app.oneshot(put_req(&hash, data)).await.unwrap(); + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); +} + +#[tokio::test] +async fn exhausted_disk_budget_is_507() { + let (app, _s, _d) = app_with( + "nospace", + Limits { + max_object_bytes: u64::MAX, + // 10 KiB of raw disk; a 50 KiB object at 1.4x needs 70 KiB. + disk_budget_bytes: Some(10_240), + }, + ) + .await; + let data = varied(50_000); + let hash = Hash::of(&data); + let resp = app.oneshot(put_req(&hash, data)).await.unwrap(); + assert_eq!(resp.status(), StatusCode::INSUFFICIENT_STORAGE); +} + +#[tokio::test] +async fn delete_frees_the_object_and_its_budget() { + let (app, storage, _d) = app("delete").await; + let data = varied(20_000); + let hash = Hash::of(&data); + app.clone().oneshot(put_req(&hash, data)).await.unwrap(); + assert!(storage.has(&hash).await); + + let resp = app + .clone() + .oneshot( + Request::builder() + .method("DELETE") + .uri(format!("/storage/v1/objects/{}", hash.to_hex())) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + assert!(!storage.has(&hash).await); + + let resp = app + .oneshot( + Request::get("/storage/v1/status") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let s: serde_json::Value = serde_json::from_slice(&body_bytes(resp).await).unwrap(); + assert_eq!(s["used_bytes"], 0); +} + +// ----------------------------------------------------------------------- auth + +fn signed_app(tag: &str, sk: &SigningKey) -> (Router, Storage, std::path::PathBuf) { + let dir = tempdir(tag); + let storage = futures::executor::block_on(async { storage_at(&dir).await }); + let auth = SignedEnvelope::new().with_key(DID, sk.verifying_key()); + let state = StorageApiState::new(storage.clone(), Arc::new(auth), Limits::default()); + (storage_api::router(state), storage, dir) +} + +#[tokio::test] +async fn writes_require_an_envelope_but_reads_do_not() { + let sk = SigningKey::from_bytes(&[42u8; 32]); + let (app, _s, _d) = signed_app("auth", &sk); + let data = varied(3_000); + let hash = Hash::of(&data); + let path = format!("/storage/v1/objects/{}", hash.to_hex()); + + // Unsigned write: refused. + let resp = app + .clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + + // Signed write: accepted. + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + let env = sign_envelope(DID, &sk, "PUT", &path, ts, &hash.to_hex()); + let resp = app + .clone() + .oneshot( + Request::builder() + .method("PUT") + .uri(&path) + .header(header::CONTENT_LENGTH, data.len()) + .header("x-coinpay-auth", env) + .body(Body::from(data.clone())) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + // Read needs nothing: the hash is the capability. + let resp = app + .clone() + .oneshot(Request::get(&path).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_bytes(resp).await, data); + + // Unsigned delete: refused. + let resp = app + .oneshot( + Request::builder() + .method("DELETE") + .uri(&path) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} + +/// An envelope minted for one object must not authorize a write to another. +#[tokio::test] +async fn envelope_cannot_be_replayed_against_another_object() { + let sk = SigningKey::from_bytes(&[42u8; 32]); + let (app, _s, _d) = signed_app("replay", &sk); + let a = varied(1_000); + let b = varied(2_000); + let (ha, hb) = (Hash::of(&a), Hash::of(&b)); + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs(); + + let env_for_a = sign_envelope( + DID, + &sk, + "PUT", + &format!("/storage/v1/objects/{}", ha.to_hex()), + ts, + &ha.to_hex(), + ); + + let resp = app + .oneshot( + Request::builder() + .method("PUT") + .uri(format!("/storage/v1/objects/{}", hb.to_hex())) + .header(header::CONTENT_LENGTH, b.len()) + .header("x-coinpay-auth", env_for_a) + .body(Body::from(b)) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} + +// ------------------------------------------------------------------ streaming + +/// Memory must be bounded by block size, not object size. The CIP's headline +/// figure is 1 GiB under 200 MB RSS; this runs a smaller object so the suite +/// stays fast, and asserts the same property. +#[cfg(target_os = "linux")] +#[tokio::test] +async fn large_object_streams_without_buffering_it_all() { + fn rss_bytes() -> u64 { + let s = std::fs::read_to_string("/proc/self/statm").unwrap(); + let pages: u64 = s.split_whitespace().nth(1).unwrap().parse().unwrap(); + pages * 4096 + } + + let (app, _s, _d) = app("bigstream").await; + let size = 64 * 1024 * 1024; + let data = varied(size); + let hash = Hash::of(&data); + + let before = rss_bytes(); + let resp = app + .clone() + .oneshot(put_req(&hash, data.clone())) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + + // Read it back a range at a time; a whole-object read would legitimately + // allocate the whole object, which is what `read_stream` exists to avoid. + let resp = app + .oneshot( + Request::get(format!("/storage/v1/objects/{}", hash.to_hex())) + .header(header::RANGE, "bytes=60000000-60001023") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::PARTIAL_CONTENT); + assert_eq!( + body_bytes(resp).await, + &data[60_000_000..60_001_024], + "range read of a large object" + ); + + let growth = rss_bytes().saturating_sub(before); + // Generous: the point is that it is not proportional to the 64 MiB object. + assert!( + growth < 48 * 1024 * 1024, + "RSS grew {growth} bytes writing+reading a {size}-byte object; \ + streaming is not bounding memory" + ); +} + +fn walkdir(dir: &std::path::Path) -> Vec { + let mut out = Vec::new(); + let Ok(rd) = std::fs::read_dir(dir) else { + return out; + }; + for e in rd.flatten() { + let p = e.path(); + if p.is_dir() { + out.extend(walkdir(&p)); + } else { + out.push(p); + } + } + out +} diff --git a/node/crates/c0mpute-store/Cargo.toml b/node/crates/c0mpute-store/Cargo.toml index e241e64..ed0e831 100644 --- a/node/crates/c0mpute-store/Cargo.toml +++ b/node/crates/c0mpute-store/Cargo.toml @@ -15,3 +15,6 @@ hex = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } reed-solomon-erasure = { workspace = true } +bytes = { workspace = true } +futures = { workspace = true } +thiserror = { workspace = true } diff --git a/node/crates/c0mpute-store/src/lib.rs b/node/crates/c0mpute-store/src/lib.rs index 5923e8b..6962293 100644 --- a/node/crates/c0mpute-store/src/lib.rs +++ b/node/crates/c0mpute-store/src/lib.rs @@ -16,8 +16,13 @@ pub mod erasure; pub mod storage; +pub mod tier; -pub use storage::{Storage, ObjectManifest, ShardEntry}; +pub use storage::{ + BlockEntry, DEFAULT_BLOCK_SIZE, MANIFEST_VERSION, MAX_BLOCK_SIZE, ObjectManifest, ShardEntry, + Storage, block_size_for, +}; +pub use tier::Tier; use std::path::{Path, PathBuf}; @@ -49,6 +54,24 @@ impl ChunkStore { .join(&hex) } + /// Write bytes, reporting whether this call *created* the chunk (`true`) + /// or found it already present (`false`). + /// + /// Callers that may need to undo a partial write must use this rather than + /// [`ChunkStore::put`]: chunks are content-addressed and therefore shared + /// between objects, so rolling back a write by deleting every hash it + /// touched will happily delete chunks a *different, intact* object still + /// depends on. Deleting only what you created is the difference between + /// discarding junk and destroying someone else's data. + pub async fn put_new(&self, bytes: &[u8]) -> Result<(Hash, bool)> { + let hash = Hash::of(bytes); + if self.has(&hash).await { + return Ok((hash, false)); + } + self.put(bytes).await?; + Ok((hash, true)) + } + /// Write bytes; the returned hash is computed and is the storage key. #[instrument(skip(self, bytes))] pub async fn put(&self, bytes: &[u8]) -> Result { @@ -134,10 +157,7 @@ mod tests { } fn tempdir() -> PathBuf { - let dir = std::env::temp_dir().join(format!( - "c0mpute-store-test-{}", - uuid_like_suffix() - )); + let dir = std::env::temp_dir().join(format!("c0mpute-store-test-{}", uuid_like_suffix())); std::fs::create_dir_all(&dir).unwrap(); dir } diff --git a/node/crates/c0mpute-store/src/storage.rs b/node/crates/c0mpute-store/src/storage.rs index 0aa090f..d1746f7 100644 --- a/node/crates/c0mpute-store/src/storage.rs +++ b/node/crates/c0mpute-store/src/storage.rs @@ -1,50 +1,196 @@ -//! High-level erasure-coded storage on top of `ChunkStore`. +//! Erasure-coded object storage on top of `ChunkStore` (CIP-002). //! -//! `Storage::put(bytes)` → -//! 1. Hash the plaintext (object_hash = blake3(plaintext)). -//! 2. RS-encode into 14 shards (k=10, parity=4). -//! 3. Write each shard into the underlying ChunkStore (keyed by its -//! own blake3 hash). -//! 4. Persist a manifest at `manifests/.json` mapping -//! object → shard hashes + indices. +//! An object is split into fixed-size **blocks**; each block is independently +//! Reed-Solomon encoded into `n` shards, and each shard is written to the +//! content-addressed chunk store under its own blake3 hash. A manifest records +//! the block and shard layout. //! -//! `Storage::get(object_hash)` → -//! 1. Load the manifest. -//! 2. Read each shard from the chunk store; missing shards are -//! tolerated up to the parity budget. -//! 3. RS-decode and return. +//! Blocks are what make the rest of the storage program possible: //! -//! Single-node today (Phase 1 of DIP-0012). Phase 2 distributes -//! shards across peers; the `[ShardEntry::host_hint]` field already -//! exists for that. +//! * memory is bounded by block size, not object size, so a 1 TiB object +//! does not need 1 TiB of RAM to write or read; +//! * `get_range` fetches only the blocks a byte range touches, which is what +//! random-access file reads (CIP-007) are built on; +//! * a single damaged block is repairable (CIP-005) without touching the +//! rest of the object. +//! +//! Block size scales with object size so that manifests stay small — see +//! [`block_size_for`]. use std::path::PathBuf; +use std::pin::Pin; -use anyhow::{Context, Result, anyhow}; +use anyhow::{Context, Result, anyhow, bail}; +use bytes::Bytes; use c0mpute_proto::Hash; +use futures::{Stream, StreamExt}; use serde::{Deserialize, Serialize}; use tokio::fs; use tracing::{debug, warn}; use crate::ChunkStore; -use crate::erasure::{self, DEFAULT_K, DEFAULT_PARITY, Shard}; +use crate::erasure::{self, Shard}; +use crate::tier::Tier; + +/// Current manifest format version. +pub const MANIFEST_VERSION: u8 = 2; + +/// Smallest (and default) block size: 4 MiB. +pub const DEFAULT_BLOCK_SIZE: u32 = 4 * 1024 * 1024; + +/// Largest block size: 256 MiB. +pub const MAX_BLOCK_SIZE: u32 = 256 * 1024 * 1024; + +/// Manifests stay small by targeting at most this many blocks per object. +pub const TARGET_BLOCKS_PER_OBJECT: u64 = 4096; + +/// Choose a block size for an object of `len` bytes. +/// +/// Fixed 4 MiB blocks would give a 1 TiB object ~262k blocks and a manifest in +/// the tens of megabytes, which then needs its own durability story. Doubling +/// the block size until the object fits in [`TARGET_BLOCKS_PER_OBJECT`] keeps +/// every manifest under about a megabyte. +pub fn block_size_for(len: u64) -> u32 { + let ideal = len.div_ceil(TARGET_BLOCKS_PER_OBJECT); + let mut size = DEFAULT_BLOCK_SIZE as u64; + while size < ideal && size < MAX_BLOCK_SIZE as u64 { + size *= 2; + } + size.min(MAX_BLOCK_SIZE as u64) as u32 +} +/// One shard of one block. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] pub struct ShardEntry { pub index: u8, pub hash: Hash, - /// Host hint for cross-node placement (peer id). `None` = local. - /// Populated by Phase 2 placement; ignored by single-node mode. + /// Peer holding this shard. `None` means local-only. Populated by + /// cross-node placement (CIP-003). + /// + /// This is a *hint*, not the source of truth — repair (CIP-005) relocates + /// shards without being able to sign the customer's manifest, so a stale + /// hint is normal and readers fall back to DHT provider records. + #[serde(default, skip_serializing_if = "Option::is_none")] pub host_hint: Option, } +/// One block of an object: its plaintext hash, plaintext length, and shards. #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub struct BlockEntry { + pub index: u32, + /// Plaintext length of this block, before RS padding. The final block of + /// an object is usually short. + pub len: u32, + /// blake3 of this block's plaintext. Lets a reader verify per block + /// instead of only at the end of a whole object. + pub hash: Hash, + pub shards: Vec, +} + +/// Maps an object hash onto the blocks and shards that reconstruct it. +#[derive(Clone, Debug, Serialize, PartialEq, Eq)] pub struct ObjectManifest { + pub version: u8, pub object_hash: Hash, pub original_len: u64, + pub block_size: u32, pub k: u8, pub parity: u8, - pub shards: Vec, + pub tier: Tier, + pub blocks: Vec, +} + +impl ObjectManifest { + pub fn n(&self) -> usize { + self.k as usize + self.parity as usize + } + + /// Total shards recorded across every block. + pub fn shard_count(&self) -> usize { + self.blocks.iter().map(|b| b.shards.len()).sum() + } + + /// Which block indices cover `[offset, offset + len)`. + fn blocks_for_range(&self, offset: u64, len: u64) -> std::ops::Range { + if len == 0 || offset >= self.original_len { + return 0..0; + } + let end = (offset + len).min(self.original_len); + let bs = self.block_size as u64; + let first = (offset / bs) as usize; + let last = ((end - 1) / bs) as usize; + first..(last + 1).min(self.blocks.len()) + } +} + +/// Wire form that accepts both v1 (flat `shards`, one implicit block) and v2 +/// manifests. There is no production v1 data to migrate, but keeping the shim +/// means the original round-trip tests stay meaningful. +#[derive(Deserialize)] +struct RawManifest { + #[serde(default = "v1_version")] + version: u8, + object_hash: Hash, + original_len: u64, + #[serde(default)] + block_size: Option, + k: u8, + parity: u8, + #[serde(default)] + tier: Option, + #[serde(default)] + blocks: Option>, + /// v1 only: shards of the single implicit block. + #[serde(default)] + shards: Option>, +} + +fn v1_version() -> u8 { + 1 +} + +impl<'de> Deserialize<'de> for ObjectManifest { + fn deserialize>(d: D) -> Result { + let raw = RawManifest::deserialize(d)?; + let tier = raw + .tier + .or_else(|| Tier::from_params(raw.k, raw.parity)) + .unwrap_or_default(); + + let blocks = match (raw.blocks, raw.shards) { + (Some(blocks), _) => blocks, + // v1: one block covering the whole object. The block plaintext is + // the object plaintext, so they share a hash. + (None, Some(shards)) => vec![BlockEntry { + index: 0, + len: u32::try_from(raw.original_len).map_err(|_| { + serde::de::Error::custom("v1 manifest longer than one block can hold") + })?, + hash: raw.object_hash, + shards, + }], + (None, None) => { + return Err(serde::de::Error::custom( + "manifest has neither blocks nor shards", + )); + } + }; + + let block_size = raw.block_size.unwrap_or_else(|| { + u32::try_from(raw.original_len.max(1)).unwrap_or(DEFAULT_BLOCK_SIZE) + }); + + Ok(ObjectManifest { + version: raw.version, + object_hash: raw.object_hash, + original_len: raw.original_len, + block_size, + k: raw.k, + parity: raw.parity, + tier, + blocks, + }) + } } #[derive(Clone, Debug)] @@ -67,106 +213,382 @@ impl Storage { .root() .join("manifests") .join(&hex[0..2]) - .join(format!("{}.json", hex)) + .join(format!("{hex}.json")) } - /// Store an object. Returns the object's blake3 hash + the manifest. + // ---------------------------------------------------------------- writes + + /// Store an in-memory object at the default tier. + /// + /// Thin wrapper over [`Storage::put_stream`] so there is one write path, + /// not two that can drift. pub async fn put(&self, data: &[u8]) -> Result { - let object_hash = Hash::of(data); - let (shards, original_len) = erasure::encode(data, DEFAULT_K, DEFAULT_PARITY)?; + self.put_tiered(data, Tier::default()).await + } - let mut entries = Vec::with_capacity(shards.len()); - for s in &shards { - // Write each shard into the underlying chunk store, keyed - // by its own hash. The chunk store handles atomic writes. - let h = self.inner.put(&s.bytes).await?; - entries.push(ShardEntry { - index: s.index, - hash: h, - host_hint: None, - }); + /// Store an in-memory object at a given tier. + pub async fn put_tiered(&self, data: &[u8], tier: Tier) -> Result { + let len = data.len() as u64; + let chunk = Bytes::copy_from_slice(data); + let stream = futures::stream::once(async move { Ok(chunk) }); + self.put_stream(stream, None, tier, Some(len)).await + } + + /// Store an object from a byte stream, RS-encoding block by block. + /// + /// `expected` is the hash the caller committed to. If the bytes hash to + /// anything else the write is rejected *and every shard it wrote is + /// removed* — a caller must not be able to leave junk behind by lying + /// about a hash. + /// + /// `size_hint` (an HTTP `Content-Length`, typically) picks the block size. + /// Without it every object gets [`DEFAULT_BLOCK_SIZE`] blocks, which is + /// correct but produces large manifests for large objects. + pub async fn put_stream( + &self, + stream: S, + expected: Option, + tier: Tier, + size_hint: Option, + ) -> Result + where + S: Stream>, + { + let block_size = size_hint.map(block_size_for).unwrap_or(DEFAULT_BLOCK_SIZE); + let (k, parity) = (tier.k(), tier.parity()); + + let mut object_hasher = blake3::Hasher::new(); + let mut blocks: Vec = Vec::new(); + // Every shard written so far, so a hash mismatch can be fully undone. + let mut written: Vec = Vec::new(); + let mut buf: Vec = Vec::with_capacity(block_size as usize); + let mut total_len: u64 = 0; + + let mut stream = Box::pin(stream); + let mut failed: Option = None; + + while let Some(item) = stream.next().await { + let chunk = match item { + Ok(c) => c, + Err(e) => { + failed = Some(e); + break; + } + }; + object_hasher.update(&chunk); + total_len += chunk.len() as u64; + let mut rest: &[u8] = &chunk; + while !rest.is_empty() { + let want = block_size as usize - buf.len(); + let take = want.min(rest.len()); + buf.extend_from_slice(&rest[..take]); + rest = &rest[take..]; + if buf.len() == block_size as usize { + match self.seal_block(&buf, blocks.len() as u32, k, parity).await { + Ok((entry, created)) => { + written.extend(created); + blocks.push(entry); + } + Err(e) => { + failed = Some(e); + break; + } + } + buf.clear(); + } + } + if failed.is_some() { + break; + } + } + + // Trailing partial block. + if failed.is_none() && !buf.is_empty() { + match self.seal_block(&buf, blocks.len() as u32, k, parity).await { + Ok((entry, created)) => { + written.extend(created); + blocks.push(entry); + } + Err(e) => failed = Some(e), + } + } + + if let Some(e) = failed { + self.rollback(&written).await; + return Err(e); + } + + let object_hash = Hash(*object_hasher.finalize().as_bytes()); + if let Some(want) = expected + && want != object_hash + { + self.rollback(&written).await; + bail!("object integrity failure: committed to {want} but body hashes to {object_hash}"); + } + + // A zero-length object still needs one (empty) block so reads have + // something to iterate. + if blocks.is_empty() { + let (entry, _) = self.seal_block(&[], 0, k, parity).await?; + blocks.push(entry); } let manifest = ObjectManifest { + version: MANIFEST_VERSION, object_hash, - original_len: original_len as u64, - k: DEFAULT_K as u8, - parity: DEFAULT_PARITY as u8, - shards: entries, + original_len: total_len, + block_size, + k: k as u8, + parity: parity as u8, + tier, + blocks, }; - self.write_manifest(&manifest).await?; debug!( object_hash = %object_hash, - shard_count = manifest.shards.len(), + blocks = manifest.blocks.len(), + shards = manifest.shard_count(), + %tier, "stored object" ); Ok(manifest) } - /// Read an object back. Tolerates up to `parity` missing shards. + /// RS-encode one block and write its shards. + /// + /// Returns the block entry plus the hashes this call actually created, so + /// a failed write can be rolled back without touching shards that were + /// already on disk for some other object. + async fn seal_block( + &self, + plaintext: &[u8], + index: u32, + k: usize, + parity: usize, + ) -> Result<(BlockEntry, Vec)> { + let (shards, _) = erasure::encode(plaintext, k, parity)?; + let mut entries = Vec::with_capacity(shards.len()); + let mut created = Vec::new(); + for s in &shards { + let (hash, is_new) = self.inner.put_new(&s.bytes).await?; + if is_new { + created.push(hash); + } + entries.push(ShardEntry { + index: s.index, + hash, + host_hint: None, + }); + } + Ok(( + BlockEntry { + index, + len: plaintext.len() as u32, + hash: Hash::of(plaintext), + shards: entries, + }, + created, + )) + } + + /// Remove the shards *this* write created, after it failed. + /// + /// Only newly-created hashes are passed in, and that distinction is + /// load-bearing rather than an optimisation. Shards are content-addressed + /// and therefore shared: PUTting the bytes of an object that already + /// exists, under a wrong committed hash, produces exactly the same shard + /// hashes. Rolling back every hash the write touched would delete the + /// intact object's shards — data loss triggered by one malformed request + /// from anyone who can obtain the content. + async fn rollback(&self, hashes: &[Hash]) { + for h in hashes { + if let Err(e) = self.inner.delete(h).await { + warn!(hash = %h, err = %e, "rollback failed to remove shard"); + } + } + } + + // ----------------------------------------------------------------- reads + + /// Read a whole object into memory. Prefer [`Storage::read_stream`] for + /// anything that might be large. pub async fn get(&self, object_hash: &Hash) -> Result> { let manifest = self.read_manifest(object_hash).await?; - let n = (manifest.k + manifest.parity) as usize; - let mut received: Vec> = vec![None; n]; + 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 {}", + manifest.object_hash, + actual + ); + } + Ok(out) + } + + /// Stream an object back, reconstructing one block at a time. + /// + /// Memory stays bounded by block size regardless of object size. + pub fn read_stream( + &self, + manifest: ObjectManifest, + ) -> Pin> + Send>> { + let storage = self.clone(); + Box::pin(futures::stream::unfold( + (storage, manifest, 0usize), + |(storage, manifest, i)| async move { + if i >= manifest.blocks.len() { + return None; + } + let item = storage.read_block(&manifest, i).await.map(Bytes::from); + Some((item, (storage, manifest, i + 1))) + }, + )) + } + + /// Read `[offset, offset + len)` of an object, touching only the blocks + /// that range covers. + pub async fn get_range(&self, object_hash: &Hash, offset: u64, len: u64) -> Result> { + let manifest = self.read_manifest(object_hash).await?; + self.get_range_with(&manifest, offset, len).await + } - for entry in &manifest.shards { - match self.inner.get(&entry.hash).await { + pub async fn get_range_with( + &self, + manifest: &ObjectManifest, + offset: u64, + len: u64, + ) -> Result> { + let range = manifest.blocks_for_range(offset, len); + if range.is_empty() { + return Ok(Vec::new()); + } + let bs = manifest.block_size as u64; + let end = (offset + len).min(manifest.original_len); + let mut out = Vec::with_capacity((end - offset) as usize); + + for i in range { + let block = self.read_block(manifest, i).await?; + let block_start = i as u64 * bs; + let block_end = block_start + block.len() as u64; + let take_from = offset.max(block_start) - block_start; + let take_to = end.min(block_end) - block_start; + out.extend_from_slice(&block[take_from as usize..take_to as usize]); + } + Ok(out) + } + + /// Reconstruct one block, tolerating up to `parity` missing shards. + 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 n = manifest.n(); + let mut received: Vec> = vec![None; n]; + let mut found = 0usize; + for shard in &entry.shards { + match self.inner.get(&shard.hash).await { Ok(bytes) => { - received[entry.index as usize] = Some(Shard { - index: entry.index, + received[shard.index as usize] = Some(Shard { + index: shard.index, bytes, }); + found += 1; } Err(e) => { warn!( - object_hash = %object_hash, - shard_index = entry.index, + object_hash = %manifest.object_hash, + block = index, + shard_index = shard.index, err = %e, "shard unreadable; falling back to parity" ); } } } + if found < manifest.k as usize { + bail!( + "block {index} of object {}: need at least {} shards, found {found}", + manifest.object_hash, + manifest.k + ); + } - let plaintext = erasure::decode( + let mut plaintext = erasure::decode( received, manifest.k as usize, manifest.parity as usize, - manifest.original_len as usize, + entry.len as usize, )?; + plaintext.truncate(entry.len as usize); - // Verify integrity end-to-end: the plaintext we just decoded - // must match the manifest's object_hash. let actual = Hash::of(&plaintext); - if actual != manifest.object_hash { - anyhow::bail!( - "object integrity failure: manifest says {} but decoded bytes hash to {}", + if actual != entry.hash { + bail!( + "block {index} of object {}: integrity failure, manifest says {} but bytes hash to {actual}", manifest.object_hash, - actual + entry.hash ); } Ok(plaintext) } + // ------------------------------------------------------------- manifests + pub async fn has(&self, object_hash: &Hash) -> bool { fs::metadata(self.manifest_path(object_hash)).await.is_ok() } - /// Delete an object's manifest + every shard it points at. + /// Every object this node holds a manifest for. + /// + /// Walks the manifest directory rather than keeping an index: the + /// authoritative object list for a customer lives in their volume + /// (CIP-004), and this is only "what is on this disk". + pub async fn list(&self) -> Result> { + let root = self.inner.root().join("manifests"); + let mut out = Vec::new(); + let mut dirs = vec![root]; + while let Some(dir) = dirs.pop() { + let mut rd = match fs::read_dir(&dir).await { + Ok(rd) => rd, + Err(_) => continue, + }; + while let Some(entry) = rd.next_entry().await? { + let path = entry.path(); + if entry.file_type().await?.is_dir() { + dirs.push(path); + } else if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) + && let Ok(h) = Hash::from_hex(stem) + { + out.push(h); + } + } + } + out.sort_by_key(|h| h.to_hex()); + Ok(out) + } + + /// Delete an object's manifest and every shard it points at. pub async fn delete(&self, object_hash: &Hash) -> Result<()> { let manifest = match self.read_manifest(object_hash).await { Ok(m) => m, Err(_) => return Ok(()), }; - for entry in &manifest.shards { - let _ = self.inner.delete(&entry.hash).await; + for block in &manifest.blocks { + for shard in &block.shards { + let _ = self.inner.delete(&shard.hash).await; + } } let _ = fs::remove_file(self.manifest_path(object_hash)).await; Ok(()) } - async fn write_manifest(&self, m: &ObjectManifest) -> Result<()> { + pub async fn write_manifest(&self, m: &ObjectManifest) -> Result<()> { let path = self.manifest_path(&m.object_hash); if let Some(parent) = path.parent() { fs::create_dir_all(parent).await?; @@ -178,18 +600,16 @@ impl Storage { Ok(()) } - async fn read_manifest(&self, object_hash: &Hash) -> Result { + pub async fn read_manifest(&self, object_hash: &Hash) -> Result { let path = self.manifest_path(object_hash); let bytes = fs::read(&path) .await .with_context(|| format!("read manifest {}", path.display()))?; - let m: ObjectManifest = serde_json::from_slice(&bytes) - .context("parse manifest JSON")?; + let m: ObjectManifest = serde_json::from_slice(&bytes).context("parse manifest JSON")?; if m.object_hash != *object_hash { return Err(anyhow!( - "manifest object_hash {} != requested {}", - m.object_hash, - object_hash + "manifest object_hash {} != requested {object_hash}", + m.object_hash )); } Ok(m) @@ -213,23 +633,12 @@ mod tests { Storage::new(cs) } - #[tokio::test] - async fn put_get_roundtrip() { - let s = store().await; - let data = b"hello c0mpute erasure-coded storage".repeat(50); - let m = s.put(&data).await.unwrap(); - assert_eq!(m.shards.len(), 14); - let out = s.get(&m.object_hash).await.unwrap(); - assert_eq!(out, data); - } - - /// Pseudo-random heterogeneous bytes. Critical for shard-level - /// failure tests because identical shard content collides in the - /// content-addressed chunk store (a real, desirable dedup property - /// but it makes "lose 4 of 14" meaningless if the 4 share a hash). + /// Pseudo-random bytes. Critical for shard-loss tests: identical shard + /// content collides in the content-addressed chunk store (real, desirable + /// dedup) which would make "lose 4 of 14" meaningless. fn varied(len: usize) -> Vec { let mut out = Vec::with_capacity(len); - let mut state: u64 = 0x9e3779b97f4a7c15; + let mut state: u64 = 0x9e37_79b9_7f4a_7c15; for _ in 0..len { state ^= state << 13; state ^= state >> 7; @@ -239,16 +648,26 @@ mod tests { out } + #[tokio::test] + async fn put_get_roundtrip() { + let s = store().await; + let data = b"hello c0mpute erasure-coded storage".repeat(50); + let m = s.put(&data).await.unwrap(); + assert_eq!(m.version, MANIFEST_VERSION); + assert_eq!(m.tier, Tier::Standard); + assert_eq!(m.shard_count(), 14); + assert_eq!(s.get(&m.object_hash).await.unwrap(), data); + } + #[tokio::test] async fn survives_four_lost_shards() { let s = store().await; let data = varied(50_000); let m = s.put(&data).await.unwrap(); - for entry in m.shards.iter().take(4) { - s.inner.delete(&entry.hash).await.unwrap(); + for shard in m.blocks[0].shards.iter().take(4) { + s.inner.delete(&shard.hash).await.unwrap(); } - let out = s.get(&m.object_hash).await.unwrap(); - assert_eq!(out, data); + assert_eq!(s.get(&m.object_hash).await.unwrap(), data); } #[tokio::test] @@ -256,13 +675,304 @@ mod tests { let s = store().await; let data = varied(20_000); let m = s.put(&data).await.unwrap(); - for entry in m.shards.iter().take(5) { - s.inner.delete(&entry.hash).await.unwrap(); + for shard in m.blocks[0].shards.iter().take(5) { + s.inner.delete(&shard.hash).await.unwrap(); } - let err = s.get(&m.object_hash).await.unwrap_err(); + let err = s.get(&m.object_hash).await.unwrap_err().to_string(); + assert!(err.contains("need at least"), "unexpected error: {err}"); + } + + #[tokio::test] + async fn empty_object_roundtrips() { + let s = store().await; + let m = s.put(b"").await.unwrap(); + assert_eq!(m.original_len, 0); + assert_eq!(s.get(&m.object_hash).await.unwrap(), Vec::::new()); + } + + #[tokio::test] + async fn multi_block_object_roundtrips() { + let s = store().await; + // Three blocks and a bit, at the smallest block size. + let data = varied(DEFAULT_BLOCK_SIZE as usize * 3 + 1234); + let m = s.put(&data).await.unwrap(); + assert_eq!(m.blocks.len(), 4); + assert_eq!(m.blocks[3].len, 1234); + assert_eq!(s.get(&m.object_hash).await.unwrap(), data); + } + + /// `hot` is RS with k=1, which makes every parity shard byte-identical to + /// the data shard — i.e. genuine 3-copy replication rather than coding. + /// + /// That identity is the point: repair is a copy from a survivor (1x + /// amplification, per CIP-001) instead of a k-shard reconstruction. It + /// also means that *on one node* the content-addressed store holds a + /// single chunk for all three, since they share a hash. Durability comes + /// from placing that hash on 3 distinct hosts (CIP-003), not from three + /// distinct byte strings on one disk. + #[tokio::test] + async fn hot_tier_is_true_replication() { + let s = store().await; + let data = varied(10_000); + let m = s.put_tiered(&data, Tier::Hot).await.unwrap(); + assert_eq!(m.shard_count(), 3); + assert_eq!(m.k, 1); + assert_eq!(m.parity, 2); + + let hashes: Vec<_> = m.blocks[0].shards.iter().map(|s| s.hash).collect(); assert!( - err.to_string().contains("at least") || err.to_string().contains("need"), - "expected decode-shortage error, got: {err}" + hashes.iter().all(|h| *h == hashes[0]), + "hot-tier shards should be identical copies, got {hashes:?}" ); + assert_eq!(s.get(&m.object_hash).await.unwrap(), data); + } + + /// Any one of the three hot-tier shards reconstructs the block. Tested at + /// the erasure layer because the chunk store dedups the three copies into + /// one file, so shard loss cannot be simulated by deleting from it. + #[test] + fn hot_tier_decodes_from_any_single_shard() { + let data = varied(4096); + let (shards, len) = erasure::encode(&data, 1, 2).unwrap(); + assert_eq!(shards.len(), 3); + for keep in 0..3 { + let received: Vec> = (0..3) + .map(|i| { + if i == keep { + Some(shards[i].clone()) + } else { + None + } + }) + .collect(); + let out = erasure::decode(received, 1, 2, len).unwrap(); + assert_eq!(out, data, "failed to rebuild from shard {keep} alone"); + } + } + + #[tokio::test] + async fn critical_tier_survives_twelve_losses() { + let s = store().await; + let data = varied(40_000); + let m = s.put_tiered(&data, Tier::Critical).await.unwrap(); + assert_eq!(m.shard_count(), 32); + for shard in m.blocks[0].shards.iter().take(12) { + s.inner.delete(&shard.hash).await.unwrap(); + } + assert_eq!(s.get(&m.object_hash).await.unwrap(), data); + } + + #[tokio::test] + async fn wrong_committed_hash_is_rejected_and_leaves_nothing() { + let s = store().await; + let data = varied(9_000); + let bogus = Hash::of(b"not the data"); + let chunk = Bytes::from(data.clone()); + let stream = futures::stream::once(async move { Ok(chunk) }); + let err = s + .put_stream(stream, Some(bogus), Tier::Standard, Some(data.len() as u64)) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("integrity failure"), "unexpected: {err}"); + + // Nothing left behind: no manifest, and the shards were rolled back. + assert!(!s.has(&Hash::of(&data)).await); + let shard_dir = s.inner.root().join("shards"); + let mut count = 0; + for entry in walk(&shard_dir) { + if entry.is_file() { + count += 1; + } + } + assert_eq!(count, 0, "rollback left {count} shards on disk"); + } + + /// Regression: a rejected write must not damage an object that already + /// holds the same content. + /// + /// Shards are content-addressed, so re-uploading existing bytes under a + /// wrong committed hash yields identical shard hashes. Rolling back + /// everything the write touched used to delete the good object's shards, + /// turning one malformed request into data loss. Found by running the + /// real HTTP server, not by the unit tests — the original test stored + /// nothing beforehand, so there was nothing to destroy. + #[tokio::test] + async fn failed_write_does_not_delete_an_existing_objects_shards() { + let s = store().await; + let data = varied(300_000); + + let good = s.put(&data).await.unwrap(); + assert_eq!(s.get(&good.object_hash).await.unwrap(), data); + + // Same bytes, wrong committed hash. + let chunk = Bytes::from(data.clone()); + let stream = futures::stream::once(async move { Ok(chunk) }); + let err = s + .put_stream( + stream, + Some(Hash::of(b"a different object entirely")), + Tier::Standard, + Some(data.len() as u64), + ) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("integrity failure"), "unexpected: {err}"); + + // The original object must be completely intact. + for block in &good.blocks { + for shard in &block.shards { + assert!( + s.inner.has(&shard.hash).await, + "rollback deleted a shard belonging to an intact object" + ); + } + } + assert_eq!(s.get(&good.object_hash).await.unwrap(), data); + } + + #[tokio::test] + async fn put_new_reports_creation_once() { + let s = store().await; + let bytes = varied(4096); + let (h1, created1) = s.inner.put_new(&bytes).await.unwrap(); + let (h2, created2) = s.inner.put_new(&bytes).await.unwrap(); + assert_eq!(h1, h2); + assert!(created1, "first write should create"); + assert!(!created2, "second write should find it present"); + } + + fn walk(dir: &std::path::Path) -> Vec { + let mut out = Vec::new(); + let Ok(rd) = std::fs::read_dir(dir) else { + return out; + }; + for e in rd.flatten() { + let p = e.path(); + if p.is_dir() { + out.extend(walk(&p)); + } else { + out.push(p); + } + } + out + } + + #[tokio::test] + async fn range_read_touches_only_needed_blocks() { + let s = store().await; + let data = varied(DEFAULT_BLOCK_SIZE as usize * 3); + let m = s.put(&data).await.unwrap(); + + let got = s.get_range(&m.object_hash, 1_000_000, 4096).await.unwrap(); + assert_eq!(got, &data[1_000_000..1_004_096]); + + // A range wholly inside block 2 must read exactly one block. + let m2 = s.read_manifest(&m.object_hash).await.unwrap(); + let r = m2.blocks_for_range(DEFAULT_BLOCK_SIZE as u64 * 2 + 10, 100); + assert_eq!(r, 2..3); + } + + #[tokio::test] + async fn range_spanning_block_boundary() { + let s = store().await; + let data = varied(DEFAULT_BLOCK_SIZE as usize * 2); + let m = s.put(&data).await.unwrap(); + let off = DEFAULT_BLOCK_SIZE as u64 - 50; + let got = s.get_range(&m.object_hash, off, 100).await.unwrap(); + assert_eq!(got, &data[off as usize..off as usize + 100]); + } + + #[tokio::test] + async fn range_past_end_is_clamped() { + let s = store().await; + let data = varied(1000); + let m = s.put(&data).await.unwrap(); + let got = s.get_range(&m.object_hash, 900, 500).await.unwrap(); + assert_eq!(got, &data[900..1000]); + assert!( + s.get_range(&m.object_hash, 5000, 10) + .await + .unwrap() + .is_empty() + ); + } + + #[tokio::test] + async fn read_stream_matches_get() { + let s = store().await; + let data = varied(DEFAULT_BLOCK_SIZE as usize + 777); + let m = s.put(&data).await.unwrap(); + let mut out = Vec::new(); + let mut stream = s.read_stream(m.clone()); + while let Some(chunk) = stream.next().await { + out.extend_from_slice(&chunk.unwrap()); + } + assert_eq!(out, data); + } + + #[tokio::test] + async fn block_size_scales_to_keep_manifests_small() { + assert_eq!(block_size_for(1024), DEFAULT_BLOCK_SIZE); + assert_eq!(block_size_for(1 << 30), DEFAULT_BLOCK_SIZE); // 1 GiB + // 1 TiB must not produce a quarter-million blocks. + let bs = block_size_for(1 << 40); + let blocks = (1u64 << 40).div_ceil(bs as u64); + assert!(blocks <= TARGET_BLOCKS_PER_OBJECT, "{blocks} blocks"); + assert!(bs <= MAX_BLOCK_SIZE); + } + + #[tokio::test] + async fn v1_manifest_still_deserialises() { + let v1 = serde_json::json!({ + "object_hash": Hash::of(b"legacy").to_hex(), + "original_len": 6, + "k": 10, + "parity": 4, + "shards": [ { "index": 0, "hash": Hash::of(b"s0").to_hex() } ], + }); + let m: ObjectManifest = serde_json::from_value(v1).unwrap(); + assert_eq!(m.version, 1); + assert_eq!(m.tier, Tier::Standard); + assert_eq!(m.blocks.len(), 1); + assert_eq!(m.blocks[0].len, 6); + assert_eq!(m.blocks[0].hash, m.object_hash); + } + + #[tokio::test] + async fn manifest_v2_roundtrips_through_json() { + let s = store().await; + let m = s.put(&varied(5000)).await.unwrap(); + let json = serde_json::to_vec(&m).unwrap(); + let back: ObjectManifest = serde_json::from_slice(&json).unwrap(); + assert_eq!(m, back); + } + + #[tokio::test] + async fn corrupted_shard_is_detected_not_returned() { + let s = store().await; + let data = varied(30_000); + let m = s.put(&data).await.unwrap(); + // Corrupt enough shards that parity cannot mask it. + for shard in m.blocks[0].shards.iter().take(5) { + let path = s.inner.root().join("shards"); + let hex = shard.hash.to_hex(); + let p = path.join(&hex[0..2]).join(&hex[2..4]).join(&hex); + tokio::fs::write(&p, b"corrupted").await.unwrap(); + } + assert!(s.get(&m.object_hash).await.is_err()); + } + + #[tokio::test] + async fn delete_removes_manifest_and_shards() { + let s = store().await; + let m = s.put(&varied(8000)).await.unwrap(); + assert!(s.has(&m.object_hash).await); + s.delete(&m.object_hash).await.unwrap(); + assert!(!s.has(&m.object_hash).await); + for shard in &m.blocks[0].shards { + assert!(!s.inner.has(&shard.hash).await); + } } } diff --git a/node/crates/c0mpute-store/src/tier.rs b/node/crates/c0mpute-store/src/tier.rs new file mode 100644 index 0000000..f33d133 --- /dev/null +++ b/node/crates/c0mpute-store/src/tier.rs @@ -0,0 +1,145 @@ +//! Storage tiers and their redundancy parameters (CIP-001). +//! +//! The tier picks `(k, parity)`, and `(k, parity)` is the cost of goods: +//! the expansion factor `n/k` is how many raw GB a provider is paid for per +//! usable GB sold. See `docs/prds/001-storage-program.md` for the pricing that +//! falls out of these numbers. + +use std::fmt; +use std::str::FromStr; + +use serde::{Deserialize, Serialize}; + +/// Redundancy tier. Determines `(k, parity)` and therefore expansion factor, +/// failure tolerance, and repair amplification. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum Tier { + /// 3-copy replication (k=1, parity=2). 3.0x expansion, 1x repair + /// amplification, tolerates 2 losses. For small, hot, frequently + /// rewritten data — filesystem metadata and inline small files. + Hot, + /// Reed-Solomon 10/14. 1.4x expansion, tolerates 4 losses. The default. + #[default] + Standard, + /// Reed-Solomon 20/32. 1.6x expansion, tolerates 12 losses. For + /// irreplaceable data with long retention. + Critical, +} + +impl Tier { + /// Data shards. Also the repair amplification factor: rebuilding one lost + /// shard requires reading `k` shards. + pub const fn k(self) -> usize { + match self { + Tier::Hot => 1, + Tier::Standard => 10, + Tier::Critical => 20, + } + } + + /// Parity shards. Also the number of simultaneous losses tolerated. + pub const fn parity(self) -> usize { + match self { + Tier::Hot => 2, + Tier::Standard => 4, + Tier::Critical => 12, + } + } + + /// Total shards per block. + pub const fn n(self) -> usize { + self.k() + self.parity() + } + + /// Raw bytes stored per usable byte. + pub fn expansion(self) -> f64 { + self.n() as f64 / self.k() as f64 + } + + /// Retail price in USD per usable GB-month (CIP-001). + pub const fn price_usd_per_gb_month(self) -> f64 { + match self { + Tier::Hot => 0.006, + Tier::Standard => 0.0035, + Tier::Critical => 0.005, + } + } + + /// Recover the tier from `(k, parity)` read off an older manifest that + /// predates the tier field. + pub fn from_params(k: u8, parity: u8) -> Option { + [Tier::Hot, Tier::Standard, Tier::Critical] + .into_iter() + .find(|t| t.k() == k as usize && t.parity() == parity as usize) + } +} + +impl fmt::Display for Tier { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Tier::Hot => "hot", + Tier::Standard => "standard", + Tier::Critical => "critical", + }) + } +} + +impl FromStr for Tier { + type Err = TierParseError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "hot" => Ok(Tier::Hot), + "standard" | "" => Ok(Tier::Standard), + "critical" => Ok(Tier::Critical), + other => Err(TierParseError(other.to_string())), + } + } +} + +#[derive(Debug, thiserror::Error)] +#[error("unknown storage tier `{0}` (expected hot, standard, or critical)")] +pub struct TierParseError(pub String); + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn expansion_factors_match_cip_001() { + assert_eq!(Tier::Hot.expansion(), 3.0); + assert!((Tier::Standard.expansion() - 1.4).abs() < 1e-9); + assert!((Tier::Critical.expansion() - 1.6).abs() < 1e-9); + } + + #[test] + fn standard_is_the_default() { + assert_eq!(Tier::default(), Tier::Standard); + assert_eq!(Tier::Standard.k(), 10); + assert_eq!(Tier::Standard.n(), 14); + } + + #[test] + fn parses_and_displays() { + assert_eq!("hot".parse::().unwrap(), Tier::Hot); + assert_eq!("CRITICAL".parse::().unwrap(), Tier::Critical); + assert_eq!(Tier::Standard.to_string(), "standard"); + assert!("glacier".parse::().is_err()); + } + + #[test] + fn recovers_tier_from_legacy_params() { + assert_eq!(Tier::from_params(10, 4), Some(Tier::Standard)); + assert_eq!(Tier::from_params(20, 12), Some(Tier::Critical)); + assert_eq!(Tier::from_params(7, 3), None); + } + + /// The cheaper tier must genuinely be cheaper to supply, or the pricing in + /// CIP-001 is upside down. + #[test] + fn standard_costs_less_to_supply_than_hot() { + let payout = 0.0015_f64; // $/raw GB-month + assert!(Tier::Standard.expansion() * payout < Tier::Hot.expansion() * payout); + } +}