send-file: stream the body instead of buffering the whole file on both sides

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

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<DaemonState>,
     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
 }

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<PathBuf>
     Ok(inbox_for(home, peer).join(name))
 }
 
-/// Send one file. The initiating half.
-pub async fn send<S>(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<S>(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<S>(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<S, R>(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<u8> = (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