perf: damage-driven idle suppression + sync send buffer reuse

CHANGELOG.md

@@ -2,6 +2,15 @@
 
 ## [Unreleased]
 
+- Idle-desktop suppression: the X11 capturer subscribes a DAMAGE object
+  on the root window (`NON_EMPTY` report level, re-armed via
+  `DamageSubtract`); while the screen is still, the producer skips the
+  capture→convert→encode path entirely, polling at 25 ms and emitting a
+  refresh frame at least once a second so teardown stays prompt and the
+  delta chain stays fresh. Backends without damage tracking report
+  `changed() == true` and behave as before.
+- Sync send path: the per-chunk `vec![0; len]` allocation is now one
+  256 KiB scratch buffer per stream — zero allocation per chunk.
 - Desktop capture/encode throughput:
   - X11 capture uses MIT-SHM (`CreateSegment` fd-passing, server ≥1.2)
     when available — the pixmap lands in a shared segment instead of an

crates/rds-desktop/Cargo.toml

@@ -19,7 +19,7 @@ rds-net.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true
-x11rb = { version = "0.14", features = ["shm", "xtest", "xfixes"], optional = true }
+x11rb = { version = "0.14", features = ["shm", "xtest", "xfixes", "damage"], optional = true }
 
 [dev-dependencies]
 noq.workspace = true

crates/rds-desktop/src/capture/x11.rs

@@ -21,6 +21,8 @@ use bytes::Bytes;
 use memmap2::{MmapMut, MmapOptions};
 use rds_core::{DesktopCaps, DisplayInfo};
 use x11rb::connection::{Connection as _, RequestConnection as _};
+use x11rb::protocol::Event;
+use x11rb::protocol::damage::{self, ConnectionExt as _, Damage, ReportLevel};
 use x11rb::protocol::shm::{self, ConnectionExt as _, Seg};
 use x11rb::protocol::xproto::{ConnectionExt, GetImageReply, ImageFormat};
 use x11rb::rust_connection::RustConnection;
@@ -46,6 +48,12 @@ pub struct X11Capturer {
     height: u16,
     screen: usize,
     shm: Option<ShmPath>,
+    /// DAMAGE object on the root window — idle detection so a still
+    /// desktop costs no capture/encode work at all.
+    damage: Option<Damage>,
+    /// Sticky flag: the first `changed` call must be true (the screen
+    /// existed before the damage object did).
+    dirty: bool,
 }
 
 impl X11Capturer {
@@ -62,13 +70,16 @@ impl X11Capturer {
         );
         let _ = default_screen;
         let shm = Self::try_shm(&conn, width, height);
+        let damage = Self::try_damage(&conn, root);
         Ok(Self {
             conn,
             root,
             width,
             height,
             screen: idx,
             shm,
+            damage,
+            dirty: true,
         })
     }
 
@@ -100,6 +111,21 @@ impl X11Capturer {
             _fd: fd,
         })
     }
+
+    /// DAMAGE (XFixes ≥ 4): one object on the root window reporting
+    /// `NON_EMPTY` transitions — enough for a changed/not-changed bit.
+    /// `None` where the extension is absent.
+    fn try_damage(conn: &RustConnection, root: x11rb::protocol::xproto::Window) -> Option<Damage> {
+        conn.extension_information(damage::X11_EXTENSION_NAME)
+            .ok()??;
+        conn.damage_query_version(1, 1).ok()?.reply().ok()?;
+        let dmg = conn.generate_id().ok()?;
+        conn.damage_create(dmg, root, ReportLevel::NON_EMPTY)
+            .ok()?
+            .check()
+            .ok()?;
+        Some(dmg)
+    }
 }
 
 impl Capturer for X11Capturer {
@@ -169,10 +195,33 @@ impl Capturer for X11Capturer {
             primary: true,
         }]
     }
+
+    /// Damage-driven change detection: drains pending events; any
+    /// `DamageNotify` on our object marks the screen dirty. `NON_EMPTY`
+    /// reports only the empty→non-empty transition, so a dirty read
+    /// subtracts the region back to empty to re-arm notification.
+    fn changed(&mut self) -> bool {
+        let mut dirty = std::mem::take(&mut self.dirty);
+        let Some(dmg) = self.damage else {
+            return true;
+        };
+        while let Ok(Some(ev)) = self.conn.poll_for_event() {
+            if matches!(ev, Event::DamageNotify(e) if e.damage == dmg) {
+                dirty = true;
+            }
+        }
+        if dirty {
+            let _ = self.conn.damage_subtract(dmg, 0u32, 0u32);
+        }
+        dirty
+    }
 }
 
 impl Drop for X11Capturer {
     fn drop(&mut self) {
+        if let Some(dmg) = self.damage.take() {
+            let _ = self.conn.damage_destroy(dmg);
+        }
         if let Some(shm) = self.shm.take() {
             let _ = self.conn.shm_detach(shm.seg);
         }
@@ -192,6 +241,8 @@ pub fn capabilities() -> Result<DesktopCaps, DesktopError> {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
     use super::*;
 
     /// Real capture needs an X server — skipped silently when `$DISPLAY`
@@ -208,11 +259,24 @@ mod tests {
         };
         if local {
             assert!(cap.shm.is_some(), "local X server without MIT-SHM 1.2?");
+            assert!(cap.damage.is_some(), "local X server without DAMAGE?");
         }
         for _ in 0..3 {
             let frame = cap.capture().unwrap();
             assert_eq!(frame.data.len(), (frame.width * frame.height * 4) as usize);
             assert_eq!(frame.stride, frame.width * 4);
         }
+        if !local {
+            return;
+        }
+        // Damage bookkeeping: first read is dirty (screen predates the
+        // damage object), a still screen then reports clean, and a
+        // server-side repaint re-dirties it.
+        assert!(cap.changed(), "first changed() must be dirty");
+        assert!(!cap.changed(), "still screen reported damage");
+        x11rb::protocol::xproto::clear_area(&cap.conn, false, cap.root, 0, 0, 100, 100).unwrap();
+        cap.conn.flush().unwrap();
+        std::thread::sleep(Duration::from_millis(100));
+        assert!(cap.changed(), "root repaint produced no damage");
     }
 }

crates/rds-desktop/src/lib.rs

@@ -85,6 +85,12 @@ pub trait Capturer: Send + 'static {
     /// Next frame in display order; blocks the calling thread as needed.
     fn capture(&mut self) -> Result<RawFrame, DesktopError>;
     fn displays(&self) -> Vec<rds_core::DisplayInfo>;
+    /// Whether display content changed since the previous `capture`.
+    /// `true` is the conservative default — backends without damage
+    /// tracking always report changed.
+    fn changed(&mut self) -> bool {
+        true
+    }
 }
 
 /// Frame encoder. Implementations must emit Annex-B streams and honor

crates/rds-desktop/src/session.rs

@@ -561,13 +561,44 @@ mod x11 {
         }
     }
 
+    impl X11Producer {
+        /// Damage-aware pause: while the screen is still, capture and
+        /// encode cost nothing. Polls every `IDLE_POLL` for a damage
+        /// event, bounded by `IDLE_MAX` so the stream still emits a
+        /// refresh frame about once a second and session teardown
+        /// (the caller's `is_closed` check) is never deferred past it.
+        /// `idr` wakes the loop early: a viewer joining or recovering
+        /// from loss asks for a keyframe and must not wait out the cap.
+        fn idle_wait(&mut self, idr: &AtomicBool) {
+            const IDLE_POLL: Duration = Duration::from_millis(25);
+            const IDLE_MAX: Duration = Duration::from_secs(1);
+            if self.capturer.changed() || idr.load(Ordering::Relaxed) {
+                return;
+            }
+            let deadline = Instant::now() + IDLE_MAX;
+            loop {
+                std::thread::sleep(IDLE_POLL);
+                if self.capturer.changed()
+                    || idr.load(Ordering::Relaxed)
+                    || Instant::now() >= deadline
+                {
+                    break;
+                }
+            }
+            // Idle time is not a cadence miss: reset the schedule so
+            // the skipped slots don't count as deadline misses.
+            self.next_due = Instant::now();
+        }
+    }
+
     impl FrameProducer for X11Producer {
         fn produce(
             &mut self,
             seq: u64,
             controls: &ProducerControls,
             clock: &SessionClock,
         ) -> Option<Produced> {
+            self.idle_wait(&controls.idr);
             self.next_due += self.interval;
             if let Some(sleep) = self.next_due.checked_duration_since(Instant::now()) {
                 std::thread::sleep(sleep);

crates/rds-sync/src/engine.rs

@@ -28,7 +28,7 @@ use crate::proto::{
     CHUNKSET_BATCH, FETCH_STREAMS, MANIFEST_BATCH, MAX_CHUNKS, SyncMsg, bits_to_indices,
     check_manifest, check_rel_path, need_bits, resolve_under,
 };
-use crate::{Manifest, manifest_of_path};
+use crate::{MAX_CHUNK, Manifest, manifest_of_path};
 
 /// No protocol read may stall longer than this — a peer that is alive
 /// but silent still must not hang a transfer forever. Generous because
@@ -461,6 +461,9 @@ async fn push_chunks(
             // tokio's fs file runs every op on the blocking pool — the
             // chunk reads below never park an async worker.
             let mut file = tokio::fs::File::open(&path).await?;
+            // One scratch per stream — chunks are ≤256 KiB, so this is
+            // a single allocation rather than one per chunk.
+            let mut buf = Vec::with_capacity(MAX_CHUNK as usize);
             for batch in mine.chunks(CHUNKSET_BATCH) {
                 write_frame(
                     &mut stream,
@@ -471,7 +474,8 @@ async fn push_chunks(
                 .await?;
                 for &index in batch {
                     let c = manifest.chunks[index as usize];
-                    let mut buf = vec![0u8; c.len as usize];
+                    buf.clear();
+                    buf.resize(c.len as usize, 0);
                     file.seek(SeekFrom::Start(c.offset)).await?;
                     file.read_exact(&mut buf).await?;
                     write_frame(