diff --git a/CHANGELOG.md b/CHANGELOG.md index d48437a..aeff67a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -257,6 +257,17 @@ EXPERIMENTAL, so on-disk formats and the CLI may change without notice. ### Fixed +- **`send-file` streams instead of holding the whole file in memory on both + daemons.** The sender read the file whole with `std::fs::read` and the receiver + allocated `header.len` bytes up front, so a 1.5 GiB transfer cost about 1.5 GiB + of resident memory on each daemon at once; on a host with a memory ceiling that + is enough to kill the daemon mid-transfer. Both sides now stream the body in + bounded chunks (`tokio::io::copy`), and the receiver writes straight to its + temp file, so neither allocates against the file size. The wire format is + unchanged, so a streaming build and an old build interoperate. A transfer that + ends short of its declared length is refused and leaves no file. Finding 7 of + the 2026-08-29 review. + - **`fabric doctor` reports whether the service is ENABLED, not just that its unit file exists.** It read `service_installed` from the unit file's presence, so a service disabled during an incident with its unit left in place said diff --git a/src/daemon.rs b/src/daemon.rs index 57dc9dd..dec232b 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -2918,19 +2918,21 @@ async fn process_control_request( ControlResponse::Ok } ControlRequest::SendFile { peer, path, name } => { - let bytes = match std::fs::read(&path) { - Ok(bytes) => bytes, + // Read the size, not the bytes: the file streams from disk during + // the transfer, so the sending daemon never holds it whole. + let bytes = match std::fs::metadata(&path) { + Ok(meta) => meta.len(), Err(error) => { return Ok(ControlResponse::Error { message: format!("could not read {}: {error}", path.display()), }); } }; - match send_file_to_peer(&state, &peer, &name, &bytes).await { + match send_file_to_peer(&state, &peer, &name, &path).await { Ok(()) => ControlResponse::SentFile { peer, name, - bytes: bytes.len() as u64, + bytes, }, Err(error) => ControlResponse::Error { message: format!("{error:#}"), @@ -3272,7 +3274,7 @@ async fn send_file_to_peer( state: &Arc, peer: &str, name: &str, - bytes: &[u8], + path: &std::path::Path, ) -> Result<()> { let addr = { let book = state.peer_book.read().await; @@ -3296,7 +3298,7 @@ async fn send_file_to_peer( .with_context(|| format!("dialling {peer}"))?; let (send, recv) = connection.open_bi().await?; let stream = tokio::io::join(recv, send); - let result = crate::sendfile::send(stream, name, bytes).await; + let result = crate::sendfile::send_file(stream, name, path).await; connection.close(0u32.into(), b"done"); result } diff --git a/src/sendfile.rs b/src/sendfile.rs index 01f2c3d..1cd061c 100644 --- a/src/sendfile.rs +++ b/src/sendfile.rs @@ -36,9 +36,10 @@ pub const SERVICE: &str = "send-file"; /// The largest file this will move in one shot. /// -/// A limit rather than none, because the receiving side allocates against it and -/// a peer should not be able to fill a disk by accident. Large enough for the -/// things people actually send one at a time. +/// A limit rather than none, because the receiving side writes it to disk and a +/// peer should not be able to fill a disk by accident. The body streams in +/// bounded chunks, so this is not a memory allocation on either side. Large +/// enough for the things people actually send one at a time. pub const MAX_FILE_BYTES: u64 = 2 * 1024 * 1024 * 1024; const MAX_HEADER_BYTES: usize = 64 * 1024; @@ -114,10 +115,47 @@ pub fn destination(home: &FabricHome, peer: &str, name: &str) -> Result Ok(inbox_for(home, peer).join(name)) } -/// Send one file. The initiating half. -pub async fn send(mut stream: S, name: &str, bytes: &[u8]) -> Result<()> +/// Send one file from a path, streaming it. The initiating half. +/// +/// Opens the file and streams its bytes to the peer in bounded chunks, so the +/// sending daemon never holds the whole file in memory. Previously the caller +/// read the file whole with `std::fs::read` and this held the whole slice, so a +/// 1.5 GiB transfer cost about 1.5 GiB on each side at once. Finding 7 of the +/// 2026-08-29 review. +pub async fn send_file(stream: S, name: &str, path: &Path) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + // The size comes from the file's metadata rather than from reading it, so + // the sender never allocates against it either. + let len = tokio::fs::metadata(path) + .await + .with_context(|| format!("could not stat {}", path.display()))? + .len(); + let file = tokio::fs::File::open(path) + .await + .with_context(|| format!("could not open {}", path.display()))?; + send_from_reader(stream, name, len, file).await +} + +/// Send one file whose bytes are already in hand. A convenience over +/// [`send_from_reader`] for small payloads and the tests. +pub async fn send(stream: S, name: &str, bytes: &[u8]) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + send_from_reader(stream, name, bytes.len() as u64, bytes).await +} + +/// Send exactly `len` bytes read from `reader`, streaming them to `stream`. +/// +/// The whole file never lands in one buffer: `tokio::io::copy` moves it in +/// bounded chunks. A source that turns out shorter than `len` is caught, and +/// one that is longer is bounded by `take(len)`. +pub async fn send_from_reader(mut stream: S, name: &str, len: u64, reader: R) -> Result<()> where S: AsyncRead + AsyncWrite + Unpin, + R: AsyncRead + Unpin, { // Checked here so the refusal reaches the person who typed it, and again on // arrival because the receiver cannot trust this one. @@ -127,20 +165,20 @@ where no parent components" ); } - if bytes.len() as u64 > MAX_FILE_BYTES { - bail!( - "{} bytes is larger than the {MAX_FILE_BYTES} byte limit for one \ - transfer", - bytes.len() - ); + if len > MAX_FILE_BYTES { + bail!("{len} bytes is larger than the {MAX_FILE_BYTES} byte limit for one transfer"); } let header = serde_json::to_vec(&Header { name: name.to_string(), - len: bytes.len() as u64, + len, })?; stream.write_all(&(header.len() as u32).to_be_bytes()).await?; stream.write_all(&header).await?; - stream.write_all(bytes).await?; + // Exactly `len` bytes, streamed rather than buffered. + let copied = tokio::io::copy(&mut reader.take(len), &mut stream).await?; + if copied != len { + bail!("read {copied} of the {len} bytes the file was said to hold"); + } stream.flush().await?; // Wait for the receiver to say it committed the file. Without this the @@ -183,17 +221,34 @@ where // courtesy; this one is the boundary. let target = destination(home, peer, &header.name)?; - let mut body = vec![0u8; header.len as usize]; - stream.read_exact(&mut body).await?; - if let Some(parent) = target.parent() { std::fs::create_dir_all(parent) .with_context(|| format!("creating {}", parent.display()))?; } // Temp then rename, so an interrupted transfer never appears at the final - // path as a complete file. + // path as a complete file. The body streams straight to the temp file in + // bounded chunks, so the receiver never allocates against `header.len`. let temp = target.with_extension("fabric-partial"); - std::fs::write(&temp, &body).with_context(|| format!("writing {}", temp.display()))?; + let copied = { + let mut file = tokio::fs::File::create(&temp) + .await + .with_context(|| format!("writing {}", temp.display()))?; + // Read EXACTLY header.len bytes. `take` stops the copy at the limit, so + // it never waits for an EOF the sender does not send before its ack. + let copied = tokio::io::copy(&mut (&mut stream).take(header.len), &mut file) + .await + .with_context(|| format!("receiving into {}", temp.display()))?; + file.flush() + .await + .with_context(|| format!("flushing {}", temp.display()))?; + copied + }; + if copied != header.len { + // The sender closed early or the transport dropped. Do not commit a + // short file, and do not ack it. + let _ = std::fs::remove_file(&temp); + bail!("received {copied} of {} bytes before the stream ended", header.len); + } std::fs::rename(&temp, &target) .with_context(|| format!("renaming into {}", target.display()))?; @@ -277,6 +332,64 @@ mod tests { ); } + /// A file larger than any single buffer streams through send_from_reader and + /// the streaming receiver intact. This exercises the finding-7 path: neither + /// side allocates against the whole size, so many chunks cross the duplex. + #[tokio::test] + async fn a_large_file_streams_through_in_chunks() { + let dir = tempfile::tempdir().unwrap(); + // A few MiB, well past the internal copy buffer, with a position-varying + // pattern so a chunk written at the wrong offset would be caught. + let payload: Vec = (0..(5 * 1024 * 1024u32)).map(|i| (i % 251) as u8).collect(); + + let (client, server) = tokio::io::duplex(64 * 1024); + let home_for_server = FabricHome::new(dir.path()); + let receiver = + tokio::spawn(async move { receive(server, &home_for_server, "hetz").await }); + // send_from_reader with a reader (not a held slice) is the streaming API + // the daemon uses for a file on disk. + send_from_reader(client, "big.bin", payload.len() as u64, payload.as_slice()) + .await + .unwrap(); + let landed = receiver.await.unwrap().unwrap(); + assert_eq!(std::fs::read(&landed).unwrap(), payload); + assert!(!landed.with_extension("fabric-partial").exists()); + } + + /// A sender that promises more bytes than it delivers must not leave a + /// committed file: the receiver reads exactly `len`, sees the stream end + /// short, and refuses rather than writing a truncated file to the inbox. + #[tokio::test] + async fn a_short_stream_is_refused_and_leaves_no_file() { + let dir = tempfile::tempdir().unwrap(); + let home = FabricHome::new(dir.path()); + let (mut client, server) = tokio::io::duplex(1 << 16); + let home_for_server = FabricHome::new(dir.path()); + let receiver = + tokio::spawn(async move { receive(server, &home_for_server, "hetz").await }); + + // Hand-write a header claiming 1000 bytes, then send 10 and close. + let header = serde_json::to_vec(&Header { + name: "short.bin".to_string(), + len: 1000, + }) + .unwrap(); + client + .write_all(&(header.len() as u32).to_be_bytes()) + .await + .unwrap(); + client.write_all(&header).await.unwrap(); + client.write_all(&[7u8; 10]).await.unwrap(); + client.shutdown().await.unwrap(); + drop(client); + + let result = receiver.await.unwrap(); + assert!(result.is_err(), "a short transfer must not be committed"); + let target = inbox_for(&home, "hetz").join("short.bin"); + assert!(!target.exists(), "a truncated file reached the inbox"); + assert!(!target.with_extension("fabric-partial").exists()); + } + /// The receiver refuses an escaping name even when the sender does not check. /// /// The sender's check is a courtesy for whoever typed the command. This one