fix(peer): size Stream Install egress deadlines

Give a provider ten setup minutes plus its catalog streamed bytes at 1 MiB/s instead of one fixed ten-minute absolute window. Large healthy games can complete while the sender still has a finite nonrenewable deadline through FIN.

Test Plan:
- just test
- just clippy
- large-provider timeout regression
- git diff --check

crates/lanspread-peer/src/stream_install.rs

@@ -71,7 +71,15 @@ const STREAM_INSTALL_MIN_BYTES_PER_SECOND: u64 = 1024 * 1024;
 const STREAM_INSTALL_TOTAL_COPY_ALLOWANCE: u64 = 2;
 
 fn stream_install_total_timeout(expected_file_bytes: u64) -> Duration {
-    let budgeted_bytes = expected_file_bytes.saturating_mul(STREAM_INSTALL_TOTAL_COPY_ALLOWANCE);
+    stream_install_sized_timeout(expected_file_bytes, STREAM_INSTALL_TOTAL_COPY_ALLOWANCE)
+}
+
+fn stream_install_egress_timeout(expected_file_bytes: u64) -> Duration {
+    stream_install_sized_timeout(expected_file_bytes, 1)
+}
+
+fn stream_install_sized_timeout(expected_file_bytes: u64, copy_allowance: u64) -> Duration {
+    let budgeted_bytes = expected_file_bytes.saturating_mul(copy_allowance);
     let transfer_seconds = budgeted_bytes.div_ceil(STREAM_INSTALL_MIN_BYTES_PER_SECOND);
     STREAM_INSTALL_TOTAL_BASE_TIMEOUT.saturating_add(Duration::from_secs(transfer_seconds))
 }
@@ -800,6 +808,16 @@ pub(crate) async fn send_game_install_stream(
         }
     };
     let (frame_tx, mut frame_rx) = mpsc::channel(FRAME_CHANNEL_DEPTH);
+    let egress_timeout = match expected_streamed_file_bytes(&manifest) {
+        Ok(bytes) => stream_install_egress_timeout(bytes),
+        Err(error) => {
+            let message = error.to_string();
+            let tx =
+                send_stream_install_error_cancellable(tx, message.clone(), game_id, &cancel_token)
+                    .await;
+            return (tx, Err(eyre::eyre!(message)));
+        }
+    };
     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();
@@ -830,7 +848,12 @@ pub(crate) async fn send_game_install_stream(
 
     let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new());
     let (egress_outcome, producer_result) = {
-        let egress = forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &producer_cancel);
+        let egress = forward_stream_install_frames_with_timeout(
+            &mut framed_tx,
+            &mut frame_rx,
+            &producer_cancel,
+            egress_timeout,
+        );
         tokio::pin!(egress);
 
         tokio::select! {
@@ -2367,6 +2390,14 @@ mod tests {
             stream_install_total_timeout(STREAM_INSTALL_MIN_BYTES_PER_SECOND + 1),
             STREAM_INSTALL_TOTAL_BASE_TIMEOUT + Duration::from_secs(3)
         );
+        assert_eq!(
+            stream_install_egress_timeout(STREAM_INSTALL_MIN_BYTES_PER_SECOND + 1),
+            STREAM_INSTALL_TOTAL_BASE_TIMEOUT + Duration::from_secs(2)
+        );
+        assert!(
+            stream_install_egress_timeout(100 * 1024 * 1024 * 1024) > Duration::from_mins(10),
+            "large healthy providers must not inherit the former ten-minute absolute cap"
+        );
     }
 
     #[tokio::test(start_paused = true)]