fix(drm): the startup cache warm does not wake sleeping displays (#16437)

* fix(drm): the startup cache warm does not wake sleeping displays

Every `_drm` connection ran the display wake before the service answered with
its list, so the cache warm that each --server start runs (the service restarts
the --server every hour) woke displays that had idled off, with no client
connected (rustdesk#16356).

The --server now opens the handshake with DrmHello{wake}, and the service wakes
and settles only when asked. The warm asks without a wake; the availability
probes, the login refresh and the capture stream keep it, since a client is
waiting for them. Without a wake the service still clears the wake latch of a
connector it sees lit, as the warm did.

A warm that finds nothing lit leaves the cache without a verdict, where its own
wake used to make it Available. So a login that the DRM path can serve now
probes on a missing verdict before anything reads the cache (the greeter check,
the PipeWire prompt, the display list). That probe wakes the displays as the
warm did, but only when a client logs in.

* fix(drm): throttle the log of a peer that does not open with DrmHello

AGENTS.md: a log call that can fire at a rate a peer controls must not use
debug or higher unthrottled. Same 5 s interval as the rejected-IPC logs in
ipc/auth.rs.

* fix(drm): the startup warm keeps a list a login published while it queried

Per review: the warm samples the state generation before its query and
publishes only if no other probe published since, the check the two
background refreshes and the login refresh already use.

* fix(drm): log a failed login settlement task instead of dropping its error

Per review: the JoinError of the blocking task was discarded with `let _ =`.

src/ipc.rs

@@ -513,13 +513,17 @@ pub enum Data {
     CmWindowClosed,
     // --- DRM/KMS capture (opt-in `drm` feature) over the `_drm` service-scoped channel ---
     // All of the following are `cfg(all(linux, drm))`, so the drm-off IPC wire is byte-identical
-    // to upstream. Protocol on `_drm`: on connect the root service sends `DrmDisplayList`, the
+    // to upstream. `_drm` protocol: the client sends `DrmHello`, the service `DrmDisplayList`, the
     // client replies `DrmStart{display}`, then the service streams `DrmFrame` + send_raw(BGRA) and
     // `DrmCursor` + send_raw(RGBA). A frame/cursor header is ALWAYS immediately followed by exactly
     // one `send_raw()` payload (the same header-then-raw pairing as `FileBlockFromCM`). This keeps
     // the header extensible. The zero-copy `DrmFrameDmabuf(DmabufDesc)` sibling below carries only a
     // small JSON metadata descriptor; the scanout dma-buf fd rides an SCM_RIGHTS ancillary message on
     // the same `DrmConn` send (see `DrmConn::send_msg`), so it has NO trailing `send_raw()` body.
+    /// Client -> service, first on `_drm`: whether the service may wake sleeping displays before
+    /// it answers. The cache warm at every `--server` start says no: no peer is waiting for it.
+    #[cfg(all(target_os = "linux", feature = "drm"))]
+    DrmHello { wake: bool },
     /// Client -> service: begin streaming the chosen display.
     #[cfg(all(target_os = "linux", feature = "drm"))]
     // `need_cpu` is set by an unprivileged consumer that could not open a render-node convert context

src/ipc/drm.rs

@@ -914,6 +914,20 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
     let mut conn = dup_to_drm_conn(&stream)?;
     drop(stream);
 
+    let wake = match conn.recv_msg_timeout2(3_000).await {
+        Some(Ok((Data::DrmHello { wake }, _fd))) => wake,
+        Some(Ok((_, _fd))) => {
+            hbb_common::throttled_log!(
+                std::time::Duration::from_secs(5),
+                info,
+                "drm: peer did not open with DrmHello; closing"
+            );
+            return Ok(());
+        }
+        Some(Err(e)) => return Err(e),
+        None => return Ok(()),
+    };
+
     let (frame_tx, mut frame_rx) = tokio::sync::mpsc::channel::<DrmProducerMsg>(2);
     let (crtc_tx, crtc_rx) = std::sync::mpsc::channel::<(String, u32, bool)>();
     let stop = Arc::new(AtomicBool::new(false));
@@ -923,7 +937,7 @@ async fn handle_drm_conn(stream: Connection) -> ResultType<()> {
     let worker_gate = frames_gated.clone();
     std::thread::Builder::new()
         .name("drm-capture".into())
-        .spawn(move || drm_capture_worker(frame_tx, crtc_rx, worker_stop, worker_gate))
+        .spawn(move || drm_capture_worker(frame_tx, crtc_rx, worker_stop, worker_gate, wake))
         .map_err(|err| anyhow::anyhow!("could not spawn the drm capture worker: {err}"))?;
 
     let displays = match frame_rx.recv().await {
@@ -1108,6 +1122,7 @@ fn drm_capture_worker(
     crtc_rx: std::sync::mpsc::Receiver<(String, u32, bool)>,
     stop: std::sync::Arc<std::sync::atomic::AtomicBool>,
     frames_gated: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    wake: bool,
 ) {
     use std::sync::atomic::Ordering;
     use std::time::Duration;
@@ -1118,7 +1133,15 @@ fn drm_capture_worker(
     let t_conn = std::time::Instant::now();
 
     // Enumerate FRESH rather than serve the cache: a cached display may no longer be driven.
-    let displays = drm_enumerate_settled("a consumer connected");
+    let displays = if wake {
+        drm_enumerate_settled("a consumer connected")
+    } else {
+        let (displays, _undriven) = drm_enumerate_all_displays();
+        // Seen lit, a connector a failed wake gave up on is wakeable again.
+        #[cfg(feature = "drm-wake")]
+        drm_wakeable_undriven(&displays, &_undriven);
+        displays
+    };
     if frame_tx
         .blocking_send(DrmProducerMsg::Displays(displays))
         .is_err()

src/server/connection.rs

@@ -2027,6 +2027,10 @@ impl Connection {
         #[cfg(target_os = "linux")]
         if self.is_remote() {
             let mut msg = "".to_string();
+            #[cfg(feature = "drm")]
+            if !crate::platform::linux::is_x11_for_drm() {
+                super::drm_capturer::settle_unknown_availability().await;
+            }
             // Refuse only while nothing can capture a Wayland greeter: the DRM path can.
             if crate::platform::linux::is_login_screen_wayland() && !drm_can_serve_login_screen() {
                 msg = crate::client::LOGIN_SCREEN_WAYLAND.to_owned()

src/server/drm_capturer.rs

@@ -600,6 +600,10 @@ async fn recv_thread(
             return;
         }
     };
+    if let Err(err) = conn.send_msg(&Data::DrmHello { wake: true }, None).await {
+        let _ = tx.send(Err(err));
+        return;
+    }
     let displays = match conn.recv_msg_timeout2(DISPLAY_LIST_TIMEOUT_MS).await {
         Some(Ok((Data::DrmDisplayList(v), _fd))) => v,
         Some(Ok((other, _fd))) => {
@@ -1536,12 +1540,12 @@ const NEGATIVE_TTL: Duration = Duration::from_secs(30);
 const POSITIVE_TTL: Duration = Duration::from_secs(15);
 
 /// Runs on a throwaway thread: a nested `#[tokio::main]` panics if called from inside a runtime.
-fn query_displays() -> ResultType<Vec<DrmDisplayInfo>> {
+fn query_displays(wake: bool) -> ResultType<Vec<DrmDisplayInfo>> {
     let (tx, rx) = std::sync::mpsc::channel();
     std::thread::Builder::new()
         .name("drm-query".into())
         .spawn(move || {
-            let _ = tx.send(query_displays_async());
+            let _ = tx.send(query_displays_async(wake));
         })
         .map_err(|err| anyhow!("could not spawn the drm display query thread: {err}"))?;
     rx.recv_timeout(Duration::from_millis(HANDSHAKE_WAIT_MS))
@@ -1549,12 +1553,13 @@ fn query_displays() -> ResultType<Vec<DrmDisplayInfo>> {
 }
 
 #[tokio::main(flavor = "current_thread")]
-async fn query_displays_async() -> ResultType<Vec<DrmDisplayInfo>> {
-    query_displays_inner().await
+async fn query_displays_async(wake: bool) -> ResultType<Vec<DrmDisplayInfo>> {
+    query_displays_inner(wake).await
 }
 
-async fn query_displays_inner() -> ResultType<Vec<DrmDisplayInfo>> {
+async fn query_displays_inner(wake: bool) -> ResultType<Vec<DrmDisplayInfo>> {
     let mut conn = connect_drm(DRM_CONNECT_TIMEOUT_MS).await?;
+    conn.send_msg(&Data::DrmHello { wake }, None).await?;
     match conn.recv_msg_timeout2(DISPLAY_LIST_TIMEOUT_MS).await {
         Some(Ok((Data::DrmDisplayList(v), _fd))) => Ok(v),
         Some(Ok((other, _fd))) => Err(anyhow!("expected DrmDisplayList, got {:?}", other)),
@@ -1625,8 +1630,8 @@ pub(crate) enum Availability {
     Unsettled,
 }
 
-/// MAY BLOCK for seconds: never a routing gate, and never on the login request path — that path
-/// reads `availability_cached`. This blocking form serves the capture-side callers through
+/// MAY BLOCK for seconds: never on an async executor (a login with no verdict runs it on the
+/// blocking pool). This blocking form serves the capture-side callers through
 /// `is_available`, where waiting out a settle is acceptable.
 fn availability() -> Availability {
     let (verdict, stale_no) = {
@@ -1709,7 +1714,7 @@ pub(crate) fn availability_cached() -> Availability {
 /// Probe synchronously and publish the outcome. The caller must hold DRM_PROBE_IN_FLIGHT.
 fn probe_and_publish() -> Availability {
     let t = Instant::now();
-    let result = query_displays();
+    let result = query_displays(true);
     let mut st = DRM_STATE.lock().unwrap();
     let answer = match result {
         Ok(list) if !list.is_empty() => {
@@ -1772,7 +1777,7 @@ fn refresh_unavailable_async() {
         .name("drm-unavail-refresh".into())
         .spawn(move || {
             let _in_flight = in_flight;
-            let result = query_displays();
+            let result = query_displays(true);
             let mut st = DRM_STATE.lock().unwrap();
             if DRM_STATE_GEN.load(Ordering::Acquire) != sampled_gen {
                 return;
@@ -1816,7 +1821,7 @@ fn refresh_available_async() {
         .name("drm-avail-refresh".into())
         .spawn(move || {
             let _in_flight = in_flight;
-            let result = query_displays();
+            let result = query_displays(true);
             let mut st = DRM_STATE.lock().unwrap();
             if DRM_STATE_GEN.load(Ordering::Acquire) != sampled_gen {
                 return;
@@ -1881,21 +1886,44 @@ pub(super) fn warm_availability() {
             std::thread::sleep(Duration::from_millis(300));
             continue;
         }
-        if matches!(&*DRM_STATE.lock().unwrap(), ProbeState::Available(..)) {
-            return;
-        }
-        match query_displays() {
-            Ok(list) if !list.is_empty() => {
-                log::info!("drm: consumer cache warmed ({} displays) at startup", list.len());
-                publish_probe_state(&mut DRM_STATE.lock().unwrap(), ProbeState::Available(Instant::now(), list));
+        let sampled_gen = {
+            let st = DRM_STATE.lock().unwrap();
+            if matches!(&*st, ProbeState::Available(..)) {
                 return;
             }
-            _ => std::thread::sleep(Duration::from_millis(300)),
+            DRM_STATE_GEN.load(Ordering::Acquire)
+        };
+        match query_displays(false) {
+            Ok(list) if !list.is_empty() => {
+                let mut st = DRM_STATE.lock().unwrap();
+                // A login probe that published while this query ran keeps its newer list.
+                if DRM_STATE_GEN.load(Ordering::Acquire) == sampled_gen {
+                    log::info!("drm: consumer cache warmed ({} displays) at startup", list.len());
+                    publish_probe_state(&mut st, ProbeState::Available(Instant::now(), list));
+                }
+                return;
+            }
+            Ok(_) => {
+                log::info!("drm: no display is lit at startup; the first login probes again");
+                return;
+            }
+            Err(_) => std::thread::sleep(Duration::from_millis(300)),
         }
     }
     log::info!("drm: consumer cache warm found no producer at startup (will probe lazily)");
 }
 
+/// With no verdict yet, a login probes now instead of answering "not yet"; that may wake displays.
+/// A login that finds another probe running does not wait for it.
+pub(super) async fn settle_unknown_availability() {
+    let unknown = matches!(&*DRM_STATE.lock().unwrap(), ProbeState::Unknown);
+    if unknown {
+        if let Err(err) = tokio::task::spawn_blocking(availability).await {
+            log::warn!("drm: login availability settlement task failed: {err}");
+        }
+    }
+}
+
 /// The service holds its answer until the topology settles. Replaces only an `Available` verdict.
 pub(super) async fn refresh_displays_for_login() {
     let sampled_gen = {
@@ -1906,7 +1934,7 @@ pub(super) async fn refresh_displays_for_login() {
         DRM_STATE_GEN.load(Ordering::Acquire)
     };
     let t = Instant::now();
-    match query_displays_inner().await {
+    match query_displays_inner(true).await {
         Ok(list) if !list.is_empty() => {
             let changed = {
                 let mut st = DRM_STATE.lock().unwrap();

src/server/wayland.rs

@@ -724,9 +724,9 @@ pub(super) fn get_capturer_for_display(
     // render-node-absent seat or a convert failure on the unprivileged side) must NOT propagate out
     // and restart-loop this per-display video service. Instead fall THROUGH to PipeWire for just this
     // display; the other DRM outputs keep streaming over DRM.
-    // The ONE gate that keeps the probing form on purpose: this runs on the plain video thread,
+    // Like the login, this gate keeps the probing form on purpose: it runs on the video thread,
     // not an async executor, and it is the capture-build path, so a definitive verdict is worth
-    // seconds here. It is also what makes a cold cache recoverable at all -- warm_availability
+    // seconds here. With the login it is what makes a cold cache recoverable -- warm_availability
     // gives up after its attempts, so if EVERY gate were cache-only a --server that started
     // before the root service would never see DRM again for the rest of its life.
     #[cfg(feature = "drm")]