Merge branch 'fix/round1-highs' into land/sweep-all

crates/pf-capture/src/linux/mod.rs

@@ -1310,21 +1310,26 @@ mod pipewire {
     /// (which Mutter delivers as metadata-only "corrupted" buffers) still refresh the position.
     fn update_cursor_meta(cursor: &mut CursorState, spa_buf: *mut spa::sys::spa_buffer) {
         // SAFETY: `spa_buf` is the live buffer we still hold (dequeued, not yet requeued).
-        // `spa_buffer_find_meta_data` scans its metadata array for a `SPA_META_Cursor` of at least
+        // `spa_buffer_find_meta` returns the `spa_meta` (type + byte `size` + `data` pointer) for
-        // `size_of::<spa_meta_cursor>()` bytes and returns a pointer into that buffer's metadata
+        // `SPA_META_Cursor`, or null. We take `find_meta` rather than `find_meta_data` specifically
-        // (or null), valid until requeue. The size argument matches the struct the result is cast to.
+        // to obtain the region's real `size`: the bitmap offset, pixel offset and stride read below
-        let cur = unsafe {
+        // are ALL producer-written, and without a bound against the actual region they drive
-            spa::sys::spa_buffer_find_meta_data(
+        // out-of-bounds pointer arithmetic and an oversized `slice::from_raw_parts` — an OOB read
-                spa_buf,
+        // that SIGSEGVs inside the PipeWire `.process` callback (a segfault `catch_unwind` cannot
-                spa::sys::SPA_META_Cursor,
+        // catch). Every offset below is validated against `region_size` with checked arithmetic,
-                std::mem::size_of::<spa::sys::spa_meta_cursor>(),
+        // mirroring the fd-length guard the main frame path already applies to xdg-desktop-portal-wlr.
-            ) as *const spa::sys::spa_meta_cursor
+        let meta =
-        };
+            unsafe { spa::sys::spa_buffer_find_meta(spa_buf, spa::sys::SPA_META_Cursor as u32) };
-        if cur.is_null() {
+        if meta.is_null() {
             return;
         }
-        // SAFETY: `cur` is non-null and points to a `spa_meta_cursor` of at least its own size
+        // SAFETY: `meta` is non-null and points into the held buffer's metadata array.
-        // inside the held buffer (guaranteed by the size arg above), so every field read is in bounds.
+        let (region_size, data) = unsafe { ((*meta).size as usize, (*meta).data as *const u8) };
+        if data.is_null() || region_size < std::mem::size_of::<spa::sys::spa_meta_cursor>() {
+            return;
+        }
+        let cur = data as *const spa::sys::spa_meta_cursor;
+        // SAFETY: `region_size >= size_of::<spa_meta_cursor>()` checked above, so every field is in bounds.
         let (id, pos_x, pos_y, hot_x, hot_y, bmp_off) = unsafe {
             (
                 (*cur).id,
@@ -1347,13 +1352,18 @@ mod pipewire {
             // Position-only update — keep the cached bitmap.
             return;
         }
-        // SAFETY: `bitmap_offset` is a byte offset from `cur` to a `spa_meta_bitmap`, which the
+        let bmp_off = bmp_off as usize;
-        // producer placed inside the same meta region it sized for this cursor (>= the size we
+        // The `spa_meta_bitmap` header must fit entirely inside the region before we read it —
-        // requested). The resulting pointer is in bounds and aligned for `spa_meta_bitmap`.
+        // `bitmap_offset` is producer-controlled and otherwise reads past the metadata.
-        let bmp =
+        match bmp_off.checked_add(std::mem::size_of::<spa::sys::spa_meta_bitmap>()) {
-            unsafe { (cur as *const u8).add(bmp_off as usize) as *const spa::sys::spa_meta_bitmap };
+            Some(end) if end <= region_size => {}
-        // SAFETY: `bmp` is the in-bounds, aligned `spa_meta_bitmap` pointer computed just above; the
+            _ => return,
-        // producer fully initialized this header, so reading its scalar fields is sound.
+        }
+        // SAFETY: `bmp_off + size_of::<spa_meta_bitmap>() <= region_size` (checked directly above),
+        // so the header is fully in bounds; the producer places it aligned as before.
+        let bmp = unsafe { data.add(bmp_off) as *const spa::sys::spa_meta_bitmap };
+        // SAFETY: `bmp` is the in-bounds `spa_meta_bitmap` header validated just above; reading its
+        // scalar fields is sound.
         let (vfmt, bw, bh, stride, pix_off) = unsafe {
             (
                 (*bmp).format,
@@ -1369,10 +1379,27 @@ mod pipewire {
         }
         let row = bw as usize * 4;
         let stride = if stride < row { row } else { stride };
-        let span = stride * (bh as usize - 1) + row;
+        // `span` is the exact byte extent the strided loop reads: `stride·(bh-1) + row`. Compute it
-        // SAFETY: the bitmap pixels live at `bmp + pix_off` for `span` bytes, within the
+        // with checked arithmetic (a producer stride near `i32::MAX` would otherwise overflow) and
-        // producer-sized meta region. `span` is the exact extent the strided copy below reads.
+        // require the whole pixel block `[bmp_off + pix_off, +span)` to lie inside the region before
-        let src = unsafe { std::slice::from_raw_parts((bmp as *const u8).add(pix_off), span) };
+        // fabricating the slice — this is the check whose absence made the read go out of bounds.
+        let span = match stride
+            .checked_mul(bh as usize - 1)
+            .and_then(|v| v.checked_add(row))
+        {
+            Some(s) => s,
+            None => return,
+        };
+        match bmp_off
+            .checked_add(pix_off)
+            .and_then(|v| v.checked_add(span))
+        {
+            Some(end) if end <= region_size => {}
+            _ => return,
+        }
+        // SAFETY: `bmp_off + pix_off + span <= region_size` (checked directly above), so the slice
+        // is fully within the producer's meta region; `span` is exactly the strided loop's extent.
+        let src = unsafe { std::slice::from_raw_parts(data.add(bmp_off + pix_off), span) };
         let mut rgba = vec![0u8; bw as usize * bh as usize * 4];
         for y in 0..bh as usize {
             for x in 0..bw as usize {
@@ -2164,36 +2191,37 @@ mod pipewire {
                 }
             })
             .process(|stream, ud| {
-                // PipeWire dispatches this from a C trampoline with no catch_unwind; a
+                // Latest-frame-only (OBS pattern): Mutter delivers buffers in bursts and recycles its
-                // panic crossing that FFI boundary would abort the whole host. Contain it.
+                // pool; an older queued buffer carries a STALE frame. Drain all queued buffers, requeue
-                let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                // the older ones, keep only the newest. This dequeue/requeue runs OUTSIDE the
-                    // Latest-frame-only (OBS pattern): Mutter delivers buffers in bursts and
+                // `catch_unwind` below — they are non-panicking C FFI pointer ops, and `newest` is
-                    // recycles its pool; an older queued buffer carries a STALE frame. Drain all
+                // requeued exactly once AFTER the panic-containing region. Previously the whole thing was
-                    // queued buffers, requeue the older ones, keep only the newest.
+                // inside the catch, so a caught panic (in `update_cursor_meta`/`consume_frame`) stranded
-                    // SAFETY: `stream` is the live stream PipeWire passes into this `.process` callback on
+                // `newest` forever, permanently shrinking the stream's fixed pool until capture wedged.
-                    // the loop thread, where `pw_stream_dequeue_buffer` is the documented call. It returns
+                // SAFETY: `stream` is the live stream PipeWire passes into this `.process` callback on the
-                    // a `*mut pw_buffer` owned by the stream (or null when the queue is drained),
+                // loop thread; `dequeue_raw_buffer` returns a stream-owned `*mut pw_buffer` or null
-                    // null-checked before any use. The loop is single-threaded, so no concurrent access.
+                // (null-checked), single-threaded so no concurrent access.
                 let mut newest = unsafe { stream.dequeue_raw_buffer() };
                 if newest.is_null() {
                     return;
                 }
                 let mut drained = 1u32;
                 loop {
-                        // SAFETY: same stream/loop-thread contract as the dequeue above; each call returns
+                    // SAFETY: same stream/loop-thread contract; returns the next stream-owned buffer or null.
-                        // the next stream-owned `*mut pw_buffer` or null (null-checked before use).
                     let next = unsafe { stream.dequeue_raw_buffer() };
                     if next.is_null() {
                         break;
                     }
-                        // SAFETY: `newest` is a non-null `*mut pw_buffer` previously dequeued from this same
+                    // SAFETY: `newest` was dequeued from this stream and not yet requeued; we immediately
-                        // stream and not yet requeued; `pw_stream_queue_buffer` hands ownership back to the
+                    // overwrite it, so the requeued pointer is never touched again.
-                        // stream. We immediately overwrite `newest = next`, so the requeued pointer is never
-                        // touched again (no use-after-requeue). Loop thread, single-threaded.
                     unsafe { stream.queue_raw_buffer(newest) };
                     newest = next;
                     drained += 1;
                 }
+                // PipeWire dispatches from a C trampoline with no catch_unwind; a panic crossing that FFI
+                // boundary would abort the whole host. Contain the inspect/consume work — the only Rust
+                // code here that can panic — and requeue `newest` unconditionally after it.
+                let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                     // SAFETY: `newest` is the non-null buffer we still own (dequeued, not requeued);
                     // `.buffer` is a `*mut spa_buffer` field libpipewire populated. This is a single field
                     // load through a valid pointer — no mutation or aliasing.
@@ -2272,19 +2300,18 @@ mod pipewire {
                                 "capture: skipped a stale CORRUPTED/cursor buffer (GNOME)"
                             );
                         }
-                        // SAFETY: `newest` is the non-null buffer we own (dequeued, never requeued on this
+                        // Skip this stale/cursor buffer — `newest` is requeued unconditionally below.
-                        // skip path); hand it back to the stream exactly once and return without touching it
-                        // again. Loop thread inside `.process`.
-                        unsafe { stream.queue_raw_buffer(newest) };
                         return;
                     }
 
                     consume_frame(ud, spa_buf);
-                    // SAFETY: `consume_frame` has finished reading `spa_buf` (and the `datas` borrows derived
-                    // from `newest`), so requeuing the owned `newest` exactly once here is sound — no
-                    // use-after-requeue. Loop thread inside `.process`.
-                    unsafe { stream.queue_raw_buffer(newest) };
                 }));
+                // Hand `newest` back to the stream exactly once, on EVERY path — normal, corrupted-skip,
+                // or a caught panic in the closure above. This single requeue is what keeps the fixed
+                // buffer pool from draining.
+                // SAFETY: all reads of `spa_buf`/`newest` (update_cursor_meta, consume_frame) completed
+                // inside the closure above; `newest` was dequeued from this stream and not yet requeued.
+                unsafe { stream.queue_raw_buffer(newest) };
                 if outcome.is_err() {
                     // In the per-frame `.process` callback: a deterministic panic (e.g. a bad
                     // format) would fire this every frame, so power-of-two throttle it — enough to

crates/pf-capture/src/windows/idd_push.rs

@@ -1341,6 +1341,12 @@ impl IddPushCapturer {
         self.out_ring.clear(); // the output format changed → rebuild lazily at the new format
         self.video_conv = None; // converters are sized + HDR-specific → rebuild at the new mode
         self.hdr_p010_conv = None;
+        // The PyroWave CSC is mode-baked too (BgraToYuvPlanes picks different SDR vs HDR shaders
+        // and R8/R8G8 vs R16/R16G16 outputs). Without this, a display_hdr flip (Downgrade point D:
+        // client_10bit=true but HDR couldn't enable at open) reused the stale SDR converter against
+        // the freshly HDR-formatted pyro ring — every frame corrupted. `ensure_pyro_conv` only
+        // builds when None, so it must be reset here like its siblings.
+        self.pyro_conv = None;
         self.pyro_ring.clear(); // PyroWave two-plane ring is sized → rebuild at the new mode
         self.pyro_last = None;
         self.out_idx = 0;

crates/pf-inject/src/inject/linux/steam_gadget.rs

@@ -17,7 +17,7 @@
 use anyhow::{bail, Context, Result};
 use std::mem::size_of;
 use std::os::fd::RawFd;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::thread::JoinHandle;
 
@@ -196,6 +196,41 @@ impl Drop for GadgetFd {
     }
 }
 
+/// The signal used to break a worker thread out of a blocking raw_gadget ioctl at teardown.
+/// `EVENT_FETCH`/`EP_WRITE` are `wait_event_interruptible` in the kernel with no timeout and no
+/// `O_NONBLOCK` honouring, and closing the fd cannot wake a thread already inside the ioctl (the
+/// in-flight syscall holds a reference to the struct file). A signal is the only reliable lever:
+/// delivered with a no-op, non-`SA_RESTART` handler it forces the ioctl to return `EINTR`, after
+/// which the loop's top-of-iteration `running` check exits. `SIGUSR1` is unused elsewhere in this
+/// process; the handler is a no-op, so a stray `SIGUSR1` becomes harmless rather than fatal.
+const WAKE_SIGNAL: libc::c_int = libc::SIGUSR1;
+
+/// Install the no-op `WAKE_SIGNAL` handler exactly once. Crucially `sa_flags = 0` (no `SA_RESTART`)
+/// so a delivered signal makes the interruptible ioctl return `EINTR` instead of auto-restarting.
+fn install_wake_handler() {
+    static ONCE: std::sync::Once = std::sync::Once::new();
+    ONCE.call_once(|| {
+        extern "C" fn noop(_: libc::c_int) {}
+        // SAFETY: installing a well-formed `sigaction` with an empty mask and a valid no-op handler
+        // for a single signal; touches only this process's disposition for `WAKE_SIGNAL`.
+        unsafe {
+            let mut sa: libc::sigaction = std::mem::zeroed();
+            sa.sa_sigaction = noop as usize;
+            libc::sigemptyset(&mut sa.sa_mask);
+            sa.sa_flags = 0;
+            libc::sigaction(WAKE_SIGNAL, &sa, std::ptr::null_mut());
+        }
+    });
+}
+
+/// Lets `Drop` wake a specific worker thread parked in a blocking ioctl. `tid` is the thread's
+/// `pthread_self()` (0 until it starts); `done` is set right before the thread returns, so `Drop`
+/// stops signalling a thread that has already exited.
+struct Waker {
+    tid: Arc<AtomicU64>,
+    done: Arc<AtomicBool>,
+}
+
 /// A virtual Steam Deck presented over the USB gadget subsystem. Dropping it stops the threads and
 /// closes the gadget (the kernel tears down the device).
 pub struct SteamDeckGadget {
@@ -203,6 +238,7 @@ pub struct SteamDeckGadget {
     feedback: Arc<Mutex<super::steam_proto::SteamFeedback>>,
     running: Arc<AtomicBool>,
     threads: Vec<JoinHandle<()>>,
+    wakers: Vec<Waker>,
     _fd: Arc<GadgetFd>,
     seq: u32,
 }
@@ -243,6 +279,18 @@ impl SteamDeckGadget {
         let ctrl_ep = Arc::new(std::sync::atomic::AtomicI32::new(-1));
         let configured = Arc::new(AtomicBool::new(false));
 
+        // The teardown wake path (see `WAKE_SIGNAL`) needs the handler installed before any thread
+        // can park in a blocking ioctl.
+        install_wake_handler();
+        let ctrl_waker = Waker {
+            tid: Arc::new(AtomicU64::new(0)),
+            done: Arc::new(AtomicBool::new(false)),
+        };
+        let stream_waker = Waker {
+            tid: Arc::new(AtomicU64::new(0)),
+            done: Arc::new(AtomicBool::new(false)),
+        };
+
         // Control thread: enumerate + answer every control transfer.
         let control = {
             let fd = fd.clone();
@@ -250,10 +298,15 @@ impl SteamDeckGadget {
             let ctrl_ep = ctrl_ep.clone();
             let configured = configured.clone();
             let feedback = feedback.clone();
+            let tid = ctrl_waker.tid.clone();
+            let done = ctrl_waker.done.clone();
             std::thread::Builder::new()
                 .name("pf-deck-gadget-ctrl".into())
                 .spawn(move || {
-                    control_loop(fd, running, ctrl_ep, configured, feedback, serial, unit_id)
+                    // SAFETY: `pthread_self` is always valid on the calling thread.
+                    tid.store(unsafe { libc::pthread_self() } as u64, Ordering::SeqCst);
+                    control_loop(fd, running, ctrl_ep, configured, feedback, serial, unit_id);
+                    done.store(true, Ordering::SeqCst);
                 })
                 .context("spawn gadget control thread")?
         };
@@ -264,9 +317,16 @@ impl SteamDeckGadget {
             let ctrl_ep = ctrl_ep.clone();
             let configured = configured.clone();
             let report = report.clone();
+            let tid = stream_waker.tid.clone();
+            let done = stream_waker.done.clone();
             std::thread::Builder::new()
                 .name("pf-deck-gadget-stream".into())
-                .spawn(move || stream_loop(fd, running, ctrl_ep, configured, report))
+                .spawn(move || {
+                    // SAFETY: `pthread_self` is always valid on the calling thread.
+                    tid.store(unsafe { libc::pthread_self() } as u64, Ordering::SeqCst);
+                    stream_loop(fd, running, ctrl_ep, configured, report);
+                    done.store(true, Ordering::SeqCst);
+                })
                 .context("spawn gadget stream thread")?
         };
 
@@ -275,6 +335,7 @@ impl SteamDeckGadget {
             feedback,
             running,
             threads: vec![control, stream],
+            wakers: vec![ctrl_waker, stream_waker],
             _fd: fd,
             seq: 0,
         })
@@ -302,6 +363,32 @@ impl SteamDeckGadget {
 impl Drop for SteamDeckGadget {
     fn drop(&mut self) {
         self.running.store(false, Ordering::SeqCst);
+        // The control thread spends steady state parked in a blocking `EVENT_FETCH` ioctl that only
+        // tests `running` at the top of its loop, so clearing the flag is not enough — it must be
+        // signalled out of the syscall (see `WAKE_SIGNAL`). Without this the join below can hang the
+        // caller (the session input thread, via `PadSlots::sweep`) indefinitely. Retry until each
+        // thread reports done, to cover the race where the signal lands just before the thread
+        // re-enters the ioctl; bounded (~1 s) so a genuinely stuck thread can't wedge teardown either.
+        for _ in 0..200 {
+            let mut all_done = true;
+            for w in &self.wakers {
+                if w.done.load(Ordering::SeqCst) {
+                    continue;
+                }
+                all_done = false;
+                let tid = w.tid.load(Ordering::SeqCst);
+                if tid != 0 {
+                    // SAFETY: the thread is joinable and not yet joined (join runs after this loop),
+                    // so `tid` names a live pthread; `pthread_kill` on a finished-but-unjoined thread
+                    // is defined (returns ESRCH), never UB.
+                    unsafe { libc::pthread_kill(tid as libc::pthread_t, WAKE_SIGNAL) };
+                }
+            }
+            if all_done {
+                break;
+            }
+            std::thread::sleep(std::time::Duration::from_millis(5));
+        }
         for t in self.threads.drain(..) {
             let _ = t.join();
         }

crates/pf-inject/src/inject/windows/dualsense_windows.rs

@@ -299,13 +299,20 @@ pub(super) fn create_swdevice(p: &SwDeviceProfile) -> Result<(HSWDEVICE, Option<
     let event = unsafe { CreateEventW(None, true, false, PCWSTR::null())? };
     // `result` starts as E_FAIL, NOT S_OK: if the wait below times out, a zero-initialised HRESULT
     // would read as success and mask the failure (found by the 2026-07 driver-health audit).
-    let mut ctx = SwCreateCtx {
+    // HEAP-allocated, deliberately: `sw_create_cb` writes `result` + up to 127 u16 of instance id
+    // through this pointer and then `SetEvent`s. The wait below is bounded (10 s), so on a wedged-PnP
+    // timeout the callback may still be PENDING — a stack context would be popped and a late callback
+    // would corrupt whatever the input thread put there next, and SetEvent a closed/recycled handle.
+    // On the timeout path we therefore LEAK the box and leave the event open (a one-off ~264 B + one
+    // HANDLE, only on that rare path) so a late callback always writes to live memory.
+    let ctx = Box::into_raw(Box::new(SwCreateCtx {
         event,
         result: E_FAIL,
         instance_id: [0; 128],
-    };
+    }));
-    // SAFETY: info + the buffers + ctx outlive the call (we wait on the event before returning);
+    // SAFETY: info + the buffers outlive the call; `ctx` is a live heap allocation that outlives every
-    // windows-rs returns the HSWDEVICE (the C out-param) as the Result value.
+    // path below (reclaimed only where the callback provably ran). windows-rs returns the HSWDEVICE
+    // (the C out-param) as the Result value.
     let hsw = match unsafe {
         SwDeviceCreate(
             w!("punktfunk"),
@@ -313,13 +320,15 @@ pub(super) fn create_swdevice(p: &SwDeviceProfile) -> Result<(HSWDEVICE, Option<
             &info,
             None,
             Some(sw_create_cb),
-            Some(&mut ctx as *mut SwCreateCtx as *const c_void),
+            Some(ctx as *const c_void),
         )
     } {
         Ok(h) => h,
         Err(e) => {
-            // SAFETY: event is valid.
+            // SAFETY: the call failed, so no callback was registered and `ctx` is ours to reclaim;
+            // `event` is valid and unreferenced.
             unsafe {
+                drop(Box::from_raw(ctx));
                 let _ = CloseHandle(event);
             }
             return Err(anyhow!("SwDeviceCreate failed: {e}"));
@@ -328,17 +337,22 @@ pub(super) fn create_swdevice(p: &SwDeviceProfile) -> Result<(HSWDEVICE, Option<
     // Block until PnP finishes enumerating (the callback signals), then check its result.
     // SAFETY: event is valid.
     let wait = unsafe { WaitForSingleObject(event, 10_000) };
-    // SAFETY: event is valid.
-    unsafe {
-        let _ = CloseHandle(event);
-    }
     if wait != WAIT_OBJECT_0 {
+        // Timed out: the callback may still fire. Intentionally leak `ctx` AND leave `event` open so
+        // its eventual write + SetEvent target live memory/handle rather than freed ones.
         // SAFETY: hsw is the handle SwDeviceCreate returned.
         unsafe { SwDeviceClose(hsw) };
         return Err(anyhow!(
             "SwDeviceCreate enumeration callback never fired (10s) — PnP may be wedged"
         ));
     }
+    // The callback ran (it is what signalled the event), so nothing else will touch `ctx`/`event`.
+    // SAFETY: `ctx` came from `Box::into_raw` above and is reclaimed exactly once here; `event` is
+    // valid and no longer referenced by a pending callback.
+    let ctx = unsafe {
+        let _ = CloseHandle(event);
+        Box::from_raw(ctx)
+    };
     if ctx.result.is_err() {
         // SAFETY: hsw is the handle SwDeviceCreate returned.
         unsafe { SwDeviceClose(hsw) };

crates/pf-inject/src/inject/windows/dualshock4_windows.rs

@@ -62,7 +62,7 @@ impl Ds4WinPad {
             std::ptr::write_unaligned(base as *mut u32, SHM_MAGIC);
         }
         let inst = format!("pf_ds4_{index}");
-        let (hsw, instance_id) = match create_swdevice(&SwDeviceProfile {
+        let (hsw, instance_id) = create_swdevice(&SwDeviceProfile {
             instance: &inst,
             container_tag: 0x5046_4453, // "PFDS"
             container_index: index,
@@ -70,13 +70,13 @@ impl Ds4WinPad {
             usb_vid_pid: "VID_054C&PID_09CC",
             usb_mi: None,
             description: "punktfunk Virtual DualShock 4",
-        }) {
+        })?; // Propagate, do NOT swallow — see below.
-            Ok((h, id)) => (Some(h), id),
+        let (hsw, instance_id) = (Some(hsw), instance_id);
-            Err(e) => {
+        // Swallowing a create failure here (the previous behaviour) latched the pad slot to
-                tracing::warn!(error = %format!("{e:#}"), "SwDeviceCreate failed; DualShock 4 devnode unavailable");
+        // `Some(pad)` with no live devnode: `PadSlots::ensure` short-circuits on `is_some()` and
-                (None, None)
+        // `gate.on_success()` cleared the backoff, so the create-gate that exists precisely to
-            }
+        // self-heal a transient PnP failure never retried. The game saw no controller for the whole
-        };
+        // session unless the client unplugged the pad. Matches the XUSB sibling, which propagates.
         let _sw = hsw.map(super::gamepad_raii::SwDevice::new);
         // Bounded eager delivery — for the DS4 this is what closes the identity race: the driver
         // must read `device_type = 1` from the delivered DATA section before hidclass asks it for

crates/pf-inject/src/inject/windows/gamepad_windows.rs

@@ -82,12 +82,16 @@ fn create_swdevice(index: u8) -> Result<(HSWDEVICE, Option<String>)> {
     let event = unsafe { CreateEventW(None, true, false, PCWSTR::null())? };
     // `result` starts as E_FAIL, NOT S_OK: if the wait below times out, a zero-initialised HRESULT
     // would read as success and mask the failure (found by the 2026-07 driver-health audit).
-    let mut ctx = SwCreateCtx {
+    // HEAP-allocated for the same reason as the DualSense sibling: the callback writes through this
+    // pointer and SetEvents, and the wait below is bounded — a stack context would be popped while a
+    // late callback still holds it. On the timeout path the box is deliberately leaked and the event
+    // left open so a late write/SetEvent always targets live memory/handle.
+    let ctx = Box::into_raw(Box::new(SwCreateCtx {
         event,
         result: E_FAIL,
         instance_id: [0; 128],
-    };
+    }));
-    // SAFETY: info + buffers + ctx outlive the call (we wait on the event before returning).
+    // SAFETY: info + buffers outlive the call; `ctx` is a live heap allocation outliving every path.
     let hsw = match unsafe {
         SwDeviceCreate(
             w!("punktfunk"),
@@ -95,13 +99,14 @@ fn create_swdevice(index: u8) -> Result<(HSWDEVICE, Option<String>)> {
             &info,
             None,
             Some(sw_create_cb),
-            Some(&mut ctx as *mut SwCreateCtx as *const c_void),
+            Some(ctx as *const c_void),
         )
     } {
         Ok(h) => h,
         Err(e) => {
-            // SAFETY: event is valid.
+            // SAFETY: the call failed, so no callback is pending and `ctx` is ours to reclaim.
             unsafe {
+                drop(Box::from_raw(ctx));
                 let _ = CloseHandle(event);
             }
             return Err(anyhow!("SwDeviceCreate(pf_xusb) failed: {e}"));
@@ -109,17 +114,20 @@ fn create_swdevice(index: u8) -> Result<(HSWDEVICE, Option<String>)> {
     };
     // SAFETY: event valid; block until PnP finishes enumerating, then check the callback result.
     let wait = unsafe { WaitForSingleObject(event, 10_000) };
-    // SAFETY: event is valid.
-    unsafe {
-        let _ = CloseHandle(event);
-    }
     if wait != WAIT_OBJECT_0 {
+        // Timed out — intentionally leak `ctx` and leave `event` open (see above).
         // SAFETY: hsw is the handle SwDeviceCreate returned.
         unsafe { SwDeviceClose(hsw) };
         return Err(anyhow!(
             "SwDeviceCreate(pf_xusb) enumeration callback never fired (10s) — PnP may be wedged"
         ));
     }
+    // The callback ran (it signalled the event), so nothing else will touch `ctx`/`event`.
+    // SAFETY: `ctx` came from `Box::into_raw` and is reclaimed exactly once here.
+    let ctx = unsafe {
+        let _ = CloseHandle(event);
+        Box::from_raw(ctx)
+    };
     if ctx.result.is_err() {
         // SAFETY: hsw is the handle SwDeviceCreate returned.
         unsafe { SwDeviceClose(hsw) };

crates/pf-inject/src/inject/windows/steam_deck_windows.rs

@@ -66,7 +66,7 @@ impl DeckWinPad {
             std::ptr::write_unaligned(base as *mut u32, SHM_MAGIC);
         }
         let inst = format!("pf_deck_{index}");
-        let (hsw, instance_id) = match create_swdevice(&SwDeviceProfile {
+        let (hsw, instance_id) = create_swdevice(&SwDeviceProfile {
             instance: &inst,
             container_tag: 0x5046_4453, // "PFDS"
             container_index: index,
@@ -77,13 +77,8 @@ impl DeckWinPad {
             // spike's run-1 failure).
             usb_mi: Some(2),
             description: "punktfunk Virtual Steam Deck",
-        }) {
+        })?; // Propagate — swallowing latched the slot to a pad with no devnode (see the DS4 twin).
-            Ok((h, i)) => (Some(h), i),
+        let (hsw, instance_id) = (Some(hsw), instance_id);
-            Err(e) => {
-                tracing::warn!(error = %format!("{e:#}"), "SwDeviceCreate failed; Steam Deck devnode unavailable");
-                (None, None)
-            }
-        };
         let _sw = hsw.map(super::gamepad_raii::SwDevice::new);
         // Bounded eager delivery — the driver must read `device_type = 3` before hidclass asks
         // it for descriptors, or the pad would enumerate with the default DualSense identity.

crates/pf-zerocopy/src/imp/client.rs

@@ -21,7 +21,7 @@ use std::path::{Path, PathBuf};
 use std::process::{Child, Command};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 /// Handshake budget: EGL + CUDA bring-up is ~200 ms; a cold driver load can take seconds.
 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);
@@ -64,11 +64,27 @@ impl Drop for Shared {
 /// Children whose worker hasn't exited yet at `RemoteImporter` drop time (it exits on socket
 /// EOF, i.e. after the last in-flight frame drops). Swept on every spawn and every drop so
 /// workers don't linger as zombies for more than one capture generation.
-static REAPER: Mutex<Vec<Child>> = Mutex::new(Vec::new());
+static REAPER: Mutex<Vec<(Child, Instant)>> = Mutex::new(Vec::new());
+
+/// How long past `REPLY_TIMEOUT` a parked worker may linger before it is force-killed. A worker
+/// wedged INSIDE a driver call never observes socket EOF, so `try_wait` alone would keep it (and
+/// its CUcontext + BufferPool — order hundreds of MB of VRAM) forever.
+const REAPER_KILL_DEADLINE: Duration = Duration::from_secs(20);
 
 fn sweep_reaper() {
     let mut list = REAPER.lock().unwrap();
-    list.retain_mut(|c| !matches!(c.try_wait(), Ok(Some(_))));
+    let now = Instant::now();
+    list.retain_mut(|(c, parked)| {
+        if matches!(c.try_wait(), Ok(Some(_))) {
+            return false; // exited on its own → reaped
+        }
+        if now.duration_since(*parked) > REAPER_KILL_DEADLINE {
+            let _ = c.kill();
+            let _ = c.wait();
+            return false; // wedged past the deadline → force-killed + reaped
+        }
+        true
+    });
 }
 
 /// Fd pinned to this process's own executable image, opened (once, lazily) via the
@@ -455,7 +471,7 @@ impl Drop for RemoteImporter {
         // gone; park the rest for the next sweep.
         if let Some(mut child) = self.child.take() {
             if !matches!(child.try_wait(), Ok(Some(_))) {
-                REAPER.lock().unwrap().push(child);
+                REAPER.lock().unwrap().push((child, Instant::now()));
             }
         }
         sweep_reaper();

crates/pf-zerocopy/src/imp/cuda.rs

@@ -1003,7 +1003,13 @@ impl RegisteredTexture {
         // SAFETY: `self.resource` is the valid `CUgraphicsResource` from a successful `register_gl`
         // (its only constructor), so the wrappers forward to the live table; the caller holds the
         // GL+CUDA contexts current (the registration's contract). `cuGraphicsMapResources` maps
-        // `count == 1` resource via `&mut self.resource` (a live field) on the default stream;
+        // `count == 1` resource via `&mut self.resource` (a live field). It is issued on
+        // `copy_stream()` — NOT the NULL stream — because map's only ordering guarantee is that
+        // prior GL work completes before subsequent CUDA work issued IN THE STREAM PASSED TO IT;
+        // the copy below runs on `copy_stream()` (a `CU_STREAM_NON_BLOCKING` stream, exempt from
+        // implicit NULL-stream ordering), so mapping on NULL left the copy free to race the GL
+        // de-tile/CSC that produced this texture (glFlush only, no fence) — intermittent torn or
+        // stale frames under GPU load. Map, copy, and unmap now all share `copy_stream()`.
         // `cuGraphicsSubResourceGetMappedArray` writes the mapped `CUarray` into the live local
         // `array` (index 0, mip 0). On failure we unmap and bail (balanced). `&copy` is a live
         // local `CUDA_MEMCPY2D` outliving the synchronous `copy_blocking`: `srcArray` is valid
@@ -1012,12 +1018,12 @@ impl RegisteredTexture {
         // we always unmap afterward (even on error), keeping the map/unmap pair balanced.
         unsafe {
             ck(
-                cuGraphicsMapResources(1, &mut self.resource, std::ptr::null_mut()),
+                cuGraphicsMapResources(1, &mut self.resource, copy_stream()),
                 "cuGraphicsMapResources",
             )?;
             let mut array: CUarray = std::ptr::null_mut();
             if cuGraphicsSubResourceGetMappedArray(&mut array, self.resource, 0, 0) != 0 {
-                let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut());
+                let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream());
                 bail!("cuGraphicsSubResourceGetMappedArray failed");
             }
             let copy = CUDA_MEMCPY2D {
@@ -1031,7 +1037,7 @@ impl RegisteredTexture {
                 ..Default::default()
             };
             let res = copy_blocking(&copy, "cuMemcpy2DAsync_v2");
-            let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut());
+            let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream());
             res
         }
     }
@@ -1058,12 +1064,12 @@ impl RegisteredTexture {
         // so the map/unmap pair stays balanced and the array outlives the copy.
         unsafe {
             ck(
-                cuGraphicsMapResources(1, &mut self.resource, std::ptr::null_mut()),
+                cuGraphicsMapResources(1, &mut self.resource, copy_stream()),
                 "cuGraphicsMapResources",
             )?;
             let mut array: CUarray = std::ptr::null_mut();
             if cuGraphicsSubResourceGetMappedArray(&mut array, self.resource, 0, 0) != 0 {
-                let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut());
+                let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream());
                 bail!("cuGraphicsSubResourceGetMappedArray failed");
             }
             let copy = CUDA_MEMCPY2D {
@@ -1077,7 +1083,7 @@ impl RegisteredTexture {
                 ..Default::default()
             };
             let res = copy_blocking(&copy, "cuMemcpy2DAsync_v2(plane)");
-            let _ = cuGraphicsUnmapResources(1, &mut self.resource, std::ptr::null_mut());
+            let _ = cuGraphicsUnmapResources(1, &mut self.resource, copy_stream());
             res
         }
     }

crates/pf-zerocopy/src/imp/egl.rs

@@ -691,6 +691,15 @@ impl EglImporter {
         width: u32,
         height: u32,
     ) -> Result<DeviceBuffer> {
+        // Even dimensions only: the UV copy walks `height.div_ceil(2)` chroma rows (the correct NV12
+        // count), but the pooled UV plane is sized at `height/2` rows — for an odd height those
+        // disagree by one row and the copy writes a full `uv_pitch` past the allocation (OOB device
+        // write / CUDA_ERROR_ILLEGAL_ADDRESS that poisons the shared context). Reject here, matching
+        // the guards `Nv12Blit::new`/`Yuv444Blit::new` already carry.
+        anyhow::ensure!(
+            width % 2 == 0 && height % 2 == 0,
+            "LINEAR NV12 needs even dimensions (got {width}x{height})"
+        );
         cuda::make_current()?;
         if self
             .linear_nv12_pool

crates/pf-zerocopy/src/imp/vulkan.rs

@@ -193,10 +193,17 @@ impl VkBridge {
 
     /// Import `fd` (dup'd internally; Vulkan owns the dup) as a transfer-src buffer of `size`.
     unsafe fn import_src(&mut self, fd: i32, size: u64) -> Result<()> {
+        use std::os::fd::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd};
         let dup = libc::dup(fd);
         if dup < 0 {
             bail!("dup(dmabuf fd)");
         }
+        // Own the dup so every early return BEFORE Vulkan consumes it (at `allocate_memory` success)
+        // closes it. `SrcBuf` holds raw handles with no Drop and is only populated on the success
+        // path, so each fallible step below must also destroy the buffer it created — otherwise a
+        // failed import (which the worker survives and the caller retries every frame) leaks a
+        // VkBuffer + VkDeviceMemory + fd per frame for the worker's whole lifetime.
+        let dup = OwnedFd::from_raw_fd(dup);
         let mut ext_info = vk::ExternalMemoryBufferCreateInfo::default()
             .handle_types(vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT);
         let buffer = self
@@ -212,41 +219,55 @@ impl VkBridge {
                     .push_next(&mut ext_info),
                 None,
             )
-            .context("create import buffer")?;
+            .context("create import buffer")?; // `dup` drops → closes on failure
         let mut fd_props = vk::MemoryFdPropertiesKHR::default();
-        self.ext_fd
+        if let Err(e) = self.ext_fd.get_memory_fd_properties(
-            .get_memory_fd_properties(
             vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT,
-                dup,
+            dup.as_raw_fd(),
             &mut fd_props,
-            )
+        ) {
-            .context("vkGetMemoryFdPropertiesKHR")?;
+            self.device.destroy_buffer(buffer, None);
+            return Err(e).context("vkGetMemoryFdPropertiesKHR");
+        }
         let reqs = self.device.get_buffer_memory_requirements(buffer);
-        let mem_type = self.memory_type(
+        let mem_type = match self.memory_type(
             reqs.memory_type_bits & fd_props.memory_type_bits,
             vk::MemoryPropertyFlags::empty(),
-        )?;
+        ) {
+            Ok(t) => t,
+            Err(e) => {
+                self.device.destroy_buffer(buffer, None);
+                return Err(e);
+            }
+        };
+        // Vulkan takes ownership of the fd on a SUCCESSFUL import: hand over the raw fd now, and on
+        // failure close it ourselves (matching the original contract) plus destroy the buffer.
+        let raw = dup.into_raw_fd();
         let mut import = vk::ImportMemoryFdInfoKHR::default()
             .handle_type(vk::ExternalMemoryHandleTypeFlags::DMA_BUF_EXT)
-            .fd(dup); // Vulkan takes ownership of `dup` on success
+            .fd(raw);
         let mut dedicated = vk::MemoryDedicatedAllocateInfo::default().buffer(buffer);
-        let memory = self
+        let memory = match self.device.allocate_memory(
-            .device
-            .allocate_memory(
             &vk::MemoryAllocateInfo::default()
                 .allocation_size(reqs.size.max(size))
                 .memory_type_index(mem_type)
                 .push_next(&mut import)
                 .push_next(&mut dedicated),
             None,
-            )
+        ) {
-            .map_err(|e| {
+            Ok(m) => m,
-                libc::close(dup); // failed import does not consume the fd
+            Err(e) => {
-                anyhow!("import dmabuf memory: {e}")
+                libc::close(raw); // failed import does not consume the fd
-            })?;
+                self.device.destroy_buffer(buffer, None);
-        self.device
+                return Err(anyhow!("import dmabuf memory: {e}"));
-            .bind_buffer_memory(buffer, memory, 0)
+            }
-            .context("bind import memory")?;
+        };
+        if let Err(e) = self.device.bind_buffer_memory(buffer, memory, 0) {
+            // `memory` owns the imported fd — freeing it releases the fd too.
+            self.device.free_memory(memory, None);
+            self.device.destroy_buffer(buffer, None);
+            return Err(e).context("bind import memory");
+        }
         self.src_cache.insert(
             fd,
             SrcBuf {
@@ -263,11 +284,11 @@ impl VkBridge {
         if self.dst.as_ref().is_some_and(|d| d.size >= size) {
             return Ok(());
         }
-        if let Some(old) = self.dst.take() {
+        // Build the replacement FULLY before retiring the old one. Previously the old dst was
-            self.device.destroy_buffer(old.buffer, None);
+        // destroyed and `self.dst` nulled up front, so a failed rebuild both dropped the working
-            self.device.free_memory(old.memory, None);
+        // buffer AND leaked every object the partial rebuild created (`buffer`/`memory` are raw ash
-            // old.cuda drops its mapping with it
+        // handles with no Drop, and `VkBridge::drop` only frees the live `self.dst`). Now every
-        }
+        // fallible step unwinds locally, and the swap happens only on full success.
         let mut ext_info = vk::ExternalMemoryBufferCreateInfo::default()
             .handle_types(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD);
         let buffer = self
@@ -285,35 +306,63 @@ impl VkBridge {
             .context("create export buffer")?;
         let reqs = self.device.get_buffer_memory_requirements(buffer);
         let mem_type =
-            self.memory_type(reqs.memory_type_bits, vk::MemoryPropertyFlags::DEVICE_LOCAL)?;
+            match self.memory_type(reqs.memory_type_bits, vk::MemoryPropertyFlags::DEVICE_LOCAL) {
+                Ok(t) => t,
+                Err(e) => {
+                    self.device.destroy_buffer(buffer, None);
+                    return Err(e);
+                }
+            };
         let mut export = vk::ExportMemoryAllocateInfo::default()
             .handle_types(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD);
         let mut dedicated = vk::MemoryDedicatedAllocateInfo::default().buffer(buffer);
-        let memory = self
+        let memory = match self.device.allocate_memory(
-            .device
-            .allocate_memory(
             &vk::MemoryAllocateInfo::default()
                 .allocation_size(reqs.size)
                 .memory_type_index(mem_type)
                 .push_next(&mut export)
                 .push_next(&mut dedicated),
             None,
-            )
+        ) {
-            .context("allocate exportable memory")?;
+            Ok(m) => m,
-        self.device
+            Err(e) => {
-            .bind_buffer_memory(buffer, memory, 0)
+                self.device.destroy_buffer(buffer, None);
-            .context("bind export memory")?;
+                return Err(e).context("allocate exportable memory");
-        let opaque_fd = self
+            }
-            .ext_fd
+        };
-            .get_memory_fd(
+        if let Err(e) = self.device.bind_buffer_memory(buffer, memory, 0) {
+            self.device.free_memory(memory, None);
+            self.device.destroy_buffer(buffer, None);
+            return Err(e).context("bind export memory");
+        }
+        let opaque_fd = match self.ext_fd.get_memory_fd(
             &vk::MemoryGetFdInfoKHR::default()
                 .memory(memory)
                 .handle_type(vk::ExternalMemoryHandleTypeFlags::OPAQUE_FD),
-            )
+        ) {
-            .context("vkGetMemoryFdKHR")?;
+            Ok(f) => f,
+            Err(e) => {
+                self.device.free_memory(memory, None);
+                self.device.destroy_buffer(buffer, None);
+                return Err(e).context("vkGetMemoryFdKHR");
+            }
+        };
         // CUDA imports (and on success owns) the exported fd. Size must match the allocation.
-        let cuda = cuda::ExternalDmabuf::import_owned_fd(opaque_fd, reqs.size)
+        // `import_owned_fd` closes `opaque_fd` on its own failure, so only the Vulkan objects unwind.
-            .context("cuImportExternalMemory(OPAQUE_FD from Vulkan)")?;
+        let cuda = match cuda::ExternalDmabuf::import_owned_fd(opaque_fd, reqs.size) {
+            Ok(c) => c,
+            Err(e) => {
+                self.device.free_memory(memory, None);
+                self.device.destroy_buffer(buffer, None);
+                return Err(e).context("cuImportExternalMemory(OPAQUE_FD from Vulkan)");
+            }
+        };
+        // Full success: retire the previous buffer now, then publish the new one.
+        if let Some(old) = self.dst.take() {
+            self.device.destroy_buffer(old.buffer, None);
+            self.device.free_memory(old.memory, None);
+            // old.cuda drops its mapping with it
+        }
         tracing::info!(size, "Vulkan→CUDA exportable staging buffer ready");
         self.dst = Some(DstBuf {
             buffer,
@@ -544,9 +593,19 @@ impl VkBridge {
             self.device
                 .queue_submit(self.queue, &[submit], self.fence)
                 .context("queue submit")?;
-            self.device
+            // Exception-safe wait: a TIMEOUT/DEVICE_LOST must not `?` out with the submission still
+            // executing — `self.cmd` and `self.fence` are reused every frame, and the caller retries
+            // on the SAME bridge (and `ensure_dst` later destroys `dst.buffer` assuming no in-flight
+            // work references it). Drain the GPU and reset the fence before propagating so the shared
+            // cmd/fence return clean.
+            if let Err(e) = self
+                .device
                 .wait_for_fences(&[self.fence], true, 1_000_000_000)
-                .context("fence wait")?;
+            {
+                let _ = self.device.device_wait_idle();
+                let _ = self.device.reset_fences(&[self.fence]);
+                return Err(e).context("fence wait");
+            }
             self.device
                 .reset_fences(&[self.fence])
                 .context("reset fence")?;
@@ -639,9 +698,19 @@ impl VkBridge {
             self.device
                 .queue_submit(self.queue, &[submit], self.fence)
                 .context("queue submit")?;
-            self.device
+            // Exception-safe wait: a TIMEOUT/DEVICE_LOST must not `?` out with the submission still
+            // executing — `self.cmd` and `self.fence` are reused every frame, and the caller retries
+            // on the SAME bridge (and `ensure_dst` later destroys `dst.buffer` assuming no in-flight
+            // work references it). Drain the GPU and reset the fence before propagating so the shared
+            // cmd/fence return clean.
+            if let Err(e) = self
+                .device
                 .wait_for_fences(&[self.fence], true, 1_000_000_000)
-                .context("fence wait")?;
+            {
+                let _ = self.device.device_wait_idle();
+                let _ = self.device.reset_fences(&[self.fence]);
+                return Err(e).context("fence wait");
+            }
             self.device
                 .reset_fences(&[self.fence])
                 .context("reset fence")?;