From 7ef1d1d31501c2fca4fe7dda851f725e3fb8a16f Mon Sep 17 00:00:00 2001 From: Mud <44410798+MudDev@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:05:26 -0700 Subject: [PATCH] feat(Bigfile): piecefield exchange + piece-aware peer selection Adds the Bigfile piecefield protocol so downloads only ask peers that actually hold a piece. - epix-xite: a Piecefield type wire-compatible with EpixNet's BigfilePiecefield - one flag per piece, packed as a little-endian u16 array of alternating run-lengths (present, absent, ...). pack/unpack are pinned to the reference run-length format, with a size guard against malicious runs. - AppState::our_piecefields computes, per served big file (keyed by sha512), which pieces we hold by verifying on-disk bytes against the piecemap. - FileService answers getPiecefields {site} with piecefields_packed, and acknowledges setPiecefields. - Connection::get_piecefields fetches a peer's piecefields (raw packed bytes, so the protocol layer stays free of epix-xite). - bigfile_fetch_range: for a multi-piece fetch, it first asks each peer which pieces it holds and skips peers that lack a given piece. Tests: piecefield pack/unpack + run-length vectors + oversize guard; and an end-to-end test where a server holding pieces 0 and 2 (piece 1 a hole) reports the exact piecefield to a client over TCP. Desktop and mobile builds green; full suite green. (Connection reuse across pieces remains a follow-up optimization.) --- crates/epix-protocol/src/connection.rs | 20 +++ crates/epix-ui/src/fileserve.rs | 17 +++ crates/epix-ui/src/state.rs | 76 +++++++++++ crates/epix-ui/tests/bigfile.rs | 65 +++++++++- crates/epix-xite/src/lib.rs | 2 + crates/epix-xite/src/piecefield.rs | 173 +++++++++++++++++++++++++ 6 files changed, 351 insertions(+), 2 deletions(-) create mode 100644 crates/epix-xite/src/piecefield.rs diff --git a/crates/epix-protocol/src/connection.rs b/crates/epix-protocol/src/connection.rs index 70432c1..ca10434 100644 --- a/crates/epix-protocol/src/connection.rs +++ b/crates/epix-protocol/src/connection.rs @@ -167,6 +167,26 @@ impl Connection { self.request("update", params).await } + /// Ask which pieces of each big file the peer holds (`getPiecefields`). + /// Returns `sha512 -> packed piecefield bytes`; the caller unpacks with + /// `epix_xite::Piecefield` (kept out of the protocol layer to avoid a cycle). + pub async fn get_piecefields( + &mut self, + xite: &str, + ) -> Result>> { + let params = vmap(vec![("site", Value::from(xite))]); + let resp = self.request("getPiecefields", params).await?; + let mut out = std::collections::HashMap::new(); + if let Some(Value::Map(entries)) = vget(&resp, "piecefields_packed") { + for (k, v) in entries { + if let (Some(sha), Value::Binary(bytes)) = (k.as_str(), v) { + out.insert(sha.to_string(), bytes.clone()); + } + } + } + Ok(out) + } + /// Download a whole file, following `location`/`size` across chunks. pub async fn get_file(&mut self, xite: &str, inner_path: &str) -> Result> { let mut out = Vec::new(); diff --git a/crates/epix-ui/src/fileserve.rs b/crates/epix-ui/src/fileserve.rs index 8e3c613..3fef6ef 100644 --- a/crates/epix-ui/src/fileserve.rs +++ b/crates/epix-ui/src/fileserve.rs @@ -28,6 +28,19 @@ impl FileService { Self { state } } + /// `getPiecefields {site}` - report which pieces of each big file we hold, + /// keyed by the file's sha512, so a downloader only asks us for pieces we + /// actually have. + async fn get_piecefields(&self, params: &Value) -> Value { + let site = vget_str(params, "site").unwrap_or_default(); + let packed = self.state.our_piecefields(&site).await; + let map: Vec<(Value, Value)> = packed + .into_iter() + .map(|(sha512, bytes)| (Value::from(sha512), Value::Binary(bytes))) + .collect(); + vmap(vec![("piecefields_packed", Value::Map(map))]) + } + async fn get_file(&self, params: &Value) -> Value { let site = vget_str(params, "site").unwrap_or_default(); let inner_path = vget_str(params, "inner_path").unwrap_or_default(); @@ -56,6 +69,10 @@ impl RequestHandler for FileService { match cmd { "ping" => vmap(vec![("body", Value::Binary(b"Pong!".to_vec()))]), "getFile" | "streamFile" => self.get_file(params).await, + "getPiecefields" => self.get_piecefields(params).await, + // A peer pushing us its piecefields: acknowledge (our downloader + // re-queries piecefields when it needs them, so we don't retain). + "setPiecefields" => vmap(vec![("ok", Value::from("Updated"))]), // Unknown/unsupported request: empty body (the server still wraps it // as a response so the peer isn't left hanging). _ => Value::Map(vec![]), diff --git a/crates/epix-ui/src/state.rs b/crates/epix-ui/src/state.rs index 4df7a88..8ba216b 100644 --- a/crates/epix-ui/src/state.rs +++ b/crates/epix-ui/src/state.rs @@ -1099,6 +1099,28 @@ impl AppState { let transport = self.transport.read().await.clone(); let peers = self.connectable_peers(address, 20).await; + // Piece-aware peer selection (Bigfile piecefields): for a multi-piece + // fetch, ask each peer up front which pieces of this file it holds, so we + // skip peers that don't have a given piece. `sha512` keys the piecefield. + let sha512 = entry.get("sha512").and_then(|v| v.as_str()).unwrap_or("").to_string(); + let mut peer_pf: std::collections::HashMap = + std::collections::HashMap::new(); + if last > first { + if let Some(t) = &transport { + for peer in &peers { + if let Ok(mut conn) = Connection::connect(t.as_ref(), peer).await { + if conn.handshake().await.is_ok() { + if let Ok(map) = conn.get_piecefields(address).await { + if let Some(bytes) = map.get(&sha512) { + peer_pf.insert(peer.to_string(), epix_xite::Piecefield::unpack(bytes)); + } + } + } + } + } + } + } + for i in first..=last { let poff = i * piece_size; let plen = piece_size.min(total - poff); @@ -1109,6 +1131,12 @@ impl AppState { let transport = transport.clone().ok_or("no transport")?; let mut got = false; for peer in &peers { + // Skip peers we know (from their piecefield) don't have this piece. + if let Some(pf) = peer_pf.get(&peer.to_string()) { + if !pf.get(i as usize) { + continue; + } + } let Ok(mut conn) = Connection::connect(transport.as_ref(), peer).await else { continue }; if conn.handshake().await.is_err() { continue; @@ -1133,6 +1161,54 @@ impl AppState { Ok(()) } + /// Which pieces of each big file we hold, keyed by the file's `sha512` + /// (`getPiecefields`). A big file is one with a `piecemap` + `piece_size` in + /// content.json; a piece counts as held when the on-disk bytes verify against + /// the piecemap. Files without their piecemap on disk are skipped. + pub async fn our_piecefields(&self, address: &str) -> std::collections::HashMap> { + let mut out = std::collections::HashMap::new(); + let content = self.content(address).await; + let Some(files_opt) = content + .as_ref() + .and_then(|c| c.get("files_optional")) + .and_then(|o| o.as_object()) + else { + return out; + }; + let Some(storage) = self.xites.read().await.get(address).map(|x| x.storage.clone()) else { + return out; + }; + for (inner_path, entry) in files_opt { + let (Some(sha512), Some(piecemap_path)) = ( + entry.get("sha512").and_then(|v| v.as_str()), + entry.get("piecemap").and_then(|v| v.as_str()), + ) else { + continue; // not a big file + }; + let piece_size = + entry.get("piece_size").and_then(|v| v.as_i64()).unwrap_or(1024 * 1024) as u64; + let total = entry.get("size").and_then(|v| v.as_i64()).unwrap_or(0) as u64; + if piece_size == 0 || total == 0 || !storage.exists(piecemap_path) { + continue; + } + let Ok(pm_bytes) = storage.read(piecemap_path) else { continue }; + let file_name = inner_path.rsplit('/').next().unwrap_or(inner_path); + let Some(hashes) = epix_xite::parse_piecemap(&pm_bytes, file_name) else { continue }; + let piece_num = total.div_ceil(piece_size); + let mut pf = epix_xite::Piecefield::new(); + for i in 0..piece_num { + let poff = i * piece_size; + let plen = piece_size.min(total - poff); + let present = hashes + .get(i as usize) + .is_some_and(|h| piece_present(&storage, inner_path, poff, plen, h)); + pf.set(i as usize, present); + } + out.insert(sha512.to_string(), pf.pack()); + } + out + } + /// Read a byte range from a xite file (for streaming big files / HTTP Range). /// Returns up to `length` bytes starting at `offset`. pub async fn read_file_range( diff --git a/crates/epix-ui/tests/bigfile.rs b/crates/epix-ui/tests/bigfile.rs index f830ded..cad16bd 100644 --- a/crates/epix-ui/tests/bigfile.rs +++ b/crates/epix-ui/tests/bigfile.rs @@ -6,10 +6,11 @@ use std::sync::Arc; use async_trait::async_trait; use epix_core::PeerAddr; -use epix_protocol::{vget, vmap, PeerServer, RequestHandler}; +use epix_protocol::{vget, vmap, Connection, PeerServer, RequestHandler}; use epix_transport::TcpTransport; +use epix_ui::fileserve::FileService; use epix_ui::{AppState, XiteEntry}; -use epix_xite::XiteStorage; +use epix_xite::{Piecefield, XiteStorage}; use rmpv::Value as Rmp; use serde_json::json; use tokio::net::TcpListener; @@ -106,3 +107,63 @@ async fn piecewise_download_pulls_only_needed_pieces() { let whole = std::fs::read(cli_dir.path().join("movie.mp4")).unwrap(); assert_eq!(whole, big, "reassembled big file matches the source byte-for-byte"); } + +#[tokio::test] +async fn get_piecefields_reports_which_pieces_a_peer_holds() { + let piece_size = 1024 * 1024u64; + let big: Vec = (0..(2 * piece_size + 500)).map(|i| (i % 251) as u8).collect(); + let ps = piece_size as usize; + + // Piece hashes over the *full* file (what the piecemap declares). + let piece_len = |off: u64| piece_size.min(big.len() as u64 - off); + let mut piece_hashes = Vec::new(); + let mut off = 0; + while off < big.len() as u64 { + let len = piece_len(off); + piece_hashes.push(Rmp::Binary(raw_hash(&big[off as usize..(off + len) as usize]))); + off += len; + } + let piecemap = Rmp::Map(vec![( + Rmp::from("movie.mp4"), + Rmp::Map(vec![ + (Rmp::from("sha512_pieces"), Rmp::Array(piece_hashes)), + (Rmp::from("piece_size"), Rmp::from(piece_size as i64)), + ]), + )]); + let mut pm_bytes = Vec::new(); + rmpv::encode::write_value(&mut pm_bytes, &piecemap).unwrap(); + + // Server holds pieces 0 and 2 but NOT piece 1 (that piece is a zero hole). + let mut on_disk = vec![0u8; big.len()]; + on_disk[0..ps].copy_from_slice(&big[0..ps]); + on_disk[2 * ps..].copy_from_slice(&big[2 * ps..]); + let dir = tempfile::tempdir().unwrap(); + let storage = XiteStorage::new(dir.path()); + storage.write("movie.mp4", &on_disk).unwrap(); + storage.write("movie.mp4.piecemap.msgpack", &pm_bytes).unwrap(); + + let sha512 = XiteStorage::hash_bytes(&big); + let content = json!({ + "files_optional": { "movie.mp4": { + "size": big.len(), "sha512": sha512, + "piecemap": "movie.mp4.piecemap.msgpack", "piece_size": piece_size, + } }, + }); + let state = AppState::new("seed"); + state.add_xite("1BigSeed", XiteEntry { storage, content: Some(content) }).await; + + // Serve via the real FileService and query piecefields from a client. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(PeerServer::new(Arc::new(FileService::new(state))).serve(listener)); + + let mut conn = Connection::connect(&TcpTransport, &PeerAddr::Ip(addr)).await.unwrap(); + conn.handshake().await.unwrap(); + let fields = conn.get_piecefields("1BigSeed").await.unwrap(); + let packed = fields.get(&sha512).expect("piecefield for the big file"); + let pf = Piecefield::unpack(packed); + assert!(pf.get(0), "piece 0 is held"); + assert!(!pf.get(1), "piece 1 is a hole"); + assert!(pf.get(2), "piece 2 is held"); + assert_eq!(pf.count_present(), 2); +} diff --git a/crates/epix-xite/src/lib.rs b/crates/epix-xite/src/lib.rs index 9c9c9cf..36660c3 100644 --- a/crates/epix-xite/src/lib.rs +++ b/crates/epix-xite/src/lib.rs @@ -2,12 +2,14 @@ pub mod announcer; pub mod settings; +pub mod piecefield; pub mod piecemap; pub mod xite; pub mod storage; pub use announcer::announce; pub use settings::{content_stats, Cache, ContentStats, XiteSettings}; +pub use piecefield::Piecefield; pub use piecemap::parse_piecemap; pub use xite::{FileEntry, Xite}; pub use storage::XiteStorage; diff --git a/crates/epix-xite/src/piecefield.rs b/crates/epix-xite/src/piecefield.rs new file mode 100644 index 0000000..9a174b5 --- /dev/null +++ b/crates/epix-xite/src/piecefield.rs @@ -0,0 +1,173 @@ +//! Bigfile piecefields - a compact record of which pieces of a big file a peer +//! (or we) hold, exchanged so downloads only ask peers that actually have a +//! piece. +//! +//! Wire-compatible with EpixNet's `BigfilePiecefield`: the unpacked form is one +//! flag per piece (present / absent); the packed form is a little-endian `u16` +//! array of run lengths that alternate present, absent, present, … starting with +//! the count of leading present pieces (0 when the first piece is absent). + +/// Safety cap on the total pieces an unpacked piecefield may describe, so a +/// malicious peer can't make us allocate an enormous vector. +const MAX_PIECES: usize = 8 * 1024 * 1024; + +/// Which pieces of one big file are present. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Piecefield { + bits: Vec, +} + +impl Piecefield { + pub fn new() -> Self { + Self { bits: Vec::new() } + } + + /// A piecefield with all `n` pieces present (a fully-downloaded file). + pub fn all_present(n: usize) -> Self { + Self { bits: vec![true; n] } + } + + /// Build from a present/absent flag per piece. + pub fn from_bits(bits: Vec) -> Self { + Self { bits } + } + + /// Whether piece `i` is present (absent past the end). + pub fn get(&self, i: usize) -> bool { + self.bits.get(i).copied().unwrap_or(false) + } + + /// Mark piece `i` present/absent, extending with absent pieces as needed. + pub fn set(&mut self, i: usize, present: bool) { + if i >= self.bits.len() { + self.bits.resize(i + 1, false); + } + self.bits[i] = present; + } + + /// Number of pieces this field describes. + pub fn len(&self) -> usize { + self.bits.len() + } + + pub fn is_empty(&self) -> bool { + self.bits.is_empty() + } + + /// Number of present pieces. + pub fn count_present(&self) -> usize { + self.bits.iter().filter(|b| **b).count() + } + + /// Pack to the wire form: a little-endian `u16` array of alternating run + /// lengths (present, absent, present, …), first run = leading present count. + pub fn pack(&self) -> Vec { + if self.bits.is_empty() { + return Vec::new(); + } + let mut runs: Vec = Vec::new(); + let mut expected = true; // the first run counts present pieces + let mut i = 0; + while i < self.bits.len() { + let mut count: u32 = 0; + while i < self.bits.len() && self.bits[i] == expected { + count += 1; + i += 1; + } + // A single run can't exceed u16::MAX; split it with a zero-length + // opposite run (which round-trips to nothing) to stay valid. + while count > u16::MAX as u32 { + runs.push(u16::MAX); + runs.push(0); + count -= u16::MAX as u32; + } + runs.push(count as u16); + expected = !expected; + } + let mut out = Vec::with_capacity(runs.len() * 2); + for r in runs { + out.extend_from_slice(&r.to_le_bytes()); + } + out + } + + /// Unpack from the wire form. Returns an empty field on malformed input or if + /// it would exceed [`MAX_PIECES`]. + pub fn unpack(packed: &[u8]) -> Self { + let mut bits = Vec::new(); + let mut present = true; // the first run is present pieces + for chunk in packed.chunks_exact(2) { + let run = u16::from_le_bytes([chunk[0], chunk[1]]) as usize; + if bits.len() + run > MAX_PIECES { + return Self::new(); + } + bits.extend(std::iter::repeat(present).take(run)); + present = !present; + } + Self { bits } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pack_unpack_roundtrip() { + for bits in [ + vec![], + vec![true], + vec![false], + vec![true, true, false, true], + vec![false, true, true], + vec![false, false, false, true, false], + vec![true; 100], + (0..500).map(|i| i % 3 == 0).collect::>(), + ] { + let pf = Piecefield::from_bits(bits.clone()); + let round = Piecefield::unpack(&pf.pack()); + assert_eq!(round, pf, "roundtrip failed for {bits:?}"); + } + } + + #[test] + fn packs_like_epixnet_run_lengths() { + // data 1,1,0,1 -> runs [2,1,1]; data 0,1,1 -> runs [0,1,2]. + let a = Piecefield::from_bits(vec![true, true, false, true]).pack(); + assert_eq!(a, [2u16, 1, 1].iter().flat_map(|r| r.to_le_bytes()).collect::>()); + let b = Piecefield::from_bits(vec![false, true, true]).pack(); + assert_eq!(b, [0u16, 1, 2].iter().flat_map(|r| r.to_le_bytes()).collect::>()); + } + + #[test] + fn get_set_and_count() { + let mut pf = Piecefield::new(); + pf.set(5, true); + assert!(pf.get(5)); + assert!(!pf.get(3)); + assert!(!pf.get(9)); + assert_eq!(pf.len(), 6); + assert_eq!(pf.count_present(), 1); + pf.set(5, false); + assert_eq!(pf.count_present(), 0); + } + + #[test] + fn all_present_packs_to_a_single_run() { + let pf = Piecefield::all_present(9); + // A single present-run of 9 -> [9]. + assert_eq!(pf.pack(), 9u16.to_le_bytes().to_vec()); + assert_eq!(Piecefield::unpack(&pf.pack()).count_present(), 9); + } + + #[test] + fn oversized_run_is_rejected() { + // A run claiming 60000 pieces repeated to exceed MAX_PIECES -> empty. + let mut packed = Vec::new(); + for _ in 0..200 { + packed.extend_from_slice(&60000u16.to_le_bytes()); + packed.extend_from_slice(&0u16.to_le_bytes()); + } + assert!(Piecefield::unpack(&packed).is_empty()); + } +}