fix(desktop): preserve responsiveness across idle and media pressure

crates/rds-cli/src/desktop.rs

@@ -244,10 +244,13 @@ mod native {
                 result = async {
                     match &mut source {
                         Source::Managed { client,session,peer,grant } => {
+                            view.stage("checking managed session");
                             if session.is_none() || client.selected(*session).await.is_err() {
+                                view.stage("connecting peer");
                                 let Reply::Connected(id) = client.request(Command::Connect { target:peer.clone(),grant:grant.clone() }).await? else { anyhow::bail!("unexpected local connection response"); };
                                 // After the first authenticated connect retain the exact peer
                                 // identity, even when the initial target was a registry name.
+                                view.stage("verifying peer");
                                 let snapshot = client.snapshot().await?;
                                 let authenticated = snapshot.sessions.into_iter().find(|s| s.id == id).ok_or_else(|| anyhow::anyhow!("connected session disappeared"))?.peer;
                                 *peer = reconnect_target(peer,authenticated)?;
@@ -277,6 +280,7 @@ mod native {
                 pending_input_acks = snapshot.report.pending_input_acks,
                 oldest_input_ack_age_ms = snapshot.report.oldest_input_ack_age_ms,
                 control_rtt_ms = snapshot.report.control_rtt_ms,
+                control_echo_age_ms = snapshot.control_echo_age_ms,
                 occluded = snapshot.occluded,
                 "desktop reconnecting"
             );
@@ -463,7 +467,8 @@ mod native {
         extent(view, session.caps(), options.display)?;
         view.status("Waiting for screen");
         let ctrl = session.control_sender();
-        let mut last_frame = Instant::now();
+        let mut last_frame = tokio::time::Instant::now();
+        let mut watchdog = control::VideoWatchdog::default();
         let mut tick = tokio::time::interval(Duration::from_secs(1));
         tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
         let result = loop {
@@ -477,7 +482,14 @@ mod native {
                     Some(ViewerInput::Close)|None => break Ok(true),
                 },
                 _ = tick.tick() => {
-                    anyhow::ensure!(last_frame.elapsed() < Duration::from_secs(15),"remote video stopped making progress");
+                    match watchdog.observe(last_frame) {
+                        control::VideoAction::Reconnect => anyhow::bail!("remote video stopped making progress"),
+                        control::VideoAction::Repair => {
+                            view.video_repair_requested();
+                            tokio::time::timeout(Duration::from_secs(2), ctrl.send(rds_core::DesktopControl::RequestIdr)).await.map_err(|_|anyhow::anyhow!("direct desktop repair stalled"))??;
+                        },
+                        control::VideoAction::Healthy => {},
+                    }
                     tokio::time::timeout(Duration::from_secs(2),ctrl.send(rds_core::DesktopControl::Heartbeat { seq: 0,ts_ms: started.elapsed().as_millis() as u64 })).await.map_err(|_|anyhow::anyhow!("direct desktop heartbeat stalled"))??;
                 },
                 event = session.events.recv() => match event {
@@ -487,7 +499,7 @@ mod native {
                     None => break Ok(false),
                 },
                 frame = session.frames.recv() => match frame {
-                    Some(raw) => { last_frame = Instant::now(); view.frame(raw,Instant::now()); },
+                    Some(raw) => { last_frame = tokio::time::Instant::now(); view.frame(raw,Instant::now()); },
                     None => break Ok(false),
                 },
             }

crates/rds-cli/src/desktop/control.rs

@@ -25,6 +25,60 @@ impl Sender for rds_client::local::ManagedControl {
     }
 }
 
+const VIDEO_STALL_TIMEOUT: Duration = Duration::from_secs(15);
+const REPAIR_AGES: [Duration; 2] = [Duration::from_secs(3), Duration::from_secs(8)];
+
+#[derive(Debug, PartialEq, Eq)]
+pub(super) enum VideoAction {
+    Healthy,
+    Repair,
+    Reconnect,
+}
+
+/// Idle video is expected to refresh about once a second. Request at most
+/// two independent images before reopening the desktop; heartbeats alone
+/// never reset actual decoded progress. A new frame rearms this policy.
+#[derive(Default)]
+pub(super) struct VideoWatchdog {
+    progress: Option<tokio::time::Instant>,
+    repairs: usize,
+}
+
+impl VideoWatchdog {
+    pub(super) fn observe(&mut self, progress: tokio::time::Instant) -> VideoAction {
+        if self.progress != Some(progress) {
+            self.progress = Some(progress);
+            self.repairs = 0;
+        }
+        let age = progress.elapsed();
+        if age >= VIDEO_STALL_TIMEOUT {
+            return VideoAction::Reconnect;
+        }
+        if REPAIR_AGES
+            .get(self.repairs)
+            .is_some_and(|threshold| age >= *threshold)
+        {
+            self.repairs += 1;
+            return VideoAction::Repair;
+        }
+        VideoAction::Healthy
+    }
+}
+
+async fn send_control(
+    sender: &impl Sender,
+    message: DesktopControl,
+    sent: &impl Fn(&DesktopControl),
+) -> anyhow::Result<()> {
+    sent(&message);
+    // A partially written control is never reused after timeout: the
+    // owner ends both legs and drops the channel before reconnecting.
+    tokio::time::timeout(Duration::from_secs(2), sender.send(message))
+        .await
+        .map_err(|_| anyhow::anyhow!("desktop control write stalled"))??;
+    Ok(())
+}
+
 pub(super) async fn pump(
     input: &mut impl Input,
     sender: &impl Sender,
@@ -34,30 +88,35 @@ pub(super) async fn pump(
 ) -> anyhow::Result<bool> {
     let mut tick = tokio::time::interval(Duration::from_secs(1));
     tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let mut watchdog = VideoWatchdog::default();
     loop {
         let message = tokio::select! {
-            message = input.recv() => match message {
-                Some(ViewerInput::Control(message)) => message,
-                Some(ViewerInput::Close) | None => return Ok(true),
-            },
+            biased;
             _ = tick.tick() => {
-                let age = progress.borrow().elapsed();
-                if age >= Duration::from_secs(15) {
-                    tracing::warn!(last_decoded_age_ms = age.as_millis() as u64,
-                        "desktop progress watchdog expired");
-                    anyhow::bail!("remote video stopped making progress");
+                let progress = *progress.borrow();
+                match watchdog.observe(progress) {
+                    VideoAction::Reconnect => {
+                        tracing::warn!(last_decoded_age_ms = progress.elapsed().as_millis() as u64,
+                            "desktop progress watchdog expired");
+                        anyhow::bail!("remote video stopped making progress");
+                    },
+                    VideoAction::Repair => {
+                        tracing::warn!(last_decoded_age_ms = progress.elapsed().as_millis() as u64,
+                            "desktop progress repair requested");
+                        send_control(sender, DesktopControl::RequestIdr, &sent).await?;
+                    },
+                    VideoAction::Healthy => {},
                 }
                 DesktopControl::Heartbeat {
                     seq: 0, ts_ms: started.elapsed().as_millis() as u64,
                 }
             },
+            message = input.recv() => match message {
+                Some(ViewerInput::Control(message)) => message,
+                Some(ViewerInput::Close) | None => return Ok(true),
+            },
         };
-        sent(&message);
-        // A partially written control is never reused after timeout: the
-        // owner ends both legs and drops the channel before reconnecting.
-        tokio::time::timeout(Duration::from_secs(2), sender.send(message))
-            .await
-            .map_err(|_| anyhow::anyhow!("desktop control write stalled"))??;
+        send_control(sender, message, &sent).await?;
     }
 }
 
@@ -110,6 +169,94 @@ mod tests {
         }
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn video_watchdog_repairs_before_reconnect_and_rearms_only_on_frames() {
+        let first = tokio::time::Instant::now();
+        let mut watchdog = VideoWatchdog::default();
+        assert_eq!(watchdog.observe(first), VideoAction::Healthy);
+        tokio::time::advance(Duration::from_secs(3)).await;
+        assert_eq!(watchdog.observe(first), VideoAction::Repair);
+        assert_eq!(watchdog.observe(first), VideoAction::Healthy);
+        tokio::time::advance(Duration::from_secs(5)).await;
+        assert_eq!(watchdog.observe(first), VideoAction::Repair);
+        assert_eq!(watchdog.observe(first), VideoAction::Healthy);
+        tokio::time::advance(Duration::from_secs(7)).await;
+        assert_eq!(watchdog.observe(first), VideoAction::Reconnect);
+        let decoded = tokio::time::Instant::now();
+        assert_eq!(watchdog.observe(decoded), VideoAction::Healthy);
+        tokio::time::advance(Duration::from_secs(3)).await;
+        assert_eq!(watchdog.observe(decoded), VideoAction::Repair);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn idle_video_repair_keeps_input_and_heartbeats_on_the_existing_channel() {
+        let (wire, mut remote) = tokio::io::duplex(256);
+        let (tx, mut input) = mpsc::channel(8);
+        let (progress, last_frame) = watch::channel(tokio::time::Instant::now());
+        let task = tokio::spawn(async move {
+            pump(
+                &mut input,
+                &Wire(Mutex::new(wire)),
+                &last_frame,
+                Instant::now(),
+                |_| {},
+            )
+            .await
+        });
+        let mut repaired = false;
+        let mut input_observed = false;
+        let started = tokio::time::Instant::now();
+        // Consume actual framed controls. A repair causes the simulated
+        // decoder to publish new progress on the original channel.
+        loop {
+            let DesktopUp::Control(message) = rds_net::read_frame(&mut remote).await.unwrap()
+            else {
+                panic!("unexpected control framing");
+            };
+            match message {
+                DesktopControl::RequestIdr => {
+                    assert!(!repaired, "one repair must not become an IDR storm");
+                    assert_eq!(started.elapsed(), Duration::from_secs(3));
+                    repaired = true;
+                    progress.send_replace(tokio::time::Instant::now());
+                    tx.send(ViewerInput::Control(DesktopControl::Input(InputEvent {
+                        seq: 37,
+                        event_ts_ms: 0,
+                        display_id: 0,
+                        kind: InputKind::PointerButton {
+                            button: 0x110,
+                            pressed: true,
+                        },
+                    })))
+                    .await
+                    .unwrap();
+                }
+                DesktopControl::Heartbeat { .. } if repaired => {
+                    progress.send_replace(tokio::time::Instant::now());
+                    if started.elapsed() >= Duration::from_secs(60) {
+                        break;
+                    }
+                }
+                DesktopControl::Input(event) => {
+                    assert_eq!(event.seq, 37);
+                    input_observed = true;
+                }
+                DesktopControl::Heartbeat { .. } => {}
+                _ => panic!("unexpected control"),
+            }
+        }
+        assert!(
+            !task.is_finished(),
+            "media recovery must retain the existing session"
+        );
+        assert!(
+            input_observed,
+            "repair must not consume or delay user input"
+        );
+        tx.send(ViewerInput::Close).await.unwrap();
+        assert!(task.await.unwrap().unwrap());
+    }
+
     // Read the actual framed controls, rather than only observing that an
     // input was dequeued. The media leg remains held until the session ends.
     #[tokio::test(start_paused = true)]
@@ -304,7 +451,18 @@ mod tests {
         });
         let started = tokio::time::Instant::now();
         for second in 0..31 {
-            let _: DesktopUp = rds_net::read_frame(&mut remote).await.unwrap();
+            // Count real heartbeat ticks, not all framed controls: soft
+            // repair is additional traffic, not a faster timer.
+            loop {
+                match rds_net::read_frame::<_, DesktopUp>(&mut remote)
+                    .await
+                    .unwrap()
+                {
+                    DesktopUp::Control(DesktopControl::Heartbeat { .. }) => break,
+                    DesktopUp::Control(DesktopControl::RequestIdr) => {}
+                    other => panic!("unexpected control: {other:?}"),
+                }
+            }
             if second == 10 || second == 20 {
                 progress.send_replace(tokio::time::Instant::now());
             }

crates/rds-desktop/src/render/platform.rs

@@ -14,15 +14,27 @@ impl Drop for RemoteActivity {
     }
 }
 
+#[cfg(target_os = "macos")]
+fn activity_options() -> objc2_foundation::NSActivityOptions {
+    use objc2_foundation::NSActivityOptions;
+    NSActivityOptions::NSActivityUserInitiated
+        | NSActivityOptions::NSActivityLatencyCritical
+        | NSActivityOptions::NSActivityAutomaticTerminationDisabled
+        | NSActivityOptions::NSActivitySuddenTerminationDisabled
+}
+
 #[cfg(target_os = "macos")]
 pub(super) fn remote_activity() -> RemoteActivity {
-    use objc2_foundation::{NSActivityOptions, NSProcessInfo, NSString};
+    use objc2_foundation::{NSProcessInfo, NSString};
     // SAFETY: Foundation owns the retained activity token; the RAII guard
     // pairs its begin/end. This user-requested stream must remain responsive
-    // while covered, without preventing idle system/display sleep or locking.
+    // while covered and inactive. A session-scoped activity prevents idle
+    // system sleep (network I/O cannot progress in suspend) and requests
+    // latency-critical timer/I/O precision. Display sleep, screen locking,
+    // explicit user sleep and lid-close policy remain under OS control.
     let token = unsafe {
         NSProcessInfo::processInfo().beginActivityWithOptions_reason(
-            NSActivityOptions::NSActivityUserInitiatedAllowingIdleSystemSleep,
+            activity_options(),
             &NSString::from_str("Remote desktop session"),
         )
     };
@@ -158,3 +170,15 @@ pub(super) fn activate_application() {
 
 #[cfg(not(target_os = "macos"))]
 pub(super) fn activate_application() {}
+
+#[cfg(all(test, target_os = "macos"))]
+mod tests {
+    #[test]
+    fn session_prevents_idle_system_sleep_without_holding_the_display() {
+        use objc2_foundation::NSActivityOptions;
+        let options = super::activity_options();
+        assert!(options.contains(NSActivityOptions::NSActivityIdleSystemSleepDisabled));
+        assert!(!options.contains(NSActivityOptions::NSActivityIdleDisplaySleepDisabled));
+        assert!(options.contains(NSActivityOptions::NSActivityLatencyCritical));
+    }
+}