From c42615bc31a539b46c1e39b111d2c6e67bd5bb0c Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Sat, 29 Aug 2026 14:52:22 +0200 Subject: [PATCH] send-file: stream the body instead of buffering the whole file on both sides Finding 7 of the 2026-08-29 review. 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. send_file opens the file and send_from_reader streams exactly len bytes with tokio::io::copy over a bounded buffer; the daemon passes the path rather than reading it. The receiver streams the body straight to its temp file with the same bounded copy, reading exactly header.len bytes via take() so it never waits for an EOF the sender does not send before its ack. Neither side 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 now refused and leaves no file, where before a short read would error on read_exact. Tests: a_large_file_streams_through_in_chunks round-trips 5 MiB over a 64 KiB duplex; a_short_stream_is_refused_and_leaves_no_file pins the truncation guard. --- CHANGELOG.md | 11 ++++ src/daemon.rs | 14 +++-- src/sendfile.rs | 149 ++++++++++++++++++++++++++++++++++++++++++------ 3 files changed, 150 insertions(+), 24 deletions(-) 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