fix(zerocopy/client): renegotiation retires the old generation's IPC mappings

Shared::mappings only ever grew: clear_cache reset sent_keys and told the
worker to drop its fd cache, but nothing closed the host-side CUDA IPC
mappings for the previous pool generation. A session that renegotiates
repeatedly (mode changes, HDR toggles, client reconnects) accumulated a
pool's worth of stale mappings each time, each pinning a host VA
reservation to peer memory the worker had already freed.

Mappings now carry a refcount and a retired flag. clear_cache closes every
unreferenced mapping immediately and marks the rest retired; a retired
mapping closes when its last in-flight DeviceBuffer releases. The worker's
half of the contract: ClearCache also forgets its VA→id map, so anything
delivered after the boundary gets a fresh id WITH its descriptor — without
that, a same-shape renegotiation (worker keeps its pool) would re-deliver
an old id whose host mapping was just closed, and the host would misread
it as a desync. Ids never repeat (next_id only counts up), so fresh ids
cannot collide with the graveyard.

(R6 from design/pf-zerocopy-sweep-handoff.md.)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -35,7 +35,7 @@ const REPLY_TIMEOUT: Duration = Duration::from_secs(10);
 /// close, which is what tells an idle worker to exit.
 struct Shared {
     sock: OwnedFd,
-    mappings: Mutex<HashMap<u32, Mapping>>,
+    mappings: Mutex<HashMap<u32, MapEntry>>,
     dead: AtomicBool,
 }
 
@@ -49,16 +49,34 @@ struct Mapping {
     height: u32,
 }
 
+/// A [`Mapping`] plus its lifecycle: how many in-flight [`DeviceBuffer`]s still point into it,
+/// and whether its pool generation was retired by a renegotiation ([`RemoteImporter::clear_cache`]).
+/// Retired-but-referenced entries linger as a graveyard and close when their last frame releases
+/// — without this, every renegotiation (mode change, HDR toggle, client reconnect) permanently
+/// pinned a pool's worth of host VA reservations to peer memory the worker had already freed.
+/// Worker buffer ids are never reused (its `next_id` only counts up), so retired entries can
+/// share the map with the next generation's.
+struct MapEntry {
+    m: Mapping,
+    refs: u32,
+    retired: bool,
+}
+
 impl Drop for Shared {
     fn drop(&mut self) {
-        // Last reference gone — no DeviceBuffer can still point into these mappings.
-        for (_, m) in self.mappings.lock().unwrap().drain() {
+        // Last reference gone — no DeviceBuffer can still point into these mappings (current
+        // generation or graveyard alike).
+        for (_, e) in self.mappings.lock().unwrap().drain() {
+            close_mapping(&e.m);
+        }
+    }
+}
+
+fn close_mapping(m: &Mapping) {
     cuda::ipc_close(m.y);
     if let Some((uv, _)) = m.uv {
         cuda::ipc_close(uv);
     }
-        }
-    }
 }
 
 /// Children whose worker hasn't exited yet at `RemoteImporter` drop time (it exits on socket
@@ -414,19 +432,24 @@ impl RemoteImporter {
                         self.mark_dead();
                         format!("open CUDA IPC mapping for worker buffer {id}")
                     })?;
-                    self.shared.mappings.lock().unwrap().insert(id, mapping);
+                    self.shared.mappings.lock().unwrap().insert(
+                        id,
+                        MapEntry {
+                            m: mapping,
+                            refs: 0,
+                            retired: false,
+                        },
+                    );
                 }
-                let m = self
-                    .shared
-                    .mappings
-                    .lock()
-                    .unwrap()
-                    .get(&id)
-                    .copied()
-                    .ok_or_else(|| {
+                let m = {
+                    let mut g = self.shared.mappings.lock().unwrap();
+                    let entry = g.get_mut(&id).ok_or_else(|| {
                         self.mark_dead();
                         anyhow::anyhow!("worker delivered unknown buffer id {id} (desync)")
                     })?;
+                    entry.refs += 1;
+                    entry.m
+                };
                 let shared = self.shared.clone();
                 Ok(DeviceBuffer::remote(
                     m.y,
@@ -439,8 +462,17 @@ impl RemoteImporter {
                     Box::new(move || {
                         // Fire-and-forget recycle; a dead worker just means EPIPE, ignored. The
                         // captured `shared` Arc is what keeps the mapping + socket alive until
-                        // the last frame drops.
+                        // the last frame drops. A retired mapping (its generation renegotiated
+                        // away) closes here with its last reference.
                         let _ = proto::send(shared.sock.as_fd(), &Request::Release { id }, None);
+                        let mut g = shared.mappings.lock().unwrap();
+                        if let Some(entry) = g.get_mut(&id) {
+                            entry.refs = entry.refs.saturating_sub(1);
+                            if entry.retired && entry.refs == 0 {
+                                let entry = g.remove(&id).expect("entry exists");
+                                close_mapping(&entry.m);
+                            }
+                        }
                     }),
                 ))
             }
@@ -452,9 +484,24 @@ impl RemoteImporter {
         }
     }
 
-    /// The PipeWire stream renegotiated — reset both sides' per-buffer caches.
+    /// The PipeWire stream renegotiated — reset both sides' per-buffer caches, and retire the
+    /// outgoing generation's CUDA IPC mappings: the worker replaces its pool, so these host-side
+    /// mappings pin VA reservations to peer memory that is about to be (or already was) freed.
+    /// Unreferenced ones close now; ones still under an in-flight frame close with its release.
     pub fn clear_cache(&mut self) {
         self.sent_keys.clear();
+        {
+            let mut g = self.shared.mappings.lock().unwrap();
+            g.retain(|_, entry| {
+                if entry.refs == 0 {
+                    close_mapping(&entry.m);
+                    false
+                } else {
+                    entry.retired = true;
+                    true
+                }
+            });
+        }
         if !self.dead() {
             if let Err(e) = proto::send(self.shared.sock.as_fd(), &Request::ClearCache, None) {
                 tracing::warn!(error = %e, "zerocopy worker ClearCache failed");

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

@@ -194,8 +194,9 @@ struct EglBackend {
     inflight: HashMap<u32, DeviceBuffer>,
     /// Buffer id per device allocation. Valid only within one pool generation: pools never free
     /// allocations while alive, so a device VA can't repeat until a size change replaces the pool
-    /// — at which point [`Self::note_dims`] clears this map (ids themselves are never reused;
-    /// `next_id` only counts up).
+    /// — at which point [`Self::note_dims`] clears this map, as does `ClearCache` (the host
+    /// retires its IPC mappings on that boundary). Ids themselves are never reused; `next_id`
+    /// only counts up.
     ids: HashMap<CUdeviceptr, u32>,
     next_id: u32,
     /// The (kind, width, height) of the last import — a change means the importer replaced its
@@ -282,6 +283,13 @@ impl ImportBackend for EglBackend {
         }
         self.fd_lru.clear();
         self.importer.clear_linear_cache();
+        // ClearCache is the generation boundary the HOST retires its CUDA IPC mappings on: it
+        // closes every mapping the renegotiated-away generation opened (keeping only ones still
+        // under an in-flight frame, until they release). Forget the VA→id map so anything
+        // delivered after this point gets a fresh id WITH its descriptor — re-delivering an old
+        // id would reference a mapping the host just closed. Ids never repeat (`next_id` only
+        // counts up), so fresh ids can't collide with the host's graveyard.
+        self.ids.clear();
     }
 }