drm: bound the two waits a peer could hold open in the root service

A review pass over the privileged side, reading src/ipc/drm.rs as a local
unprivileged attacker. Two findings, both confirmed by tracing every link.

The wire had a deadline in one direction only. Every read has been bounded since
the beginning, and next_raw_into even carries the argument for it: a peer that
writes a header and then stops pins the other end forever on a readiness wait.
The write side had no deadline at all. That asymmetry costs more here, because
the parked task is in the root service: a peer that simply stops reading - a
kill -STOP on its own --server, a ptrace stop, a frozen cgroup - leaves the send
blocked inside the forward loop, so the loop top is never reached again. The
credit stall, the per-frame reauthorization and the topology-generation check
all live at that loop top, and the connection slot, the worker thread and its
DRM context stay pinned until the peer chooses to resume. drm_write_all is the
single funnel for both directions, so one deadline there covers every send; the
consumer's frame-ack write had the same shape and gets the same bound.

And drain_frame_acks looped until WouldBlock, which is a promise the peer gets
to keep. It is synchronous on the single-threaded _drm runtime, so a peer that
writes a continuous stream instead of one ack byte per frame keeps the receive
queue non-empty, never yields, and pins that thread at 100% CPU - starving every
other stream on it, which on a multi-monitor client means one connection wedging
its own siblings. Capped per call, with an early return once the credit budget
is full; anything left stays queued for the next pass.

Three comments were describing a mechanism that no longer exists. Two still said
a delivered frame drops the whole health entry, which stopped being true when
that was narrowed to zeroing the streak; the third, written in that same change,
pointed at drm_clear_prefer_cpu, a function deleted several commits earlier. The
convert verdict having no clearing site is correct and now says why: it is keyed
by connector identity, so a monitor that moves to another GPU arrives under a new
key and starts clean.

Also, the new regression test held the process-wide health mutex across its
assertions, so the one failure it exists to report would have poisoned that mutex
and buried itself under unrelated PoisonErrors in its sibling tests. It copies
the record out and releases the guard first, as the module's own helper does.

src/ipc/drm.rs

@@ -1778,6 +1778,17 @@ const MAX_DRM_JSON_BYTES: usize = 8 * 1024 * 1024;
 /// a full frame on the CPU path (33 MB at 4K) but it crosses a unix socket, so it is milliseconds in
 /// practice and this only has to bound a peer that stopped.
 const DRM_BODY_TIMEOUT_MS: u64 = 5_000;
+/// Total budget for one wire SEND. The read side has bounded every wait since the beginning, with
+/// the argument written at `next_raw_into`: a peer that stops mid-message pins the other end's
+/// thread forever on a readiness wait. The WRITE side had no deadline at all, and the asymmetry
+/// matters more here, because the parked task is in the ROOT service: a peer that simply stops
+/// reading (a `kill -STOP` on its own `--server`, a ptrace stop, a cgroup freeze) leaves the send
+/// blocked inside the forward loop, so the loop top is never reached again -- and the loop top is
+/// where the credit stall, the per-frame reauthorization and the topology-generation check live.
+/// The connection slot, the worker thread and its DRM context stay pinned until the peer chooses to
+/// resume. Generous for the same reason as the read budget: this only has to bound a peer that
+/// stopped, not pace a healthy one.
+const DRM_SEND_TIMEOUT_MS: u64 = 5_000;
 
 /// Cap on a raw body read by `next_raw_into` (CPU-fallback BGRA / cursor RGBA). Covers a 256 MiB 8K
 /// scanout (`DrmReader` bounds a frame to that) with margin.
@@ -1904,7 +1915,16 @@ async fn drm_write_all(
     mut pass_fd: Option<RawFd>,
 ) -> ResultType<()> {
     while !buf.is_empty() {
-        stream.writable().await?;
+        // Bounded, and a timeout is a hard error rather than a retry: a partial write has already
+        // desynchronized the framing and cannot be resumed, which is the same argument
+        // `next_raw_into` makes for the read side.
+        match timeout(DRM_SEND_TIMEOUT_MS, stream.writable()).await {
+            Ok(r) => r?,
+            Err(_) => bail!(
+                "drm: peer did not accept the remaining {} byte(s) within {DRM_SEND_TIMEOUT_MS}ms; closing",
+                buf.len()
+            ),
+        }
         let raw = stream.as_raw_fd();
         let chunk = buf;
         let fd_now = pass_fd;
@@ -2022,7 +2042,15 @@ impl DrmConn {
     /// Uses `&self.stream` directly, so it never conflicts with a concurrent `send_msg`/`recv_msg`.
     pub async fn send_frame_ack(&self) -> ResultType<()> {
         loop {
-            self.stream.writable().await?;
+            // Same bound as drm_write_all, same reason: an unbounded readiness wait lets a wedged
+            // peer pin this thread. Here the parked side is the consumer, so the cost is one stalled
+            // stream rather than a root slot, but the shape is identical and so is the fix.
+            match timeout(DRM_SEND_TIMEOUT_MS, self.stream.writable()).await {
+                Ok(r) => r?,
+                Err(_) => bail!(
+                    "drm: _drm frame-ack was not accepted within {DRM_SEND_TIMEOUT_MS}ms; closing"
+                ),
+            }
             match self.stream.try_write(&[1u8]) {
                 Ok(n) if n > 0 => return Ok(()),
                 // A non-empty write returning 0 means the write half is shut down (the producer is
@@ -2040,14 +2068,31 @@ impl DrmConn {
     /// (a single `try_read` that returns WouldBlock).
     pub fn drain_frame_acks(&self, credit: &mut i32, max: i32) -> ResultType<()> {
         let mut buf = [0u8; 64];
-        loop {
+        // BOUNDED, and the bound is the point: this is a synchronous loop on the single-threaded
+        // `_drm` runtime, so "drain until WouldBlock" is a promise the PEER gets to keep. A peer
+        // that writes a continuous stream instead of one ack byte per frame keeps the receive queue
+        // non-empty forever, WouldBlock never happens, and this never returns -- pinning the root
+        // service's runtime thread at 100% CPU and starving every other stream on it, which on a
+        // multi-monitor client is one connection wedging its own siblings.
+        // Credit is capped at `max` anyway, so a full budget is reached long before this cap; there
+        // is nothing to gain by reading further in one call, and whatever is left stays queued for
+        // the next pass.
+        const MAX_ACK_READS: usize = 64;
+        for _ in 0..MAX_ACK_READS {
             match self.stream.try_read(&mut buf) {
                 Ok(0) => bail!("drm: _drm frame-ack peer closed"),
-                Ok(n) => *credit = (*credit + n as i32).min(max),
+                Ok(n) => {
+                    *credit = (*credit + n as i32).min(max);
+                    // Full budget: nothing further to gain from this pass.
+                    if *credit >= max {
+                        return Ok(());
+                    }
+                }
                 Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => return Ok(()),
                 Err(e) => return Err(e.into()),
             }
         }
+        Ok(())
     }
 
     /// Producer: await until the socket is readable (a frame ack arrived, or the peer errored/closed).

src/server/drm_capturer.rs

@@ -215,8 +215,9 @@ const RAPID_REBUILD_MAX: u32 = 6;
 /// waits 30 s, is advertised online again, fails its four sessions in a few seconds and is demoted
 /// again, which is PeerInfo churn on a ~35 s cycle for as long as the process lives. Doubling turns
 /// that into a handful of retries and then near-silence, while keeping the property that made the
-/// cooldown recoverable in the first place, because the entry is dropped entirely the moment the
-/// display delivers a frame (see `frame()`), not decayed by time.
+/// cooldown recoverable in the first place, because a delivered frame zeroes the demote count (see
+/// `frame()`), not decayed by time. It zeroes ONLY the streak verdicts: the rebuild cadence and the
+/// convert verdict survive on purpose, since a first frame says nothing about either.
 fn demote_cooldown(demotes: u32) -> Duration {
     DEMOTE_COOLDOWN * (1u32 << demotes.saturating_sub(1).min(DEMOTE_BACKOFF_MAX_SHIFT))
 }
@@ -509,8 +510,13 @@ impl TraitCapturer for IpcDrmCapturer {
                     //    session. Worse, the set happens on the recv thread and this delete on the
                     //    encoder thread, so a convert failure racing a queued frame could destroy
                     //    the bit inside the very session that learned it.
-                    // Only `drm_clear_prefer_cpu` (on a topology change, where the mapping really
-                    // can have changed) may clear the convert verdict.
+                    // Nothing clears the convert verdict, and that is deliberate: which GPU exports
+                    // a given monitor is a property of the HOST, so the bit is keyed by connector
+                    // identity (`device:name`) and follows that monitor for the process run, exactly
+                    // as its own doc says. A monitor that moves to another GPU arrives under a new
+                    // identity and starts clean, so there is nothing to invalidate on a topology
+                    // change either. (An earlier revision pointed at `drm_clear_prefer_cpu` here;
+                    // that function no longer exists.)
                     self.got_frame = true;
                     if let Some(key) = &self.connector {
                         if let Some(h) = DRM_DISPLAY_HEALTH.lock().unwrap().get_mut(key) {
@@ -1872,7 +1878,8 @@ pub(super) fn get_capturer_info(
         // Refuse a demoted display UNLESS its demotion has aged past the cooldown for its demote
         // count, in which case clear the failure streak so the display retries DRM (recoverable).
         // The demote count itself is KEPT, so a display that fails again waits twice as long; only a
-        // delivered frame erases it (frame() drops the entry outright).
+        // delivered frame erases it (frame() zeroes `demotes`; it deliberately leaves the rebuild
+        // cadence and the convert verdict alone).
         let mut map = DRM_DISPLAY_HEALTH.lock().unwrap();
         if let Some(h) = key.as_ref().and_then(|k| map.get_mut(k)) {
             if h.zero_frame_streak >= DRM_GRAB_MAX_FAILURES {
@@ -2031,8 +2038,14 @@ mod drm_capturer_tests {
         put_frame(&c, 64, 32);
         assert!(matches!(c.frame(Duration::from_millis(50)), Ok(_)));
 
-        let map = DRM_DISPLAY_HEALTH.lock().unwrap();
-        let h = map.get(key).expect("the entry must SURVIVE a delivered frame");
+        // Copy the fields out and RELEASE the guard before asserting: DRM_DISPLAY_HEALTH is
+        // process-wide, and a failing assertion while holding it poisons the mutex for every
+        // sibling test in the binary -- so the one failure this test exists to report would arrive
+        // buried under unrelated PoisonErrors. `zero_frame_streak_of` above does the same.
+        let h = {
+            let map = DRM_DISPLAY_HEALTH.lock().unwrap();
+            *map.get(key).expect("the entry must SURVIVE a delivered frame")
+        };
         assert_eq!(h.zero_frame_streak, 0, "a delivered frame refutes the zero-frame streak");
         assert_eq!(h.demotes, 0, "and the demotion count that streak drove");
         assert_eq!(