fix(stream-install)!: stream archive payloads as raw frames

Streamed installs were sending FileChunk payloads through the shared JSON Message impl. serde_json serializes bytes as arrays of integers, which bloats wire traffic and burns CPU on large archives. Replace StreamInstallFrame encoding with tagged frames: JSON control frames keep their shape under tag 0, while file chunks carry raw bytes under tag 1. The stream install metadata now carries unpacked archive size and mandatory CRC32. The CLI unrar provider validates CRCs up front, runs one archive-wide unrar p stream, splits stdout by listed file sizes, and refuses trailing or missing bytes. That avoids solid archive re-decompression and sidesteps unrar wildcard masks for path arguments. Receivers now sample existing download progress events for streamed installs, report staging-relative chunk paths, and retry trusted peers with a fresh streamed-install transaction after a failed attempt. The current protocol policy does not preserve compatibility with older stream-install builds. Test Plan: - just fmt - just test - just clippy - git diff --check - git diff --cached --check BREAKING CHANGE: StreamInstallFrame now uses tagged frames with raw chunk payloads and requires current peers on both sides of streamed installs. Refs: NEXT_STEPS_CLAUDES_REVIEW.md
2026-06-07 21:12:15 +02:00
parent cc147def73
commit 5dd356eca8
5 changed files with 450 additions and 165 deletions
@@ -15,7 +15,10 @@ use lanspread_peer::{StreamInstallFuture, StreamInstallProvider, UnpackFuture, U
 use lanspread_proto::StreamInstallFrame;
 use serde::Serialize;
 use serde_json::{Value, json};
-use tokio::{io::AsyncReadExt, sync::mpsc};
+use tokio::{
+    io::{AsyncRead, AsyncReadExt},
+    sync::mpsc,
+};
 use tokio_util::sync::CancellationToken;

 pub const DEFAULT_FIXTURE_VERSION: &str = "20250101";
@@ -298,53 +301,19 @@ impl StreamInstallProvider for ExternalUnrarStreamProvider {
                StreamInstallFrame::ArchiveBegin {
                    archive_name: archive_name.clone(),
                    solid: listing.solid,
+                    unpacked_size: listing.unpacked_size(),
                },
            )
            .await?;

-            for entry in listing.entries {
-                if cancel_token.is_cancelled() {
-                    eyre::bail!("streamed archive {} was cancelled", archive.display());
-                }
-
-                match entry.kind {
-                    RarEntryKind::Directory => {
-                        send_stream_frame(
-                            &frames,
-                            StreamInstallFrame::Directory {
-                                relative_path: entry.relative_path,
-                            },
-                        )
-                        .await?;
-                    }
-                    RarEntryKind::File => {
-                        send_stream_frame(
-                            &frames,
-                            StreamInstallFrame::FileBegin {
-                                relative_path: entry.relative_path.clone(),
-                                size: entry.size,
-                                crc32: entry.crc32,
-                            },
-                        )
-                        .await?;
-                        stream_unrar_file(
-                            &self.program,
-                            archive,
-                            &entry.relative_path,
-                            &frames,
-                            cancel_token.clone(),
-                        )
-                        .await?;
-                        send_stream_frame(
-                            &frames,
-                            StreamInstallFrame::FileEnd {
-                                relative_path: entry.relative_path,
-                            },
-                        )
-                        .await?;
-                    }
-                }
-            }
+            stream_unrar_entries(
+                &self.program,
+                archive,
+                &listing.entries,
+                &frames,
+                cancel_token.clone(),
+            )
+            .await?;

            send_stream_frame(&frames, StreamInstallFrame::ArchiveEnd { archive_name }).await
        })
@@ -357,6 +326,16 @@ struct RarListing {
    entries: Vec<RarEntry>,
 }

+impl RarListing {
+    fn unpacked_size(&self) -> u64 {
+        self.entries
+            .iter()
+            .filter(|entry| entry.kind == RarEntryKind::File)
+            .map(|entry| entry.size)
+            .sum()
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct RarEntry {
    relative_path: String,
@@ -448,26 +427,32 @@ fn push_rar_entry(entries: &mut Vec<RarEntry>, draft: RarEntryDraft) -> eyre::Re
        return Ok(());
    };

-    let size = match kind {
-        RarEntryKind::File => draft
-            .size
-            .ok_or_else(|| eyre::eyre!("RAR file entry {relative_path} has no Size"))?,
-        RarEntryKind::Directory => 0,
+    let (size, crc32) = match kind {
+        RarEntryKind::File => {
+            let size = draft
+                .size
+                .ok_or_else(|| eyre::eyre!("RAR file entry {relative_path} has no Size"))?;
+            let crc32 = draft
+                .crc32
+                .ok_or_else(|| eyre::eyre!("RAR file entry {relative_path} has no CRC32"))?;
+            (size, Some(crc32))
+        }
+        RarEntryKind::Directory => (0, None),
    };

    entries.push(RarEntry {
        relative_path,
        kind,
        size,
-        crc32: draft.crc32,
+        crc32,
    });
    Ok(())
 }

-async fn stream_unrar_file(
+async fn stream_unrar_entries(
    program: &Path,
    archive: &Path,
-    relative_path: &str,
+    entries: &[RarEntry],
    frames: &mpsc::Sender<StreamInstallFrame>,
    cancel_token: CancellationToken,
 ) -> eyre::Result<()> {
@@ -476,28 +461,112 @@ async fn stream_unrar_file(
        .arg("-inul")
        .arg("-cfg-")
        .arg(archive)
-        .arg(relative_path)
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()?;

-    let mut stdout = child
-        .stdout
-        .take()
-        .ok_or_eyre("unrar stdout was not captured")?;
-    let mut buffer = vec![0_u8; STREAM_CHUNK_SIZE];
+    let result = async {
+        let mut stdout = child
+            .stdout
+            .take()
+            .ok_or_eyre("unrar stdout was not captured")?;
+        let mut buffer = vec![0_u8; STREAM_CHUNK_SIZE];

-    loop {
-        let read = tokio::select! {
-            () = cancel_token.cancelled() => {
-                let _ = child.kill().await;
-                eyre::bail!("streaming {relative_path} from {} was cancelled", archive.display());
+        for entry in entries {
+            if cancel_token.is_cancelled() {
+                eyre::bail!("streamed archive {} was cancelled", archive.display());
            }
-            read = stdout.read(&mut buffer) => read?,
-        };

+            match entry.kind {
+                RarEntryKind::Directory => {
+                    send_stream_frame(
+                        frames,
+                        StreamInstallFrame::Directory {
+                            relative_path: entry.relative_path.clone(),
+                        },
+                    )
+                    .await?;
+                }
+                RarEntryKind::File => {
+                    let Some(crc32) = entry.crc32 else {
+                        eyre::bail!("RAR file entry {} has no CRC32", entry.relative_path);
+                    };
+                    send_stream_frame(
+                        frames,
+                        StreamInstallFrame::FileBegin {
+                            relative_path: entry.relative_path.clone(),
+                            size: entry.size,
+                            crc32,
+                        },
+                    )
+                    .await?;
+                    stream_unrar_file_from_stdout(
+                        &mut stdout,
+                        archive,
+                        entry,
+                        frames,
+                        &mut buffer,
+                        &cancel_token,
+                    )
+                    .await?;
+                    send_stream_frame(
+                        frames,
+                        StreamInstallFrame::FileEnd {
+                            relative_path: entry.relative_path.clone(),
+                        },
+                    )
+                    .await?;
+                }
+            }
+        }
+
+        let extra =
+            read_unrar_stdout(&mut stdout, &mut buffer[..1], &cancel_token, archive).await?;
+        if extra != 0 {
+            eyre::bail!(
+                "unrar produced bytes after listed entries for {}",
+                archive.display()
+            );
+        }
+
+        let status = wait_unrar_child(&mut child, &cancel_token, archive).await?;
+        if !status.success() {
+            eyre::bail!(
+                "unrar p failed for {} with status {status}",
+                archive.display()
+            );
+        }
+
+        Ok(())
+    }
+    .await;
+
+    if result.is_err() {
+        let _ = child.kill().await;
+    }
+
+    result
+}
+
+async fn stream_unrar_file_from_stdout(
+    stdout: &mut (impl AsyncRead + Unpin),
+    archive: &Path,
+    entry: &RarEntry,
+    frames: &mpsc::Sender<StreamInstallFrame>,
+    buffer: &mut [u8],
+    cancel_token: &CancellationToken,
+) -> eyre::Result<()> {
+    let mut remaining = entry.size;
+    while remaining > 0 {
+        let read_len = usize::try_from(remaining.min(buffer.len() as u64))?;
+        let read =
+            read_unrar_stdout(stdout, &mut buffer[..read_len], cancel_token, archive).await?;
        if read == 0 {
-            break;
+            eyre::bail!(
+                "unrar ended while streaming {} from {}; {remaining} bytes missing",
+                entry.relative_path,
+                archive.display()
+            );
        }

        send_stream_frame(
@@ -507,20 +576,40 @@ async fn stream_unrar_file(
            },
        )
        .await?;
-    }
-
-    let status = child.wait().await?;
-    if !status.success() {
-        eyre::bail!(
-            "unrar p failed for {}:{} with status {status}",
-            archive.display(),
-            relative_path
-        );
+        remaining = remaining.saturating_sub(u64::try_from(read)?);
    }

    Ok(())
 }

+async fn read_unrar_stdout(
+    stdout: &mut (impl AsyncRead + Unpin),
+    buffer: &mut [u8],
+    cancel_token: &CancellationToken,
+    archive: &Path,
+) -> eyre::Result<usize> {
+    tokio::select! {
+        () = cancel_token.cancelled() => {
+            eyre::bail!("streamed archive {} was cancelled", archive.display());
+        }
+        read = stdout.read(buffer) => Ok(read?),
+    }
+}
+
+async fn wait_unrar_child(
+    child: &mut tokio::process::Child,
+    cancel_token: &CancellationToken,
+    archive: &Path,
+) -> eyre::Result<std::process::ExitStatus> {
+    tokio::select! {
+        () = cancel_token.cancelled() => {
+            let _ = child.kill().await;
+            eyre::bail!("streamed archive {} was cancelled", archive.display());
+        }
+        status = child.wait() => Ok(status?),
+    }
+}
+
 async fn send_stream_frame(
    frames: &mpsc::Sender<StreamInstallFrame>,
    frame: StreamInstallFrame,
@@ -639,7 +728,7 @@ mod tests {
        let listing = parse_unrar_listing(
            r#"
 Archive: game.eti
-Details: RAR 5
+Details: RAR 5, solid

        Name: bin/payload.bin
        Type: File
@@ -652,7 +741,7 @@ Details: RAR 5
        )
        .expect("listing should parse");

-        assert!(!listing.solid);
+        assert!(listing.solid);
        assert_eq!(
            listing.entries,
            vec![
@@ -672,6 +761,23 @@ Details: RAR 5
        );
    }

+    #[test]
+    fn rejects_unrar_file_entries_without_crc32() {
+        let err = parse_unrar_listing(
+            r#"
+Archive: game.eti
+Details: RAR 5
+
+        Name: bin/payload.bin
+        Type: File
+        Size: 123
+"#,
+        )
+        .expect_err("file entries without CRC32 should be rejected");
+
+        assert!(err.to_string().contains("has no CRC32"));
+    }
+
    #[tokio::test]
    async fn fixture_unpacker_creates_install_payload() {
        let temp = TempDir::new("lanspread-peer-cli-fixture");