fix(zerocopy): error-path leaks, exception-safe fence wait, odd-height NV12 overrun, worker reaper deadline

Seven medium findings from the round-1 sweep, all in pf-zerocopy. Adjudicated
against source; all zero-copy-preserving (no GPU→CPU→GPU roundtrip introduced).

- import_src (vulkan.rs) leaked a VkBuffer + VkDeviceMemory + dup'd fd on every
  fallible step before the success-only src_cache.insert. A failed import is
  survived by the worker and RETRIED by the caller every frame, so this leaked
  per frame for the worker's whole lifetime. Now owns the dup fd via OwnedFd
  (closes on early return until Vulkan consumes it at allocate_memory) and
  destroys the buffer + frees memory on each error path.

- ensure_dst destroyed the old exportable buffer and nulled self.dst BEFORE
  building the replacement, so a failed rebuild both dropped the working buffer
  and leaked every object the partial rebuild created (raw ash handles, no Drop;
  VkBridge::drop only frees the live self.dst). Now builds fully, unwinds each
  error locally, and swaps only on success.

- The submit→wait→reset sequence in import_linear_nv12 / import_linear `?`-ed out
  of wait_for_fences BEFORE reset_fences on a TIMEOUT/DEVICE_LOST, leaving the
  shared self.cmd PENDING and self.fence IN-USE while the caller retries on the
  same bridge (and ensure_dst later destroys dst.buffer assuming nothing is in
  flight). Now drains the GPU (device_wait_idle) and resets the fence before
  propagating — fixing the reported ensure_dst UAF at its source.

- copy_pitched_nv12_to_buffer writes height.div_ceil(2) chroma rows into a UV
  plane sized at height/2, so an odd height overruns by one uv_pitch row (OOB
  device write / CUDA_ERROR_ILLEGAL_ADDRESS). Added the even-dimension guard to
  import_linear_nv12, matching Nv12Blit/Yuv444Blit.

- sweep_reaper only reaped already-exited workers, so a worker wedged inside a
  driver call lingered forever holding its CUcontext + BufferPool (hundreds of MB
  VRAM). Now force-kills a child parked past REAPER_KILL_DEADLINE (20s).

Compile + tests green on Linux .21 (RTX 5070 Ti): pf-zerocopy 17/0. The error
paths themselves are not fault-injected; the fixes are structural.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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/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
-            .get_memory_fd_properties(
+        if let Err(e) = self.ext_fd.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
-            .device
-            .allocate_memory(
+        let memory = match self.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| {
-                libc::close(dup); // failed import does not consume the fd
-                anyhow!("import dmabuf memory: {e}")
-            })?;
-        self.device
-            .bind_buffer_memory(buffer, memory, 0)
-            .context("bind import memory")?;
+        ) {
+            Ok(m) => m,
+            Err(e) => {
+                libc::close(raw); // failed import does not consume the fd
+                self.device.destroy_buffer(buffer, None);
+                return Err(anyhow!("import dmabuf memory: {e}"));
+            }
+        };
+        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() {
-            self.device.destroy_buffer(old.buffer, None);
-            self.device.free_memory(old.memory, None);
-            // old.cuda drops its mapping with it
-        }
+        // Build the replacement FULLY before retiring the old one. Previously the old dst was
+        // destroyed and `self.dst` nulled up front, so a failed rebuild both dropped the working
+        // buffer AND leaked every object the partial rebuild created (`buffer`/`memory` are raw ash
+        // 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
-            .device
-            .allocate_memory(
+        let memory = match self.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")?;
-        self.device
-            .bind_buffer_memory(buffer, memory, 0)
-            .context("bind export memory")?;
-        let opaque_fd = self
-            .ext_fd
-            .get_memory_fd(
+        ) {
+            Ok(m) => m,
+            Err(e) => {
+                self.device.destroy_buffer(buffer, None);
+                return Err(e).context("allocate exportable memory");
+            }
+        };
+        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)
-            .context("cuImportExternalMemory(OPAQUE_FD from Vulkan)")?;
+        // `import_owned_fd` closes `opaque_fd` on its own failure, so only the Vulkan objects unwind.
+        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")?;