test(peer-cli): align scenarios with catalog versions

Remote aggregation now filters to catalog-version roots, but the checked-in
peer-cli fixtures and skew scenarios still stamped synthetic future versions.
That hid fixture rows in S3 and left scenario docs asserting latest-version
behavior.

Teach the harness the catalog versions for fixture game IDs, stamp generated
fixtures with catalog versions by default, and update skew, mesh, propagation,
and throughput scenarios to expect only catalog-version peers. Also wire S38
into the executable matrix so the documented first-play launch-setting scenario
is covered by the same full run as S1-S47.

This keeps stale peers as negative coverage: they are absent from list-games and
cannot provide descriptors, votes, or chunks. The fixture version.ini updates
are checked in so alpha, bravo, charlie, and persona roots advertise
downloadable catalog games again.

Test Plan:
- python3 -m py_compile
  crates/lanspread-peer-cli/scripts/run_extended_scenarios.py
- python3 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py \
  S3 S8 S14 S15 S16 S17 S21 S22 S23 S24 S29 S30 S31 S34 S36 S37 \
  S39 S40 S41 S42 S43 S44 S45 S46 S47 --build-image
- python3 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py S38
- python3 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py
- git diff --check
- git diff --cached --check

Docs: PEER_CLI_SCENARIOS.md

Cargo.lock

@@ -1521,9 +1521,9 @@ dependencies = [
 
 [[package]]
 name = "http"
-version = "1.4.2"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
+checksum = "8be7462df143984c4598a256ef469b251d7d7f9e271135073e78fc535414f3d0"
 dependencies = [
  "bytes",
  "itoa",
@@ -1937,12 +1937,13 @@ dependencies = [
 
 [[package]]
 name = "js-sys"
-version = "0.3.100"
+version = "0.3.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f2025f20d7a4fa7785846e7b63d10a76d3f1cee98ee5cb79ea59703f95e42162"
+checksum = "142bc4740e452c1e57ade0cbc129f139c9093e354346f0872ef985f4f5cf5f11"
 dependencies = [
  "cfg-if",
  "futures-util",
+ "once_cell",
  "wasm-bindgen",
 ]
 
@@ -3103,9 +3104,9 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.12.4"
+version = "1.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba"
+checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -3126,9 +3127,9 @@ dependencies = [
 
 [[package]]
 name = "regex-syntax"
-version = "0.8.11"
+version = "0.8.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
+checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
 
 [[package]]
 name = "reqwest"
@@ -3274,9 +3275,9 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
 
 [[package]]
 name = "s2n-codec"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a650d3f187901f3519ec8a1fe7da3faccc0b2fb40f350eda2c7851fdf2bda0f6"
+checksum = "d197a3c92bbe21fc00ba8366f6ba14edb8685316b6c8c14c622d3aba0a3816d8"
 dependencies = [
  "byteorder",
  "bytes",
@@ -3285,9 +3286,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic"
-version = "1.82.0"
+version = "1.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c27c34127facefcd3e5530c4de5739a62cd4a593710b1194dacbd8e884b6be92"
+checksum = "8728244102e791769cebe44a4abace966d8826f3266e9691c4233f47921b94b8"
 dependencies = [
  "bytes",
  "cfg-if",
@@ -3309,9 +3310,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-core"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "79fbc3f06797d985363f74de105d18554b5a272b924b166d73a6564943da1230"
+checksum = "6cc69861a4909ea508b26309504899f4b0f77bb35348f6a36b7de9a28b1a4b92"
 dependencies = [
  "atomic-waker",
  "byteorder",
@@ -3331,9 +3332,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-crypto"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e58ea5aa39eecc29559d1e1bb4a5d55a747fa7b80cff5a3400c57489510644e3"
+checksum = "5a3ce7f399a87be4b49d76895cdddb987620d34f334072d011bcac913d20fe69"
 dependencies = [
  "aws-lc-rs",
  "cfg-if",
@@ -3345,9 +3346,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-platform"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4eebb6007139cfffdf3d473d39f01a214032c339432a6293b16b0f7b25343f40"
+checksum = "fa9004809ae3a778b8e015581a47e9fb389f9ec230456a24b81c6287b000fefe"
 dependencies = [
  "cfg-if",
  "futures",
@@ -3360,9 +3361,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-rustls"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eb0084afa65eefae2c37d9ab44118a14dfc5bb78dbf997c0f5176f7cf8d2e633"
+checksum = "cf7c34876c77f7560ee4385cd5ff0510acade2eb66dc237a45f7c63d2e7f1af3"
 dependencies = [
  "bytes",
  "rustls",
@@ -3374,9 +3375,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-tls"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91150b25ce824ffea581b449ad04acf9b4aef2fa68a46f667cdc9cc6f7b87823"
+checksum = "fc7b14505cff3d9e39b930c31c150fe2965ee5fe1b654f7c6d33b1f50680ac0b"
 dependencies = [
  "bytes",
  "errno",
@@ -3389,9 +3390,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-tls-default"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e1f5ae64863972facee778dc80a24317e613f035296631f267b71f225e569c22"
+checksum = "3297fc8531b3c19f339a3ce1969fdc0e9928cfe439ca6c9f9d9d7ca4522a3b8c"
 dependencies = [
  "s2n-quic-rustls",
  "s2n-quic-tls",
@@ -3399,9 +3400,9 @@ dependencies = [
 
 [[package]]
 name = "s2n-quic-transport"
-version = "0.82.0"
+version = "0.81.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b82fca53ce1734cc1d1dca96cc9ceb65ed528f27cb43b7de865215b6cf17908"
+checksum = "d1ddd739c1776770dd2ab0b33da1cf372a395500252ae5250c08e2d6bf51b38f"
 dependencies = [
  "bytes",
  "futures-channel",
@@ -5002,9 +5003,9 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
 
 [[package]]
 name = "uuid"
-version = "1.23.3"
+version = "1.23.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7"
+checksum = "d258b83ceec21034727ecee8c382cfa6c3e133699b0742c64571814fb420c9f7"
 dependencies = [
  "getrandom 0.4.2",
  "js-sys",
@@ -5107,9 +5108,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.123"
+version = "0.2.122"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a254a4b10c19a76f09a27640e7ffbf9bc30bf67e16a3bf28aaefa4920fe81563"
+checksum = "3ed04576f974d2b2fba0f38c51dbc5518011e38c36bf1143164be765528fd409"
 dependencies = [
  "cfg-if",
  "once_cell",
@@ -5120,9 +5121,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.73"
+version = "0.4.72"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "54568702fabf5d4849ce2b90fadfa64168a097eaf4b351ce9df8b687a0086aaf"
+checksum = "9473dbd2991ae90b6291c3c32c30c6187ac49aa32f9905d1cce280ec1e110b0f"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -5130,9 +5131,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.123"
+version = "0.2.122"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "24a40fc75b0ec6f3746ceb10d36f53a93dcd68a93b11b6445983945d79eba0dc"
+checksum = "916151b09da36bd82f6615cbf3a419e2f0ba23a03c6160e8e92eb6bd4aa1dec6"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -5140,9 +5141,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.123"
+version = "0.2.122"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "908f34bd9b9ce3d4caf07b72dfab63d61504d156856c6bd3cd87fa350cf3985b"
+checksum = "299047362ccbfce148b67ab7e73349f77748e00c8296f9542adfad2ad82c5c5e"
 dependencies = [
  "bumpalo",
  "proc-macro2",
@@ -5153,9 +5154,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.123"
+version = "0.2.122"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7acbf7616c27b194bbb550bf77ed0c2c3e5b7fd1260a93082b95fb7f47959b92"
+checksum = "9a929b2c61f11ba3e9bc35b50c1f25cb38e0e892c0c231ae2b8cf78d5dad4437"
 dependencies = [
  "unicode-ident",
 ]
@@ -5209,9 +5210,9 @@ dependencies = [
 
 [[package]]
 name = "web-sys"
-version = "0.3.100"
+version = "0.3.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e0871acf327f283dc6da28a1696cdc64fb355ba9f935d052021fa77f35cce69"
+checksum = "6d621441cfc37b84979402712047321980c178f299193a3589d05b99e8763436"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -6018,18 +6019,18 @@ dependencies = [
 
 [[package]]
 name = "zerocopy"
-version = "0.8.52"
+version = "0.8.50"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce1022995ff5ff5d841ad7d994facc23098cd40152f2c1d11cd607c6f530653f"
+checksum = "3b065d4f0e55f82fae73202e189638116a87c55ab6b8e6c2721e13dd9d854ad1"
 dependencies = [
  "zerocopy-derive",
 ]
 
 [[package]]
 name = "zerocopy-derive"
-version = "0.8.52"
+version = "0.8.50"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930"
+checksum = "0b631b19d36a892ab55420c92dbc83ccd79274f25be714855d3074aa71cab639"
 dependencies = [
  "proc-macro2",
  "quote",

NEXT_STEPS.md

@@ -61,6 +61,5 @@ product-ready.
    modal status shows `Installed, not shareable`. Downloaded-and-installed games
    keep the normal `Installed` label.
 
-The remaining production-readiness step is additive: move from sender-owned RAR
+My recommended next slice: make the provider abstraction final-ish, then
-metadata to catalog-owned archive or extracted-file hashes, then verify those
+implement a real one-pass provider. Everything else builds cleanly on that.
-at the receiver before committing the streamed install.

PEER_CLI_SCENARIOS.md

@@ -46,7 +46,7 @@ for deterministic local runs; mDNS/macvlan remains an environment smoke path.
 | S36 | Catalog singleton beats stale majority | Five peers advertise one game; one peer has the catalog version and four peers have stale versions. | `list-games` reports `peer_count=1` and the catalog `eti_game_version`; all descriptors and chunks come from the singleton catalog-version peer, while stale peers remain hidden and contribute zero bytes. |
 | S37 | Single-source download throughput | A source peer advertises a temporary catalog game with one sparse `2 GiB` `.eti`; an empty client downloads it with `install=false`. | The client emits `download-finished` with throughput measurements (`bytes`, `duration_ms`, `mib_per_s`, `mbit_per_s`), and the downloaded archive size matches the source. |
 | S38 | First-play launch-setting stamping | `fixture-persona/css` ships a real RAR `.eti` whose tree buries a CRLF `SmartSteamEmu.ini` with a stub `PersonaName` line under `engine/bin/win64/steam_settings/`, plus a stub `account_name.txt` and `language.txt` under `profiles/local/`. A peer installs `css` (with `--unrar`), then sends `play css` with a username and language, then `play css` again. | After install the marker `games/css/launch_settings_applied` is absent and the stub files are intact under `local/`. The first `play` returns `already_applied=false` with `account_name_written`, `language_written`, and `persona_name_written` all true; the deep `SmartSteamEmu.ini` `PersonaName` value becomes the username with its `\r\n` ending and sibling lines preserved, `account_name.txt` becomes the username, `language.txt` becomes the passed language, and the marker now exists. A second `play` returns `already_applied=true`, rewrites nothing, and leaves the files untouched even if their values were reset externally. |
-| S39 | Streamed install without keeping archive payload | Empty client connects to `fixture-bravo`, then sends `stream-install cnctw`. The source has real RAR `.eti` payload entries under `bin/` and `data/`; the receiver uses the container-bundled `unrar` stream provider. | Client emits `download-begin`, streamed `download-chunk-finished`, `download-finished`, `install-begin`, and `install-finished`. Local `cnctw` is `downloaded=false`, `installed=true`, `availability=LocalOnly`; root `version.ini` and `.eti` are absent; `local/bin/cnctw-payload.bin` and `local/data/cnctw-assets.dat` match `unrar p` output by SHA-256; the source reports no active outbound transfer for `cnctw` after completion. |
+| S39 | Streamed install without keeping archive payload | Empty client connects to `fixture-bravo`, then sends `stream-install cnctw`. The source has real RAR `.eti` payload entries under `bin/` and `data/`; the receiver uses the container-bundled `unrar` stream provider. | Client emits `got-game-files`, `download-begin`, streamed `download-chunk-finished`, `download-finished`, `install-begin`, and `install-finished`. Local `cnctw` is `downloaded=false`, `installed=true`, `availability=LocalOnly`; root `version.ini` and `.eti` are absent; `local/bin/cnctw-payload.bin` and `local/data/cnctw-assets.dat` match `unrar p` output by SHA-256. |
 | S40 | Streamed install receiver is not a peer source | After S39, a third peer connects only to the streamed-install receiver. | The third peer may see the receiver's local-only summary in peer snapshots, but `list-games` remote aggregation does not expose `cnctw` as downloadable, `peer_count` remains zero/absent, and attempting `download cnctw` fails with no local files created. |
 | S41 | Solid archive streamed install | Empty client connects to a peer serving `fixture-solid/cnctw`, whose `.eti` is a real solid RAR archive. The receiver uses the container-bundled `unrar` stream provider. | The fixture is verified as solid with `unrar lt`; streamed install finishes with `downloaded=false`, `installed=true`, `availability=LocalOnly`; root archive and `version.ini` are absent; streamed byte count equals the extracted solid entries; local payload SHA-256 hashes match `unrar p` output. |
 | S42 | Streamed install whole-stream retry | Empty client connects to two peers serving the same catalog-version `cnctw`: one broken source whose `--unrar` path is missing, followed by one good source. | The broken source sorts before the good source in retry order, contributes zero chunks, and the good source completes a fresh whole-stream attempt. The final state is local-only installed, no root archive/sentinel, no `.local.installing`, byte count matches the extracted entries, and payload hashes match the good source. |

crates/lanspread-peer-cli/README.md

@@ -42,7 +42,3 @@ echoed back on the result or error line.
 {"id":"u1","cmd":"uninstall","game_id":"fixture-one"}
 {"id":"q1","cmd":"shutdown"}
 ```
-
-The `status` result includes receiver-side `active_operations` and
-sender-side `active_outbound_transfers` counts by game ID, which the scenario
-runner uses to verify transfer lifecycle cleanup.

crates/lanspread-peer-cli/scripts/run_extended_scenarios.py

@@ -1208,6 +1208,12 @@ class Runner:
         wait_remote_game(client, "cnctw", peer_count=1)
         waiter = LineWaiter(len(client.output))
         client.send({"cmd": "stream-install", "game_id": "cnctw"})
+        client.wait_for(
+            event_is("got-game-files", "cnctw"),
+            timeout=20,
+            description="got cnctw files",
+            waiter=waiter,
+        )
         client.wait_for(
             event_is("download-begin", "cnctw"),
             timeout=20,
@@ -1270,11 +1276,9 @@ class Runner:
                 f"streamed byte count mismatch: {streamed_bytes} != {expected_bytes}"
             )
 
-        wait_no_outbound_transfer(source, "cnctw")
-
         return (
             "cnctw streamed into local/ only; root archive and version.ini absent; "
-            f"payload hashes={actual}; source outbound transfer drained"
+            f"payload hashes={actual}"
         )
 
     def s40_streamed_receiver_not_source(self) -> str:
@@ -1316,6 +1320,12 @@ class Runner:
 
         waiter = LineWaiter(len(client.output))
         client.send({"cmd": "stream-install", "game_id": "cnctw"})
+        client.wait_for(
+            event_is("got-game-files", "cnctw"),
+            timeout=20,
+            description="got solid cnctw files",
+            waiter=waiter,
+        )
         client.wait_for(
             event_is("download-finished", "cnctw"),
             timeout=60,
@@ -1399,6 +1409,12 @@ class Runner:
 
         waiter = LineWaiter(len(client.output))
         client.send({"cmd": "stream-install", "game_id": "cnctw"})
+        client.wait_for(
+            event_is("got-game-files", "cnctw"),
+            timeout=20,
+            description="got retry cnctw files",
+            waiter=waiter,
+        )
         client.wait_for(
             event_is("download-finished", "cnctw"),
             timeout=60,
@@ -1867,20 +1883,6 @@ def wait_no_active(peer: Peer, game_id: str, timeout: float = 20) -> None:
     raise ScenarioError(f"{peer.name} still has active operation for {game_id}: {last_active}")
 
 
-def wait_no_outbound_transfer(peer: Peer, game_id: str, timeout: float = 20) -> None:
-    deadline = time.monotonic() + timeout
-    last_active: dict[str, int] = {}
-    while time.monotonic() < deadline:
-        active = peer.status()["active_outbound_transfers"]
-        last_active = active
-        if active.get(game_id, 0) == 0:
-            return
-        time.sleep(0.4)
-    raise ScenarioError(
-        f"{peer.name} still has outbound transfer for {game_id}: {last_active}"
-    )
-
-
 def assert_game_state(
     game: dict[str, Any],
     *,

crates/lanspread-peer-cli/src/main.rs

@@ -19,7 +19,6 @@ use lanspread_peer::{
     ExternalUnrarStreamProvider,
     InstallOperation,
     NoopStreamInstallProvider,
-    OutboundTransfers,
     PeerCommand,
     PeerEvent,
     PeerGameDB,
@@ -120,7 +119,6 @@ struct SharedState {
     state: RwLock<CliState>,
     peer_game_db: Arc<RwLock<PeerGameDB>>,
     catalog: Arc<RwLock<GameCatalog>>,
-    active_outbound_transfers: OutboundTransfers,
     notify: Notify,
     games_dir: PathBuf,
     state_dir: PathBuf,
@@ -139,7 +137,6 @@ async fn main() -> eyre::Result<()> {
     let (tx_events, rx_events) = mpsc::unbounded_channel();
     let peer_game_db = Arc::new(RwLock::new(PeerGameDB::new()));
     let catalog = Arc::new(RwLock::new(catalog));
-    let active_outbound_transfers: OutboundTransfers = Arc::new(RwLock::new(HashMap::new()));
     let unrar_for_streaming = args.unrar.clone().or_else(default_unrar_program);
     let unpacker: Arc<dyn lanspread_peer::Unpacker> = match args.unrar.clone() {
         Some(path) => Arc::new(ExternalUnrarUnpacker::new(path)),
@@ -158,7 +155,7 @@ async fn main() -> eyre::Result<()> {
         catalog.clone(),
         PeerStartOptions {
             state_dir: Some(args.state_dir.clone()),
-            active_outbound_transfers: Some(active_outbound_transfers.clone()),
+            active_outbound_transfers: None,
             stream_install_provider: Some(stream_install_provider),
         },
     )?;
@@ -168,7 +165,6 @@ async fn main() -> eyre::Result<()> {
         state: RwLock::new(CliState::default()),
         peer_game_db,
         catalog: catalog.clone(),
-        active_outbound_transfers,
         notify: Notify::new(),
         games_dir: args.games_dir.clone(),
         state_dir: args.state_dir.clone(),
@@ -265,6 +261,7 @@ async fn handle_command(
         CliCommand::StreamInstall { game_id } => {
             ensure_catalog_game(shared, game_id).await?;
             ensure_no_active_operation(shared, game_id).await?;
+            let _ = game_files_for_download(sender, shared, game_id).await?;
             sender.send(PeerCommand::StreamInstallGame {
                 id: game_id.clone(),
             })?;
@@ -317,20 +314,12 @@ async fn handle_command(
 async fn status(shared: &SharedState) -> eyre::Result<Value> {
     let state = shared.state.read().await;
     let peer_count = shared.peer_game_db.read().await.peer_snapshots().len();
-    let active_outbound_transfers = {
-        let active = shared.active_outbound_transfers.read().await;
-        active
-            .iter()
-            .map(|(game_id, transfers)| (game_id.clone(), transfers.len()))
-            .collect::<HashMap<_, _>>()
-    };
     Ok(json!({
         "local_peer": state.local_peer.clone(),
         "peer_count": peer_count,
         "local_games": state.local_games.len(),
         "remote_games": state.remote_games.len(),
         "active_operations": active_operations_json(&state.active_operations),
-        "active_outbound_transfers": active_outbound_transfers,
     }))
 }
 

crates/lanspread-peer/README.md

@@ -14,8 +14,8 @@ It is designed to run headless – other crates (most notably
   roots are announced or served.
 - `PeerCommand` represents the small control surface exposed to the UI layer:
   `ListGames`, `GetGame`, `FetchLatestFromPeers`, `DownloadGameFiles`,
-  `StreamInstallGame`, `InstallGame`, `UninstallGame`, `RemoveDownloadedGame`,
+  `InstallGame`, `UninstallGame`, `RemoveDownloadedGame`, `CancelDownload`,
-  `CancelDownload`, `SetGameDir`, and `GetPeerCount`.
+  `SetGameDir`, and `GetPeerCount`.
 - `PeerEvent` enumerates everything the peer runtime reports back to the UI:
   library snapshots, download/install/uninstall lifecycle updates, runtime
   failures, and peer membership changes.
@@ -28,8 +28,8 @@ lifetime of the process:
 
 1. **Server component** (`run_server_component`) – listens for QUIC connections,
    advertises via mDNS, and serves `Request::ListGames`, `Request::GetGame`,
-   `Request::GetGameFileData`, `Request::GetGameFileChunk`, and
+   `Request::GetGameFileData`, and `Request::GetGameFileChunk` by reading from
-   `Request::StreamInstall` by reading from the local game directory.
+   the local game directory.
 2. **Discovery loop** (`run_peer_discovery`) – uses the `lanspread-mdns`
    helper to discover other peers. The blocking mDNS work is executed on a
    dedicated thread via `tokio::task::spawn_blocking` so that the Tokio runtime
@@ -87,26 +87,6 @@ When the UI asks to download a game:
 7. After a successful sentinel commit, `PeerEvent::DownloadGameFilesFinished`
    is emitted and the peer auto-runs the install transaction.
 
-### Streamed Install Pipeline
-
-Low-disk installs use `PeerCommand::StreamInstallGame` instead of the normal
-archive download pipeline. The peer core owns the whole operation: it refreshes
-file metadata from catalog-version peers, runs the same majority file-size
-validation used by normal downloads, selects a validated peer list, and emits
-the regular download/install lifecycle events while streaming archive-expanded
-bytes directly into a `StreamedInstallTransaction`.
-
-The sender-side `StreamInstallProvider` writes control and chunk frames through
-a cancellable `StreamInstallFrameSink`. If the QUIC writer fails because the
-receiver cancelled or disconnected, the sink wakes any producer blocked on the
-bounded frame channel and lets the transfer guard drop normally.
-
-Each failed peer attempt rolls back its staging directory before trying the next
-validated peer. A transaction that created a previously missing game root
-removes that root again when rollback leaves it empty. Once staging has been
-renamed to `local/`, post-promote intent or launch-settings cleanup failures are
-logged for startup recovery rather than reported as a failed install.
-
 `PeerCommand::CancelDownload` cancels the tracked download token for an active
 transfer. The transfer task remains responsible for clearing `active_operations`,
 discarding partial payload files, and refreshing the settled local snapshot, so

crates/lanspread-peer/src/handlers.rs

@@ -471,6 +471,38 @@ pub async fn handle_stream_install_game_command(
         return;
     }
 
+    let expected_version = catalog_expected_version(ctx, &id).await;
+    let mut peers = {
+        match ctx
+            .peer_game_db
+            .read()
+            .await
+            .validate_file_sizes_majority(&id, expected_version.as_deref())
+        {
+            Ok((validated_files, peer_whitelist, _)) if !validated_files.is_empty() => {
+                peer_whitelist
+            }
+            Ok(_) => {
+                log::error!("No trusted peers available for streamed install of {id}");
+                send_download_failed(tx_notify_ui, &id);
+                return;
+            }
+            Err(err) => {
+                log::error!(
+                    "File size majority validation failed for streamed install {id}: {err}"
+                );
+                send_download_failed(tx_notify_ui, &id);
+                return;
+            }
+        }
+    };
+    peers.sort();
+    if peers.is_empty() {
+        log::error!("No peer selected for streamed install of {id}");
+        send_download_failed(tx_notify_ui, &id);
+        return;
+    }
+
     match begin_operation(ctx, tx_notify_ui, &id, OperationKind::Downloading).await {
         BeginOperationResult::Started => {}
         BeginOperationResult::AlreadyActive => {
@@ -484,7 +516,6 @@ pub async fn handle_stream_install_game_command(
         }
     }
 
-    let expected_version = catalog_expected_version(ctx, &id).await;
     let cancel_token = ctx.shutdown.child_token();
     ctx.active_downloads
         .write()
@@ -494,14 +525,7 @@ pub async fn handle_stream_install_game_command(
     let ctx_clone = ctx.clone();
     let tx_notify_ui = tx_notify_ui.clone();
     ctx.task_tracker.spawn(async move {
-        run_stream_install_operation(
+        run_stream_install_operation(ctx_clone, tx_notify_ui, id, game_root, peers, cancel_token)
-            ctx_clone,
-            tx_notify_ui,
-            id,
-            game_root,
-            expected_version,
-            cancel_token,
-        )
             .await;
     });
 }
@@ -551,7 +575,7 @@ async fn run_stream_install_operation(
     tx_notify_ui: UnboundedSender<PeerEvent>,
     id: String,
     game_root: PathBuf,
-    expected_version: Option<String>,
+    peer_addrs: Vec<SocketAddr>,
     cancel_token: CancellationToken,
 ) {
     let download_guard = OperationGuard::download(
@@ -566,43 +590,42 @@ async fn run_stream_install_operation(
         PeerEvent::DownloadGameFilesBegin { id: id.clone() },
     );
 
-    let peer_addrs =
+    let mut last_receive_error = None;
-        match select_stream_install_peers(&ctx, &id, expected_version.as_deref(), &cancel_token)
+    for peer_addr in peer_addrs {
-            .await
+        if cancel_token.is_cancelled() {
-        {
+            last_receive_error = Some(eyre::eyre!("streamed install for {id} was cancelled"));
-            Ok(peers) => peers,
+            break;
-            Err(err) => {
-                let download_was_cancelled = cancel_token.is_cancelled();
-                if download_was_cancelled {
-                    log::info!("Streamed install preflight cancelled for {id}: {err}");
-                } else {
-                    log::error!("Streamed install preflight failed for {id}: {err}");
         }
-                finish_failed_stream_download(
+
-                    &ctx,
+        let transaction =
-                    &tx_notify_ui,
+            match install::begin_streamed_install(&game_root, ctx.state_dir.as_ref(), &id).await {
-                    &id,
+                Ok(transaction) => transaction,
-                    download_guard,
+                Err(err) => {
-                    download_was_cancelled,
+                    log::error!("Failed to prepare streamed install for {id}: {err}");
-                )
+                    finish_failed_stream_download(&ctx, &tx_notify_ui, &id, download_guard, false)
                         .await;
                     return;
                 }
             };
 
-    match receive_streamed_install_from_peers(
+        let receive_result = receive_streamed_install(
+            peer_addr,
+            &id,
+            transaction.staging_dir(),
+            tx_notify_ui.clone(),
+            cancel_token.clone(),
+        )
+        .await;
+
+        match receive_result {
+            Ok(()) => {
+                if transition_download_to_install(
                     &ctx,
                     &tx_notify_ui,
                     &id,
-        &game_root,
+                    OperationKind::Installing,
-        &peer_addrs,
-        &cancel_token,
                 )
                 .await
-    {
-        Ok(transaction) => {
-            if transition_download_to_install(&ctx, &tx_notify_ui, &id, OperationKind::Installing)
-                .await
                 {
                     clear_active_download(&ctx, &id).await;
                     send_download_finished(&tx_notify_ui, &id);
@@ -614,60 +637,18 @@ async fn run_stream_install_operation(
                 if let Err(err) = transaction.rollback().await {
                     log::error!("Failed to roll back streamed install for {id}: {err}");
                 }
-            finish_failed_stream_download(&ctx, &tx_notify_ui, &id, download_guard, false).await;
+                finish_failed_stream_download(&ctx, &tx_notify_ui, &id, download_guard, false)
-        }
-        Err(err) => {
-            let download_was_cancelled = cancel_token.is_cancelled();
-            if download_was_cancelled {
-                log::info!("Streamed install download cancelled for {id}: {err}");
-            } else {
-                log::error!("Streamed install download failed for {id}: {err}");
-            }
-            finish_failed_stream_download(
-                &ctx,
-                &tx_notify_ui,
-                &id,
-                download_guard,
-                download_was_cancelled,
-            )
                     .await;
+                return;
             }
-    }
-}
-
-async fn receive_streamed_install_from_peers(
-    ctx: &Ctx,
-    tx_notify_ui: &UnboundedSender<PeerEvent>,
-    id: &str,
-    game_root: &Path,
-    peer_addrs: &[SocketAddr],
-    cancel_token: &CancellationToken,
-) -> eyre::Result<install::StreamedInstallTransaction> {
-    let mut last_receive_error = None;
-    for &peer_addr in peer_addrs {
-        if cancel_token.is_cancelled() {
-            eyre::bail!("streamed install for {id} was cancelled");
-        }
-
-        let transaction =
-            install::begin_streamed_install(game_root, ctx.state_dir.as_ref(), id).await?;
-        let receive_result = receive_streamed_install(
-            peer_addr,
-            id,
-            transaction.staging_dir(),
-            tx_notify_ui.clone(),
-            cancel_token.clone(),
-        )
-        .await;
-
-        match receive_result {
-            Ok(()) => return Ok(transaction),
             Err(err) => {
                 if let Err(rollback_err) = transaction.rollback().await {
                     log::error!("Failed to roll back streamed install for {id}: {rollback_err}");
                 }
                 if cancel_token.is_cancelled() {
-                    return Err(err);
+                    log::info!("Streamed install download cancelled for {id}: {err}");
+                    last_receive_error = Some(err);
+                    break;
                 }
 
                 log::warn!(
@@ -678,84 +659,24 @@ async fn receive_streamed_install_from_peers(
         }
     }
 
-    Err(last_receive_error.unwrap_or_else(|| {
+    let download_was_cancelled = cancel_token.is_cancelled();
-        eyre::eyre!("streamed install download failed for {id}: no peer attempts were made")
+    if let Some(err) = last_receive_error {
-    }))
+        if download_was_cancelled {
-}
+            log::info!("Streamed install download cancelled for {id}: {err}");
-
+        } else {
-async fn select_stream_install_peers(
+            log::error!("Streamed install download failed for {id}: {err}");
-    ctx: &Ctx,
-    id: &str,
-    expected_version: Option<&str>,
-    cancel_token: &CancellationToken,
-) -> eyre::Result<Vec<SocketAddr>> {
-    let mut metadata_peers = {
-        ctx.peer_game_db
-            .read()
-            .await
-            .peers_with_expected_version(id, expected_version)
-    };
-    metadata_peers.sort();
-    if metadata_peers.is_empty() {
-        eyre::bail!("no peers have game {id}");
         }
-
+    } else {
-    refresh_stream_install_file_details(ctx, id, &metadata_peers, cancel_token).await?;
+        log::error!("Streamed install download failed for {id}: no peer attempts were made");
-
-    let mut peers = match ctx
-        .peer_game_db
-        .read()
-        .await
-        .validate_file_sizes_majority(id, expected_version)
-    {
-        Ok((validated_files, peer_whitelist, _)) if !validated_files.is_empty() => peer_whitelist,
-        Ok(_) => {
-            eyre::bail!("no trusted peers available for streamed install of {id}");
     }
-        Err(err) => {
+    finish_failed_stream_download(
-            return Err(err.wrap_err(format!(
+        &ctx,
-                "file size majority validation failed for streamed install {id}"
+        &tx_notify_ui,
-            )));
+        &id,
-        }
+        download_guard,
-    };
+        download_was_cancelled,
-    peers.sort();
+    )
-    if peers.is_empty() {
+    .await;
-        eyre::bail!("no peer selected for streamed install of {id}");
-    }
-
-    Ok(peers)
-}
-
-async fn refresh_stream_install_file_details(
-    ctx: &Ctx,
-    id: &str,
-    peers: &[SocketAddr],
-    cancel_token: &CancellationToken,
-) -> eyre::Result<()> {
-    let mut fetched_any = false;
-    for &peer_addr in peers {
-        if cancel_token.is_cancelled() {
-            eyre::bail!("streamed install for {id} was cancelled");
-        }
-
-        match request_game_details_and_update(peer_addr, id, ctx.peer_game_db.clone()).await {
-            Ok(_) => {
-                log::info!("Fetched streamed-install file list for {id} from peer {peer_addr}");
-                fetched_any = true;
-            }
-            Err(err) => {
-                log::error!(
-                    "Failed to fetch streamed-install files for {id} from {peer_addr}: {err}"
-                );
-            }
-        }
-    }
-
-    if !fetched_any {
-        eyre::bail!("failed to retrieve game files for {id} from any peer");
-    }
-
-    Ok(())
 }
 
 async fn finish_failed_stream_download(

crates/lanspread-peer/src/install/transaction.rs

@@ -39,7 +39,6 @@ pub struct StreamedInstallTransaction {
     id: String,
     staging: PathBuf,
     eti_version: Option<String>,
-    created_game_root: bool,
 }
 
 impl StreamedInstallTransaction {
@@ -50,61 +49,40 @@ impl StreamedInstallTransaction {
 
     pub async fn commit(self) -> eyre::Result<()> {
         let local = local_dir(&self.game_root);
-        if let Err(err) = tokio::fs::rename(&self.staging, &local)
+        let result = async {
+            tokio::fs::rename(&self.staging, &local)
                 .await
-            .wrap_err_with(|| format!("failed to promote streamed install for {}", self.id))
+                .wrap_err_with(|| format!("failed to promote streamed install for {}", self.id))?;
-        {
+            reset_launch_settings_marker(&self.state_dir, &self.id).await?;
+            write_intent(
+                &self.state_dir,
+                &self.id,
+                &InstallIntent::none(&self.id, self.eti_version.clone()),
+            )
+            .await
+        }
+        .await;
+
+        if result.is_err() {
             if let Err(cleanup_err) = remove_dir_all_if_exists(&self.staging).await {
                 log::warn!(
                     "Failed to clean streamed install staging {}: {cleanup_err}",
                     self.staging.display()
                 );
             }
-            if let Err(cleanup_err) =
-                remove_created_empty_game_root(&self.game_root, self.created_game_root).await
-            {
-                log::warn!(
-                    "Failed to clean streamed install game root {}: {cleanup_err}",
-                    self.game_root.display()
-                );
-            }
             let _ = write_intent(
                 &self.state_dir,
                 &self.id,
                 &InstallIntent::none(&self.id, self.eti_version.clone()),
             )
             .await;
-            return Err(err);
         }
 
-        if let Err(err) = reset_launch_settings_marker(&self.state_dir, &self.id).await {
+        result
-            log::error!(
-                "Streamed install for {} was promoted but launch-settings marker reset failed: {err}",
-                self.id
-            );
-        }
-        if let Err(err) = write_intent(
-            &self.state_dir,
-            &self.id,
-            &InstallIntent::none(&self.id, self.eti_version.clone()),
-        )
-        .await
-        {
-            log::error!(
-                "Streamed install for {} was promoted but intent cleanup failed: {err}",
-                self.id
-            );
-        }
-
-        Ok(())
     }
 
     pub async fn rollback(self) -> eyre::Result<()> {
-        let cleanup_result = async {
+        let staging_result = remove_dir_all_if_exists(&self.staging).await;
-            remove_dir_all_if_exists(&self.staging).await?;
-            remove_created_empty_game_root(&self.game_root, self.created_game_root).await
-        }
-        .await;
         let intent_result = write_intent(
             &self.state_dir,
             &self.id,
@@ -112,7 +90,7 @@ impl StreamedInstallTransaction {
         )
         .await;
 
-        cleanup_result?;
+        staging_result?;
         intent_result
     }
 }
@@ -126,36 +104,18 @@ pub async fn begin_streamed_install(
         eyre::bail!("game {id} is already installed");
     }
 
-    let created_game_root = !path_exists(game_root).await;
     tokio::fs::create_dir_all(game_root).await?;
     let eti_version = read_downloaded_version(game_root).await;
-    if let Err(err) = write_intent(
+    write_intent(
         state_dir,
         id,
         &InstallIntent::new(id, InstallIntentState::Installing, eti_version.clone()),
     )
-    .await
+    .await?;
-    {
-        if let Err(cleanup_err) = remove_created_empty_game_root(game_root, created_game_root).await
-        {
-            log::warn!(
-                "Failed to clean streamed install game root {}: {cleanup_err}",
-                game_root.display()
-            );
-        }
-        return Err(err);
-    }
 
     let staging = installing_dir(game_root);
     if let Err(err) = prepare_owned_empty_dir(&staging).await {
         let _ = write_intent(state_dir, id, &InstallIntent::none(id, eti_version)).await;
-        if let Err(cleanup_err) = remove_created_empty_game_root(game_root, created_game_root).await
-        {
-            log::warn!(
-                "Failed to clean streamed install game root {}: {cleanup_err}",
-                game_root.display()
-            );
-        }
         return Err(err);
     }
 
@@ -167,7 +127,6 @@ pub async fn begin_streamed_install(
         id: id.to_string(),
         staging,
         eti_version,
-        created_game_root,
     })
 }
 
@@ -627,28 +586,6 @@ async fn remove_dir_all_if_exists(path: &Path) -> eyre::Result<()> {
     }
 }
 
-async fn remove_created_empty_game_root(game_root: &Path, created: bool) -> eyre::Result<()> {
-    if !created {
-        return Ok(());
-    }
-    remove_empty_dir_if_exists(game_root).await
-}
-
-async fn remove_empty_dir_if_exists(path: &Path) -> eyre::Result<()> {
-    match tokio::fs::remove_dir(path).await {
-        Ok(()) => Ok(()),
-        Err(err)
-            if matches!(
-                err.kind(),
-                ErrorKind::NotFound | ErrorKind::DirectoryNotEmpty
-            ) =>
-        {
-            Ok(())
-        }
-        Err(err) => Err(err.into()),
-    }
-}
-
 async fn path_is_dir(path: &Path) -> bool {
     tokio::fs::metadata(path)
         .await
@@ -790,74 +727,6 @@ mod tests {
         assert!(!launch_settings_applied_path(state.path(), "game").exists());
     }
 
-    #[tokio::test]
-    async fn streamed_install_rollback_removes_new_empty_game_root() {
-        let temp = TempDir::new("lanspread-install");
-        let state = test_state();
-        let root = temp.path().join("streamed-game");
-
-        let transaction = begin_streamed_install(&root, state.path(), "streamed-game")
-            .await
-            .expect("streamed transaction should begin");
-        assert!(transaction.staging_dir().is_dir());
-
-        transaction
-            .rollback()
-            .await
-            .expect("streamed rollback should succeed");
-
-        assert!(!root.exists());
-        let intent = read_intent(state.path(), "streamed-game").await;
-        assert_eq!(intent.state, InstallIntentState::None);
-    }
-
-    #[tokio::test]
-    async fn streamed_install_rollback_keeps_existing_game_root() {
-        let temp = TempDir::new("lanspread-install");
-        let state = test_state();
-        let root = temp.game_root();
-        write_file(&root.join("version.ini"), b"20250101");
-
-        let transaction = begin_streamed_install(&root, state.path(), "game")
-            .await
-            .expect("streamed transaction should begin");
-
-        transaction
-            .rollback()
-            .await
-            .expect("streamed rollback should succeed");
-
-        assert!(root.is_dir());
-        assert!(root.join("version.ini").is_file());
-        assert!(!root.join(INSTALLING_DIR).exists());
-    }
-
-    #[tokio::test]
-    async fn streamed_install_commit_succeeds_when_post_promote_intent_cleanup_fails() {
-        let temp = TempDir::new("lanspread-install");
-        let state = test_state();
-        let root = temp.game_root();
-        let transaction = begin_streamed_install(&root, state.path(), "game")
-            .await
-            .expect("streamed transaction should begin");
-        write_file(&transaction.staging_dir().join("payload.txt"), b"installed");
-
-        let game_state_dir = crate::state_paths::game_state_dir(state.path(), "game");
-        std::fs::remove_dir_all(&game_state_dir).expect("game state dir should be removed");
-        write_file(&game_state_dir, b"not a directory");
-
-        transaction
-            .commit()
-            .await
-            .expect("promoted streamed install should be reported as success");
-
-        assert_eq!(
-            std::fs::read(root.join(LOCAL_DIR).join("payload.txt"))
-                .expect("promoted payload should be present"),
-            b"installed"
-        );
-    }
-
     #[tokio::test]
     async fn install_unpacks_multiple_root_eti_archives_in_sorted_order() {
         let temp = TempDir::new("lanspread-install");

crates/lanspread-peer/src/lib.rs

@@ -80,14 +80,12 @@ use crate::{
     state_paths::resolve_state_dir,
 };
 pub use crate::{
-    context::OutboundTransfers,
     launch_settings::{LaunchSettingsOutcome, apply_launch_settings_once},
     startup::PeerRuntimeHandle,
     state_paths::{launch_settings_applied_path, setup_done_path},
     stream_install::{
         ExternalUnrarStreamProvider,
         NoopStreamInstallProvider,
-        StreamInstallFrameSink,
         StreamInstallFuture,
         StreamInstallProvider,
     },

crates/lanspread-peer/src/stream_install.rs

@@ -77,37 +77,11 @@ impl SenderArchiveIntegrity {
 
 pub type StreamInstallFuture<'a> = Pin<Box<dyn Future<Output = eyre::Result<()>> + Send + 'a>>;
 
-#[derive(Clone)]
-pub struct StreamInstallFrameSink {
-    frames: mpsc::Sender<StreamInstallFrame>,
-    cancel_token: CancellationToken,
-}
-
-impl StreamInstallFrameSink {
-    fn new(frames: mpsc::Sender<StreamInstallFrame>, cancel_token: CancellationToken) -> Self {
-        Self {
-            frames,
-            cancel_token,
-        }
-    }
-
-    pub async fn send(&self, frame: StreamInstallFrame) -> eyre::Result<()> {
-        tokio::select! {
-            () = self.cancel_token.cancelled() => {
-                eyre::bail!("streamed install frame send was cancelled");
-            }
-            result = self.frames.send(frame) => {
-                result.map_err(|_| eyre::eyre!("streamed install frame receiver closed"))
-            }
-        }
-    }
-}
-
 pub trait StreamInstallProvider: Send + Sync {
     fn stream_archive<'a>(
         &'a self,
         archive: &'a Path,
-        frames: StreamInstallFrameSink,
+        frames: mpsc::Sender<StreamInstallFrame>,
         cancel_token: CancellationToken,
     ) -> StreamInstallFuture<'a>;
 }
@@ -119,7 +93,7 @@ impl StreamInstallProvider for NoopStreamInstallProvider {
     fn stream_archive<'a>(
         &'a self,
         archive: &'a Path,
-        _frames: StreamInstallFrameSink,
+        _frames: mpsc::Sender<StreamInstallFrame>,
         _cancel_token: CancellationToken,
     ) -> StreamInstallFuture<'a> {
         Box::pin(async move {
@@ -147,7 +121,7 @@ impl StreamInstallProvider for ExternalUnrarStreamProvider {
     fn stream_archive<'a>(
         &'a self,
         archive: &'a Path,
-        frames: StreamInstallFrameSink,
+        frames: mpsc::Sender<StreamInstallFrame>,
         cancel_token: CancellationToken,
     ) -> StreamInstallFuture<'a> {
         Box::pin(async move {
@@ -158,12 +132,14 @@ impl StreamInstallProvider for ExternalUnrarStreamProvider {
                 .unwrap_or("archive.eti")
                 .to_string();
 
-            frames
+            send_stream_frame(
-                .send(StreamInstallFrame::ArchiveBegin {
+                &frames,
+                StreamInstallFrame::ArchiveBegin {
                     archive_name: archive_name.clone(),
                     solid: listing.solid,
                     unpacked_size: listing.unpacked_size(),
-                })
+                },
+            )
             .await?;
 
             stream_unrar_entries(
@@ -175,9 +151,7 @@ impl StreamInstallProvider for ExternalUnrarStreamProvider {
             )
             .await?;
 
-            frames
+            send_stream_frame(&frames, StreamInstallFrame::ArchiveEnd { archive_name }).await
-                .send(StreamInstallFrame::ArchiveEnd { archive_name })
-                .await
         })
     }
 }
@@ -294,13 +268,9 @@ fn push_rar_entry(entries: &mut Vec<RarEntry>, draft: RarEntryDraft) -> eyre::Re
             let size = draft
                 .size
                 .ok_or_else(|| eyre::eyre!("RAR file entry {relative_path} has no Size"))?;
-            let crc32 = match (size, draft.crc32) {
+            let crc32 = draft
-                (_, Some(crc32)) => crc32,
+                .crc32
-                (0, None) => 0,
+                .ok_or_else(|| eyre::eyre!("RAR file entry {relative_path} has no CRC32"))?;
-                (_, None) => {
-                    eyre::bail!("RAR file entry {relative_path} has no CRC32");
-                }
-            };
             (size, Some(crc32))
         }
         RarEntryKind::Directory => (0, None),
@@ -319,7 +289,7 @@ async fn stream_unrar_entries(
     program: &Path,
     archive: &Path,
     entries: &[RarEntry],
-    frames: &StreamInstallFrameSink,
+    frames: &mpsc::Sender<StreamInstallFrame>,
     cancel_token: CancellationToken,
 ) -> eyre::Result<()> {
     let mut child = Command::new(program)
@@ -345,22 +315,26 @@ async fn stream_unrar_entries(
 
             match entry.kind {
                 RarEntryKind::Directory => {
-                    frames
+                    send_stream_frame(
-                        .send(StreamInstallFrame::Directory {
+                        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);
                     };
-                    frames
+                    send_stream_frame(
-                        .send(StreamInstallFrame::FileBegin {
+                        frames,
+                        StreamInstallFrame::FileBegin {
                             relative_path: entry.relative_path.clone(),
                             size: entry.size,
                             crc32,
-                        })
+                        },
+                    )
                     .await?;
                     stream_unrar_file_from_stdout(
                         &mut stdout,
@@ -371,10 +345,12 @@ async fn stream_unrar_entries(
                         &cancel_token,
                     )
                     .await?;
-                    frames
+                    send_stream_frame(
-                        .send(StreamInstallFrame::FileEnd {
+                        frames,
+                        StreamInstallFrame::FileEnd {
                             relative_path: entry.relative_path.clone(),
-                        })
+                        },
+                    )
                     .await?;
                 }
             }
@@ -412,7 +388,7 @@ async fn stream_unrar_file_from_stdout(
     stdout: &mut (impl AsyncRead + Unpin),
     archive: &Path,
     entry: &RarEntry,
-    frames: &StreamInstallFrameSink,
+    frames: &mpsc::Sender<StreamInstallFrame>,
     buffer: &mut [u8],
     cancel_token: &CancellationToken,
 ) -> eyre::Result<()> {
@@ -429,10 +405,12 @@ async fn stream_unrar_file_from_stdout(
             );
         }
 
-        frames
+        send_stream_frame(
-            .send(StreamInstallFrame::FileChunk {
+            frames,
+            StreamInstallFrame::FileChunk {
                 bytes: Bytes::copy_from_slice(&buffer[..read]),
-            })
+            },
+        )
         .await?;
         remaining = remaining.saturating_sub(u64::try_from(read)?);
     }
@@ -468,6 +446,16 @@ async fn wait_unrar_child(
     }
 }
 
+async fn send_stream_frame(
+    frames: &mpsc::Sender<StreamInstallFrame>,
+    frame: StreamInstallFrame,
+) -> eyre::Result<()> {
+    frames
+        .send(frame)
+        .await
+        .map_err(|_| eyre::eyre!("streamed install frame receiver closed"))
+}
+
 pub(crate) async fn send_stream_install_error(
     tx: SendStream,
     message: impl Into<String>,
@@ -513,7 +501,6 @@ pub(crate) async fn send_game_install_stream(
 
     let (frame_tx, mut frame_rx) = mpsc::channel(FRAME_CHANNEL_DEPTH);
     let producer_cancel = cancel_token.child_token();
-    let frame_sink = StreamInstallFrameSink::new(frame_tx, producer_cancel.clone());
     let game_id_for_producer = game_id.to_string();
     let producer = tokio::spawn({
         let provider = provider.clone();
@@ -525,16 +512,16 @@ pub(crate) async fn send_game_install_stream(
                 }
 
                 if let Err(err) = provider
-                    .stream_archive(&archive, frame_sink.clone(), producer_cancel.clone())
+                    .stream_archive(&archive, frame_tx.clone(), producer_cancel.clone())
                     .await
                 {
                     let message = err.to_string();
-                    let _ = frame_sink.send(StreamInstallFrame::Error { message }).await;
+                    let _ = frame_tx.send(StreamInstallFrame::Error { message }).await;
                     return Err(err);
                 }
             }
 
-            let _ = frame_sink.send(StreamInstallFrame::Complete).await;
+            let _ = frame_tx.send(StreamInstallFrame::Complete).await;
             Ok(())
         }
     });
@@ -549,7 +536,6 @@ pub(crate) async fn send_game_install_stream(
             break;
         }
     }
-    drop(frame_rx);
 
     let close_result = framed_tx
         .close()
@@ -890,31 +876,6 @@ Details: RAR 5
         assert!(err.to_string().contains("has no CRC32"));
     }
 
-    #[test]
-    fn accepts_zero_size_unrar_file_entries_without_crc32() {
-        let listing = parse_unrar_listing(
-            r#"
-Archive: game.eti
-Details: RAR 5
-
-        Name: bin/empty.cfg
-        Type: File
-        Size: 0
-"#,
-        )
-        .expect("empty file without CRC32 should parse as CRC32 zero");
-
-        assert_eq!(
-            listing.entries,
-            vec![RarEntry {
-                relative_path: "bin/empty.cfg".to_string(),
-                kind: RarEntryKind::File,
-                size: 0,
-                crc32: Some(0),
-            }]
-        );
-    }
-
     #[test]
     fn sender_archive_integrity_accepts_matching_size_and_crc32() {
         let bytes = b"payload";

crates/lanspread-tauri-deno-ts/src-tauri/src/lib.rs

@@ -85,6 +85,7 @@ struct LanSpreadState {
     peer_runtime: Arc<RwLock<Option<PeerRuntimeHandle>>>,
     games: Arc<RwLock<GameDB>>,
     active_operations: Arc<RwLock<HashMap<String, UiOperationKind>>>,
+    pending_stream_installs: Arc<RwLock<HashSet<String>>>,
     games_folder: Arc<RwLock<String>>,
     peer_game_db: Arc<RwLock<PeerGameDB>>,
     catalog: Arc<RwLock<GameCatalog>>,
@@ -258,6 +259,16 @@ async fn install_game(
         log::warn!("Game already has an active operation: {id}");
         return Ok(false);
     }
+    if state
+        .inner()
+        .pending_stream_installs
+        .read()
+        .await
+        .contains(&id)
+    {
+        log::warn!("Game already has a pending streamed install: {id}");
+        return Ok(false);
+    }
 
     let peer_ctrl_arc = state.inner().peer_ctrl.clone();
     let peer_ctrl = peer_ctrl_arc.read().await.clone();
@@ -312,6 +323,16 @@ async fn stream_install_game(
         log::warn!("Game already has an active operation: {id}");
         return Ok(false);
     }
+    if state
+        .inner()
+        .pending_stream_installs
+        .read()
+        .await
+        .contains(&id)
+    {
+        log::warn!("Game already has a pending streamed install: {id}");
+        return Ok(false);
+    }
 
     let Some((downloaded, installed, peer_count)) = state
         .inner()
@@ -339,8 +360,19 @@ async fn stream_install_game(
         return Ok(false);
     };
 
-    if let Err(e) = peer_ctrl.send(PeerCommand::StreamInstallGame { id }) {
+    {
-        log::error!("Failed to send PeerCommand::StreamInstallGame: {e:?}");
+        let mut pending = state.inner().pending_stream_installs.write().await;
+        pending.insert(id.clone());
+    }
+
+    if let Err(e) = peer_ctrl.send(PeerCommand::GetGame(id.clone())) {
+        log::error!("Failed to send PeerCommand::GetGame for streamed install: {e:?}");
+        state
+            .inner()
+            .pending_stream_installs
+            .write()
+            .await
+            .remove(&id);
         return Ok(false);
     }
 
@@ -2060,6 +2092,7 @@ async fn handle_peer_event(app_handle: &AppHandle, event: PeerEvent) {
         }
         PeerEvent::NoPeersHaveGame { id } => {
             log::warn!("PeerEvent::NoPeersHaveGame received for {id}");
+            clear_pending_stream_install(app_handle, &id).await;
             emit_game_id_event(
                 app_handle,
                 "game-no-peers",
@@ -2098,6 +2131,7 @@ async fn handle_peer_event(app_handle: &AppHandle, event: PeerEvent) {
         }
         PeerEvent::DownloadGameFilesFailed { id } => {
             log::warn!("PeerEvent::DownloadGameFilesFailed received");
+            clear_pending_stream_install(app_handle, &id).await;
             emit_game_id_event(
                 app_handle,
                 "game-download-failed",
@@ -2107,6 +2141,7 @@ async fn handle_peer_event(app_handle: &AppHandle, event: PeerEvent) {
         }
         PeerEvent::DownloadGameFilesAllPeersGone { id } => {
             log::warn!("PeerEvent::DownloadGameFilesAllPeersGone received for {id}");
+            clear_pending_stream_install(app_handle, &id).await;
             emit_game_id_event(
                 app_handle,
                 "game-download-peers-gone",
@@ -2245,17 +2280,27 @@ async fn handle_got_game_files(
     );
 
     let state = app_handle.state::<LanSpreadState>();
+    let stream_install = state.pending_stream_installs.write().await.remove(&id);
     let peer_ctrl = state.peer_ctrl.read().await.clone();
     if let Some(peer_ctrl) = peer_ctrl
-        && let Err(e) = peer_ctrl.send(PeerCommand::DownloadGameFiles {
+        && let Err(e) = if stream_install {
+            peer_ctrl.send(PeerCommand::StreamInstallGame { id })
+        } else {
+            peer_ctrl.send(PeerCommand::DownloadGameFiles {
                 id,
                 file_descriptions,
             })
+        }
     {
         log::error!("Failed to continue queued game transfer: {e}");
     }
 }
 
+async fn clear_pending_stream_install(app_handle: &AppHandle, id: &str) {
+    let state = app_handle.state::<LanSpreadState>();
+    state.pending_stream_installs.write().await.remove(id);
+}
+
 fn handle_download_finished(app_handle: &AppHandle, id: String) {
     log::info!("PeerEvent::DownloadGameFilesFinished received");
     emit_game_id_event(