Bigfile piecefields: exchange + piece-aware peer selection

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<std::collections::HashMap<String, Vec<u8>>> {
+        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<Vec<u8>> {
         let mut out = Vec::new();

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![]),

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<String, epix_xite::Piecefield> =
+            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<String, Vec<u8>> {
+        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(

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<u8> = (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);
+}

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;