fix(peer): retain Stream Install egress inactivity

Keep the catalog-sized absolute provider deadline and independently reset a ten-minute no-progress deadline after each successful frame write. Large valid games can run past ten minutes, while stalled producers, readers, and FIN still release their permits promptly.

Test Plan:
- just test
- just clippy
- long-total stalled-provider and slow-reader regressions
- git diff --check

crates/lanspread-peer/src/stream_install.rs

@@ -848,11 +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_with_timeout(
+        let egress = forward_stream_install_frames_with_deadlines(
             &mut framed_tx,
             &mut frame_rx,
             &producer_cancel,
             egress_timeout,
+            STREAM_INSTALL_INACTIVITY_TIMEOUT,
         );
         tokio::pin!(egress);
 
@@ -1011,18 +1012,42 @@ async fn forward_stream_install_frames_with_timeout<W>(
 where
     W: tokio::io::AsyncWrite + Unpin,
 {
-    // Producer waits, every frame write, and the final FIN share this deadline.
+    forward_stream_install_frames_with_deadlines(
-    // Successfully forwarding a frame never renews the transfer authority.
+        framed_tx,
-    let deadline = TokioInstant::now() + egress_timeout;
+        frame_rx,
+        cancel_token,
+        egress_timeout,
+        egress_timeout,
+    )
+    .await
+}
+
+async fn forward_stream_install_frames_with_deadlines<W>(
+    framed_tx: &mut FramedWrite<W, LengthDelimitedCodec>,
+    frame_rx: &mut mpsc::Receiver<StreamInstallFrame>,
+    cancel_token: &CancellationToken,
+    total_timeout: Duration,
+    inactivity_timeout: Duration,
+) -> StreamInstallEgressOutcome
+where
+    W: tokio::io::AsyncWrite + Unpin,
+{
+    let total_deadline = TokioInstant::now() + total_timeout;
+    let mut inactivity_deadline = TokioInstant::now() + inactivity_timeout;
     loop {
         let frame = tokio::select! {
             biased;
             () = cancel_token.cancelled() => {
                 return StreamInstallEgressOutcome::Cancelled;
             }
-            () = time::sleep_until(deadline) => {
+            () = time::sleep_until(total_deadline) => {
                 return StreamInstallEgressOutcome::Failed(eyre::eyre!(
-                    "streamed install producer timed out after {egress_timeout:?} without a frame"
+                    "streamed install producer timed out after {total_timeout:?} without a frame"
+                ));
+            }
+            () = time::sleep_until(inactivity_deadline) => {
+                return StreamInstallEgressOutcome::Failed(eyre::eyre!(
+                    "streamed install producer timed out after {inactivity_timeout:?} without a frame"
                 ));
             }
             frame = frame_rx.recv() => frame,
@@ -1036,9 +1061,14 @@ where
             () = cancel_token.cancelled() => {
                 return StreamInstallEgressOutcome::Cancelled;
             }
-            () = time::sleep_until(deadline) => {
+            () = time::sleep_until(total_deadline) => {
                 return StreamInstallEgressOutcome::Failed(eyre::eyre!(
-                    "streamed install frame send timed out after {egress_timeout:?}"
+                    "streamed install frame send timed out after {total_timeout:?}"
+                ));
+            }
+            () = time::sleep_until(inactivity_deadline) => {
+                return StreamInstallEgressOutcome::Failed(eyre::eyre!(
+                    "streamed install frame send timed out after {inactivity_timeout:?}"
                 ));
             }
             result = framed_tx.send(frame.encode()) => result,
@@ -1048,14 +1078,20 @@ where
                 "failed to send streamed install frame: {err}"
             ));
         }
+        inactivity_deadline = TokioInstant::now() + inactivity_timeout;
     }
 
     tokio::select! {
         biased;
         () = cancel_token.cancelled() => StreamInstallEgressOutcome::Cancelled,
-        () = time::sleep_until(deadline) => {
+        () = time::sleep_until(total_deadline) => {
             StreamInstallEgressOutcome::Failed(eyre::eyre!(
-                "streamed install close timed out after {egress_timeout:?}"
+                "streamed install close timed out after {total_timeout:?}"
+            ))
+        }
+        () = time::sleep_until(inactivity_deadline) => {
+            StreamInstallEgressOutcome::Failed(eyre::eyre!(
+                "streamed install close timed out after {inactivity_timeout:?}"
             ))
         }
         result = framed_tx.close() => match result {
@@ -2839,10 +2875,11 @@ mod tests {
         let (_frame_tx, mut frame_rx) = mpsc::channel(1);
         let mut framed_tx = FramedWrite::new(tokio::io::sink(), LengthDelimitedCodec::new());
 
-        let outcome = forward_stream_install_frames_with_timeout(
+        let outcome = forward_stream_install_frames_with_deadlines(
             &mut framed_tx,
             &mut frame_rx,
             &CancellationToken::new(),
+            Duration::from_hours(1),
             Duration::from_millis(25),
         )
         .await;
@@ -2913,10 +2950,11 @@ mod tests {
             LengthDelimitedCodec::new(),
         );
 
-        let outcome = forward_stream_install_frames_with_timeout(
+        let outcome = forward_stream_install_frames_with_deadlines(
             &mut framed_tx,
             &mut frame_rx,
             &CancellationToken::new(),
+            Duration::from_hours(1),
             Duration::from_millis(25),
         )
         .await;