diff --git a/.gitignore b/.gitignore index 2b4cb6592..05e2cf89a 100644 --- a/.gitignore +++ b/.gitignore @@ -37,7 +37,10 @@ curvine-libsdk/java/target/ .staging/ .summary/ .claude/ -docs +docs/* +!docs/plans/ +docs/plans/* +!docs/plans/2026-03-26-p2p-pr-split-execution-plan.md __pycache__ diff --git a/Cargo.lock b/Cargo.lock index b5db1b63d..8b43c1ddf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1044,6 +1044,7 @@ dependencies = [ "serde", "serde_json", "serde_with", + "sha2", "tokio", ] diff --git a/curvine-cli/Cargo.toml b/curvine-cli/Cargo.toml index 06ea253be..a8ef94005 100644 --- a/curvine-cli/Cargo.toml +++ b/curvine-cli/Cargo.toml @@ -19,3 +19,4 @@ bigdecimal = { workspace = true } chrono = { workspace = true } bytes = { workspace = true } reqwest = { version = "0.11", features = ["json", "rustls-tls"], default-features = false} +sha2 = "0.10" diff --git a/curvine-cli/src/bin/p2p_proof_driver.rs b/curvine-cli/src/bin/p2p_proof_driver.rs new file mode 100644 index 000000000..b93585224 --- /dev/null +++ b/curvine-cli/src/bin/p2p_proof_driver.rs @@ -0,0 +1,944 @@ +use bytes::BytesMut; +use clap::{Parser, ValueEnum}; +use curvine_client::unified::UnifiedFileSystem; +use curvine_common::conf::ClusterConf; +use curvine_common::fs::{FileSystem, Path, Reader, Writer}; +use orpc::common::{Logger, Utils}; +use orpc::io::net::InetAddr; +use orpc::runtime::RpcRuntime; +use orpc::{err_box, CommonResult}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::fs::{self, OpenOptions}; +use std::hash::{Hash, Hasher}; +use std::io::Write; +use std::path::{Path as FsPath, PathBuf}; +use std::sync::Arc; +use tokio::time::{sleep, Duration, Instant}; + +#[derive(Copy, Clone, Eq, PartialEq, ValueEnum)] +enum Role { + Provider, + Consumer, + Mixed, +} + +#[derive(Parser)] +#[command(name = "p2p-proof-driver")] +struct Args { + #[arg(long)] + role: Role, + #[arg(long)] + conf: String, + #[arg(long)] + master_addrs: String, + #[arg(long)] + remote_path: String, + #[arg(long)] + input_file: Option, + #[arg(long)] + output_file: PathBuf, + #[arg(long)] + bootstrap_file: Option, + #[arg(long)] + peer_file: Option, + #[arg(long)] + bootstrap_host: Option, + #[arg(long, default_value_t = 1)] + warm_reads: u32, + #[arg(long, default_value_t = 180)] + hold_secs: u64, + #[arg(long, default_value_t = 120)] + bootstrap_wait_secs: u64, + #[arg(long, default_value_t = 2)] + report_interval_secs: u64, + #[arg(long, default_value_t = 1)] + consumer_reads: u32, + #[arg(long, default_value_t = 0)] + consumer_read_interval_ms: u64, + #[arg(long, default_value_t = 0)] + consumer_startup_wait_ms: u64, + #[arg(long)] + client_id: Option, + #[arg(long)] + manifest_dir: Option, + #[arg(long)] + read_manifest_dir: Option, + #[arg(long)] + write_manifest_dir: Option, + #[arg(long, default_value = "64KB,1MB,16MB")] + file_size_spec: String, + #[arg(long, default_value_t = 16)] + file_count: u32, + #[arg(long, default_value_t = 60)] + duration_secs: u64, + #[arg(long, default_value_t = 70)] + read_ratio: u8, + #[arg(long, default_value_t = 80)] + hotset_ratio: u8, + #[arg(long, default_value_t = 0)] + op_interval_ms: u64, + #[arg(long, default_value_t = false)] + warm_written: bool, + #[arg(long, default_value_t = 0)] + post_run_hold_secs: u64, +} + +#[derive(Clone, Serialize, Deserialize)] +struct ManifestEntry { + path: String, + size_bytes: usize, + sha256: String, + owner: String, + write_seq: u64, + created_at_ms: u64, +} + +#[derive(Clone, Serialize)] +struct LatencySummary { + count: usize, + avg_ms: f64, + p50_ms: f64, + p95_ms: f64, + p99_ms: f64, + max_ms: f64, +} + +#[derive(Serialize)] +struct WorkloadSummary { + role: String, + client_id: String, + duration_secs: u64, + size_spec: Vec, + file_count: u32, + read_ratio: u8, + hotset_ratio: u8, + total_ops: u64, + read_ops: u64, + write_ops: u64, + read_ok: u64, + write_ok: u64, + read_errors: u64, + write_errors: u64, + first_read_error: Option, + first_write_error: Option, + sha_mismatches: u64, + bytes_read: u64, + bytes_written: u64, + manifest_entries_seen: u64, + p2p_recv_delta: u64, + p2p_sent_delta: u64, + cached_chunks_delta: u64, + read_latency: LatencySummary, + write_latency: LatencySummary, +} + +struct WorkloadRng { + state: u64, +} + +impl WorkloadRng { + fn new(seed: u64) -> Self { + Self { state: seed.max(1) } + } + + fn next_u64(&mut self) -> u64 { + self.state = self.state.wrapping_add(0x9e3779b97f4a7c15); + let mut value = self.state; + value = (value ^ (value >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); + value = (value ^ (value >> 27)).wrapping_mul(0x94d049bb133111eb); + value ^ (value >> 31) + } + + fn next_index(&mut self, len: usize) -> usize { + if len <= 1 { + return 0; + } + (self.next_u64() % len as u64) as usize + } + + fn next_percent(&mut self) -> u8 { + (self.next_u64() % 100) as u8 + } +} + +fn parse_master_addrs(raw: &str) -> CommonResult> { + let mut addrs = Vec::new(); + for node in raw.split(',') { + let pair: Vec<&str> = node.trim().split(':').collect(); + if pair.len() != 2 { + return err_box!( + "invalid master_addrs format: '{}', expected 'host1:port1,host2:port2'", + raw + ); + } + let host = pair[0].to_string(); + let port = pair[1] + .parse::() + .map_err(|_| format!("invalid master port '{}'", pair[1]))?; + addrs.push(InetAddr::new(host, port)); + } + Ok(addrs) +} + +fn load_conf(args: &Args) -> CommonResult { + let mut conf = ClusterConf::from(&args.conf)?; + conf.client.master_addrs = parse_master_addrs(&args.master_addrs)?; + conf.client.init()?; + Ok(conf) +} + +fn parent_of(remote: &str) -> String { + match remote.rfind('/') { + Some(0) | None => "/".to_string(), + Some(idx) => remote[..idx].to_string(), + } +} + +fn normalize_remote_dir(remote: &str) -> String { + if remote == "/" { + return "/".to_string(); + } + remote.trim_end_matches('/').to_string() +} + +fn parse_size_spec(spec: &str) -> CommonResult> { + let mut sizes = Vec::new(); + for token in spec + .split(',') + .map(str::trim) + .filter(|token| !token.is_empty()) + { + sizes.push(parse_size_token(token)?); + } + if sizes.is_empty() { + return err_box!("file_size_spec is empty"); + } + Ok(sizes) +} + +fn parse_size_token(token: &str) -> CommonResult { + let normalized = token.trim().to_ascii_uppercase(); + let split_at = normalized + .find(|ch: char| !ch.is_ascii_digit()) + .unwrap_or(normalized.len()); + if split_at == 0 { + return err_box!("invalid size token: {}", token); + } + let value = normalized[..split_at] + .parse::() + .map_err(|_| format!("invalid size token: {}", token))?; + let unit = normalized[split_at..].trim(); + let multiplier = match unit { + "" | "B" => 1u64, + "K" | "KB" => 1024u64, + "M" | "MB" => 1024u64 * 1024, + "G" | "GB" => 1024u64 * 1024 * 1024, + _ => return err_box!("unsupported size unit in token: {}", token), + }; + let size = value + .checked_mul(multiplier) + .ok_or_else(|| format!("size token overflow: {}", token))?; + usize::try_from(size).map_err(|_| format!("size token overflow: {}", token).into()) +} + +fn choose_manifest_index(total: usize, hotset_ratio: u8, rng: &mut WorkloadRng) -> usize { + if total <= 1 { + return 0; + } + let hotset_ratio = hotset_ratio.clamp(1, 100) as usize; + let hotset_len = total + .saturating_mul(hotset_ratio) + .div_ceil(100) + .clamp(1, total); + let start = total.saturating_sub(hotset_len); + start + rng.next_index(hotset_len) +} + +fn sha256_hex(data: &[u8]) -> String { + let mut digest = Sha256::new(); + digest.update(data); + digest + .finalize() + .iter() + .map(|value| format!("{:02x}", value)) + .collect() +} + +fn summarize_latencies(mut latencies: Vec) -> LatencySummary { + if latencies.is_empty() { + return LatencySummary { + count: 0, + avg_ms: 0.0, + p50_ms: 0.0, + p95_ms: 0.0, + p99_ms: 0.0, + max_ms: 0.0, + }; + } + latencies.sort_by(|left, right| left.partial_cmp(right).unwrap_or(std::cmp::Ordering::Equal)); + let count = latencies.len(); + let avg_ms = latencies.iter().sum::() / count as f64; + let percentile = |ratio: f64| { + let index = ((count.saturating_sub(1)) as f64 * ratio).round() as usize; + latencies[index] + }; + LatencySummary { + count, + avg_ms, + p50_ms: percentile(0.50), + p95_ms: percentile(0.95), + p99_ms: percentile(0.99), + max_ms: *latencies.last().unwrap_or(&0.0), + } +} + +fn client_id(args: &Args) -> String { + args.client_id + .clone() + .unwrap_or_else(|| format!("{}-{}", std::process::id(), Utils::uuid())) +} + +fn manifest_file(manifest_dir: &FsPath, client_id: &str) -> PathBuf { + manifest_dir.join(format!("{}.jsonl", client_id)) +} + +fn resolve_manifest_dirs(args: &Args) -> CommonResult<(PathBuf, PathBuf)> { + let read_manifest_dir = args + .read_manifest_dir + .clone() + .or_else(|| args.manifest_dir.clone()) + .or_else(|| args.write_manifest_dir.clone()) + .ok_or_else(|| "mixed role requires --manifest-dir or --read-manifest-dir".to_string())?; + let write_manifest_dir = args + .write_manifest_dir + .clone() + .or_else(|| args.manifest_dir.clone()) + .or_else(|| args.read_manifest_dir.clone()) + .ok_or_else(|| "mixed role requires --manifest-dir or --write-manifest-dir".to_string())?; + Ok((read_manifest_dir, write_manifest_dir)) +} + +fn append_manifest_entry( + manifest_dir: &FsPath, + client_id: &str, + entry: &ManifestEntry, +) -> CommonResult<()> { + fs::create_dir_all(manifest_dir)?; + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(manifest_file(manifest_dir, client_id))?; + serde_json::to_writer(&mut file, entry)?; + writeln!(file)?; + Ok(()) +} + +fn load_manifest_entries(manifest_dir: &FsPath) -> CommonResult> { + if !manifest_dir.exists() { + return Ok(Vec::new()); + } + let mut entries: Vec = Vec::new(); + for item in fs::read_dir(manifest_dir)? { + let path = item?.path(); + if path.extension().and_then(|value| value.to_str()) != Some("jsonl") { + continue; + } + let content = fs::read_to_string(path)?; + for line in content + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + { + entries.push(serde_json::from_str(line)?); + } + } + entries.sort_by(|left, right| { + left.created_at_ms + .cmp(&right.created_at_ms) + .then_with(|| left.owner.cmp(&right.owner)) + .then_with(|| left.write_seq.cmp(&right.write_seq)) + }); + Ok(entries) +} + +fn payload_seed(client_id: &str, write_seq: u64, size: usize) -> u64 { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + client_id.hash(&mut hasher); + write_seq.hash(&mut hasher); + size.hash(&mut hasher); + hasher.finish() +} + +fn build_payload(client_id: &str, write_seq: u64, size: usize) -> Vec { + let mut payload = vec![0u8; size]; + let mut rng = WorkloadRng::new(payload_seed(client_id, write_seq, size)); + let mut offset = 0usize; + while offset < payload.len() { + let bytes = rng.next_u64().to_le_bytes(); + let end = (offset + bytes.len()).min(payload.len()); + payload[offset..end].copy_from_slice(&bytes[..end - offset]); + offset = end; + } + payload +} + +fn write_workload_summary(path: &PathBuf, summary: &WorkloadSummary) -> CommonResult<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + fs::write(path, serde_json::to_vec_pretty(summary)?)?; + Ok(()) +} + +async fn read_all( + fs: &UnifiedFileSystem, + path: &Path, + metric_prefix: Option<&str>, +) -> CommonResult> { + let status_start = Instant::now(); + let status = fs.get_status(path).await?; + if let Some(prefix) = metric_prefix { + println!( + "{}_STATUS_MS={:.3}", + prefix, + status_start.elapsed().as_secs_f64() * 1000.0 + ); + } + let open_start = Instant::now(); + let mut reader = fs.open(path).await?; + if let Some(prefix) = metric_prefix { + println!( + "{}_OPEN_MS={:.3}", + prefix, + open_start.elapsed().as_secs_f64() * 1000.0 + ); + } + let mut buf = BytesMut::zeroed(status.len as usize); + let read_start = Instant::now(); + let size = reader.read_full(&mut buf).await?; + if let Some(prefix) = metric_prefix { + println!( + "{}_READ_MS={:.3}", + prefix, + read_start.elapsed().as_secs_f64() * 1000.0 + ); + } + let complete_start = Instant::now(); + reader.complete().await?; + if let Some(prefix) = metric_prefix { + println!( + "{}_COMPLETE_MS={:.3}", + prefix, + complete_start.elapsed().as_secs_f64() * 1000.0 + ); + } + buf.truncate(size); + Ok(buf.to_vec()) +} + +async fn write_all(fs: &UnifiedFileSystem, path: &Path, payload: &[u8]) -> CommonResult<()> { + let mut writer = fs.create(path, true).await?; + writer.write(payload).await?; + writer.complete().await?; + Ok(()) +} + +async fn hold_after_workload(fs: &UnifiedFileSystem, hold_secs: u64, report_interval_secs: u64) { + if hold_secs == 0 { + return; + } + let report_interval = Duration::from_secs(report_interval_secs.max(1)); + let deadline = Instant::now() + Duration::from_secs(hold_secs); + while Instant::now() < deadline { + let _ = fs.cv().metrics_report().await; + sleep(report_interval).await; + } +} + +async fn wait_bootstrap_addr(fs: &UnifiedFileSystem, wait_secs: u64) -> Option { + let deadline = Instant::now() + Duration::from_secs(wait_secs.max(1)); + while Instant::now() < deadline { + if let Some(addr) = fs.p2p_bootstrap_peer_addr() { + return Some(addr); + } + sleep(Duration::from_millis(100)).await; + } + None +} + +fn rewrite_bootstrap_host(raw: &str, host: Option<&str>) -> String { + let Some(host) = host else { + return raw.to_string(); + }; + let mut parts: Vec = raw + .split('/') + .map(std::string::ToString::to_string) + .collect(); + if parts.len() > 3 && matches!(parts[1].as_str(), "ip4" | "ip6" | "dns" | "dns4" | "dns6") { + parts[2] = host.to_string(); + return parts.join("/"); + } + raw.to_string() +} + +fn peer_id_from_bootstrap_addr(addr: &str) -> Option<&str> { + addr.rsplit_once("/p2p/").map(|(_, peer_id)| peer_id) +} + +async fn publish_bootstrap_identity( + fs: &UnifiedFileSystem, + args: &Args, + prefix: &str, +) -> CommonResult> { + if args.bootstrap_file.is_none() && args.peer_file.is_none() { + return Ok(None); + } + let bootstrap_addr_raw = wait_bootstrap_addr(fs, args.bootstrap_wait_secs) + .await + .ok_or_else(|| "bootstrap address not ready".to_string())?; + let bootstrap_addr = + rewrite_bootstrap_host(&bootstrap_addr_raw, args.bootstrap_host.as_deref()); + let peer_id = fs + .p2p_peer_id() + .or_else(|| peer_id_from_bootstrap_addr(&bootstrap_addr).map(str::to_string)) + .ok_or_else(|| format!("invalid bootstrap address: {}", bootstrap_addr))?; + if let Some(bootstrap_file) = args.bootstrap_file.as_ref() { + fs::write(bootstrap_file, bootstrap_addr.as_bytes())?; + } + if let Some(peer_file) = args.peer_file.as_ref() { + fs::write(peer_file, peer_id.as_bytes())?; + } + println!("{}_PEER_ID={}", prefix, peer_id); + println!("{}_BOOTSTRAP_ADDR={}", prefix, bootstrap_addr); + Ok(Some((peer_id, bootstrap_addr))) +} + +async fn run_provider(fs: UnifiedFileSystem, args: &Args) -> CommonResult<()> { + let input_file = args + .input_file + .as_ref() + .ok_or_else(|| "provider requires --input-file".to_string())?; + if !fs.p2p_enabled() { + return err_box!("p2p service is not enabled"); + } + if args.bootstrap_file.is_none() || args.peer_file.is_none() { + return err_box!("provider requires --bootstrap-file and --peer-file"); + } + + let parent = Path::from_str(parent_of(&args.remote_path).as_str())?; + let remote = Path::from_str(&args.remote_path)?; + let payload = fs::read(input_file)?; + + let _ = fs.delete(&parent, true).await; + fs.mkdir(&parent, true).await?; + write_all(&fs, &remote, payload.as_slice()).await?; + + let mut warmed = Vec::new(); + for _ in 0..args.warm_reads.max(1) { + warmed = read_all(&fs, &remote, None).await?; + } + fs::write(&args.output_file, warmed)?; + + let _ = publish_bootstrap_identity(&fs, args, "PROVIDER").await?; + println!( + "PROVIDER_CACHE_CHUNKS={}", + fs.p2p_stats_snapshot() + .map(|snapshot| snapshot.cached_chunks_count) + .unwrap_or(0) + ); + + hold_after_workload(&fs, args.hold_secs, args.report_interval_secs).await; + let _ = fs.cv().metrics_report().await; + println!("PROVIDER_STATE_OK=1"); + Ok(()) +} + +async fn run_consumer(fs: UnifiedFileSystem, args: &Args) -> CommonResult<()> { + let remote = Path::from_str(&args.remote_path)?; + let read_times = args.consumer_reads.max(1); + let read_interval = Duration::from_millis(args.consumer_read_interval_ms); + let startup_wait = Duration::from_millis(args.consumer_startup_wait_ms); + if !startup_wait.is_zero() { + sleep(startup_wait).await; + } + let before = fs.p2p_stats_snapshot(); + let mut payload = Vec::new(); + let read_begin = Instant::now(); + for index in 0..read_times { + let read_start = Instant::now(); + let metric_prefix = format!("CONSUMER_READ_{}", index + 1); + payload = read_all(&fs, &remote, Some(metric_prefix.as_str())).await?; + println!( + "CONSUMER_READ_{}_MS={:.3}", + index + 1, + read_start.elapsed().as_secs_f64() * 1000.0 + ); + if index + 1 < read_times && !read_interval.is_zero() { + sleep(read_interval).await; + } + } + let read_elapsed_ms = read_begin.elapsed().as_secs_f64() * 1000.0; + std::fs::write(&args.output_file, payload)?; + let after = fs.p2p_stats_snapshot(); + + let _ = fs.cv().metrics_report().await; + + let recv_delta = after + .as_ref() + .zip(before.as_ref()) + .map(|(after, before)| after.bytes_recv.saturating_sub(before.bytes_recv)) + .unwrap_or(0); + let cache_delta = after + .as_ref() + .zip(before.as_ref()) + .map(|(after, before)| { + after + .cached_chunks_count + .saturating_sub(before.cached_chunks_count) + }) + .unwrap_or(0); + let p2p_enabled = u8::from(fs.p2p_enabled()); + + println!("CONSUMER_P2P_ENABLED={}", p2p_enabled); + println!("CONSUMER_P2P_RECV_DELTA={}", recv_delta); + println!("CONSUMER_CACHE_CHUNKS_DELTA={}", cache_delta); + println!("CONSUMER_READS={}", read_times); + println!("CONSUMER_READ_ELAPSED_MS={:.3}", read_elapsed_ms); + println!("CONSUMER_STATE_OK=1"); + Ok(()) +} + +async fn run_mixed(fs: UnifiedFileSystem, args: &Args) -> CommonResult<()> { + if !fs.p2p_enabled() { + return err_box!("mixed role requires p2p to be enabled"); + } + let (read_manifest_dir, write_manifest_dir) = resolve_manifest_dirs(args)?; + let client_id = client_id(args); + let size_spec = parse_size_spec(&args.file_size_spec)?; + let remote_dir = normalize_remote_dir(&args.remote_path); + let remote_root = Path::from_str(&remote_dir)?; + let _ = fs.mkdir(&remote_root, true).await; + let client_root = Path::from_str(format!("{}/{}", remote_dir, client_id))?; + let _ = fs.mkdir(&client_root, true).await; + let _ = publish_bootstrap_identity(&fs, args, "MIXED").await?; + + let before = fs + .p2p_stats_snapshot() + .ok_or_else(|| "mixed role requires p2p snapshot".to_string())?; + let deadline = Instant::now() + Duration::from_secs(args.duration_secs.max(1)); + let op_interval = Duration::from_millis(args.op_interval_ms); + let read_ratio = args.read_ratio.clamp(0, 100); + let hotset_ratio = args.hotset_ratio.clamp(1, 100); + let mut rng = WorkloadRng::new(payload_seed( + &client_id, + args.duration_secs, + size_spec.len(), + )); + let mut write_seq = 0u64; + let mut total_ops = 0u64; + let mut read_ops = 0u64; + let mut write_ops = 0u64; + let mut read_ok = 0u64; + let mut write_ok = 0u64; + let mut read_errors = 0u64; + let mut write_errors = 0u64; + let mut sha_mismatches = 0u64; + let mut bytes_read = 0u64; + let mut bytes_written = 0u64; + let mut first_read_error = None; + let mut first_write_error = None; + let mut manifest_entries_seen = 0u64; + let mut read_latencies = Vec::new(); + let mut write_latencies = Vec::new(); + + while Instant::now() < deadline { + let manifest = load_manifest_entries(read_manifest_dir.as_path())?; + manifest_entries_seen = manifest_entries_seen.max(manifest.len() as u64); + let should_read = !manifest.is_empty() && rng.next_percent() < read_ratio; + if should_read { + let entry = &manifest[choose_manifest_index(manifest.len(), hotset_ratio, &mut rng)]; + let path = Path::from_str(&entry.path)?; + let start = Instant::now(); + read_ops += 1; + match read_all(&fs, &path, None).await { + Ok(payload) => { + let actual = sha256_hex(&payload); + bytes_read = bytes_read.saturating_add(payload.len() as u64); + if actual == entry.sha256 { + read_ok = read_ok.saturating_add(1); + } else { + sha_mismatches = sha_mismatches.saturating_add(1); + read_errors = read_errors.saturating_add(1); + } + } + Err(err) => { + if first_read_error.is_none() { + first_read_error = Some(err.to_string()); + } + read_errors = read_errors.saturating_add(1); + } + } + read_latencies.push(start.elapsed().as_secs_f64() * 1000.0); + } else { + let slot = write_seq % u64::from(args.file_count.max(1)); + let size = size_spec[rng.next_index(size_spec.len())]; + let payload = build_payload(&client_id, write_seq, size); + let sha256 = sha256_hex(&payload); + let remote_path = format!( + "{}/{}/slot-{:04}-seq-{:08}.bin", + remote_dir, client_id, slot, write_seq + ); + let path = Path::from_str(&remote_path)?; + let start = Instant::now(); + write_ops = write_ops.saturating_add(1); + match write_all(&fs, &path, &payload).await { + Ok(()) => { + if args.warm_written { + match read_all(&fs, &path, None).await { + Ok(warmed) => { + let warmed_sha256 = sha256_hex(&warmed); + if warmed_sha256 != sha256 { + if first_write_error.is_none() { + first_write_error = Some(format!( + "warm_written checksum mismatch for {}: {} != {}", + remote_path, warmed_sha256, sha256 + )); + } + write_errors = write_errors.saturating_add(1); + write_latencies.push(start.elapsed().as_secs_f64() * 1000.0); + total_ops = total_ops.saturating_add(1); + if !op_interval.is_zero() { + sleep(op_interval).await; + } + continue; + } + } + Err(err) => { + if first_write_error.is_none() { + first_write_error = Some(err.to_string()); + } + write_errors = write_errors.saturating_add(1); + write_latencies.push(start.elapsed().as_secs_f64() * 1000.0); + total_ops = total_ops.saturating_add(1); + if !op_interval.is_zero() { + sleep(op_interval).await; + } + continue; + } + } + } + let entry = ManifestEntry { + path: remote_path, + size_bytes: size, + sha256, + owner: client_id.clone(), + write_seq, + created_at_ms: orpc::common::LocalTime::mills(), + }; + append_manifest_entry(write_manifest_dir.as_path(), &client_id, &entry)?; + write_ok = write_ok.saturating_add(1); + bytes_written = bytes_written.saturating_add(size as u64); + write_seq = write_seq.saturating_add(1); + } + Err(err) => { + if first_write_error.is_none() { + first_write_error = Some(err.to_string()); + } + write_errors = write_errors.saturating_add(1); + } + } + write_latencies.push(start.elapsed().as_secs_f64() * 1000.0); + } + total_ops = total_ops.saturating_add(1); + if !op_interval.is_zero() { + sleep(op_interval).await; + } + } + + let _ = fs.cv().metrics_report().await; + let after = fs + .p2p_stats_snapshot() + .ok_or_else(|| "mixed role requires p2p snapshot".to_string())?; + let summary = WorkloadSummary { + role: "mixed".to_string(), + client_id: client_id.clone(), + duration_secs: args.duration_secs.max(1), + size_spec, + file_count: args.file_count.max(1), + read_ratio, + hotset_ratio, + total_ops, + read_ops, + write_ops, + read_ok, + write_ok, + read_errors, + write_errors, + first_read_error, + first_write_error, + sha_mismatches, + bytes_read, + bytes_written, + manifest_entries_seen, + p2p_recv_delta: after.bytes_recv.saturating_sub(before.bytes_recv), + p2p_sent_delta: after.bytes_sent.saturating_sub(before.bytes_sent), + cached_chunks_delta: after + .cached_chunks_count + .saturating_sub(before.cached_chunks_count) as u64, + read_latency: summarize_latencies(read_latencies), + write_latency: summarize_latencies(write_latencies), + }; + write_workload_summary(&args.output_file, &summary)?; + + println!("MIXED_CLIENT_ID={}", summary.client_id); + println!("MIXED_TOTAL_OPS={}", summary.total_ops); + println!("MIXED_READ_OK={}", summary.read_ok); + println!("MIXED_WRITE_OK={}", summary.write_ok); + println!("MIXED_SHA_MISMATCHES={}", summary.sha_mismatches); + println!("MIXED_P2P_RECV_DELTA={}", summary.p2p_recv_delta); + println!("MIXED_P2P_SENT_DELTA={}", summary.p2p_sent_delta); + if let Some(first_write_error) = summary.first_write_error.as_ref() { + println!("MIXED_FIRST_WRITE_ERROR={}", first_write_error); + } + if let Some(first_read_error) = summary.first_read_error.as_ref() { + println!("MIXED_FIRST_READ_ERROR={}", first_read_error); + } + hold_after_workload(&fs, args.post_run_hold_secs, args.report_interval_secs).await; + let _ = fs.cv().metrics_report().await; + println!("MIXED_STATE_OK=1"); + Ok(()) +} + +fn main() -> CommonResult<()> { + let args = Args::parse(); + Utils::set_panic_exit_hook(); + let conf = load_conf(&args)?; + Logger::init(conf.log.clone()); + + let rt = Arc::new(conf.client_rpc_conf().create_runtime()); + let fs = UnifiedFileSystem::with_rt(conf, rt.clone())?; + rt.block_on(async move { + let res = match args.role { + Role::Provider => run_provider(fs, &args).await, + Role::Consumer => run_consumer(fs, &args).await, + Role::Mixed => run_mixed(fs, &args).await, + }; + if let Err(e) = &res { + eprintln!("{}", e); + } + res + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn mixed_args() -> Args { + Args { + role: Role::Mixed, + conf: "conf".to_string(), + master_addrs: "127.0.0.1:8995".to_string(), + remote_path: "/tmp".to_string(), + input_file: None, + output_file: PathBuf::from("out.json"), + bootstrap_file: None, + peer_file: None, + bootstrap_host: None, + warm_reads: 1, + hold_secs: 180, + bootstrap_wait_secs: 120, + report_interval_secs: 2, + consumer_reads: 1, + consumer_read_interval_ms: 0, + consumer_startup_wait_ms: 0, + client_id: Some("client-1".to_string()), + manifest_dir: Some(PathBuf::from("/tmp/manifest")), + read_manifest_dir: None, + write_manifest_dir: None, + file_size_spec: "64KB,1MB,16MB".to_string(), + file_count: 16, + duration_secs: 60, + read_ratio: 70, + hotset_ratio: 80, + op_interval_ms: 0, + warm_written: false, + post_run_hold_secs: 0, + } + } + + #[test] + fn parse_size_spec_accepts_mixed_units() { + let parsed = parse_size_spec("4KB, 1MB, 2GB").expect("size spec should parse"); + assert_eq!(parsed, vec![4 * 1024, 1024 * 1024, 2 * 1024 * 1024 * 1024]); + } + + #[test] + fn choose_manifest_index_stays_inside_hotset_tail() { + let mut rng = WorkloadRng::new(7); + for _ in 0..128 { + let index = choose_manifest_index(10, 30, &mut rng); + assert!(index >= 7); + assert!(index < 10); + } + } + + #[test] + fn manifest_dirs_default_to_manifest_dir() { + let args = mixed_args(); + let (read_dir, write_dir) = + resolve_manifest_dirs(&args).expect("manifest dirs should resolve"); + assert_eq!(read_dir, PathBuf::from("/tmp/manifest")); + assert_eq!(write_dir, PathBuf::from("/tmp/manifest")); + } + + #[test] + fn manifest_dirs_allow_split_read_and_write_paths() { + let mut args = mixed_args(); + args.read_manifest_dir = Some(PathBuf::from("/tmp/read-manifest")); + args.write_manifest_dir = Some(PathBuf::from("/tmp/write-manifest")); + let (read_dir, write_dir) = + resolve_manifest_dirs(&args).expect("manifest dirs should resolve"); + assert_eq!(read_dir, PathBuf::from("/tmp/read-manifest")); + assert_eq!(write_dir, PathBuf::from("/tmp/write-manifest")); + } + + #[test] + fn post_run_hold_defaults_to_zero() { + let args = Args::parse_from([ + "p2p-proof-driver", + "--role", + "mixed", + "--conf", + "conf", + "--master-addrs", + "127.0.0.1:8995", + "--remote-path", + "/tmp", + "--output-file", + "out.json", + ]); + assert_eq!(args.post_run_hold_secs, 0); + } + + #[test] + fn post_run_hold_can_be_configured() { + let args = Args::parse_from([ + "p2p-proof-driver", + "--role", + "mixed", + "--conf", + "conf", + "--master-addrs", + "127.0.0.1:8995", + "--remote-path", + "/tmp", + "--output-file", + "out.json", + "--post-run-hold-secs", + "12", + ]); + assert_eq!(args.post_run_hold_secs, 12); + } +} diff --git a/curvine-tests/tests/p2p_read_acceleration_test.rs b/curvine-tests/tests/p2p_read_acceleration_test.rs new file mode 100644 index 000000000..975c1579b --- /dev/null +++ b/curvine-tests/tests/p2p_read_acceleration_test.rs @@ -0,0 +1,662 @@ +// Copyright 2025 OPPO. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use bytes::BytesMut; +use curvine_common::error::FsError; +use curvine_common::fs::{Path, Reader, Writer}; +use curvine_common::FsResult; +use curvine_tests::Testing; +use once_cell::sync::Lazy; +use orpc::common::Utils; +use orpc::runtime::{AsyncRuntime, RpcRuntime}; +use std::sync::Arc; +use std::sync::{Mutex, MutexGuard}; +use tokio::time::{sleep, Duration}; + +static P2P_TEST_LOCK: Lazy> = Lazy::new(|| Mutex::new(())); + +fn p2p_test_guard() -> MutexGuard<'static, ()> { + P2P_TEST_LOCK.lock().unwrap() +} + +async fn wait_bootstrap_addr(fs: &curvine_client::file::CurvineFileSystem) -> Option { + for _ in 0..400 { + if let Some(addr) = fs.p2p_bootstrap_peer_addr() { + return Some(addr); + } + sleep(Duration::from_millis(25)).await; + } + None +} + +async fn wait_policy_version( + fs: &curvine_client::file::CurvineFileSystem, + min_version: u64, +) -> bool { + for _ in 0..200 { + if fs.p2p_runtime_policy_version().unwrap_or_default() >= min_version { + return true; + } + sleep(Duration::from_millis(25)).await; + } + false +} + +async fn wait_policy_version_with_deadline( + fs: &curvine_client::file::CurvineFileSystem, + min_version: u64, + deadline: Duration, +) -> bool { + let started = tokio::time::Instant::now(); + while started.elapsed() < deadline { + if fs.p2p_runtime_policy_version().unwrap_or_default() >= min_version { + return true; + } + sleep(Duration::from_millis(25)).await; + } + false +} + +async fn wait_policy_rejects( + fs: &curvine_client::file::CurvineFileSystem, + min_rejects: u64, +) -> bool { + for _ in 0..200 { + if fs + .p2p_stats_snapshot() + .is_some_and(|snapshot| snapshot.policy_rejects >= min_rejects) + { + return true; + } + sleep(Duration::from_millis(25)).await; + } + false +} + +async fn read_all(fs: &curvine_client::file::CurvineFileSystem, path: &Path) -> FsResult> { + let status = fs.get_status(path).await?; + let mut reader = fs.open(path).await?; + let mut buf = BytesMut::zeroed(status.len as usize); + let size = reader.read_full(&mut buf).await?; + reader.complete().await?; + buf.truncate(size); + Ok(buf.to_vec()) +} + +#[test] +fn test_minicluster_p2p_read_acceleration() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf.clone(); + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + let bootstrap = rt + .block_on(wait_bootstrap_addr(&provider_fs)) + .expect("provider bootstrap address should be ready"); + + let mut consumer_conf = base_conf; + consumer_conf.client.p2p.enable = true; + consumer_conf.client.p2p.enable_mdns = false; + consumer_conf.client.p2p.enable_dht = false; + consumer_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + consumer_conf.client.p2p.bootstrap_peers = vec![bootstrap]; + consumer_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-consumer-cache-{}", + Utils::uuid() + ); + let consumer_fs = testing.get_fs(Some(rt.clone()), Some(consumer_conf))?; + + let path = Path::from_str("/p2p/minicluster-e2e.log")?; + rt.block_on(async move { + let parent = Path::from_str("/p2p")?; + let _ = provider_fs.delete(&parent, true).await; + provider_fs.mkdir(&parent, true).await?; + + let payload = Utils::rand_str(256 * 1024); + provider_fs.write_string(&path, payload.as_str()).await?; + + let provider_data = read_all(&provider_fs, &path).await?; + assert_eq!(provider_data, payload.as_bytes()); + + let provider_before = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + let consumer_data = read_all(&consumer_fs, &path).await?; + let provider_after = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + let consumer_after = consumer_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + + if consumer_data != payload.as_bytes() { + let expected = payload.as_bytes(); + let first_diff = consumer_data + .iter() + .zip(expected.iter()) + .position(|(left, right)| left != right); + let left_head_len = consumer_data.len().min(64); + let right_head_len = expected.len().min(64); + panic!( + "consumer payload mismatch: left_len={}, right_len={}, first_diff={:?}, left_head={:?}, right_head={:?}", + consumer_data.len(), + expected.len(), + first_diff, + &consumer_data[..left_head_len], + &expected[..right_head_len] + ); + } + assert!(provider_after.bytes_sent > provider_before.bytes_sent); + assert!(consumer_after.bytes_recv > 0); + assert!(consumer_after.cached_chunks_count > 0); + + Ok::<(), FsError>(()) + })?; + + Ok(()) +} + +#[test] +fn test_minicluster_p2p_runtime_policy_sync_from_master() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .mutate_conf(|conf| { + conf.master.p2p_policy_version = 1; + conf.master.p2p_policy_signing_key = "policy-secret".to_string(); + conf.master.p2p_tenant_whitelist = vec!["tenant-a".to_string()]; + conf.master.p2p_peer_whitelist = vec![]; + }) + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf.clone(); + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.tenant_whitelist = vec!["tenant-b".to_string()]; + provider_conf.client.p2p.policy_hmac_key = "policy-secret".to_string(); + provider_conf.client.clean_task_interval = Duration::from_millis(100); + provider_conf.client.clean_task_interval_str = "100ms".to_string(); + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + let bootstrap = rt + .block_on(wait_bootstrap_addr(&provider_fs)) + .expect("provider bootstrap address should be ready"); + + let mut consumer_conf = base_conf; + consumer_conf.client.p2p.enable = true; + consumer_conf.client.p2p.enable_mdns = false; + consumer_conf.client.p2p.enable_dht = false; + consumer_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + consumer_conf.client.p2p.bootstrap_peers = vec![bootstrap]; + consumer_conf.client.p2p.policy_hmac_key = "policy-secret".to_string(); + consumer_conf.client.clean_task_interval = Duration::from_millis(100); + consumer_conf.client.clean_task_interval_str = "100ms".to_string(); + consumer_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-consumer-cache-{}", + Utils::uuid() + ); + let consumer_fs = testing.get_fs(Some(rt.clone()), Some(consumer_conf))?; + + rt.block_on(async move { + assert!(wait_policy_version(&provider_fs, 1).await); + + let parent = Path::from_str("/p2p-policy-sync")?; + let _ = provider_fs.delete(&parent, true).await; + provider_fs.mkdir(&parent, true).await?; + + let path_allow = Path::from_str("/p2p-policy-sync/allow.log")?; + let payload_allow = Utils::rand_str(128 * 1024); + let opts_allow = provider_fs + .create_opts_builder() + .create_parent(true) + .x_attr("tenant_id".to_string(), "tenant-a".as_bytes().to_vec()) + .build(); + let mut writer_allow = provider_fs + .create_with_opts(&path_allow, opts_allow, true) + .await?; + writer_allow.write(payload_allow.as_bytes()).await?; + writer_allow.complete().await?; + let provider_allow = read_all(&provider_fs, &path_allow).await?; + assert_eq!(provider_allow, payload_allow.as_bytes()); + + let allow_before = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + let allow_data = read_all(&consumer_fs, &path_allow).await?; + let allow_after = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + assert_eq!(allow_data, payload_allow.as_bytes()); + assert!(allow_after > allow_before); + + let path_deny = Path::from_str("/p2p-policy-sync/deny.log")?; + let payload_deny = Utils::rand_str(128 * 1024); + let opts_deny = provider_fs + .create_opts_builder() + .create_parent(true) + .x_attr("tenant_id".to_string(), "tenant-b".as_bytes().to_vec()) + .build(); + let mut writer_deny = provider_fs + .create_with_opts(&path_deny, opts_deny, true) + .await?; + writer_deny.write(payload_deny.as_bytes()).await?; + writer_deny.complete().await?; + let provider_deny = read_all(&provider_fs, &path_deny).await?; + assert_eq!(provider_deny, payload_deny.as_bytes()); + + let deny_before = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + let deny_data = read_all(&consumer_fs, &path_deny).await?; + let deny_after = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + assert_eq!(deny_data, payload_deny.as_bytes()); + assert_eq!(deny_after, deny_before); + + Ok::<(), FsError>(()) + })?; + + Ok(()) +} + +#[test] +fn test_minicluster_p2p_runtime_policy_sync_applies_on_startup() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .mutate_conf(|conf| { + conf.master.p2p_policy_version = 1; + conf.master.p2p_policy_signing_key = "policy-secret".to_string(); + conf.master.p2p_tenant_whitelist = vec!["tenant-a".to_string()]; + conf.master.p2p_peer_whitelist = vec![]; + }) + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf; + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.tenant_whitelist = vec!["tenant-b".to_string()]; + provider_conf.client.p2p.policy_hmac_key = "policy-secret".to_string(); + provider_conf.client.clean_task_interval = Duration::from_secs(5); + provider_conf.client.clean_task_interval_str = "5s".to_string(); + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-startup-policy-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + rt.block_on(async move { + assert!( + wait_policy_version_with_deadline(&provider_fs, 1, Duration::from_millis(800)).await + ); + Ok::<(), FsError>(()) + })?; + + Ok(()) +} + +#[test] +fn test_minicluster_p2p_runtime_policy_dual_key_rotation_window() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .mutate_conf(|conf| { + conf.master.p2p_policy_version = 1; + conf.master.p2p_policy_signing_key = "new-secret".to_string(); + conf.master.p2p_policy_transition_signing_key = "old-secret".to_string(); + conf.master.p2p_tenant_whitelist = vec!["tenant-a".to_string()]; + conf.master.p2p_peer_whitelist = vec![]; + }) + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf.clone(); + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.tenant_whitelist = vec!["tenant-b".to_string()]; + provider_conf.client.p2p.policy_hmac_key = "old-secret".to_string(); + provider_conf.client.clean_task_interval = Duration::from_millis(100); + provider_conf.client.clean_task_interval_str = "100ms".to_string(); + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-rotation-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + let bootstrap = rt + .block_on(wait_bootstrap_addr(&provider_fs)) + .expect("provider bootstrap address should be ready"); + + let mut consumer_conf = base_conf; + consumer_conf.client.p2p.enable = true; + consumer_conf.client.p2p.enable_mdns = false; + consumer_conf.client.p2p.enable_dht = false; + consumer_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + consumer_conf.client.p2p.bootstrap_peers = vec![bootstrap]; + consumer_conf.client.p2p.policy_hmac_key = "old-secret".to_string(); + consumer_conf.client.clean_task_interval = Duration::from_millis(100); + consumer_conf.client.clean_task_interval_str = "100ms".to_string(); + consumer_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-consumer-rotation-cache-{}", + Utils::uuid() + ); + let consumer_fs = testing.get_fs(Some(rt.clone()), Some(consumer_conf))?; + + rt.block_on(async move { + assert!(wait_policy_version(&provider_fs, 1).await); + + let parent = Path::from_str("/p2p-policy-rotation-window")?; + let _ = provider_fs.delete(&parent, true).await; + provider_fs.mkdir(&parent, true).await?; + + let path = Path::from_str("/p2p-policy-rotation-window/allow.log")?; + let payload = Utils::rand_str(128 * 1024); + let opts = provider_fs + .create_opts_builder() + .create_parent(true) + .x_attr("tenant_id".to_string(), "tenant-a".as_bytes().to_vec()) + .build(); + let mut writer = provider_fs.create_with_opts(&path, opts, true).await?; + writer.write(payload.as_bytes()).await?; + writer.complete().await?; + let provider_data = read_all(&provider_fs, &path).await?; + assert_eq!(provider_data, payload.as_bytes()); + + let before = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + let data = read_all(&consumer_fs, &path).await?; + let after = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + assert_eq!(data, payload.as_bytes()); + assert!(after > before); + + Ok::<(), FsError>(()) + })?; + + Ok(()) +} + +#[test] +fn test_minicluster_p2p_runtime_policy_signature_mismatch_rejected() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .mutate_conf(|conf| { + conf.master.p2p_policy_version = 1; + conf.master.p2p_policy_signing_key = "policy-secret".to_string(); + conf.master.p2p_tenant_whitelist = vec!["tenant-a".to_string()]; + conf.master.p2p_peer_whitelist = vec![]; + }) + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf.clone(); + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.tenant_whitelist = vec!["tenant-b".to_string()]; + provider_conf.client.p2p.policy_hmac_key = "wrong-secret".to_string(); + provider_conf.client.clean_task_interval = Duration::from_millis(100); + provider_conf.client.clean_task_interval_str = "100ms".to_string(); + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + let bootstrap = rt + .block_on(wait_bootstrap_addr(&provider_fs)) + .expect("provider bootstrap address should be ready"); + + let mut consumer_conf = base_conf; + consumer_conf.client.p2p.enable = true; + consumer_conf.client.p2p.enable_mdns = false; + consumer_conf.client.p2p.enable_dht = false; + consumer_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + consumer_conf.client.p2p.bootstrap_peers = vec![bootstrap]; + consumer_conf.client.p2p.policy_hmac_key = "wrong-secret".to_string(); + consumer_conf.client.clean_task_interval = Duration::from_millis(100); + consumer_conf.client.clean_task_interval_str = "100ms".to_string(); + consumer_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-consumer-cache-{}", + Utils::uuid() + ); + let consumer_fs = testing.get_fs(Some(rt.clone()), Some(consumer_conf))?; + + rt.block_on(async move { + assert!(!wait_policy_version(&provider_fs, 1).await); + assert!(wait_policy_rejects(&provider_fs, 1).await); + + let parent = Path::from_str("/p2p-policy-signature-reject")?; + let _ = provider_fs.delete(&parent, true).await; + provider_fs.mkdir(&parent, true).await?; + + let path = Path::from_str("/p2p-policy-signature-reject/deny.log")?; + let payload = Utils::rand_str(128 * 1024); + let opts = provider_fs + .create_opts_builder() + .create_parent(true) + .x_attr("tenant_id".to_string(), "tenant-a".as_bytes().to_vec()) + .build(); + let mut writer = provider_fs.create_with_opts(&path, opts, true).await?; + writer.write(payload.as_bytes()).await?; + writer.complete().await?; + + let baseline = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + let data = read_all(&consumer_fs, &path).await?; + let after = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test") + .bytes_sent; + assert_eq!(data, payload.as_bytes()); + assert_eq!(after, baseline); + + Ok::<(), FsError>(()) + })?; + + Ok(()) +} + +#[test] +fn test_minicluster_p2p_miss_fallbacks_to_worker() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf.clone(); + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + let bootstrap = rt + .block_on(wait_bootstrap_addr(&provider_fs)) + .expect("provider bootstrap address should be ready"); + + let mut consumer_conf = base_conf; + consumer_conf.client.p2p.enable = true; + consumer_conf.client.p2p.enable_mdns = false; + consumer_conf.client.p2p.enable_dht = false; + consumer_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + consumer_conf.client.p2p.bootstrap_peers = vec![bootstrap]; + consumer_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-consumer-cache-{}", + Utils::uuid() + ); + let consumer_fs = testing.get_fs(Some(rt.clone()), Some(consumer_conf))?; + + rt.block_on(async move { + let path = Path::from_str("/p2p-fallback/miss-then-worker.log")?; + let payload = Utils::rand_str(128 * 1024); + provider_fs.write_string(&path, payload.as_str()).await?; + + let provider_before = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + let consumer_before = consumer_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + let out = read_all(&consumer_fs, &path).await?; + let provider_after = provider_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + let consumer_after = consumer_fs + .p2p_stats_snapshot() + .expect("p2p should be enabled in this test"); + + assert_eq!(out, payload.as_bytes()); + assert_eq!( + provider_after.bytes_sent, provider_before.bytes_sent, + "p2p miss should fallback to worker without provider p2p transfer" + ); + assert_eq!( + consumer_after.bytes_recv, consumer_before.bytes_recv, + "consumer should not receive p2p bytes on fallback-worker path" + ); + Ok::<(), FsError>(()) + })?; + + Ok(()) +} + +#[test] +fn test_minicluster_p2p_miss_fails_when_fallback_disabled() -> FsResult<()> { + let _guard = p2p_test_guard(); + let rt = Arc::new(AsyncRuntime::single()); + let testing = Testing::builder() + .workers(2) + .with_base_conf_path("../etc/curvine-cluster.toml") + .build()?; + testing.start_cluster()?; + let base_conf = testing.get_active_cluster_conf()?; + + let mut provider_conf = base_conf.clone(); + provider_conf.client.p2p.enable = true; + provider_conf.client.p2p.enable_mdns = false; + provider_conf.client.p2p.enable_dht = false; + provider_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + provider_conf.client.p2p.bootstrap_peers = vec![]; + provider_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-provider-cache-{}", + Utils::uuid() + ); + let provider_fs = testing.get_fs(Some(rt.clone()), Some(provider_conf))?; + + let bootstrap = rt + .block_on(wait_bootstrap_addr(&provider_fs)) + .expect("provider bootstrap address should be ready"); + + let mut consumer_conf = base_conf; + consumer_conf.client.p2p.enable = true; + consumer_conf.client.p2p.enable_mdns = false; + consumer_conf.client.p2p.enable_dht = false; + consumer_conf.client.p2p.listen_addrs = vec!["/ip4/127.0.0.1/tcp/0".to_string()]; + consumer_conf.client.p2p.bootstrap_peers = vec![bootstrap]; + consumer_conf.client.p2p.fallback_worker_on_fail = false; + consumer_conf.client.p2p.cache_dir = format!( + "../testing/curvine-tests/p2p-consumer-cache-{}", + Utils::uuid() + ); + let consumer_fs = testing.get_fs(Some(rt.clone()), Some(consumer_conf))?; + + rt.block_on(async move { + let path = Path::from_str("/p2p-fallback/miss-no-worker.log")?; + let payload = Utils::rand_str(128 * 1024); + provider_fs.write_string(&path, payload.as_str()).await?; + + let err = read_all(&consumer_fs, &path) + .await + .expect_err("p2p miss should fail when worker fallback is disabled"); + let err_msg = err.to_string(); + let err_msg_lower = err_msg.to_lowercase(); + assert!( + err_msg_lower.contains("fallback") + || err_msg_lower.contains("p2p read miss") + || err_msg_lower.contains("buffer read"), + "unexpected error message: {}", + err_msg + ); + Ok::<(), FsError>(()) + })?; + + Ok(()) +} diff --git a/docs/plans/2026-03-26-p2p-pr-split-execution-plan.md b/docs/plans/2026-03-26-p2p-pr-split-execution-plan.md new file mode 100644 index 000000000..27afaa0bf --- /dev/null +++ b/docs/plans/2026-03-26-p2p-pr-split-execution-plan.md @@ -0,0 +1,677 @@ +# P2P PR Split Execution Plan + +> **For Claude/Codex:** This document is a strict execution guide for rebuilding the current `feature/p2p` branch into a sequence of reviewable PRs from `github/main`. Do not improvise the PR boundaries. Do not optimize the order. Do not merge behavior-changing P2P read-path code before the required foundation PRs are in place. + +**Goal:** Split the current P2P branch into a sequence of independently mergeable PRs that preserve mainline behavior at every step and keep P2P fully gated until the first end-to-end slice is intentionally enabled. + +**Architecture:** The split must be done as vertical capability slices, not by commit order and not by file ownership. The first slices add protocol/config/foundation only. The first PR that changes read behavior must still be guarded by `client.p2p.enable = false` by default and must preserve worker fallback. + +**Tech Stack:** Rust, libp2p, protobuf/prost, Curvine client/master/server, shell-based verification scripts, minicluster integration tests. + +--- + +## 1. Why This Plan Exists + +The current `feature/p2p` branch is too large to review safely as one PR. The branch contains protocol changes, config changes, client read-path changes, runtime policy, data-plane redirect, cache backends, performance work, test harnesses, and rollout material all mixed together. + +A naive split will fail for one of three reasons: + +1. It will split by commit order and produce PRs that compile but have incoherent behavior. +2. It will split by file and produce PRs that do not build because `service.rs`, `block_reader.rs`, `fs_context.rs`, `proto`, and config changes are tightly coupled. +3. It will expose partial P2P behavior on mainline before fallback and policy mechanisms are present. + +This plan avoids those failures by enforcing four rules: + +1. Use `github/main` as the only valid split base. +2. Keep P2P default-off until the dedicated end-to-end slice lands. +3. Preserve worker fallback in every behavior-changing PR. +4. Keep verification proportional: no meaningless test spam, but every PR must prove its contract. + +--- + +## 2. Hard Rules + +These rules are mandatory. + +1. **Base branch rule** + - Every PR branch must be created from `github/main`, not local `main`. + - Reason: local `main` is behind `github/main` and will contaminate the split. + +2. **Default-off rule** + - `client.p2p.enable` must remain `false` by default until the full feature is intentionally enabled by configuration. + - No PR may change the default rollout posture. + +3. **Fallback rule** + - Every PR that touches read behavior must preserve direct worker read as the safe fallback path. + - No PR may create a situation where `enable=true` can bypass worker fallback without explicit design intent. + +4. **No fat cherry-pick rule** + - Do not cherry-pick the original large commits as-is. + - The original commits are too wide and will recreate the same review problem. + - Rebuild each PR by selectively restoring files or hunks from `feature/p2p`. + +5. **No standalone cleanup PR rule** + - Do not create separate PRs for style-only changes, test trimming, or scope cleanup. + - Absorb necessary cleanup into the owning PR. + +6. **Verification rule** + - Each PR must run only the smallest meaningful verification set for that slice. + - Do not run noisy or irrelevant tests just to inflate confidence. + +7. **Behavior continuity rule** + - The first six PRs must not change mainline client read behavior. + - The first behavior-changing PR must still be safe to merge because the feature remains opt-in. + +--- + +## 3. Execution Model + +### 3.0 Preflight + +Before creating any PR worktree, verify the split sources exist locally and are current: + +```bash +git fetch origin +git fetch github +git branch --list feature/p2p +git rev-parse github/main +``` + +If `feature/p2p` does not exist locally, create a local ref first: + +```bash +git branch feature/p2p origin/feature/p2p +``` + +Do not run any extraction step until both `github/main` and `feature/p2p` resolve locally. + +### 3.1 Branch Naming + +Create exactly these PR branches: + +1. `p2p-01-file-version-foundation` +2. `p2p-02-master-policy-protocol` +3. `p2p-03-p2p-config-surface` +4. `p2p-04-client-p2p-module-shell` +5. `p2p-05-fetched-cache-backend` +6. `p2p-06-service-lifecycle-shell` +7. `p2p-07-read-acceleration-baseline` +8. `p2p-08-runtime-policy-sync` +9. `p2p-09-data-plane-redirect` +10. `p2p-10-correctness-hardening` +11. `p2p-11-reader-and-listener-lifecycle-hardening` +12. `p2p-12-unified-boundary-hardening` +13. `p2p-13-data-plane-fast-path` +14. `p2p-14-adaptive-read-performance` +15. `p2p-15-runtime-service-hardening` +16. `p2p-16-metrics-label-telemetry` +17. `p2p-17-validation-harness-and-samples` + +### 3.2 Recommended Workspace Setup + +Use separate worktrees. Example: + +```bash +git fetch github +git worktree add ../curvine-p2p-01 github/main +cd ../curvine-p2p-01 +git switch -c p2p-01-file-version-foundation +``` + +Repeat per PR. Do not stack 14 branches in one dirty worktree. + +### 3.3 Extraction Strategy + +Preferred extraction order for each PR: + +```bash +# from the PR worktree rooted at github/main +git restore --source feature/p2p -- path/to/file +# or, when only part of a file belongs in the PR +git restore -p --source feature/p2p -- path/to/file +``` + +Then manually prune any hunk that belongs to a later PR. + +If the local source branch name is different, substitute it explicitly: + +```bash +git restore --source origin/feature/p2p -- path/to/file +``` + +Do not use `git cherry-pick` on the original fat commits unless the commit maps 1:1 to the target PR, which is rare here. + +--- + +## 4. Global Verification Matrix + +Use the smallest meaningful gate for each PR. + +### 4.1 Foundation PRs (`p2p-01` to `p2p-06`) + +Run only: + +```bash +cargo fmt --all +cargo test -p curvine-common --lib +cargo test -p curvine-client --lib +``` + +Add targeted tests only if the PR introduces new protocol/config behavior. + +### 4.2 Behavior PRs (`p2p-07` to `p2p-13`) + +Minimum gate: + +```bash +cargo fmt --all +cargo test -p curvine-client --test p2p_e2e_test -- --nocapture +cargo test -p curvine-tests --test p2p_read_acceleration_test -- --nocapture +bash scripts/tests/p2p-production-gate.sh +``` + +If the PR does not yet include the relevant script or test, run the narrowest equivalent meaningful subset. + +### 4.3 Tail PRs (`p2p-14` to `p2p-17`) + +Use the smallest meaningful subset: + +```bash +cargo fmt --all +cargo test -p curvine-client --lib --no-run +``` + +Then add only the targeted checks required by the tail slice: +- `p2p-14`: adaptive-path targeted tests +- `p2p-15`: runtime-service targeted tests + focused `p2p_e2e_test` +- `p2p-16`: metrics/label telemetry targeted tests +- `p2p-17`: proof-driver build, harness script syntax/forwarder checks, and `p2p_read_acceleration_test --no-run` or stronger + +--- + +## 5. PR Sequence + +## PR 1: `p2p-01-file-version-foundation` + +**Purpose:** Establish shared file-version metadata required by the P2P read path without changing read behavior. + +**Include:** +- `curvine-common/proto/common.proto` +- `curvine-common/src/state/file_status.rs` +- `curvine-common/src/utils/proto_utils.rs` only the `FileStatus`/`version_epoch` portion + +**Exclude:** +- Any P2P-specific policy fields +- Any client read-path changes + +**Behavior expectation:** No user-visible change. + +**Required verification:** +```bash +cargo test -p curvine-common proto_roundtrip_keeps -- --nocapture +``` + +**Merge condition:** Safe on its own. No client/server behavior shift. + +--- + +## PR 2: `p2p-02-master-policy-protocol` + +**Purpose:** Add wire-level master policy structures and signing utilities. + +**Include:** +- `curvine-common/proto/master.proto` +- `curvine-common/src/state/master_info.rs` +- `curvine-common/src/utils/common_utils.rs` only policy-signing helpers +- minimal `proto_utils.rs` support for master policy fields + +**Exclude:** +- client sync logic +- master runtime application logic + +**Behavior expectation:** Existing clients remain unaffected. + +**Required verification:** targeted proto/signature tests only. + +**Merge condition:** Protocol definitions compile and decode legacy payloads. + +--- + +## PR 3: `p2p-03-p2p-config-surface` + +**Purpose:** Add P2P config schema and validation while keeping runtime disabled by default. + +**Include:** +- `curvine-common/src/conf/client_conf.rs` +- `curvine-common/src/conf/master_conf.rs` +- `curvine-common/src/conf/mod.rs` +- any minimal cargo/proto glue required for config compilation + +**Exclude:** +- service wiring +- read-path interception + +**Behavior expectation:** P2P remains unused unless explicitly enabled. + +**Required verification:** config parsing and default-value tests. + +**Merge condition:** Minimal config loads with `enable=false`; no runtime path change. + +--- + +## PR 4: `p2p-04-client-p2p-module-shell` + +**Purpose:** Introduce the client P2P module namespace and base types without read-path integration. + +**Include:** +- `curvine-client/src/lib.rs` +- `curvine-client/src/p2p/mod.rs` +- `curvine-client/src/p2p/discovery.rs` +- `curvine-client/src/p2p/transfer.rs` +- `curvine-client/src/file/mod.rs` only exports needed by the shell + +**Exclude:** +- `service.rs` behavior +- cache manager +- read-path hooks + +**Behavior expectation:** Pure scaffolding. + +**Required verification:** `cargo test -p curvine-client --lib` + +**Merge condition:** Module compiles; runtime behavior unchanged. + +--- + +## PR 5: `p2p-05-fetched-cache-backend` + +**Purpose:** Add the fetched-cache backend as infrastructure before it is used by the main read path. + +**Include:** +- `curvine-client/src/p2p/cache_manager.rs` +- `curvine-client/Cargo.toml` dependency additions required by this backend + +**Exclude:** +- any `BlockReader` / `FsContext` integration +- data-plane logic + +**Behavior expectation:** Backend exists but is dormant. + +**Required verification:** backend unit tests only. + +**Merge condition:** Standalone cache backend compiles and passes its local contract tests. + +--- + +## PR 6: `p2p-06-service-lifecycle-shell` + +**Purpose:** Introduce `P2pService` lifecycle wiring into the client, but do not intercept reads yet. + +**Include:** +- early `service.rs` lifecycle skeleton only +- `fs_context.rs` service ownership/start/stop shell +- `curvine_filesystem.rs` creation/plumbing +- minimal metrics scaffolding needed to compile + +**Exclude:** +- actual read acceleration +- policy sync behavior +- redirect/data-plane logic + +**Behavior expectation:** With `enable=false`, no change. With `enable=true`, service may start but must not take over reads. + +**Required verification:** targeted lifecycle tests only. + +**Merge condition:** start/stop is clean and isolated from mainline reads. + +--- + +## PR 7: `p2p-07-read-acceleration-baseline` + +**Purpose:** First end-to-end P2P read path behind the feature flag, with worker fallback intact. + +**Include:** +- baseline `BlockReader`/`FsReader` integration +- enough `service.rs` logic to serve chunk fetches +- minimal block/file-side plumbing required for a P2P read hit + +**Primary files:** +- `curvine-client/src/block/block_reader.rs` +- `curvine-client/src/block/block_reader_remote.rs` +- `curvine-client/src/file/fs_reader_base.rs` +- `curvine-client/src/file/fs_reader_buffer.rs` +- `curvine-client/src/p2p/service.rs` + +**Exclude:** +- runtime master policy sync +- data-plane redirect +- advanced perf optimizations + +**Behavior expectation:** +- `enable=false`: identical to current mainline +- `enable=true`: P2P may serve reads, but worker fallback must remain correct + +**Required verification:** +- minimal e2e read-acceleration proof +- no broad stress harness yet + +**Merge condition:** This is the first PR that intentionally changes enabled behavior. + +--- + +## PR 8: `p2p-08-runtime-policy-sync` + +**Purpose:** Add master-driven runtime policy, signature verification, startup sync, periodic sync, and rollback to local policy. + +**Include:** +- client sync logic in `fs_context.rs` / `service.rs` +- master handlers and filesystem support +- runtime allowlist application and version persistence + +**Primary files:** +- `curvine-server/src/master/master_handler.rs` +- `curvine-server/src/master/fs/master_filesystem.rs` +- `curvine-client/src/file/fs_context.rs` +- `curvine-client/src/file/curvine_filesystem.rs` +- `curvine-client/src/p2p/service.rs` + +**Exclude:** +- data-plane performance work + +**Required verification:** +- startup sync applies immediately +- version `0` rollback restores local policy +- policy failure surfaces as failure, not silent success + +--- + +## PR 9: `p2p-09-data-plane-redirect` + +**Purpose:** Introduce redirect-to-data-plane architecture and ticketed follow-up fetches. + +**Include:** +- control-plane redirect response +- data-plane listener/bind lifecycle +- request/response framing needed for redirected fetch +- block follow-up read path + +**Primary files:** +- `curvine-client/src/p2p/service.rs` +- `curvine-client/src/p2p/transfer.rs` if framing types evolve + +**Exclude:** +- correctness hardening that can be isolated later +- advanced perf tuning + +**Required verification:** +- redirect only when first chunk is serviceable +- follow-up response never issues pointless ticket +- worker fallback remains available + +**Execution note:** This branch is intermediate-only. It establishes redirect/ticket plumbing and some fetched-cache visibility needed to make the redirect path function, but it is not rollout-usable on its own. Deployment-safe redirect behavior does not exist until `p2p-11`, after `p2p-10` correctness fixes and `p2p-11` listener/reader lifecycle hardening. + +--- + +## PR 10: `p2p-10-correctness-hardening` + +**Purpose:** Fix correctness gaps discovered after redirect architecture exists. + +**Include:** +- avoid redirecting empty peers +- serve published chunks before persist completes +- scope cached data-plane tickets correctly by request context +- refreshed provider advertisement via identify +- pending-fetched visibility and related correctness glue + +**Primary file:** +- `curvine-client/src/p2p/service.rs` + +**Required verification:** +- no stale ticket reuse across tenant/mtime +- no `response_not_found_or_empty` window after publish +- provider refresh works after republish + +**Execution note:** This branch closes redirect correctness gaps, but it is still intermediate-only for rollout. Data-plane stop/restart correctness, transient `accept()` recovery, and reader/pool lifecycle safety land in `p2p-11`. + +--- + +## PR 11: `p2p-11-reader-and-listener-lifecycle-hardening` + +**Purpose:** Fix failure handling and lifecycle safety in reader paths and data-plane listener management. + +**Include:** +- reader fallback correctness +- last-replica failure behavior +- remote reader cleanup and pool hygiene +- listener retry, shutdown, generation isolation +- `FsContext`/pool lifetime fixes directly needed by P2P lifecycle + +**Primary files:** +- `curvine-client/src/block/block_reader.rs` +- `curvine-client/src/block/block_reader_remote.rs` +- `curvine-client/src/block/block_client_pool.rs` +- `curvine-client/src/file/fs_context.rs` +- `curvine-client/src/p2p/service.rs` + +**Required verification:** +- broken remote readers do not leak back into pool +- stop/start listener is clean +- transient listener failure self-recovers + +**Execution note:** This is the first redirect slice that is reasonable to describe as rollout-usable, and only when stacked on top of `p2p-10`. + +--- + +## PR 12: `p2p-12-unified-boundary-hardening` + +**Purpose:** Fix the one unified boundary issue that is both independently mergeable and relevant to mounted-path correctness: mount lookup must prefer the deepest matching mount. + +**Include:** +- `mount_cache.rs` + +**Exclude:** +- `cache_sync_writer.rs` changes that depend on later mount-write-type evolution +- `unified_filesystem.rs` route changes that depend on later mount model expansion +- unrelated unified refactors + +**Required verification:** +- mounted path selection correctness + +**Dependency note:** This PR is intentionally detached from the core P2P chain. It stacks on `p2p-11`, but `p2p-13` and later branches continue from `p2p-11`, not from `p2p-12`. Merge it independently when ready. + +**Note:** During execution, broader unified changes were found to depend on non-P2P mount-model drift (`WriteType`, consistency semantics, and proto/display conversions). Those changes must not be force-fit into this PR. + +--- + +## PR 13: `p2p-13-data-plane-fast-path` + +**Purpose:** Land the 0ms-critical data-plane optimizations after the fetched-cache path is stable. + +**Include:** +- avoid zeroing response buffers +- stream first data-plane chunk +- 0ms data-plane read-path shaping that does not change rollout posture + +**Primary files:** +- `curvine-client/src/p2p/service.rs` +- `curvine-client/src/block/block_reader_remote.rs` if required by the first-chunk path + +**Required verification:** +- production gate +- 0ms benchmark sanity +- redirected-read targeted tests + +**Rule:** Do not mix adaptive bypass, metrics-label work, or test harness expansion into this PR. + +--- + +## PR 14: `p2p-14-adaptive-read-performance` + +**Purpose:** Land read-side optimization work that changes selection heuristics only after the 0ms data-plane path is already isolated. + +**Include:** +- cached-hit read-flight bypass +- adaptive worker bypass +- adaptive-sample hygiene + +**Primary files:** +- `curvine-client/src/block/block_reader.rs` +- `curvine-client/src/file/fs_context.rs` +- `curvine-client/src/p2p/service.rs` only if required for sample attribution + +**Required verification:** +- production gate +- 0ms benchmark sanity +- targeted adaptive-behavior tests + +**Rule:** Do not mix metrics-label work, validation harnesses, or rollout material into this PR. + +--- + +## PR 15: `p2p-15-runtime-service-hardening` + +**Purpose:** Finish the remaining runtime hardening in `service.rs` after adaptive read behavior has landed. + +**Include:** +- data-plane tenant whitelist runtime update handling +- data-plane listener restart/stop generation safety +- cached ticket request-context scoping +- a compact `p2p_e2e_test` suite that locks the highest-value runtime regressions in service lifecycle, negative provider cache, and provider rediscovery + +**Exclude:** +- metrics-label telemetry +- proof/stress harnesses +- config samples and rollout material + +**Required verification:** +- targeted runtime-service tests +- compact core `p2p_e2e_test` suite + +**Rule:** This PR may still change product code, but it must stay inside runtime service hardening only. + +--- + +## PR 16: `p2p-16-metrics-label-telemetry` + +**Purpose:** Add observability for read-source attribution, fallback reasons, and labeled P2P metrics without changing core protocol behavior. + +**Include:** +- `curvine-client/src/client_metrics.rs` +- read-source / fallback telemetry plumbing in `block_reader.rs` +- tenant/job label extraction in `fs_reader_base.rs` +- metrics label policy setup and P2P snapshot syncing in `fs_context.rs` + +**Exclude:** +- `curvine_filesystem.rs` API churn +- `fs_client.rs` cleanup or rename work +- product-path behavior changes unrelated to telemetry + +**Required verification:** +- labeled read-source metrics tests +- labeled fallback metrics tests +- one cached-read path regression test +- `cargo test -p curvine-client --lib --no-run` + +--- + +## PR 17: `p2p-17-validation-harness-and-samples` + +**Purpose:** Bring in verification harnesses, sample configs, rollout material, and operational scripts after product logic is settled. + +**Include:** +- `curvine-tests/tests/p2p_read_acceleration_test.rs` +- `curvine-cli/src/bin/p2p_proof_driver.rs` +- `scripts/tests/p2p-*.sh` +- `scripts/tests/master-forwarder.py` +- `etc/curvine-cluster-p2p-*.toml` + +**Exclude:** +- unrelated historical script churn from other topics +- new product-path behavior + +**Required verification:** +- `cargo build -p curvine-cli --bin p2p_proof_driver` +- `cargo test -p curvine-tests --test p2p_read_acceleration_test --no-run` +- `bash scripts/tests/test-master-forwarder.sh` +- shell syntax checks for the added scripts + +**Rule:** This PR is the end of the split, not the beginning. Do not use it to smuggle more product changes. + +--- + +## 6. Commits That Must Not Become Standalone PRs + +Do **not** create separate PRs for these categories: + +1. style-only cleanup +2. test trimming only +3. review-scope cleanup only +4. local experimental benchmark artifacts +5. docs-only drift caused by previous sessions + +In the current branch, the following commits are examples of changes that should be absorbed, dropped, or rewritten into the owning PR rather than preserved as standalone PRs: + +- `b3eb0d1` +- `e10c3ae` +- `4f950c7` +- `c51153f` +- `6da2486` + +--- + +## 7. Extraction Checklist Per PR + +For every PR branch, execute this checklist in order. + +1. Create worktree from `github/main`. +2. Create the PR branch with the exact branch name from this plan. +3. Restore only the files/hunks that belong to that PR. +4. Delete any later-stage code that slipped in. +5. Run `cargo fmt --all`. +6. Run only the verification required by that PR. +7. Confirm `client.p2p.enable` default remains `false` unless the PR is only config/schema and still not behavior-changing. +8. Confirm worker fallback still exists for any read-path PR. +9. Confirm the diff contains no unrelated script churn. +10. Commit with a title matching the PR purpose. +11. Open PR against `github/main`. +12. After merge, rebase or recreate the next PR branch from fresh `github/main` plus merged PRs as needed. + +--- + +## 8. Reviewer Guidance + +Each PR description should explicitly answer these four questions: + +1. What capability becomes possible after this PR? +2. What still remains disabled or incomplete after this PR? +3. Why is this PR safe to merge even if the rest of the split has not landed yet? +4. What exact tests prove that safety claim? + +If the PR description cannot answer those four questions cleanly, the PR is too large or incorrectly scoped. + +--- + +## 9. Stop Conditions + +Stop and rescope if any of the following happens: + +1. A PR requires more than one unrelated behavior change to build. +2. A PR cannot be explained without referencing more than two later PRs. +3. A PR forces `enable=true` behavior to be partially active before fallback/policy/correctness are ready. +4. A PR review discussion starts focusing on unrelated performance tuning instead of the PR's stated contract. + +When a stop condition triggers, split the PR further or move code to a later PR. Do not push through with a messy boundary. + +--- + +## 10. Final Recommendation + +Use **17 PRs**, not 20. + +That is the smallest number that keeps: +- protocol/config foundations understandable, +- the first enabled read-path PR reviewable, +- correctness hardening separate from optimization, +- validation/ops separate from product logic. + +The extra tail split is intentional: runtime-service hardening, telemetry, and validation assets review more cleanly as separate concerns than they did in the original 14-PR draft. The detached `p2p-12` side PR is also intentional: it fixes a real mounted-path bug without forcing artificial dependency depth into the core P2P chain. diff --git a/etc/curvine-cluster-p2p-bootstrap.toml b/etc/curvine-cluster-p2p-bootstrap.toml new file mode 100644 index 000000000..3147e4b1c --- /dev/null +++ b/etc/curvine-cluster-p2p-bootstrap.toml @@ -0,0 +1,51 @@ + +format_master = false +format_worker = false + +[master] +meta_dir = "/data/curvine/master-meta" +log = { level = "info", log_dir = "stdout", file_name = "master.log" } +p2p_policy_version = 1 +p2p_policy_signing_key = "replace-with-strong-secret-32-chars-min" +p2p_tenant_whitelist = [] + +[journal] +journal_addrs = [ + {id = 1, hostname = "localhost", port = 8996} +] +journal_dir = "/data/curvine/journal" + +[worker] +dir_reserved = "0" +data_dir = [ + "[DISK]/data/curvine/worker-data", +] +log = { level = "info", log_dir = "stdout", file_name = "worker.log" } +enable_s3_gateway = false + +[client] + +[client.p2p] +enable = true +cache_dir = "/data/curvine/p2p-cache" +listen_addrs = ["/ip4/0.0.0.0/udp/4101/quic-v1"] +policy_hmac_key = "replace-with-strong-secret-32-chars-min" +bootstrap_peers = [] + +[fuse] + +[log] +level = "info" +log_dir = "stdout" +file_name = "curvine.log" + +[s3_gateway] +listen = "0.0.0.0:9900" +region = "us-east-1" +put_temp_dir = "/tmp/curvine-temp" +put_memory_buffer_threshold = 1048576 +put_max_memory_buffer = 16777216 +get_chunk_size_mb = 1.0 + +[cli] +log = { level = "warn", log_dir = "stdout", file_name = "cli.log" } diff --git a/etc/curvine-cluster-p2p-client.toml b/etc/curvine-cluster-p2p-client.toml new file mode 100644 index 000000000..832988756 --- /dev/null +++ b/etc/curvine-cluster-p2p-client.toml @@ -0,0 +1,52 @@ + +format_master = false +format_worker = false + +[master] +meta_dir = "/data/curvine/master-meta" +log = { level = "info", log_dir = "stdout", file_name = "master.log" } +p2p_policy_version = 1 +p2p_policy_signing_key = "replace-with-strong-secret-32-chars-min" +p2p_tenant_whitelist = [] + +[journal] +journal_addrs = [ + {id = 1, hostname = "localhost", port = 8996} +] +journal_dir = "/data/curvine/journal" + +[worker] +dir_reserved = "0" +data_dir = [ + "[DISK]/data/curvine/worker-data", +] +log = { level = "info", log_dir = "stdout", file_name = "worker.log" } +enable_s3_gateway = false + +[client] + +[client.p2p] +enable = true +cache_dir = "/data/curvine/p2p-cache" +policy_hmac_key = "replace-with-strong-secret-32-chars-min" +bootstrap_peers = [ + "/ip4/10.0.0.11/udp/4101/quic-v1/p2p/12D3KooWH4NJL41x4kBnKmsiQjiGYMEtrhJfmJjKEmVQwZWwcJer", +] + +[fuse] + +[log] +level = "info" +log_dir = "stdout" +file_name = "curvine.log" + +[s3_gateway] +listen = "0.0.0.0:9900" +region = "us-east-1" +put_temp_dir = "/tmp/curvine-temp" +put_memory_buffer_threshold = 1048576 +put_max_memory_buffer = 16777216 +get_chunk_size_mb = 1.0 + +[cli] +log = { level = "warn", log_dir = "stdout", file_name = "cli.log" } diff --git a/etc/curvine-cluster-p2p-minimal.toml b/etc/curvine-cluster-p2p-minimal.toml new file mode 100644 index 000000000..6109551e1 --- /dev/null +++ b/etc/curvine-cluster-p2p-minimal.toml @@ -0,0 +1,50 @@ + +format_master = false +format_worker = false + +[master] +meta_dir = "/data/curvine/master-meta" +log = { level = "info", log_dir = "stdout", file_name = "master.log" } +p2p_policy_version = 1 +p2p_policy_signing_key = "replace-with-strong-secret-32-chars-min" +p2p_tenant_whitelist = [] + +[journal] +journal_addrs = [ + {id = 1, hostname = "localhost", port = 8996} +] +journal_dir = "/data/curvine/journal" + +[worker] +dir_reserved = "0" +data_dir = [ + "[DISK]/data/curvine/worker-data", +] +log = { level = "info", log_dir = "stdout", file_name = "worker.log" } +enable_s3_gateway = false + +[client] + +[client.p2p] +enable = true +cache_dir = "/data/curvine/p2p-cache" +policy_hmac_key = "replace-with-strong-secret-32-chars-min" +bootstrap_peers = [] + +[fuse] + +[log] +level = "info" +log_dir = "stdout" +file_name = "curvine.log" + +[s3_gateway] +listen = "0.0.0.0:9900" +region = "us-east-1" +put_temp_dir = "/tmp/curvine-temp" +put_memory_buffer_threshold = 1048576 +put_max_memory_buffer = 16777216 +get_chunk_size_mb = 1.0 + +[cli] +log = { level = "warn", log_dir = "stdout", file_name = "cli.log" } diff --git a/scripts/tests/master-forwarder.py b/scripts/tests/master-forwarder.py new file mode 100644 index 000000000..c4f1febff --- /dev/null +++ b/scripts/tests/master-forwarder.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +import argparse +import socket +import threading + + +def pipe(src, dst): + try: + while True: + data = src.recv(65536) + if not data: + break + dst.sendall(data) + except Exception: + pass + finally: + try: + dst.shutdown(socket.SHUT_WR) + except Exception: + pass + + +def handle(client, target_host, target_port): + try: + upstream = socket.create_connection((target_host, target_port), timeout=5) + upstream.settimeout(None) + except Exception: + client.close() + return + threading.Thread(target=pipe, args=(client, upstream), daemon=True).start() + threading.Thread(target=pipe, args=(upstream, client), daemon=True).start() + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--listen-host", default="0.0.0.0") + parser.add_argument("--listen-port", type=int, required=True) + parser.add_argument("--target-host", default="127.0.0.1") + parser.add_argument("--target-port", type=int, required=True) + args = parser.parse_args() + + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind((args.listen_host, args.listen_port)) + listener.listen(512) + + while True: + conn, _ = listener.accept() + threading.Thread( + target=handle, + args=(conn, args.target_host, args.target_port), + daemon=True, + ).start() + + +if __name__ == "__main__": + main() diff --git a/scripts/tests/p2p-performance-ab.sh b/scripts/tests/p2p-performance-ab.sh new file mode 100755 index 000000000..078f87c4c --- /dev/null +++ b/scripts/tests/p2p-performance-ab.sh @@ -0,0 +1,458 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +DRIVER_BIN="$ROOT/target/debug/p2p_proof_driver" +EVID_ROOT="${P2P_PERF_EVID_ROOT:-$ROOT/evidence/p2p-perf}" +TS="$(date +%Y%m%d-%H%M%S)" +EVID="$EVID_ROOT/$TS" +RUNS="${P2P_PERF_RUNS:-8}" +DATA_MB="${P2P_PERF_DATA_MB:-128}" +GW_DELAY_MS="${P2P_PERF_GW_DELAY_MS:-40}" +WORKER_RATE_MBIT="${P2P_PERF_WORKER_RATE_MBIT:-0}" +SHORT_CIRCUIT="${P2P_PERF_SHORT_CIRCUIT:-false}" +REMOTE_PATH="${P2P_PERF_REMOTE_PATH:-/p2p-proof/perf-$TS.bin}" +RUNTIME_IMAGE="${P2P_RUNTIME_IMAGE:-curvine/p2p-proof-runner:latest}" +READ_CHUNK_SIZE="${P2P_PERF_READ_CHUNK_SIZE:-1MB}" +READ_CHUNK_NUM="${P2P_PERF_READ_CHUNK_NUM:-1}" +CONSUMER_READS="${P2P_PERF_CONSUMER_READS:-2}" +CONSUMER_READ_INTERVAL_MS="${P2P_PERF_CONSUMER_READ_INTERVAL_MS:-0}" +LOG_LEVEL="${P2P_PERF_LOG_LEVEL:-info}" +TRACE_ENABLE="${P2P_PERF_TRACE_ENABLE:-false}" +mkdir -p "$EVID" + +log() { + echo "[p2p-perf-ab] $*" +} + +sum_metric() { + local file="$1" + local name="$2" + awk -v n="$name" '$1 ~ ("^" n "($|\\{)") {sum+=$2} END {print sum+0}' "$file" +} + +can_manage_cluster() { + [[ -x "$ROOT/build/bin/local-cluster.sh" ]] \ + && [[ -x "$ROOT/build/bin/curvine-master.sh" ]] \ + && [[ -x "$ROOT/build/bin/curvine-worker.sh" ]] \ + && [[ -f "$ROOT/build/conf/curvine-env.sh" ]] +} + +ensure_runtime_image() { + if docker image inspect "$RUNTIME_IMAGE" >/dev/null 2>&1; then + return + fi + log "building runtime image: $RUNTIME_IMAGE" + docker build -t "$RUNTIME_IMAGE" - <<'DOCKERFILE' +FROM ubuntu:24.04 +RUN apt-get update \ + && apt-get install -y --no-install-recommends iproute2 ca-certificates \ + && rm -rf /var/lib/apt/lists/* +DOCKERFILE +} + +cleanup() { + set +e + if [[ -n "${FWD_PID:-}" ]]; then + kill "$FWD_PID" >/dev/null 2>&1 || true + wait "$FWD_PID" >/dev/null 2>&1 || true + fi + docker rm -f perf-provider >/dev/null 2>&1 || true + docker network rm cv-p2p-perf-net >/dev/null 2>&1 || true +} + +write_consumer_conf() { + local mode="$1" + local round="$2" + local conf_path="$3" + local p2p_enable="false" + local bootstrap_line='bootstrap_peers = []' + if [[ "$mode" == "on" ]]; then + p2p_enable="true" + bootstrap_line="bootstrap_peers = [\"$BOOTSTRAP_ADDR\"]" + fi + + cat > "$conf_path" < '"$driver_log"' 2>&1 +' >/dev/null + + local elapsed_ms + elapsed_ms="$(sed -n 's/^CONSUMER_READ_ELAPSED_MS=\([0-9.][0-9.]*\)$/\1/p' "$driver_log_file" | tail -n1)" + if [[ -z "$elapsed_ms" ]]; then + echo "missing CONSUMER_READ_ELAPSED_MS in $driver_log_file" >&2 + return 1 + fi + local elapsed + elapsed="$(awk -v ms="$elapsed_ms" 'BEGIN{printf "%.6f", ms/1000.0}')" + + local p2p_hit=0 + if [[ "$mode" == "on" ]]; then + local recv_delta + recv_delta="$(sed -n 's/^CONSUMER_P2P_RECV_DELTA=\([0-9][0-9]*\)$/\1/p' "$driver_log_file" | tail -n1)" + recv_delta="${recv_delta:-0}" + if awk -v v="$recv_delta" 'BEGIN{exit !(v>0)}'; then + p2p_hit=1 + fi + fi + + local output_sha + output_sha="$(sha256sum "$EVID/output-${mode}-${round}.bin" | awk '{print $1}')" + printf '%s,%s,%s,%s,%s\n' "$round" "$mode" "$elapsed" "$p2p_hit" "$output_sha" >> "$EVID/timings.csv" +} + +trap cleanup EXIT + +if [[ ! -x "$DRIVER_BIN" ]]; then + log "building p2p_proof_driver" + (cd "$ROOT" && cargo build -p curvine-cli --bin p2p_proof_driver) +fi +ensure_runtime_image + +if can_manage_cluster; then + if timeout 1 bash -lc '/dev/null 2>&1; then + log "restarting local cluster for clean baseline" + bash "$ROOT/build/bin/local-cluster.sh" stop || true + sleep 2 + fi + log "starting local cluster" + bash "$ROOT/build/bin/local-cluster.sh" start + sleep 6 +else + log "skip cluster restart/start: local-cluster scripts unavailable or not executable" +fi + +if ! timeout 1 bash -lc '/dev/null 2>&1; then + echo "master 127.0.0.1:8995 is not reachable" >&2 + exit 1 +fi + +python3 -u "$ROOT/scripts/tests/master-forwarder.py" \ + --listen-port 18995 \ + --target-port 8995 \ + >"$EVID/master-forward.log" 2>&1 & +FWD_PID=$! +sleep 1 + +if ! timeout 1 bash -lc '/dev/null 2>&1; then + echo "master forwarder 127.0.0.1:18995 is not reachable" >&2 + exit 1 +fi + +log "prepare docker network" +docker rm -f perf-provider >/dev/null 2>&1 || true +docker network rm cv-p2p-perf-net >/dev/null 2>&1 || true +docker network create --subnet 172.29.0.0/16 cv-p2p-perf-net >/dev/null + +log "prepare source data ${DATA_MB}MB" +dd if=/dev/urandom of="$EVID/source.bin" bs=1M count="$DATA_MB" status=none +SOURCE_SHA="$(sha256sum "$EVID/source.bin" | awk '{print $1}')" +printf 'round,mode,elapsed_sec,p2p_hit,output_sha\n' > "$EVID/timings.csv" + +touch "$EVID/provider-driver.log" +cat > "$EVID/provider.toml" < /evidence/provider-driver.log 2>&1 +' >/dev/null + +for _ in $(seq 1 240); do + if [[ -s "$EVID/provider_bootstrap_addr.raw.txt" && -s "$EVID/provider_peer_id.txt" ]]; then + break + fi + sleep 0.5 +done +if [[ ! -s "$EVID/provider_bootstrap_addr.raw.txt" || ! -s "$EVID/provider_peer_id.txt" ]]; then + docker logs perf-provider > "$EVID/provider.container.log" 2>&1 || true + echo "provider bootstrap/peer not ready" >&2 + exit 1 +fi + +BOOTSTRAP_ADDR="$(cat "$EVID/provider_bootstrap_addr.raw.txt")" +log "provider bootstrap: $BOOTSTRAP_ADDR" + +if [[ ! -s "$EVID/provider.out" ]]; then + echo "provider output not ready" >&2 + exit 1 +fi +PROVIDER_SHA="$(sha256sum "$EVID/provider.out" | awk '{print $1}')" +if [[ "$PROVIDER_SHA" != "$SOURCE_SHA" ]]; then + echo "provider output sha mismatch: source=$SOURCE_SHA provider=$PROVIDER_SHA" >&2 + exit 1 +fi + +sleep 6 +curl -sS http://127.0.0.1:9000/metrics > "$EVID/metrics.before.prom" + +for round in $(seq 1 "$RUNS"); do + log "round=$round mode=off" + run_case off "$round" + log "round=$round mode=on" + run_case on "$round" +done + +curl -sS http://127.0.0.1:9000/metrics > "$EVID/metrics.after.prom" + +B_SENT_BEFORE="$(sum_metric "$EVID/metrics.before.prom" client_p2p_bytes_sent)" +B_SENT_AFTER="$(sum_metric "$EVID/metrics.after.prom" client_p2p_bytes_sent)" +B_RECV_BEFORE="$(sum_metric "$EVID/metrics.before.prom" client_p2p_bytes_recv)" +B_RECV_AFTER="$(sum_metric "$EVID/metrics.after.prom" client_p2p_bytes_recv)" +DELTA_SENT="$(awk -v a="$B_SENT_AFTER" -v b="$B_SENT_BEFORE" 'BEGIN{print a-b}')" +DELTA_RECV="$(awk -v a="$B_RECV_AFTER" -v b="$B_RECV_BEFORE" 'BEGIN{print a-b}')" + +EVID="$EVID" \ +RUNS="$RUNS" \ +GW_DELAY_MS="$GW_DELAY_MS" \ +WORKER_RATE_MBIT="$WORKER_RATE_MBIT" \ +DATA_MB="$DATA_MB" \ +READ_CHUNK_SIZE="$READ_CHUNK_SIZE" \ +READ_CHUNK_NUM="$READ_CHUNK_NUM" \ +CONSUMER_READS="$CONSUMER_READS" \ +CONSUMER_READ_INTERVAL_MS="$CONSUMER_READ_INTERVAL_MS" \ +LOG_LEVEL="$LOG_LEVEL" \ +TRACE_ENABLE="$TRACE_ENABLE" \ +SOURCE_SHA="$SOURCE_SHA" \ +PROVIDER_SHA="$PROVIDER_SHA" \ +REMOTE_PATH="$REMOTE_PATH" \ +DELTA_SENT="$DELTA_SENT" \ +DELTA_RECV="$DELTA_RECV" \ +python3 - <<'PY' +import csv +import json +import os +from pathlib import Path + + +def percentile(data, p): + if not data: + return 0.0 + data = sorted(data) + if len(data) == 1: + return float(data[0]) + rank = (len(data) - 1) * p + lo = int(rank) + hi = min(lo + 1, len(data) - 1) + frac = rank - lo + return float(data[lo] + (data[hi] - data[lo]) * frac) + + +def stats(values): + values = [float(v) for v in values] + if not values: + return { + "count": 0, + "avg_sec": 0.0, + "p50_sec": 0.0, + "p95_sec": 0.0, + "min_sec": 0.0, + "max_sec": 0.0, + } + return { + "count": len(values), + "avg_sec": sum(values) / len(values), + "p50_sec": percentile(values, 0.5), + "p95_sec": percentile(values, 0.95), + "min_sec": min(values), + "max_sec": max(values), + } + + +evid = Path(os.environ["EVID"]) +source_sha = os.environ["SOURCE_SHA"] +rows = list(csv.DictReader((evid / "timings.csv").open("r", encoding="utf-8"))) + +off_times = [float(r["elapsed_sec"]) for r in rows if r["mode"] == "off"] +on_times = [float(r["elapsed_sec"]) for r in rows if r["mode"] == "on"] +on_hits = sum(1 for r in rows if r["mode"] == "on" and r["p2p_hit"] == "1") +sha_mismatches = [r for r in rows if r["output_sha"] != source_sha] + +off_stats = stats(off_times) +on_stats = stats(on_times) + +speedup = (off_stats["avg_sec"] / on_stats["avg_sec"]) if on_stats["avg_sec"] > 0 else 0.0 +latency_reduction_pct = (1.0 - (on_stats["avg_sec"] / off_stats["avg_sec"])) * 100 if off_stats["avg_sec"] > 0 else 0.0 + +all_on_hit = on_hits == len(on_times) and len(on_times) > 0 +all_sha_match = len(sha_mismatches) == 0 + +report = { + "runs_per_mode": int(os.environ["RUNS"]), + "data_mb": int(os.environ["DATA_MB"]), + "remote_path": os.environ["REMOTE_PATH"], + "read_chunk_size": os.environ["READ_CHUNK_SIZE"], + "read_chunk_num": int(os.environ["READ_CHUNK_NUM"]), + "consumer_reads_per_run": int(os.environ["CONSUMER_READS"]), + "consumer_read_interval_ms": int(os.environ["CONSUMER_READ_INTERVAL_MS"]), + "log_level": os.environ["LOG_LEVEL"], + "trace_enable": os.environ["TRACE_ENABLE"].lower() == "true", + "worker_path_gateway_delay_ms": int(os.environ["GW_DELAY_MS"]), + "worker_path_rate_limit_mbit": int(os.environ["WORKER_RATE_MBIT"]), + "mode_off": off_stats, + "mode_on": on_stats, + "speedup_x": speedup, + "avg_latency_reduction_pct": latency_reduction_pct, + "all_on_runs_p2p_hit": all_on_hit, + "on_hit_count": on_hits, + "on_total": len(on_times), + "all_output_sha_match_source": all_sha_match, + "sha_mismatch_count": len(sha_mismatches), + "provider_output_sha_match_source": os.environ["PROVIDER_SHA"] == source_sha, + "metrics_delta": { + "client_p2p_bytes_sent": float(os.environ["DELTA_SENT"]), + "client_p2p_bytes_recv": float(os.environ["DELTA_RECV"]), + }, +} + +(evid / "perf-report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n") +print(json.dumps(report, indent=2, ensure_ascii=False)) +PY + +cat "$EVID/perf-report.json" + +SPEEDUP="$(python3 - <<'PY' "$EVID/perf-report.json" +import json,sys +r=json.load(open(sys.argv[1])) +print(r['speedup_x']) +PY +)" +ON_HIT_OK="$(python3 - <<'PY' "$EVID/perf-report.json" +import json,sys +r=json.load(open(sys.argv[1])) +print('1' if r['all_on_runs_p2p_hit'] else '0') +PY +)" +SHA_OK="$(python3 - <<'PY' "$EVID/perf-report.json" +import json,sys +r=json.load(open(sys.argv[1])) +print('1' if r['all_output_sha_match_source'] else '0') +PY +)" +if [[ "$ON_HIT_OK" != "1" || "$SHA_OK" != "1" ]]; then + echo "p2p hit or sha validation failed" >&2 + exit 1 +fi +if ! awk -v v="$SPEEDUP" 'BEGIN{exit !(v>1.0)}'; then + echo "no performance gain detected" >&2 + exit 1 +fi + +log "PASS" +log "evidence_dir=$EVID" diff --git a/scripts/tests/p2p-production-gate.sh b/scripts/tests/p2p-production-gate.sh new file mode 100755 index 000000000..523e9a3e9 --- /dev/null +++ b/scripts/tests/p2p-production-gate.sh @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +set -euo pipefail + +run() { + echo "[p2p-gate] $*" + "$@" +} + +run cargo test -p curvine-client --test p2p_e2e_test -- --nocapture +run cargo test -p curvine-tests --test p2p_read_acceleration_test -- --nocapture + +echo "[p2p-gate] all required checks passed" diff --git a/scripts/tests/p2p-stress-lib.sh b/scripts/tests/p2p-stress-lib.sh new file mode 100644 index 000000000..2a5e2df12 --- /dev/null +++ b/scripts/tests/p2p-stress-lib.sh @@ -0,0 +1,243 @@ +#!/usr/bin/env bash + +p2p_stress_init() { + local script_path="$1" + ROOT="$(cd "$(dirname "$script_path")/../.." && pwd)" + DRIVER_BIN="$ROOT/target/debug/p2p_proof_driver" + EVID_ROOT="${P2P_STRESS_EVID_ROOT:-$ROOT/evidence/p2p-stress}" + TS="$(date +%Y%m%d-%H%M%S)" + RUN_ROOT="$EVID_ROOT/$TS" + RUNTIME_IMAGE="${P2P_RUNTIME_IMAGE:-curvine/p2p-proof-runner:latest}" + CLIENTS="${P2P_STRESS_CLIENTS:-4}" + DURATION_SECS="${P2P_STRESS_DURATION_SECS:-20}" + OP_INTERVAL_MS="${P2P_STRESS_OP_INTERVAL_MS:-10}" + BOOTSTRAP_WAIT_SECS="${P2P_STRESS_BOOTSTRAP_WAIT_SECS:-120}" + MASTER_PORT="${P2P_STRESS_MASTER_PORT:-8995}" + FORWARD_PORT="${P2P_STRESS_FORWARD_PORT:-18995}" + READ_CHUNK_SIZE="${P2P_STRESS_READ_CHUNK_SIZE:-1MB}" + READ_CHUNK_NUM="${P2P_STRESS_READ_CHUNK_NUM:-1}" + TRACE_ENABLE="${P2P_STRESS_TRACE_ENABLE:-false}" + LOG_LEVEL="${P2P_STRESS_LOG_LEVEL:-info}" + NETWORK_NAME="${P2P_STRESS_NETWORK_NAME:-cv-p2p-stress-net}" + NETWORK_CIDR="${P2P_STRESS_NETWORK_CIDR:-}" + SEED_IP="" + mkdir -p "$RUN_ROOT" +} + +log() { + echo "[p2p-stress] $*" +} + +sum_metric() { + local file="$1" + local name="$2" + awk -v n="$name" '$1 ~ ("^" n "($|\\{)") {sum+=$2} END {print sum+0}' "$file" +} + +can_manage_cluster() { + [[ -f "$ROOT/build/bin/local-cluster.sh" ]] \ + && [[ -f "$ROOT/build/bin/curvine-master.sh" ]] \ + && [[ -f "$ROOT/build/bin/curvine-worker.sh" ]] \ + && [[ -f "$ROOT/build/conf/curvine-env.sh" ]] +} + +network_prefix() { + local cidr="$1" + local addr="${cidr%%/*}" + IFS='.' read -r a b c _ <<<"$addr" + printf '%s.%s.%s' "$a" "$b" "$c" +} + +auto_network_cidr() { + local used + used="$( + docker network inspect $(docker network ls -q) \ + --format '{{range .IPAM.Config}}{{.Subnet}} {{end}}' 2>/dev/null || true + )" + local octet + for octet in $(seq 60 250); do + local candidate="172.31.${octet}.0/24" + if ! grep -qw "$candidate" <<<"$used"; then + printf '%s\n' "$candidate" + return 0 + fi + done + return 1 +} + +ensure_network_plan() { + if [[ -z "$NETWORK_CIDR" ]]; then + NETWORK_CIDR="$(auto_network_cidr)" || { + echo "unable to find free docker subnet for stress network" >&2 + exit 1 + } + fi + SEED_IP="$(network_prefix "$NETWORK_CIDR").11" +} + +ensure_runtime_image() { + if docker image inspect "$RUNTIME_IMAGE" >/dev/null 2>&1; then + return + fi + log "building runtime image: $RUNTIME_IMAGE" + docker build -t "$RUNTIME_IMAGE" - <<'DOCKERFILE' +FROM ubuntu:24.04 +RUN apt-get update \ + && apt-get install -y --no-install-recommends iproute2 ca-certificates \ + && rm -rf /var/lib/apt/lists/* +DOCKERFILE +} + +cleanup() { + set +e + if [[ -n "${FWD_PID:-}" ]]; then + kill "$FWD_PID" >/dev/null 2>&1 || true + wait "$FWD_PID" >/dev/null 2>&1 || true + fi + docker ps -a --format '{{.Names}}' | rg '^stress-' | xargs -r docker rm -f >/dev/null 2>&1 || true + docker network rm "$NETWORK_NAME" >/dev/null 2>&1 || true +} + +write_client_conf() { + local conf_path="$1" + local client_name="$2" + local bootstrap_addr="${3:-}" + local listen_addr="${4:-/ip4/0.0.0.0/tcp/0}" + local log_file="${5:-${client_name}.log}" + local bootstrap_peers='[]' + if [[ -n "$bootstrap_addr" ]]; then + bootstrap_peers="[\"${bootstrap_addr}\"]" + fi + cat > "$conf_path" <"$RUN_ROOT/master-forward.log" 2>&1 & + FWD_PID=$! + sleep 1 +} + +prepare_env() { + if [[ ! -x "$DRIVER_BIN" ]]; then + log "building p2p_proof_driver" + (cd "$ROOT" && cargo build -p curvine-cli --bin p2p_proof_driver) + fi + ensure_runtime_image + + if can_manage_cluster; then + if timeout 1 bash -lc "/dev/null 2>&1; then + log "restarting local cluster for clean stress baseline" + bash "$ROOT/build/bin/local-cluster.sh" stop || true + sleep 2 + fi + log "starting local cluster" + bash "$ROOT/build/bin/local-cluster.sh" start + sleep 6 + else + log "skip cluster restart/start: local-cluster scripts unavailable or not executable" + fi + + if ! timeout 1 bash -lc "/dev/null 2>&1; then + echo "master 127.0.0.1:${MASTER_PORT} is not reachable" >&2 + exit 1 + fi + + export FORWARD_PORT MASTER_PORT + start_forwarder + if ! timeout 1 bash -lc "/dev/null 2>&1; then + echo "master forwarder 127.0.0.1:${FORWARD_PORT} is not reachable" >&2 + exit 1 + fi + + ensure_network_plan + docker network rm "$NETWORK_NAME" >/dev/null 2>&1 || true + docker network create --subnet "$NETWORK_CIDR" "$NETWORK_NAME" >/dev/null +} + +wait_for_file() { + local path="$1" + local label="$2" + local attempts="${3:-240}" + local sleep_secs="${4:-0.5}" + for _ in $(seq 1 "$attempts"); do + if [[ -s "$path" ]]; then + return 0 + fi + sleep "$sleep_secs" + done + echo "${label} not ready: $path" >&2 + return 1 +} + +count_manifest_entries() { + local manifest_dir="$1" + python3 - <<'PY' "$manifest_dir" +from pathlib import Path +import sys + +manifest_dir = Path(sys.argv[1]) +total = 0 +if manifest_dir.exists(): + for path in manifest_dir.glob("*.jsonl"): + with path.open("r", encoding="utf-8", errors="ignore") as handle: + total += sum(1 for line in handle if line.strip()) +print(total) +PY +} + +wait_for_manifest_entries() { + local manifest_dir="$1" + local min_entries="$2" + local label="$3" + local attempts="${4:-240}" + local sleep_secs="${5:-0.5}" + local total=0 + for _ in $(seq 1 "$attempts"); do + total="$(count_manifest_entries "$manifest_dir")" + if [[ "$total" -ge "$min_entries" ]]; then + return 0 + fi + sleep "$sleep_secs" + done + echo "${label} not ready: $manifest_dir expected>=$min_entries actual=$total" >&2 + return 1 +} + +snapshot_manifest_dir() { + local src_dir="$1" + local dst_dir="$2" + mkdir -p "$dst_dir" + find "$dst_dir" -type f -name '*.jsonl' -delete + if find "$src_dir" -maxdepth 1 -type f -name '*.jsonl' | grep -q .; then + find "$src_dir" -maxdepth 1 -type f -name '*.jsonl' -exec cp {} "$dst_dir"/ \; + fi +} diff --git a/scripts/tests/p2p-stress-matrix.sh b/scripts/tests/p2p-stress-matrix.sh new file mode 100755 index 000000000..be20eb34b --- /dev/null +++ b/scripts/tests/p2p-stress-matrix.sh @@ -0,0 +1,201 @@ +#!/usr/bin/env bash +set -euo pipefail + +source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/p2p-stress-lib.sh" +p2p_stress_init "${BASH_SOURCE[0]}" + +run_case() { + local name="$1" + local size_spec="$2" + local read_ratio="$3" + local hotset_ratio="$4" + local file_count="$5" + local duration_secs="$6" + local case_root="$RUN_ROOT/$name" + local manifest_dir="$case_root/manifest" + local remote_root="/p2p-stress/$name" + mkdir -p "$case_root" "$manifest_dir" + + log "case=$name sizes=$size_spec read_ratio=$read_ratio hotset_ratio=$hotset_ratio" + curl -sS http://127.0.0.1:9000/metrics > "$case_root/metrics.before.prom" + + write_client_conf "$case_root/seed.toml" seed "" "/ip4/0.0.0.0/tcp/31001" "seed.log" + + docker run -d --name "stress-${name}-seed" \ + --network "$NETWORK_NAME" --ip "$SEED_IP" \ + --cap-add NET_ADMIN \ + --add-host host.docker.internal:host-gateway \ + -v "$DRIVER_BIN:/usr/local/bin/p2p_proof_driver:ro" \ + -v "$case_root:/evidence" \ + "$RUNTIME_IMAGE" \ + bash -lc ' +set -euo pipefail +/usr/local/bin/p2p_proof_driver \ + --role mixed \ + --conf /evidence/seed.toml \ + --master-addrs host.docker.internal:'"$FORWARD_PORT"' \ + --bootstrap-host '"$SEED_IP"' \ + --remote-path '"$remote_root"' \ + --manifest-dir /evidence/manifest \ + --client-id seed \ + --file-size-spec '"$size_spec"' \ + --file-count '"$file_count"' \ + --duration-secs '"$duration_secs"' \ + --read-ratio 0 \ + --hotset-ratio '"$hotset_ratio"' \ + --op-interval-ms '"$OP_INTERVAL_MS"' \ + --warm-written \ + --bootstrap-file /evidence/bootstrap.txt \ + --peer-file /evidence/peer.txt \ + --bootstrap-wait-secs '"$BOOTSTRAP_WAIT_SECS"' \ + --output-file /evidence/seed-summary.json \ + > /evidence/seed-driver.log 2>&1 +' >/dev/null + + wait_for_file "$case_root/bootstrap.txt" "seed bootstrap for case $name" + local bootstrap_addr + bootstrap_addr="$(cat "$case_root/bootstrap.txt")" + + for client_idx in $(seq 1 "$CLIENTS"); do + local client_name="client-${client_idx}" + write_client_conf "$case_root/${client_name}.toml" "$client_name" "$bootstrap_addr" "/ip4/0.0.0.0/tcp/0" "${client_name}.log" + docker run -d --name "stress-${name}-${client_idx}" \ + --network "$NETWORK_NAME" \ + --cap-add NET_ADMIN \ + --add-host host.docker.internal:host-gateway \ + -v "$DRIVER_BIN:/usr/local/bin/p2p_proof_driver:ro" \ + -v "$case_root:/evidence" \ + "$RUNTIME_IMAGE" \ + bash -lc ' +set -euo pipefail +/usr/local/bin/p2p_proof_driver \ + --role mixed \ + --conf /evidence/'"${client_name}.toml"' \ + --master-addrs host.docker.internal:'"$FORWARD_PORT"' \ + --remote-path '"$remote_root"' \ + --manifest-dir /evidence/manifest \ + --client-id '"$client_name"' \ + --file-size-spec '"$size_spec"' \ + --file-count '"$file_count"' \ + --duration-secs '"$duration_secs"' \ + --read-ratio '"$read_ratio"' \ + --hotset-ratio '"$hotset_ratio"' \ + --op-interval-ms '"$OP_INTERVAL_MS"' \ + --bootstrap-wait-secs '"$BOOTSTRAP_WAIT_SECS"' \ + --output-file /evidence/'"${client_name}-summary.json"' \ + > /evidence/'"${client_name}-driver.log"' 2>&1 +' >/dev/null + done + + local rc=0 + docker wait "stress-${name}-seed" >/dev/null || rc=1 + for client_idx in $(seq 1 "$CLIENTS"); do + docker wait "stress-${name}-${client_idx}" >/dev/null || rc=1 + done + + curl -sS http://127.0.0.1:9000/metrics > "$case_root/metrics.after.prom" + python3 - <<'PY' "$case_root" "$name" "$rc" "$READ_CHUNK_SIZE" "$READ_CHUNK_NUM" +import json +import sys +from pathlib import Path + +case_root = Path(sys.argv[1]) +name = sys.argv[2] +rc = int(sys.argv[3]) +read_chunk_size = sys.argv[4] +read_chunk_num = int(sys.argv[5]) +summaries = [] +for path in sorted(case_root.glob('*-summary.json')): + summaries.append(json.loads(path.read_text())) +report = { + 'case': name, + 'exit_code': rc, + 'summary_count': len(summaries), + 'read_chunk_size': read_chunk_size, + 'read_chunk_num': read_chunk_num, + 'clients': summaries, + 'total_ops': sum(item.get('total_ops', 0) for item in summaries), + 'read_ok': sum(item.get('read_ok', 0) for item in summaries), + 'write_ok': sum(item.get('write_ok', 0) for item in summaries), + 'read_errors': sum(item.get('read_errors', 0) for item in summaries), + 'write_errors': sum(item.get('write_errors', 0) for item in summaries), + 'sha_mismatches': sum(item.get('sha_mismatches', 0) for item in summaries), + 'p2p_recv_delta': sum(item.get('p2p_recv_delta', 0) for item in summaries), + 'p2p_sent_delta': sum(item.get('p2p_sent_delta', 0) for item in summaries), + 'any_p2p_observed': any(item.get('p2p_recv_delta', 0) > 0 or item.get('p2p_sent_delta', 0) > 0 for item in summaries), +} +(case_root / 'scenario-summary.json').write_text(json.dumps(report, indent=2, ensure_ascii=False) + '\n') +print(case_root / 'scenario-summary.json') +PY + + local before_sent after_sent before_recv after_recv any_p2p_observed + before_sent="$(sum_metric "$case_root/metrics.before.prom" client_p2p_bytes_sent)" + after_sent="$(sum_metric "$case_root/metrics.after.prom" client_p2p_bytes_sent)" + before_recv="$(sum_metric "$case_root/metrics.before.prom" client_p2p_bytes_recv)" + after_recv="$(sum_metric "$case_root/metrics.after.prom" client_p2p_bytes_recv)" + any_p2p_observed="$(python3 - <<'PY' "$case_root/scenario-summary.json" +import json +import sys +with open(sys.argv[1], 'r', encoding='utf-8') as f: + report = json.load(f) +print(1 if report.get('any_p2p_observed') else 0) +PY +)" + printf '%s,%s,%s,%s,%s,%s\n' "$name" "$((after_sent-before_sent))" "$((after_recv-before_recv))" "$CLIENTS" "$rc" "$any_p2p_observed" >> "$RUN_ROOT/report.csv" + + if [[ "$rc" -ne 0 ]]; then + return 1 + fi + python3 - <<'PY' "$case_root/scenario-summary.json" +import json +import sys +with open(sys.argv[1], 'r', encoding='utf-8') as f: + report = json.load(f) +raise SystemExit(0 if report['sha_mismatches'] == 0 and report['read_errors'] == 0 and report['write_errors'] == 0 and report['summary_count'] > 0 else 1) +PY +} + +main() { + trap cleanup EXIT + prepare_env + printf 'case,p2p_bytes_sent_delta,p2p_bytes_recv_delta,clients,exit_code,any_p2p_observed\n' > "$RUN_ROOT/report.csv" + + run_case small_fixed_cost "64KB,256KB,1MB" 90 80 12 "$DURATION_SECS" + run_case mixed_balanced "256KB,1MB,16MB" 70 70 16 "$DURATION_SECS" + run_case large_hotset "16MB,64MB,128MB" 80 90 8 "$DURATION_SECS" + + python3 - <<'PY' "$RUN_ROOT" "$READ_CHUNK_SIZE" "$READ_CHUNK_NUM" +import json +import sys +from pathlib import Path + +run_root = Path(sys.argv[1]) +read_chunk_size = sys.argv[2] +read_chunk_num = int(sys.argv[3]) +reports = [] +for path in sorted(run_root.glob('*/scenario-summary.json')): + reports.append(json.loads(path.read_text())) +final = { + 'run_root': str(run_root), + 'case_count': len(reports), + 'all_passed': all(item.get('exit_code') == 0 and item.get('sha_mismatches') == 0 and item.get('read_errors') == 0 and item.get('write_errors') == 0 for item in reports), + 'p2p_observed_cases': sum(1 for item in reports if item.get('any_p2p_observed')), + 'any_p2p_observed': any(item.get('any_p2p_observed') for item in reports), + 'read_chunk_size': read_chunk_size, + 'read_chunk_num': read_chunk_num, + 'cases': reports, +} +(run_root / 'final-report.json').write_text(json.dumps(final, indent=2, ensure_ascii=False) + '\n') +print(run_root / 'final-report.json') +PY + cat "$RUN_ROOT/final-report.json" + python3 - <<'PY' "$RUN_ROOT/final-report.json" +import json +import sys +with open(sys.argv[1], 'r', encoding='utf-8') as f: + report = json.load(f) +raise SystemExit(0 if report.get('all_passed') and report.get('any_p2p_observed') else 1) +PY +} + +main "$@" diff --git a/scripts/tests/p2p-training-stability-matrix.sh b/scripts/tests/p2p-training-stability-matrix.sh new file mode 100644 index 000000000..1c7da2437 --- /dev/null +++ b/scripts/tests/p2p-training-stability-matrix.sh @@ -0,0 +1,473 @@ +#!/usr/bin/env bash +set -euo pipefail + +source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/p2p-stress-lib.sh" +p2p_stress_init "${BASH_SOURCE[0]}" + +TRAIN_DURATION_SECS="${P2P_TRAIN_DURATION_SECS:-$DURATION_SECS}" +TRAINERS="${P2P_TRAIN_TRAINERS:-4}" +CHECKPOINTERS="${P2P_TRAIN_CHECKPOINTERS:-1}" +EVALUATORS="${P2P_TRAIN_EVALUATORS:-1}" +DATASET_FILE_COUNT="${P2P_TRAIN_DATASET_FILE_COUNT:-24}" +CHECKPOINT_FILE_COUNT="${P2P_TRAIN_CHECKPOINT_FILE_COUNT:-4}" +REQUIRE_P2P="${P2P_TRAIN_REQUIRE_P2P:-false}" +DATASET_WARMUP_ENTRIES="${P2P_TRAIN_DATASET_WARMUP_ENTRIES:-4}" +CHECKPOINT_WARMUP_ENTRIES="${P2P_TRAIN_CHECKPOINT_WARMUP_ENTRIES:-1}" +DATASET_SETTLE_SECS="${P2P_TRAIN_DATASET_SETTLE_SECS:-3}" +CHECKPOINT_SETTLE_SECS="${P2P_TRAIN_CHECKPOINT_SETTLE_SECS:-2}" +SEED_OP_INTERVAL_MS="${P2P_TRAIN_SEED_OP_INTERVAL_MS:-25}" +TRAINER_OP_INTERVAL_MS="${P2P_TRAIN_TRAINER_OP_INTERVAL_MS:-5}" +CHECKPOINT_OP_INTERVAL_MS="${P2P_TRAIN_CHECKPOINT_OP_INTERVAL_MS:-200}" +EVALUATOR_OP_INTERVAL_MS="${P2P_TRAIN_EVALUATOR_OP_INTERVAL_MS:-20}" +PROVIDER_HOLD_BUFFER_SECS="${P2P_TRAIN_PROVIDER_HOLD_BUFFER_SECS:-15}" +SEED_POST_RUN_HOLD_SECS="${P2P_TRAIN_SEED_POST_RUN_HOLD_SECS:-$((TRAIN_DURATION_SECS * 2 + DATASET_SETTLE_SECS + CHECKPOINT_SETTLE_SECS + PROVIDER_HOLD_BUFFER_SECS))}" +CHECKPOINT_POST_RUN_HOLD_SECS="${P2P_TRAIN_CHECKPOINT_POST_RUN_HOLD_SECS:-$((TRAIN_DURATION_SECS + CHECKPOINT_SETTLE_SECS + PROVIDER_HOLD_BUFFER_SECS))}" + +run_mixed_container() { + local container_name="$1" + local case_root="$2" + local conf_name="$3" + local remote_root="$4" + local client_id="$5" + local size_spec="$6" + local file_count="$7" + local duration_secs="$8" + local read_ratio="$9" + local hotset_ratio="${10}" + local op_interval_ms="${11}" + local output_name="${12}" + local driver_log="${13}" + local read_manifest_rel="${14:-}" + local write_manifest_rel="${15:-}" + local warm_written="${16:-false}" + local bootstrap_host="${17:-}" + local bootstrap_file="${18:-}" + local peer_file="${19:-}" + local post_run_hold_secs="${20:-0}" + + local manifest_flags="" + local warm_flag="" + local bootstrap_host_flag="" + local bootstrap_file_flag="" + local peer_file_flag="" + local post_run_hold_flag="" + local docker_ip_args=() + [[ -n "$read_manifest_rel" ]] && manifest_flags+=" --read-manifest-dir /evidence/${read_manifest_rel}" + [[ -n "$write_manifest_rel" ]] && manifest_flags+=" --write-manifest-dir /evidence/${write_manifest_rel}" + [[ "$warm_written" == "true" ]] && warm_flag=" --warm-written" + [[ -n "$bootstrap_host" ]] && bootstrap_host_flag=" --bootstrap-host ${bootstrap_host}" + [[ -n "$bootstrap_file" ]] && bootstrap_file_flag=" --bootstrap-file /evidence/${bootstrap_file}" + [[ -n "$peer_file" ]] && peer_file_flag=" --peer-file /evidence/${peer_file}" + [[ "$post_run_hold_secs" -gt 0 ]] && post_run_hold_flag=" --post-run-hold-secs ${post_run_hold_secs}" + [[ -n "$bootstrap_host" ]] && docker_ip_args=(--ip "$bootstrap_host") + + docker run -d --name "$container_name" \ + --network "$NETWORK_NAME" \ + "${docker_ip_args[@]}" \ + --cap-add NET_ADMIN \ + --add-host host.docker.internal:host-gateway \ + -v "$DRIVER_BIN:/usr/local/bin/p2p_proof_driver:ro" \ + -v "$case_root:/evidence" \ + "$RUNTIME_IMAGE" \ + bash -lc ' +set -euo pipefail +/usr/local/bin/p2p_proof_driver \ + --role mixed \ + --conf /evidence/'"$conf_name"' \ + --master-addrs host.docker.internal:'"$FORWARD_PORT"' \ + --remote-path '"$remote_root"' \ + --client-id '"$client_id"' \ + --file-size-spec '"$size_spec"' \ + --file-count '"$file_count"' \ + --duration-secs '"$duration_secs"' \ + --read-ratio '"$read_ratio"' \ + --hotset-ratio '"$hotset_ratio"' \ + --op-interval-ms '"$op_interval_ms"' \ + --bootstrap-wait-secs '"$BOOTSTRAP_WAIT_SECS"' \ + --output-file /evidence/'"$output_name"''"$manifest_flags"''"$warm_flag"''"$bootstrap_host_flag"''"$bootstrap_file_flag"''"$peer_file_flag"''"$post_run_hold_flag"' \ + > /evidence/'"$driver_log"' 2>&1 +' >/dev/null +} + +summarize_training_case() { + local case_root="$1" + local case_name="$2" + local rc="$3" + local expected_clients="$4" + python3 - <<'PY' "$case_root" "$case_name" "$rc" "$expected_clients" "$READ_CHUNK_SIZE" "$READ_CHUNK_NUM" "$REQUIRE_P2P" +import json +import sys +from pathlib import Path + +case_root = Path(sys.argv[1]) +case_name = sys.argv[2] +rc = int(sys.argv[3]) +expected_clients = int(sys.argv[4]) +read_chunk_size = sys.argv[5] +read_chunk_num = int(sys.argv[6]) +require_p2p = sys.argv[7].lower() == 'true' + +summaries = [] +for path in sorted(case_root.glob('*-summary.json')): + data = json.loads(path.read_text()) + client_id = data.get('client_id', '') + if client_id == 'seed': + group = 'seed' + elif client_id.startswith('trainer-'): + group = 'trainer' + elif client_id.startswith('checkpoint-'): + group = 'checkpoint' + elif client_id.startswith('eval-'): + group = 'evaluator' + else: + group = 'unknown' + data['group'] = group + summaries.append(data) + +log_counts = { + 'response_retry': 0, + 'data_plane_failure': 0, + 'miss': 0, + 'response_ok': 0, + 'response_redirect': 0, + 'dht_query': 0, + 'dht_result': 0, + 'fatal_lines': 0, +} +for path in sorted(case_root.glob('*.log*')): + if path.name.endswith('driver.log') or '.driver.' in path.name: + continue + text = path.read_text(errors='ignore') + log_counts['response_retry'] += text.count('event="response_retry"') + log_counts['data_plane_failure'] += text.count('event="data_plane_failure"') + log_counts['miss'] += text.count('event="miss"') + log_counts['response_ok'] += text.count('event="response_ok"') + log_counts['response_redirect'] += text.count('event="response_redirect"') + log_counts['dht_query'] += text.count('event="dht_query"') + log_counts['dht_result'] += text.count('event="dht_result"') + for line in text.splitlines(): + lower = line.lower() + if 'panicked at' in lower or ' fatal ' in lower or ('thread \'' in lower and 'panicked' in lower): + log_counts['fatal_lines'] += 1 + +trainers = [item for item in summaries if item['group'] == 'trainer'] +checkpoints = [item for item in summaries if item['group'] == 'checkpoint'] +evaluators = [item for item in summaries if item['group'] == 'evaluator'] +read_clients = [item for item in summaries if item['group'] in {'trainer', 'evaluator'}] +read_p99 = [item.get('read_latency', {}).get('p99_ms', 0.0) for item in read_clients if item.get('read_latency', {}).get('count', 0) > 0] +report = { + 'case': case_name, + 'exit_code': rc, + 'expected_clients': expected_clients, + 'summary_count': len(summaries), + 'p2p_required': require_p2p, + 'read_chunk_size': read_chunk_size, + 'read_chunk_num': read_chunk_num, + 'clients': summaries, + 'trainer_count': len(trainers), + 'checkpoint_count': len(checkpoints), + 'evaluator_count': len(evaluators), + 'total_ops': sum(item.get('total_ops', 0) for item in summaries), + 'read_ok': sum(item.get('read_ok', 0) for item in summaries), + 'write_ok': sum(item.get('write_ok', 0) for item in summaries), + 'read_errors': sum(item.get('read_errors', 0) for item in summaries), + 'write_errors': sum(item.get('write_errors', 0) for item in summaries), + 'sha_mismatches': sum(item.get('sha_mismatches', 0) for item in summaries), + 'p2p_recv_delta': sum(item.get('p2p_recv_delta', 0) for item in summaries), + 'p2p_sent_delta': sum(item.get('p2p_sent_delta', 0) for item in summaries), + 'any_p2p_observed': any(item.get('p2p_recv_delta', 0) > 0 or item.get('p2p_sent_delta', 0) > 0 for item in summaries), + 'max_reader_p99_ms': max(read_p99) if read_p99 else 0.0, + 'log_counts': log_counts, +} +report['all_stable'] = ( + report['exit_code'] == 0 + and report['summary_count'] == report['expected_clients'] + and report['sha_mismatches'] == 0 + and report['read_errors'] == 0 + and report['write_errors'] == 0 + and report['log_counts']['fatal_lines'] == 0 + and report['log_counts']['data_plane_failure'] == 0 +) +report['has_p2p_gap'] = report['all_stable'] and not report['any_p2p_observed'] +report['status'] = ( + 'stable_accelerated' if report['all_stable'] and report['any_p2p_observed'] + else 'stable_worker_fallback' if report['all_stable'] + else 'unstable' +) +(case_root / 'scenario-summary.json').write_text(json.dumps(report, indent=2, ensure_ascii=False) + '\n') +print(case_root / 'scenario-summary.json') +PY +} + +run_training_case() { + local name="$1" + local dataset_sizes="$2" + local dataset_hotset="$3" + local checkpoint_sizes="$4" + local checkpoint_hotset="$5" + local evaluator_hotset="$6" + local evaluator_manifest_kind="${7:-dataset}" + local case_root="$RUN_ROOT/$name" + local dataset_manifest_live_rel="manifest-dataset-live" + local dataset_manifest_read_rel="manifest-dataset-read" + local checkpoint_manifest_live_rel="manifest-checkpoint-live" + local checkpoint_manifest_read_rel="manifest-checkpoint-read" + local remote_root="/p2p-training/$name" + mkdir -p \ + "$case_root/$dataset_manifest_live_rel" \ + "$case_root/$dataset_manifest_read_rel" \ + "$case_root/$checkpoint_manifest_live_rel" \ + "$case_root/$checkpoint_manifest_read_rel" + + log "case=$name dataset_sizes=$dataset_sizes checkpoint_sizes=$checkpoint_sizes" + curl -sS http://127.0.0.1:9000/metrics > "$case_root/metrics.before.prom" + + write_client_conf "$case_root/seed.toml" seed "" "/ip4/0.0.0.0/tcp/31001" "seed.log" + run_mixed_container \ + "stress-${name}-seed" \ + "$case_root" \ + "seed.toml" \ + "$remote_root" \ + seed \ + "$dataset_sizes" \ + "$DATASET_FILE_COUNT" \ + "$TRAIN_DURATION_SECS" \ + 0 \ + "$dataset_hotset" \ + "$SEED_OP_INTERVAL_MS" \ + "seed-summary.json" \ + "seed-driver.log" \ + "$dataset_manifest_live_rel" \ + "$dataset_manifest_live_rel" \ + true \ + "$SEED_IP" \ + "bootstrap.txt" \ + "peer.txt" \ + "$SEED_POST_RUN_HOLD_SECS" + + wait_for_file "$case_root/bootstrap.txt" "seed bootstrap for case $name" + wait_for_manifest_entries \ + "$case_root/$dataset_manifest_live_rel" \ + "$DATASET_WARMUP_ENTRIES" \ + "dataset manifest for case $name" + sleep "$DATASET_SETTLE_SECS" + snapshot_manifest_dir \ + "$case_root/$dataset_manifest_live_rel" \ + "$case_root/$dataset_manifest_read_rel" + local bootstrap_addr + bootstrap_addr="$(cat "$case_root/bootstrap.txt")" + + if [[ "$evaluator_manifest_kind" == "checkpoint" ]]; then + for idx in $(seq 1 "$CHECKPOINTERS"); do + local client_name="checkpoint-$(printf '%02d' "$idx")" + write_client_conf "$case_root/${client_name}.toml" "$client_name" "$bootstrap_addr" "/ip4/0.0.0.0/tcp/0" "${client_name}.log" + run_mixed_container \ + "stress-${name}-${client_name}" \ + "$case_root" \ + "${client_name}.toml" \ + "$remote_root" \ + "$client_name" \ + "$checkpoint_sizes" \ + "$CHECKPOINT_FILE_COUNT" \ + "$TRAIN_DURATION_SECS" \ + 0 \ + "$checkpoint_hotset" \ + "$CHECKPOINT_OP_INTERVAL_MS" \ + "${client_name}-summary.json" \ + "${client_name}-driver.log" \ + "$dataset_manifest_read_rel" \ + "$checkpoint_manifest_live_rel" \ + true \ + "" \ + "" \ + "" \ + "$CHECKPOINT_POST_RUN_HOLD_SECS" + done + wait_for_manifest_entries \ + "$case_root/$checkpoint_manifest_live_rel" \ + "$CHECKPOINT_WARMUP_ENTRIES" \ + "checkpoint manifest for case $name" + sleep "$CHECKPOINT_SETTLE_SECS" + snapshot_manifest_dir \ + "$case_root/$checkpoint_manifest_live_rel" \ + "$case_root/$checkpoint_manifest_read_rel" + fi + + local idx rc expected_clients + expected_clients=$((1 + TRAINERS + CHECKPOINTERS + EVALUATORS)) + for idx in $(seq 1 "$TRAINERS"); do + local client_name="trainer-$(printf '%02d' "$idx")" + write_client_conf "$case_root/${client_name}.toml" "$client_name" "$bootstrap_addr" "/ip4/0.0.0.0/tcp/0" "${client_name}.log" + run_mixed_container \ + "stress-${name}-${client_name}" \ + "$case_root" \ + "${client_name}.toml" \ + "$remote_root" \ + "$client_name" \ + "$dataset_sizes" \ + "$DATASET_FILE_COUNT" \ + "$TRAIN_DURATION_SECS" \ + 100 \ + "$dataset_hotset" \ + "$TRAINER_OP_INTERVAL_MS" \ + "${client_name}-summary.json" \ + "${client_name}-driver.log" \ + "$dataset_manifest_read_rel" \ + "$dataset_manifest_read_rel" + done + + if [[ "$evaluator_manifest_kind" != "checkpoint" ]]; then + for idx in $(seq 1 "$CHECKPOINTERS"); do + local client_name="checkpoint-$(printf '%02d' "$idx")" + write_client_conf "$case_root/${client_name}.toml" "$client_name" "$bootstrap_addr" "/ip4/0.0.0.0/tcp/0" "${client_name}.log" + run_mixed_container \ + "stress-${name}-${client_name}" \ + "$case_root" \ + "${client_name}.toml" \ + "$remote_root" \ + "$client_name" \ + "$checkpoint_sizes" \ + "$CHECKPOINT_FILE_COUNT" \ + "$TRAIN_DURATION_SECS" \ + 0 \ + "$checkpoint_hotset" \ + "$CHECKPOINT_OP_INTERVAL_MS" \ + "${client_name}-summary.json" \ + "${client_name}-driver.log" \ + "$dataset_manifest_read_rel" \ + "$checkpoint_manifest_live_rel" \ + true \ + "" \ + "" \ + "" \ + "$CHECKPOINT_POST_RUN_HOLD_SECS" + done + fi + + for idx in $(seq 1 "$EVALUATORS"); do + local client_name="eval-$(printf '%02d' "$idx")" + local eval_read_manifest_rel="$dataset_manifest_read_rel" + [[ "$evaluator_manifest_kind" == "checkpoint" ]] && eval_read_manifest_rel="$checkpoint_manifest_read_rel" + write_client_conf "$case_root/${client_name}.toml" "$client_name" "$bootstrap_addr" "/ip4/0.0.0.0/tcp/0" "${client_name}.log" + run_mixed_container \ + "stress-${name}-${client_name}" \ + "$case_root" \ + "${client_name}.toml" \ + "$remote_root" \ + "$client_name" \ + "$dataset_sizes" \ + "$DATASET_FILE_COUNT" \ + "$TRAIN_DURATION_SECS" \ + 100 \ + "$evaluator_hotset" \ + "$EVALUATOR_OP_INTERVAL_MS" \ + "${client_name}-summary.json" \ + "${client_name}-driver.log" \ + "$eval_read_manifest_rel" \ + "$eval_read_manifest_rel" + done + + rc=0 + docker wait "stress-${name}-seed" >/dev/null || rc=1 + for idx in $(seq 1 "$TRAINERS"); do + docker wait "stress-${name}-trainer-$(printf '%02d' "$idx")" >/dev/null || rc=1 + done + for idx in $(seq 1 "$CHECKPOINTERS"); do + docker wait "stress-${name}-checkpoint-$(printf '%02d' "$idx")" >/dev/null || rc=1 + done + for idx in $(seq 1 "$EVALUATORS"); do + docker wait "stress-${name}-eval-$(printf '%02d' "$idx")" >/dev/null || rc=1 + done + + curl -sS http://127.0.0.1:9000/metrics > "$case_root/metrics.after.prom" + summarize_training_case "$case_root" "$name" "$rc" "$expected_clients" + + local before_sent after_sent before_recv after_recv any_p2p_observed max_p99 all_stable data_plane_failures response_retry + before_sent="$(sum_metric "$case_root/metrics.before.prom" client_p2p_bytes_sent)" + after_sent="$(sum_metric "$case_root/metrics.after.prom" client_p2p_bytes_sent)" + before_recv="$(sum_metric "$case_root/metrics.before.prom" client_p2p_bytes_recv)" + after_recv="$(sum_metric "$case_root/metrics.after.prom" client_p2p_bytes_recv)" + read -r any_p2p_observed max_p99 all_stable data_plane_failures response_retry <<<"$(python3 - <<'PY' "$case_root/scenario-summary.json" +import json +import sys +with open(sys.argv[1], 'r', encoding='utf-8') as f: + report = json.load(f) +print( + int(report.get('any_p2p_observed', False)), + report.get('max_reader_p99_ms', 0.0), + int(report.get('all_stable', False)), + report.get('log_counts', {}).get('data_plane_failure', 0), + report.get('log_counts', {}).get('response_retry', 0), +) +PY +)" + printf '%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s\n' \ + "$name" \ + "$((after_sent-before_sent))" \ + "$((after_recv-before_recv))" \ + "$TRAINERS" \ + "$CHECKPOINTERS" \ + "$EVALUATORS" \ + "$REQUIRE_P2P" \ + "$any_p2p_observed" \ + "$all_stable" \ + "$data_plane_failures" \ + "$response_retry" >> "$RUN_ROOT/report.csv" + + python3 - <<'PY' "$case_root/scenario-summary.json" +import json +import sys +with open(sys.argv[1], 'r', encoding='utf-8') as f: + report = json.load(f) +raise SystemExit(0 if report.get('all_stable') else 1) +PY +} + +main() { + trap cleanup EXIT + prepare_env + printf 'case,p2p_bytes_sent_delta,p2p_bytes_recv_delta,trainers,checkpointers,evaluators,p2p_required,any_p2p_observed,all_stable,data_plane_failures,response_retry\n' > "$RUN_ROOT/report.csv" + + run_training_case epoch_hotset "16MB,64MB" 100 "64MB" 100 100 dataset + run_training_case checkpoint_overlap "16MB,64MB" 100 "64MB,256MB" 100 100 checkpoint + run_training_case shuffle_eval_mix "4MB,16MB,64MB" 100 "64MB" 100 100 dataset + + python3 - <<'PY' "$RUN_ROOT" "$READ_CHUNK_SIZE" "$READ_CHUNK_NUM" "$REQUIRE_P2P" +import json +import sys +from pathlib import Path + +run_root = Path(sys.argv[1]) +read_chunk_size = sys.argv[2] +read_chunk_num = int(sys.argv[3]) +require_p2p = sys.argv[4].lower() == 'true' +reports = [] +for path in sorted(run_root.glob('*/scenario-summary.json')): + reports.append(json.loads(path.read_text())) +final = { + 'run_root': str(run_root), + 'case_count': len(reports), + 'all_stable': all(item.get('all_stable') for item in reports), + 'p2p_required': require_p2p, + 'p2p_observed_cases': sum(1 for item in reports if item.get('any_p2p_observed')), + 'any_p2p_observed': any(item.get('any_p2p_observed') for item in reports), + 'coverage_gap_cases': sum(1 for item in reports if item.get('has_p2p_gap')), + 'read_chunk_size': read_chunk_size, + 'read_chunk_num': read_chunk_num, + 'cases': reports, +} +(run_root / 'final-report.json').write_text(json.dumps(final, indent=2, ensure_ascii=False) + '\n') +print(run_root / 'final-report.json') +PY + cat "$RUN_ROOT/final-report.json" + python3 - <<'PY' "$RUN_ROOT/final-report.json" "$REQUIRE_P2P" +import json +import sys +with open(sys.argv[1], 'r', encoding='utf-8') as f: + report = json.load(f) +require_p2p = sys.argv[2].lower() == 'true' +raise SystemExit(0 if report.get('all_stable') and (report.get('any_p2p_observed') or not require_p2p) else 1) +PY +} + +main "$@" diff --git a/scripts/tests/test-master-forwarder.sh b/scripts/tests/test-master-forwarder.sh new file mode 100644 index 000000000..e3406be2a --- /dev/null +++ b/scripts/tests/test-master-forwarder.sh @@ -0,0 +1,86 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/../.." && pwd)" +TMP_DIR="$(mktemp -d)" +cleanup() { + set +e + [[ -n "${FWD_PID:-}" ]] && kill "$FWD_PID" >/dev/null 2>&1 || true + [[ -n "${BACKEND_PID:-}" ]] && kill "$BACKEND_PID" >/dev/null 2>&1 || true + wait "${FWD_PID:-}" >/dev/null 2>&1 || true + wait "${BACKEND_PID:-}" >/dev/null 2>&1 || true + rm -rf "$TMP_DIR" +} +trap cleanup EXIT + +read -r FORWARD_PORT BACKEND_PORT < <(python3 - <<'PY' +import socket +ports = [] +for _ in range(2): + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + ports.append(s.getsockname()[1]) + s.close() +print(*ports) +PY +) + +python3 -u - <<'PY' "$BACKEND_PORT" >"$TMP_DIR/backend.log" 2>&1 & +import socket +import sys +import time + +port = int(sys.argv[1]) +ls = socket.socket(socket.AF_INET, socket.SOCK_STREAM) +ls.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) +ls.bind(("127.0.0.1", port)) +ls.listen(1) +while True: + conn, _ = ls.accept() + data = conn.recv(4) + if data != b"ping": + conn.close() + continue + time.sleep(6.2) + conn.sendall(b"pong") + conn.close() + break +ls.close() +PY +BACKEND_PID=$! + +python3 -u "$ROOT/scripts/tests/master-forwarder.py" \ + --listen-host 127.0.0.1 \ + --listen-port "$FORWARD_PORT" \ + --target-host 127.0.0.1 \ + --target-port "$BACKEND_PORT" \ + >"$TMP_DIR/forwarder.log" 2>&1 & +FWD_PID=$! + +python3 - <<'PY' "$FORWARD_PORT" +import socket +import sys +import time + +port = int(sys.argv[1]) +deadline = time.time() + 5 +while time.time() < deadline: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + if sock.connect_ex(("127.0.0.1", port)) == 0: + raise SystemExit(0) + time.sleep(0.1) +raise SystemExit("forwarder did not start") +PY + +python3 - <<'PY' "$FORWARD_PORT" +import socket +import sys + +port = int(sys.argv[1]) +with socket.create_connection(("127.0.0.1", port), timeout=2) as sock: + sock.settimeout(8) + sock.sendall(b"ping") + data = sock.recv(4) +if data != b"pong": + raise SystemExit(f"unexpected response: {data!r}") +PY