feat(peer): prototype streamed installs

Add a streamed-install prototype that can receive archive-derived install bytes
straight into local/ without first storing the peer-owned root archive payload.
This is intended for low-disk clients that want to install a game but opt out of
becoming a downloadable peer source for that game.

The protocol gains a current-version-only StreamInstall request and framed
StreamInstallFrame responses. The peer core owns the generic transport,
transaction, path validation, size checks, CRC32 verification, and lifecycle
state. The archive-specific work is hidden behind StreamInstallProvider so the
prototype can use unrar while the final implementation can swap in a better
provider without rewriting the peer command path.

The receiver writes into .local.installing and only promotes to local/ after the
full stream verifies. It deliberately does not write the root version.ini or
archive files, so the settled local state is installed=true, downloaded=false,
and availability=LocalOnly. That preserves the existing rule that local/ is not
served to peers and makes streamed receivers non-sources by construction.

The CLI is the only caller for now. It exposes stream-install and provides the
prototype unrar implementation with unrar lt for entry metadata and unrar p for
file bytes. This is simple and good enough to prove non-solid archive streaming,
but it is not the production provider shape for solid archives because per-file
unrar p would repeatedly decompress prefixes. The Tauri app explicitly passes
stream_install_provider: None, so the GUI behavior stays unchanged until a real
product path is designed.

Document the production-readiness work in NEXT_STEPS.md. The main follow-up is
to make the provider abstraction final-ish and replace the per-file CLI unrar
provider with a one-pass archive provider, then wire a deliberate GUI low-disk
mode, retry semantics, and broader failure scenarios.

Test Plan:
- just fmt
- RUSTC_WRAPPER= CARGO_BUILD_RUSTC_WRAPPER= just test
- python3 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py \
  S39 S40 --build-image
- RUSTC_WRAPPER= CARGO_BUILD_RUSTC_WRAPPER= just clippy
- git diff --check
- git diff --cached --check

Follow-up: NEXT_STEPS.md

crates/lanspread-peer/Cargo.toml

@@ -14,6 +14,7 @@ lanspread-utils = { path = "../lanspread-utils" }
 
 # external
 bytes = { workspace = true }
+crc32fast = { workspace = true }
 eyre = { workspace = true }
 futures = { workspace = true }
 gethostname = { workspace = true }

crates/lanspread-peer/src/context.rs

@@ -6,7 +6,14 @@ use lanspread_db::db::{GameCatalog, GameDB};
 use tokio::sync::{RwLock, mpsc::UnboundedSender};
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
 
-use crate::{PeerEvent, Unpacker, events, library::LocalLibraryState, peer_db::PeerGameDB};
+use crate::{
+    PeerEvent,
+    StreamInstallProvider,
+    Unpacker,
+    events,
+    library::LocalLibraryState,
+    peer_db::PeerGameDB,
+};
 
 /// Thread-safe map of active outbound file transfers grouped by game ID.
 pub type OutboundTransfers = Arc<RwLock<HashMap<String, Vec<(u64, CancellationToken)>>>>;
@@ -38,6 +45,7 @@ pub struct Ctx {
     pub active_operations: Arc<RwLock<HashMap<String, OperationKind>>>,
     pub active_downloads: Arc<RwLock<HashMap<String, CancellationToken>>>,
     pub unpacker: Arc<dyn Unpacker>,
+    pub stream_install_provider: Arc<dyn StreamInstallProvider>,
     pub catalog: Arc<RwLock<GameCatalog>>,
     pub peer_id: Arc<String>,
     pub shutdown: CancellationToken,
@@ -57,6 +65,7 @@ pub struct PeerCtx {
     pub catalog: Arc<RwLock<GameCatalog>>,
     pub peer_id: Arc<String>,
     pub tx_notify_ui: tokio::sync::mpsc::UnboundedSender<PeerEvent>,
+    pub stream_install_provider: Arc<dyn StreamInstallProvider>,
     pub shutdown: CancellationToken,
     pub task_tracker: TaskTracker,
     pub active_outbound_transfers: OutboundTransfers,
@@ -86,6 +95,7 @@ impl Ctx {
         task_tracker: TaskTracker,
         catalog: Arc<RwLock<GameCatalog>>,
         active_outbound_transfers: OutboundTransfers,
+        stream_install_provider: Arc<dyn StreamInstallProvider>,
     ) -> Self {
         Self {
             game_dir: Arc::new(RwLock::new(game_dir)),
@@ -97,6 +107,7 @@ impl Ctx {
             active_operations: Arc::new(RwLock::new(HashMap::new())),
             active_downloads: Arc::new(RwLock::new(HashMap::new())),
             unpacker,
+            stream_install_provider,
             catalog,
             peer_id: Arc::new(peer_id),
             shutdown,
@@ -120,6 +131,7 @@ impl Ctx {
             catalog: self.catalog.clone(),
             peer_id: self.peer_id.clone(),
             tx_notify_ui,
+            stream_install_provider: self.stream_install_provider.clone(),
             shutdown: self.shutdown.clone(),
             task_tracker: self.task_tracker.clone(),
             active_outbound_transfers: self.active_outbound_transfers.clone(),

crates/lanspread-peer/src/handlers.rs

@@ -11,6 +11,7 @@ use std::{
 
 use lanspread_db::db::{GameDB, GameFileDescription};
 use tokio::sync::{RwLock, mpsc::UnboundedSender};
+use tokio_util::sync::CancellationToken;
 
 use crate::{
     InstallOperation,
@@ -33,6 +34,7 @@ use crate::{
     peer_db::PeerGameDB,
     remote_peer::ensure_peer_id_for_addr,
     services::{HandshakeCtx, perform_handshake_with_peer},
+    stream_install::receive_streamed_install,
 };
 
 // =============================================================================
@@ -450,6 +452,91 @@ pub async fn handle_install_game_command(
     spawn_install_operation(ctx, tx_notify_ui, id);
 }
 
+pub async fn handle_stream_install_game_command(
+    ctx: &Ctx,
+    tx_notify_ui: &UnboundedSender<PeerEvent>,
+    id: String,
+) {
+    if !catalog_contains(ctx, &id).await {
+        log::warn!("Ignoring streamed install command for non-catalog game {id}");
+        send_download_failed(tx_notify_ui, &id);
+        return;
+    }
+
+    let games_folder = { ctx.game_dir.read().await.clone() };
+    let game_root = games_folder.join(&id);
+    if local_dir_is_directory(&game_root).await {
+        log::warn!("Ignoring streamed install command for already-installed game {id}");
+        send_download_failed(tx_notify_ui, &id);
+        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();
+    let Some(peer_addr) = peers.into_iter().next() else {
+        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 => {
+            log::warn!("Operation for {id} already in progress; ignoring streamed install request");
+            return;
+        }
+        BeginOperationResult::DrainTimedOut => {
+            log::error!("Timed out waiting for outbound transfers before streamed install of {id}");
+            send_download_failed(tx_notify_ui, &id);
+            return;
+        }
+    }
+
+    let cancel_token = ctx.shutdown.child_token();
+    ctx.active_downloads
+        .write()
+        .await
+        .insert(id.clone(), cancel_token.clone());
+
+    let ctx_clone = ctx.clone();
+    let tx_notify_ui = tx_notify_ui.clone();
+    ctx.task_tracker.spawn(async move {
+        run_stream_install_operation(
+            ctx_clone,
+            tx_notify_ui,
+            id,
+            game_root,
+            peer_addr,
+            cancel_token,
+        )
+        .await;
+    });
+}
+
 /// Handles the `UninstallGame` command.
 pub async fn handle_uninstall_game_command(
     ctx: &Ctx,
@@ -490,6 +577,151 @@ pub async fn handle_cancel_download_command(
     cancel_token.cancel();
 }
 
+async fn run_stream_install_operation(
+    ctx: Ctx,
+    tx_notify_ui: UnboundedSender<PeerEvent>,
+    id: String,
+    game_root: PathBuf,
+    peer_addr: SocketAddr,
+    cancel_token: CancellationToken,
+) {
+    let download_guard = OperationGuard::download(
+        id.clone(),
+        ctx.active_operations.clone(),
+        ctx.active_downloads.clone(),
+        tx_notify_ui.clone(),
+    );
+
+    events::send(
+        &tx_notify_ui,
+        PeerEvent::DownloadGameFilesBegin { id: id.clone() },
+    );
+
+    let transaction = match install::begin_streamed_install(&game_root, ctx.state_dir.as_ref(), &id)
+        .await
+    {
+        Ok(transaction) => transaction,
+        Err(err) => {
+            log::error!("Failed to prepare streamed install for {id}: {err}");
+            finish_failed_stream_download(&ctx, &tx_notify_ui, &id, download_guard, false).await;
+            return;
+        }
+    };
+
+    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, OperationKind::Installing)
+                .await
+            {
+                clear_active_download(&ctx, &id).await;
+                send_download_finished(&tx_notify_ui, &id);
+                download_guard.disarm();
+                commit_streamed_install(&ctx, &tx_notify_ui, id, transaction).await;
+            } else {
+                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;
+            }
+        }
+        Err(err) => {
+            if let Err(rollback_err) = transaction.rollback().await {
+                log::error!("Failed to roll back streamed install for {id}: {rollback_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;
+        }
+    }
+}
+
+async fn finish_failed_stream_download(
+    ctx: &Ctx,
+    tx_notify_ui: &UnboundedSender<PeerEvent>,
+    id: &str,
+    guard: OperationGuard,
+    cancelled: bool,
+) {
+    if let Err(err) = refresh_local_game_for_ending_operation(ctx, tx_notify_ui, id).await {
+        log::error!("Failed to refresh local library after streamed install failure: {err}");
+    }
+    end_download_operation(ctx, tx_notify_ui, id).await;
+    guard.disarm();
+    send_download_failed_unless_cancelled(tx_notify_ui, id, cancelled);
+}
+
+async fn commit_streamed_install(
+    ctx: &Ctx,
+    tx_notify_ui: &UnboundedSender<PeerEvent>,
+    id: String,
+    transaction: install::StreamedInstallTransaction,
+) {
+    let operation_guard = OperationGuard::new(
+        id.clone(),
+        ctx.active_operations.clone(),
+        tx_notify_ui.clone(),
+    );
+    events::send(
+        tx_notify_ui,
+        PeerEvent::InstallGameBegin {
+            id: id.clone(),
+            operation: InstallOperation::Installing,
+        },
+    );
+
+    match transaction.commit().await {
+        Ok(()) => {
+            if let Err(err) = refresh_local_game_for_ending_operation(ctx, tx_notify_ui, &id).await
+            {
+                log::error!("Failed to refresh local library after streamed install: {err}");
+            }
+            end_operation(ctx, tx_notify_ui, &id).await;
+            operation_guard.disarm();
+            events::send(
+                tx_notify_ui,
+                PeerEvent::InstallGameFinished { id: id.clone() },
+            );
+        }
+        Err(err) => {
+            log::error!("Streamed install commit failed for {id}: {err}");
+            if let Err(refresh_err) =
+                refresh_local_game_for_ending_operation(ctx, tx_notify_ui, &id).await
+            {
+                log::error!(
+                    "Failed to refresh local library after streamed install commit failure: {refresh_err}"
+                );
+            }
+            end_operation(ctx, tx_notify_ui, &id).await;
+            operation_guard.disarm();
+            events::send(
+                tx_notify_ui,
+                PeerEvent::InstallGameFailed { id: id.clone() },
+            );
+        }
+    }
+}
+
 fn spawn_install_operation(ctx: &Ctx, tx_notify_ui: &UnboundedSender<PeerEvent>, id: String) {
     let ctx = ctx.clone();
     let tx_notify_ui = tx_notify_ui.clone();
@@ -1264,6 +1496,7 @@ mod tests {
             TaskTracker::new(),
             Arc::new(RwLock::new(GameCatalog::from_ids(["game".to_string()]))),
             Arc::new(RwLock::new(HashMap::new())),
+            Arc::new(crate::NoopStreamInstallProvider),
         )
     }
 

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

@@ -4,5 +4,13 @@ mod transaction;
 pub mod unpack;
 
 pub use remove::remove_downloaded;
-pub use transaction::{install, recover_on_startup, uninstall, update};
+pub(crate) use transaction::root_eti_archives;
+pub use transaction::{
+    StreamedInstallTransaction,
+    begin_streamed_install,
+    install,
+    recover_on_startup,
+    uninstall,
+    update,
+};
 pub use unpack::{UnpackFuture, Unpacker};

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

@@ -33,6 +33,103 @@ struct InstallFsState {
     backup: FsEntryState,
 }
 
+pub struct StreamedInstallTransaction {
+    game_root: PathBuf,
+    state_dir: PathBuf,
+    id: String,
+    staging: PathBuf,
+    eti_version: Option<String>,
+}
+
+impl StreamedInstallTransaction {
+    #[must_use]
+    pub fn staging_dir(&self) -> &Path {
+        &self.staging
+    }
+
+    pub async fn commit(self) -> eyre::Result<()> {
+        let local = local_dir(&self.game_root);
+        let result = async {
+            tokio::fs::rename(&self.staging, &local)
+                .await
+                .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()
+                );
+            }
+            let _ = write_intent(
+                &self.state_dir,
+                &self.id,
+                &InstallIntent::none(&self.id, self.eti_version.clone()),
+            )
+            .await;
+        }
+
+        result
+    }
+
+    pub async fn rollback(self) -> eyre::Result<()> {
+        let staging_result = remove_dir_all_if_exists(&self.staging).await;
+        let intent_result = write_intent(
+            &self.state_dir,
+            &self.id,
+            &InstallIntent::none(&self.id, self.eti_version.clone()),
+        )
+        .await;
+
+        staging_result?;
+        intent_result
+    }
+}
+
+pub async fn begin_streamed_install(
+    game_root: &Path,
+    state_dir: &Path,
+    id: &str,
+) -> eyre::Result<StreamedInstallTransaction> {
+    if path_is_dir(&local_dir(game_root)).await {
+        eyre::bail!("game {id} is already installed");
+    }
+
+    tokio::fs::create_dir_all(game_root).await?;
+    let eti_version = read_downloaded_version(game_root).await;
+    write_intent(
+        state_dir,
+        id,
+        &InstallIntent::new(id, InstallIntentState::Installing, eti_version.clone()),
+    )
+    .await?;
+
+    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;
+        return Err(err);
+    }
+
+    let staging = tokio::fs::canonicalize(&staging).await.unwrap_or(staging);
+
+    Ok(StreamedInstallTransaction {
+        game_root: game_root.to_path_buf(),
+        state_dir: state_dir.to_path_buf(),
+        id: id.to_string(),
+        staging,
+        eti_version,
+    })
+}
+
 pub async fn install(
     game_root: &Path,
     state_dir: &Path,
@@ -258,7 +355,7 @@ async fn unpack_archives(
     Ok(())
 }
 
-async fn root_eti_archives(game_root: &Path) -> eyre::Result<Vec<PathBuf>> {
+pub(crate) async fn root_eti_archives(game_root: &Path) -> eyre::Result<Vec<PathBuf>> {
     let mut entries = tokio::fs::read_dir(game_root).await?;
     let mut archives = Vec::new();
     while let Some(entry) = entries.next_entry().await? {

crates/lanspread-peer/src/lib.rs

@@ -32,6 +32,7 @@ mod remote_peer;
 mod services;
 mod startup;
 mod state_paths;
+mod stream_install;
 #[cfg(test)]
 mod test_support;
 
@@ -82,6 +83,7 @@ pub use crate::{
     launch_settings::{LaunchSettingsOutcome, apply_launch_settings_once},
     startup::PeerRuntimeHandle,
     state_paths::{launch_settings_applied_path, setup_done_path},
+    stream_install::{NoopStreamInstallProvider, StreamInstallFuture, StreamInstallProvider},
 };
 
 // =============================================================================
@@ -243,6 +245,8 @@ pub enum PeerCommand {
         file_descriptions: Vec<GameFileDescription>,
         install_after_download: bool,
     },
+    /// Stream archive-expanded bytes directly into `local/` without keeping root archives.
+    StreamInstallGame { id: String },
     /// Install already-downloaded archives into `local/`.
     InstallGame { id: String },
     /// Remove only the `local/` install for a game.
@@ -260,11 +264,29 @@ pub enum PeerCommand {
 }
 
 /// Optional startup settings for non-GUI callers and tests.
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Default)]
 pub struct PeerStartOptions {
     /// Directory used for peer identity and other state.
     pub state_dir: Option<PathBuf>,
     pub active_outbound_transfers: Option<crate::context::OutboundTransfers>,
+    /// Provider used to stream archive entries for low-disk streamed installs.
+    pub stream_install_provider: Option<Arc<dyn StreamInstallProvider>>,
+}
+
+impl std::fmt::Debug for PeerStartOptions {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("PeerStartOptions")
+            .field("state_dir", &self.state_dir)
+            .field(
+                "active_outbound_transfers",
+                &self.active_outbound_transfers.as_ref().map(|_| "..."),
+            )
+            .field(
+                "stream_install_provider",
+                &self.stream_install_provider.as_ref().map(|_| "..."),
+            )
+            .finish()
+    }
 }
 
 // =============================================================================
@@ -314,11 +336,14 @@ pub fn start_peer_with_options(
     let PeerStartOptions {
         state_dir,
         active_outbound_transfers,
+        stream_install_provider,
     } = options;
     let state_dir = resolve_state_dir(state_dir.as_deref());
     let game_dir = game_dir.into();
     let active_outbound_transfers = active_outbound_transfers
         .unwrap_or_else(|| Arc::new(RwLock::new(std::collections::HashMap::new())));
+    let stream_install_provider =
+        stream_install_provider.unwrap_or_else(|| Arc::new(NoopStreamInstallProvider));
     log::info!(
         "Starting peer system with game directory: {}",
         game_dir.display()
@@ -338,6 +363,7 @@ pub fn start_peer_with_options(
         unpacker,
         catalog,
         active_outbound_transfers,
+        stream_install_provider,
     ))
 }
 
@@ -355,6 +381,7 @@ async fn run_peer(
     task_tracker: TaskTracker,
     catalog: Arc<RwLock<GameCatalog>>,
     active_outbound_transfers: crate::context::OutboundTransfers,
+    stream_install_provider: Arc<dyn StreamInstallProvider>,
 ) -> eyre::Result<()> {
     let ctx = Ctx::new(
         peer_game_db,
@@ -366,6 +393,7 @@ async fn run_peer(
         task_tracker,
         catalog,
         active_outbound_transfers,
+        stream_install_provider,
     );
     if let Err(err) = load_local_library(&ctx, &tx_notify_ui).await {
         log::error!("Failed to load initial local game database: {err}");
@@ -439,6 +467,9 @@ async fn handle_peer_commands(
                 )
                 .await;
             }
+            PeerCommand::StreamInstallGame { id } => {
+                handlers::handle_stream_install_game_command(ctx, tx_notify_ui, id).await;
+            }
             PeerCommand::InstallGame { id } => {
                 handle_install_game_command(ctx, tx_notify_ui, id).await;
             }

crates/lanspread-peer/src/services/handshake.rs

@@ -319,6 +319,7 @@ mod tests {
             TaskTracker::new(),
             Arc::new(RwLock::new(catalog)),
             Arc::new(RwLock::new(HashMap::new())),
+            Arc::new(crate::NoopStreamInstallProvider),
         );
         *ctx.local_peer_addr.write().await = Some(addr([127, 0, 0, 1], 4000));
 

crates/lanspread-peer/src/services/local_monitor.rs

@@ -384,6 +384,7 @@ mod tests {
             TaskTracker::new(),
             Arc::new(RwLock::new(catalog)),
             Arc::new(RwLock::new(std::collections::HashMap::new())),
+            Arc::new(crate::NoopStreamInstallProvider),
         )
     }
 

crates/lanspread-peer/src/services/stream.rs

@@ -15,6 +15,7 @@ use crate::{
     local_games::{get_game_file_descriptions, is_local_dir_name, local_download_matches_catalog},
     peer::{send_game_file_chunk, send_game_file_data},
     services::handshake::{HandshakeCtx, accept_inbound_hello, spawn_library_resync},
+    stream_install::{send_game_install_stream, send_stream_install_error},
 };
 
 type ResponseWriter = FramedWrite<SendStream, LengthDelimitedCodec>;
@@ -99,6 +100,9 @@ async fn dispatch_request(
         } => {
             handle_file_chunk_request(ctx, game_id, relative_path, offset, length, framed_tx).await
         }
+        Request::StreamInstall { game_id } => {
+            handle_stream_install_request(ctx, game_id, framed_tx).await
+        }
         Request::Goodbye { peer_id } => {
             handle_goodbye(ctx, remote_addr, peer_id).await;
             framed_tx
@@ -386,6 +390,49 @@ async fn handle_file_chunk_request(
     FramedWrite::new(tx, LengthDelimitedCodec::new())
 }
 
+async fn handle_stream_install_request(
+    ctx: &PeerCtx,
+    game_id: String,
+    framed_tx: ResponseWriter,
+) -> ResponseWriter {
+    log::info!("Received StreamInstall request for {game_id} from peer");
+
+    let (guard, cancel_token) = TransferGuard::new(
+        game_id.clone(),
+        ctx.active_outbound_transfers.clone(),
+        ctx.tx_notify_ui.clone(),
+        &ctx.shutdown,
+    )
+    .await;
+
+    let mut tx = framed_tx.into_inner();
+    let game_dir = ctx.game_dir.read().await.clone();
+    if !can_serve_game(ctx, &game_dir, &game_id).await {
+        log::info!(
+            "Declining StreamInstall for {game_id} because the game is not currently transferable"
+        );
+        tx = send_stream_install_error(tx, format!("game {game_id} is not transferable")).await;
+        drop(guard);
+        return FramedWrite::new(tx, LengthDelimitedCodec::new());
+    }
+
+    let game_root = game_dir.join(&game_id);
+    let (returned_tx, result) = send_game_install_stream(
+        ctx.stream_install_provider.clone(),
+        tx,
+        &game_root,
+        &game_id,
+        cancel_token,
+    )
+    .await;
+    if let Err(err) = result {
+        log::warn!("StreamInstall for {game_id} ended with error: {err}");
+    }
+
+    drop(guard);
+    FramedWrite::new(returned_tx, LengthDelimitedCodec::new())
+}
+
 async fn handle_goodbye(ctx: &PeerCtx, _remote_addr: Option<SocketAddr>, peer_id: String) {
     log::info!("Received Goodbye from peer {peer_id}");
     let removed = { ctx.peer_game_db.write().await.remove_peer(&peer_id) };
@@ -442,6 +489,7 @@ mod tests {
             TaskTracker::new(),
             Arc::new(RwLock::new(catalog)),
             Arc::new(RwLock::new(std::collections::HashMap::new())),
+            Arc::new(crate::NoopStreamInstallProvider),
         )
         .to_peer_ctx(tx_notify_ui)
     }

crates/lanspread-peer/src/startup.rs

@@ -23,6 +23,7 @@ use crate::{
     PeerCommand,
     PeerEvent,
     PeerRuntimeComponent,
+    StreamInstallProvider,
     Unpacker,
     context::Ctx,
     events,
@@ -87,6 +88,7 @@ pub(crate) fn spawn_peer_runtime(
     unpacker: Arc<dyn Unpacker>,
     catalog: Arc<RwLock<GameCatalog>>,
     active_outbound_transfers: crate::context::OutboundTransfers,
+    stream_install_provider: Arc<dyn StreamInstallProvider>,
 ) -> PeerRuntimeHandle {
     let shutdown = CancellationToken::new();
     let task_tracker = TaskTracker::new();
@@ -107,6 +109,7 @@ pub(crate) fn spawn_peer_runtime(
             runtime_tracker.clone(),
             catalog,
             active_outbound_transfers,
+            stream_install_provider,
         )
         .await
         {

crates/lanspread-peer/src/stream_install.rs

@@ -0,0 +1,372 @@
+use std::{
+    future::Future,
+    net::SocketAddr,
+    path::{Path, PathBuf},
+    pin::Pin,
+    sync::Arc,
+};
+
+use bytes::Bytes;
+use crc32fast::Hasher;
+use futures::{SinkExt, StreamExt};
+use lanspread_proto::{Message, Request, StreamInstallFrame};
+use s2n_quic::stream::SendStream;
+use tokio::{
+    fs::File,
+    io::AsyncWriteExt,
+    sync::{mpsc, mpsc::UnboundedSender},
+};
+use tokio_util::{
+    codec::{FramedRead, FramedWrite, LengthDelimitedCodec},
+    sync::CancellationToken,
+};
+
+use crate::{
+    PeerEvent,
+    install::root_eti_archives,
+    network::connect_to_peer,
+    path_validation::validate_game_file_path,
+};
+
+const FRAME_CHANNEL_DEPTH: usize = 16;
+
+pub type StreamInstallFuture<'a> = Pin<Box<dyn Future<Output = eyre::Result<()>> + Send + 'a>>;
+
+pub trait StreamInstallProvider: Send + Sync {
+    fn stream_archive<'a>(
+        &'a self,
+        archive: &'a Path,
+        frames: mpsc::Sender<StreamInstallFrame>,
+        cancel_token: CancellationToken,
+    ) -> StreamInstallFuture<'a>;
+}
+
+#[derive(Debug, Default)]
+pub struct NoopStreamInstallProvider;
+
+impl StreamInstallProvider for NoopStreamInstallProvider {
+    fn stream_archive<'a>(
+        &'a self,
+        archive: &'a Path,
+        _frames: mpsc::Sender<StreamInstallFrame>,
+        _cancel_token: CancellationToken,
+    ) -> StreamInstallFuture<'a> {
+        Box::pin(async move {
+            eyre::bail!(
+                "streamed install provider is not configured for {}",
+                archive.display()
+            )
+        })
+    }
+}
+
+pub(crate) async fn send_stream_install_error(
+    tx: SendStream,
+    message: impl Into<String>,
+) -> SendStream {
+    let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new());
+    if let Err(err) = framed_tx
+        .send(
+            StreamInstallFrame::Error {
+                message: message.into(),
+            }
+            .encode(),
+        )
+        .await
+    {
+        log::warn!("Failed to send streamed install error frame: {err}");
+    }
+    if let Err(err) = framed_tx.close().await {
+        log::debug!("Failed to close streamed install error response: {err}");
+    }
+    framed_tx.into_inner()
+}
+
+pub(crate) async fn send_game_install_stream(
+    provider: Arc<dyn StreamInstallProvider>,
+    tx: SendStream,
+    game_root: &Path,
+    game_id: &str,
+    cancel_token: CancellationToken,
+) -> (SendStream, eyre::Result<()>) {
+    let archives = match root_eti_archives(game_root).await {
+        Ok(archives) => archives,
+        Err(err) => {
+            let message = err.to_string();
+            let tx = send_stream_install_error(tx, message.clone()).await;
+            return (tx, Err(eyre::eyre!(message)));
+        }
+    };
+    if archives.is_empty() {
+        let message = format!("no .eti archives found for {game_id}");
+        let tx = send_stream_install_error(tx, message.clone()).await;
+        return (tx, Err(eyre::eyre!(message)));
+    }
+
+    let (frame_tx, mut frame_rx) = mpsc::channel(FRAME_CHANNEL_DEPTH);
+    let producer_cancel = cancel_token.child_token();
+    let game_id_for_producer = game_id.to_string();
+    let producer = tokio::spawn({
+        let provider = provider.clone();
+        let producer_cancel = producer_cancel.clone();
+        async move {
+            for archive in archives {
+                if producer_cancel.is_cancelled() {
+                    eyre::bail!("streamed install for {game_id_for_producer} was cancelled");
+                }
+
+                if let Err(err) = provider
+                    .stream_archive(&archive, frame_tx.clone(), producer_cancel.clone())
+                    .await
+                {
+                    let message = err.to_string();
+                    let _ = frame_tx.send(StreamInstallFrame::Error { message }).await;
+                    return Err(err);
+                }
+            }
+
+            let _ = frame_tx.send(StreamInstallFrame::Complete).await;
+            Ok(())
+        }
+    });
+
+    let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new());
+    let mut send_result = Ok(());
+
+    while let Some(frame) = frame_rx.recv().await {
+        if let Err(err) = framed_tx.send(frame.encode()).await {
+            producer_cancel.cancel();
+            send_result = Err(eyre::eyre!("failed to send streamed install frame: {err}"));
+            break;
+        }
+    }
+
+    let close_result = framed_tx
+        .close()
+        .await
+        .map_err(|err| eyre::eyre!("failed to close streamed install stream: {err}"));
+    let tx = framed_tx.into_inner();
+    let producer_result = match producer.await {
+        Ok(result) => result,
+        Err(err) => Err(eyre::eyre!("streamed install producer task failed: {err}")),
+    };
+    let result = send_result.and(producer_result).and(close_result);
+
+    (tx, result)
+}
+
+pub(crate) async fn receive_streamed_install(
+    peer_addr: SocketAddr,
+    game_id: &str,
+    staging_dir: &Path,
+    tx_notify_ui: UnboundedSender<PeerEvent>,
+    cancel_token: CancellationToken,
+) -> eyre::Result<()> {
+    let staging_dir = tokio::fs::canonicalize(staging_dir)
+        .await
+        .unwrap_or_else(|_| staging_dir.to_path_buf());
+    let mut conn = connect_to_peer(peer_addr).await?;
+    let stream = conn.open_bidirectional_stream().await?;
+    let (rx, tx) = stream.split();
+    let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new());
+
+    framed_tx
+        .send(
+            Request::StreamInstall {
+                game_id: game_id.to_string(),
+            }
+            .encode(),
+        )
+        .await?;
+    framed_tx.close().await?;
+
+    let mut framed_rx = FramedRead::new(rx, LengthDelimitedCodec::new());
+    let mut current_file: Option<IncomingFile> = None;
+
+    loop {
+        let next = tokio::select! {
+            () = cancel_token.cancelled() => eyre::bail!("streamed install for {game_id} was cancelled"),
+            next = framed_rx.next() => next,
+        };
+
+        let Some(frame) = next else {
+            eyre::bail!("streamed install ended before Complete");
+        };
+        let frame = frame?.freeze();
+        let frame = StreamInstallFrame::decode(frame);
+
+        match frame {
+            StreamInstallFrame::ArchiveBegin {
+                archive_name,
+                solid,
+            } => {
+                log::info!(
+                    "Receiving streamed install archive {archive_name} for {game_id} (solid={solid})"
+                );
+            }
+            StreamInstallFrame::Directory { relative_path } => {
+                let path = resolve_stream_path(&staging_dir, &relative_path)?;
+                tokio::fs::create_dir_all(path).await?;
+            }
+            StreamInstallFrame::FileBegin {
+                relative_path,
+                size,
+                crc32,
+            } => {
+                if current_file.is_some() {
+                    eyre::bail!("received FileBegin for {relative_path} before previous FileEnd");
+                }
+                let path = resolve_stream_path(&staging_dir, &relative_path)?;
+                if let Some(parent) = path.parent() {
+                    tokio::fs::create_dir_all(parent).await?;
+                }
+                let file = File::create(&path).await?;
+                current_file = Some(IncomingFile::new(relative_path, path, size, crc32, file));
+            }
+            StreamInstallFrame::FileChunk { bytes } => {
+                let Some(file) = current_file.as_mut() else {
+                    eyre::bail!("received FileChunk without FileBegin");
+                };
+                file.write_chunk(game_id, peer_addr, &tx_notify_ui, bytes)
+                    .await?;
+            }
+            StreamInstallFrame::FileEnd { relative_path } => {
+                let Some(file) = current_file.take() else {
+                    eyre::bail!("received FileEnd for {relative_path} without FileBegin");
+                };
+                file.finish(&relative_path).await?;
+            }
+            StreamInstallFrame::ArchiveEnd { archive_name } => {
+                log::info!("Finished streamed install archive {archive_name} for {game_id}");
+            }
+            StreamInstallFrame::Complete => {
+                if current_file.is_some() {
+                    eyre::bail!("streamed install completed with an open file");
+                }
+                return Ok(());
+            }
+            StreamInstallFrame::Error { message } => {
+                eyre::bail!("streamed install sender failed: {message}");
+            }
+        }
+    }
+}
+
+struct IncomingFile {
+    relative_path: String,
+    path: PathBuf,
+    expected_size: u64,
+    expected_crc32: Option<u32>,
+    received: u64,
+    hasher: Hasher,
+    file: File,
+}
+
+impl IncomingFile {
+    fn new(
+        relative_path: String,
+        path: PathBuf,
+        expected_size: u64,
+        expected_crc32: Option<u32>,
+        file: File,
+    ) -> Self {
+        Self {
+            relative_path,
+            path,
+            expected_size,
+            expected_crc32,
+            received: 0,
+            hasher: Hasher::new(),
+            file,
+        }
+    }
+
+    async fn write_chunk(
+        &mut self,
+        game_id: &str,
+        peer_addr: SocketAddr,
+        tx_notify_ui: &UnboundedSender<PeerEvent>,
+        bytes: Bytes,
+    ) -> eyre::Result<()> {
+        let offset = self.received;
+        let length = u64::try_from(bytes.len())?;
+        if offset.saturating_add(length) > self.expected_size {
+            eyre::bail!(
+                "streamed file {} exceeded expected size {}",
+                self.relative_path,
+                self.expected_size
+            );
+        }
+        self.file.write_all(&bytes).await?;
+        self.hasher.update(&bytes);
+        self.received = self.received.saturating_add(length);
+
+        let _ = tx_notify_ui.send(PeerEvent::DownloadGameFileChunkFinished {
+            id: game_id.to_string(),
+            peer_addr,
+            relative_path: format!("{game_id}/local/{}", self.relative_path),
+            offset,
+            length,
+        });
+        Ok(())
+    }
+
+    async fn finish(mut self, relative_path: &str) -> eyre::Result<()> {
+        if self.relative_path != relative_path {
+            eyre::bail!(
+                "streamed file end mismatch: began {}, ended {relative_path}",
+                self.relative_path
+            );
+        }
+        self.file.flush().await?;
+
+        if self.received != self.expected_size {
+            eyre::bail!(
+                "streamed file {} size mismatch: got {}, expected {}",
+                self.relative_path,
+                self.received,
+                self.expected_size
+            );
+        }
+
+        if let Some(expected) = self.expected_crc32 {
+            let actual = self.hasher.finalize();
+            if actual != expected {
+                eyre::bail!(
+                    "streamed file {} CRC32 mismatch: got {actual:08X}, expected {expected:08X}",
+                    self.relative_path
+                );
+            }
+        }
+
+        log::debug!(
+            "Received streamed file {} -> {}",
+            self.relative_path,
+            self.path.display()
+        );
+        Ok(())
+    }
+}
+
+fn resolve_stream_path(staging_dir: &Path, relative_path: &str) -> eyre::Result<PathBuf> {
+    validate_game_file_path(staging_dir, relative_path)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_support::TempDir;
+
+    #[test]
+    fn stream_paths_stay_inside_staging_dir() {
+        let temp = TempDir::new("lanspread-stream-install-path");
+        let staging = temp.path().join("staging");
+        std::fs::create_dir_all(&staging).expect("staging should be created");
+        let staging = std::fs::canonicalize(staging).expect("staging should canonicalize");
+
+        assert!(resolve_stream_path(&staging, "bin/game.exe").is_ok());
+        assert!(resolve_stream_path(&staging, "../outside").is_err());
+        assert!(resolve_stream_path(&staging, "/absolute").is_err());
+        assert!(resolve_stream_path(&staging, "C:/windows").is_err());
+    }
+}