Merge branch 'fix/encode-medium-tier' into land/sweep-all

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

@@ -1861,6 +1861,7 @@ impl IddPushCapturer {
                     cbcr,
                     fence_handle,
                     fence_value,
+                    ring_gen: self.generation,
                 }),
             )
         } else {
@@ -1919,6 +1920,7 @@ impl IddPushCapturer {
                         cbcr: dst_cbcr,
                         fence_handle,
                         fence_value,
+                        ring_gen: self.generation,
                     }),
                 }),
                 cursor: None,

crates/pf-encode/src/enc/codec.rs

@@ -293,6 +293,16 @@ pub trait Encoder: Send {
     /// flagged [`EncodedFrame::chunk_aligned`] and the session marks them on the wire.
     /// Default: no-op (the H.26x backends' bitstreams cannot be cut losslessly).
     fn set_wire_chunking(&mut self, _shard_payload: usize) {}
+    /// How many frames the CAPTURER guarantees the encoder may hold in flight before it starts
+    /// reusing an input texture (`Capturer::pipeline_depth`). Backends that encode the capturer's
+    /// textures IN PLACE — no `CopyResource` — must not pipeline deeper than this: the capturer
+    /// rotates its output ring per delivered frame with no regard for encode completion, so a
+    /// deeper pipeline lets it overwrite a texture mid-encode. That is visual corruption (torn or
+    /// mixed frames), not UB, so it fails silently and intermittently.
+    ///
+    /// Called once by the session glue after the capturer is known; a backend that copies its
+    /// input, or is synchronous, ignores it. Default: no-op.
+    fn set_input_ring_depth(&mut self, _depth: usize) {}
     /// Signal end-of-stream. After this, drain the remaining AUs with [`poll`](Self::poll)
     /// until it returns `None` — NVENC buffers frames internally even at `delay=0`.
     fn flush(&mut self) -> Result<()>;

crates/pf-encode/src/enc/linux/pyrowave.rs

@@ -1176,7 +1176,24 @@ impl Encoder for PyroWaveEncoder {
     fn submit(&mut self, frame: &CapturedFrame) -> Result<()> {
         // SAFETY: single-threaded encoder; `encode_frame` records/submits on handles this
         // struct owns and waits its own fence before touching results.
-        unsafe { self.encode_frame(frame) }
+        let r = unsafe { self.encode_frame(frame) };
+        if r.is_err() {
+            // `encode_frame` opens the recording window early and has several fallible steps
+            // inside it (cursor prep, dmabuf import, format mapping, the CPU-RGB staging path,
+            // an unsupported-payload bail, and the encode call itself). Every one returns with
+            // `self.cmd` still RECORDING, and nothing downstream repairs it — there is exactly
+            // one `begin_command_buffer` in this file and `reset()`/`Drop` never touch `cmd` —
+            // so the NEXT frame would call `begin` on a recording buffer, which is invalid usage.
+            // Legal here on every path: the pool carries RESET_COMMAND_BUFFER and the buffer is
+            // not pending (we never reached the submit, or the submit itself failed).
+            // SAFETY: `self.cmd` is owned by this encoder and, on these paths, not in flight.
+            unsafe {
+                let _ = self
+                    .device
+                    .reset_command_buffer(self.cmd, vk::CommandBufferResetFlags::empty());
+            }
+        }
+        r
     }
 
     fn caps(&self) -> EncoderCaps {

crates/pf-encode/src/enc/linux/vulkan_video.rs

@@ -37,6 +37,12 @@ const DPB_SLOTS: u32 = 8;
 /// latency — on-glass validated as rock-solid at 1080p@240, so it is the real-time default;
 /// backpressure kicks in at the 2nd unread frame. Distinct from `DPB_SLOTS` (reference pool).
 const RING_DEFAULT: usize = 2;
+
+/// Ceiling on any blocking GPU fence wait on the encode thread (5 s). Generous against a real
+/// encode (single-digit ms even on a loaded GPU) and against a driver hiccup, but finite: this is
+/// the thread the stall watchdog's `reset()` runs on, so an unbounded wait would deadlock the very
+/// path that recovers the session. Matches the Windows NVENC retrieve-thread budget.
+const ENCODE_FENCE_TIMEOUT_NS: u64 = 5_000_000_000;
 /// AV1 base quantizer index (0..=255) seeded into every frame. CBR rate control overrides it per
 /// frame; it only matters as the starting point and for the (rate-control-ignored) constant-Q path.
 const AV1_BASE_Q_IDX: u8 = 128;
@@ -868,6 +874,15 @@ impl VulkanVideoEncoder {
         let (img, mem, view) = self.import_dmabuf(d, cw, ch)?;
         // Bound the cache; evict oldest (FIFO). A stable PipeWire pool never trips this in steady state
         // (all imports resident); it only cycles across a pool change (which also rebuilds the session).
+        // Up to `ring_depth - 1` submitted frames may still be executing against a cached image
+        // (`enqueue` only drains down to `frames.len()`, and `record_submit` imports before it
+        // records), so destroying an evicted import here is a GPU-side use-after-free. `Drop` and
+        // `reset()` both idle the device first; this was the one unguarded destroy. Guarded on the
+        // length test so the steady-state path — where the cache is resident and never evicts —
+        // pays nothing.
+        if self.import_cache.len() >= IMPORT_CACHE_CAP {
+            let _ = self.device.device_wait_idle();
+        }
         while self.import_cache.len() >= IMPORT_CACHE_CAP {
             let (_, _, oi, om, ov) = self.import_cache.remove(0);
             self.device.destroy_image_view(ov, None);
@@ -1849,8 +1864,25 @@ impl VulkanVideoEncoder {
     unsafe fn read_slot(&mut self, slot: usize) -> Result<EncodedFrame> {
         let dev = self.device.clone();
         let f = &self.frames[slot];
-        let mut fb = [[0u32; 2]; 1];
+        // Ask for the operation status alongside the two feedback words: without it a FAILED encode
-        dev.get_query_pool_results(f.query_pool, 0, &mut fb, vk::QueryResultFlags::WAIT)?;
+        // is indistinguishable from a successful one, and its offset/bytes-written are read as if
+        // they described real bitstream. The status rides as a trailing element (signed:
+        // `VkQueryResultStatusKHR` is >0 COMPLETE, 0 NOT_READY, <0 error).
+        let mut fb = [[0i32; 3]; 1];
+        dev.get_query_pool_results(
+            f.query_pool,
+            0,
+            &mut fb,
+            vk::QueryResultFlags::WAIT | vk::QueryResultFlags::WITH_STATUS_KHR,
+        )?;
+        let status = fb[0][2];
+        if status <= 0 {
+            anyhow::bail!(
+                "vulkan-encode: encode feedback for slot {slot} reports status {status} \
+                 (not COMPLETE) — dropping the frame rather than shipping its bitstream"
+            );
+        }
+        let fb = [[fb[0][0] as u32, fb[0][1] as u32]];
         // The (offset, bytes-written) pair is driver-reported: validate it against the bitstream
         // allocation BEFORE mapping, or the `from_raw_parts` below reads outside the buffer and
         // ships whatever it finds straight onto the wire. Checked in u64 so the add cannot wrap,
@@ -1892,8 +1924,25 @@ impl VulkanVideoEncoder {
         // and free it — that oldest slot is exactly the round-robin `ring` cursor we reuse next.
         while self.in_flight.len() >= self.frames.len() {
             let slot = self.in_flight.pop_front().unwrap();
-            self.device
+            // Bounded, not `u64::MAX`: this runs ON the host encode thread, which is also the
-                .wait_for_fences(&[self.frames[slot].fence], true, u64::MAX)?;
+            // thread the stall watchdog's `reset()` would run on. An infinite wait against a
+            // wedged GPU/driver therefore parks the one thread that could recover the session —
+            // it never errors, never resets, and teardown blocks joining it. Surfacing expiry as
+            // an error hands control back to the existing recovery path (same convention as the
+            // pyrowave and Windows NVENC backends).
+            match self.device.wait_for_fences(
+                &[self.frames[slot].fence],
+                true,
+                ENCODE_FENCE_TIMEOUT_NS,
+            ) {
+                Ok(()) => {}
+                Err(vk::Result::TIMEOUT) => anyhow::bail!(
+                    "vulkan-encode: fence for slot {slot} did not signal within {} ms — GPU or \
+                     driver wedged; failing the submit so the session can reset",
+                    ENCODE_FENCE_TIMEOUT_NS / 1_000_000
+                ),
+                Err(e) => return Err(e.into()),
+            }
             let done = self.read_slot(slot)?;
             self.pending.push_back(done);
         }

crates/pf-encode/src/enc/windows/nvenc.rs

@@ -409,6 +409,12 @@ pub struct NvencD3d11Encoder {
     events: Vec<usize>,
     /// Async mode: the retrieve thread + its channels (`None` = classic same-thread sync retrieve).
     async_rt: Option<AsyncRetrieve>,
+    /// The capturer's `pipeline_depth` (`set_input_ring_depth`). This backend encodes the
+    /// capturer's textures IN PLACE, so it is a HARD ceiling on async in-flight depth: the
+    /// capturer rotates its ring per delivered frame regardless of encode completion, so
+    /// pipelining deeper lets it overwrite a texture mid-encode (torn frames). `None` until the
+    /// session glue reports it — treated as "unknown, don't pipeline past the env cap".
+    input_ring_depth: Option<usize>,
     /// `NV_ENC_CAPS_ASYNC_ENCODE_SUPPORT` from the caps probe — gates the async retrieve mode.
     async_supported: bool,
     /// (bitstream, mapped input resource to unmap after retrieval, pts_ns, recovery-anchor) per
@@ -505,6 +511,7 @@ impl NvencD3d11Encoder {
             bitstreams: Vec::new(),
             events: Vec::new(),
             async_rt: None,
+            input_ring_depth: None,
             async_supported: false,
             pending: VecDeque::new(),
             frame_idx: 0,
@@ -1156,11 +1163,21 @@ impl Encoder for NvencD3d11Encoder {
         // index, which is non-zero on a mid-session encoder rebuild's first frame.
         let opening = self.next == 0;
         // Async backpressure: never hand NVENC an output bitstream that is still in flight, and
-        // keep in-flight depth within the capturer's texture ring (see `async_inflight_cap`). At
+        // keep in-flight depth within the capturer's texture ring. At the cap, block on the OLDEST
-        // the cap, block on the OLDEST completion (the retrieve thread is already waiting on its
+        // completion (the retrieve thread is already waiting on its event) before submitting more —
-        // event) before submitting more — bounding depth exactly like the sync path's per-tick
+        // bounding depth exactly like the sync path's per-tick blocking poll, just `cap` deep
-        // blocking poll, just `cap` deep instead of 1.
+        // instead of 1.
-        while self.async_rt.is_some() && self.pending.len() >= async_inflight_cap() {
+        //
+        // The ring term is the one that matters for correctness: `async_inflight_cap()` is only the
+        // output-bitstream-pool ceiling plus an env knob, and consults NOTHING about the capturer,
+        // despite this comment previously claiming otherwise. Since this backend encodes the
+        // capturer's textures in place, exceeding the capturer's declared `pipeline_depth` lets it
+        // rotate a texture out from under a live encode — torn frames, silently.
+        let cap = match self.input_ring_depth {
+            Some(d) => async_inflight_cap().min(d.max(1)),
+            None => async_inflight_cap(),
+        };
+        while self.async_rt.is_some() && self.pending.len() >= cap {
             let done = {
                 let rt = self.async_rt.as_mut().expect("checked in loop condition");
                 rt.done_rx
@@ -1336,6 +1353,17 @@ impl Encoder for NvencD3d11Encoder {
         self.submit(frame)
     }
 
+    fn set_input_ring_depth(&mut self, depth: usize) {
+        // This backend registers and encodes the capturer's textures in place (no CopyResource),
+        // so the capturer's ring depth is a hard ceiling on how deep async may pipeline.
+        self.input_ring_depth = Some(depth);
+        tracing::debug!(
+            depth,
+            env_cap = async_inflight_cap(),
+            "NVENC: capturer input-ring depth reported — async in-flight bounded by the smaller"
+        );
+    }
+
     fn request_keyframe(&mut self) {
         self.force_kf = true;
     }

crates/pf-encode/src/enc/windows/pyrowave.rs

@@ -43,6 +43,9 @@ const BS_SLACK: usize = 256 * 1024;
 /// (a desktop-switch device recreate), in which case the stale imports are evicted + destroyed.
 const IMPORT_CACHE_CAP: usize = 8;
 
+/// Plane-import cache key: the texture's COM address plus the extent it was imported at.
+type PlaneKey = (isize, u32, u32);
+
 // --- Vulkan enum values not surfaced by pyrowave-sys' bindgen (only enums *reachable* from the
 // pyrowave C API are generated; these plain #define / flags-typedef values are stable spec
 // constants). bindgen renders every reachable Vulkan enum as a `u32` type alias, so these u32
@@ -136,8 +139,12 @@ pub struct PyroWaveEncoder {
     // Imported plane textures, cached by the out-ring texture's raw pointer (stable per ring slot):
     // the full-res R8 Y plane and the half-res R8G8 CbCr plane, imported SEPARATELY (a single planar
     // NV12 import is unreliable on NVIDIA at arbitrary sizes).
-    y_images: Vec<(isize, pw::pyrowave_image)>,
+    /// The capturer ring generation the cached plane imports below belong to. A recreate bumps it,
-    cbcr_images: Vec<(isize, pw::pyrowave_image)>,
+    /// and every cached import is destroyed — the COM addresses they are keyed on can be recycled
+    /// by the allocator after a recreate, so identity cannot rest on the pointer alone.
+    ring_gen: Option<u32>,
+    y_images: Vec<(PlaneKey, pw::pyrowave_image)>,
+    cbcr_images: Vec<(PlaneKey, pw::pyrowave_image)>,
 
     width: u32,
     height: u32,
@@ -268,6 +275,7 @@ impl PyroWaveEncoder {
                 pw_dev,
                 pw_enc,
                 sync: std::ptr::null_mut(),
+                ring_gen: None,
                 y_images: Vec::new(),
                 cbcr_images: Vec::new(),
                 width,
@@ -351,10 +359,16 @@ impl PyroWaveEncoder {
     ///
     /// # Safety
     /// Same contract as [`import_plane`].
+    /// Keyed on `(texture address, width, height)` rather than the bare address: the COM pointer
+    /// carries no reference here, so a released texture's address can be recycled by a later
+    /// allocation and return an import describing the WRONG surface. Folding the extent in means a
+    /// recycled address at a different size can never alias. (A recycle at the SAME size is still
+    /// possible in principle — the complete fix is to key on the capturer's ring generation, which
+    /// needs that generation plumbed onto `PyroFrameShare`.)
     unsafe fn cached_plane(
-        cache: &mut Vec<(isize, pw::pyrowave_image)>,
+        cache: &mut Vec<(PlaneKey, pw::pyrowave_image)>,
         make: impl FnOnce() -> Result<pw::pyrowave_image>,
-        key: isize,
+        key: PlaneKey,
     ) -> Result<pw::pyrowave_image> {
         if let Some((_, img)) = cache.iter().find(|(k, _)| *k == key) {
             return Ok(*img);
@@ -423,6 +437,21 @@ impl PyroWaveEncoder {
             !self.pw_enc.is_null(),
             "pyrowave: encode after a failed reset (encoder was destroyed and not rebuilt)"
         );
+        // The plane textures are imported at the encoder's CONFIGURED extent, not the frame's, so a
+        // capture that changed size would be read under a stale `VkImageCreateInfo`. This is
+        // reachable without any client Reconfigure: the IDD capturer autonomously recreates its ring
+        // on a confirmed display-descriptor change (e.g. a fullscreen game mode-setting the virtual
+        // display). Refuse instead — the session must reopen the encoder at the new mode. Mirrors
+        // the guard the QSV and AMF backends already carry.
+        anyhow::ensure!(
+            frame.width == self.width && frame.height == self.height,
+            "pyrowave: captured frame {}x{} != encoder {}x{} (the capturer recreated its ring at a \
+             new mode — the encoder must be reopened)",
+            frame.width,
+            frame.height,
+            self.width,
+            self.height
+        );
         let FramePayload::D3d11(d3d) = &frame.payload else {
             bail!("pyrowave (Windows) needs a D3D11 frame (the capturer must be in pyrowave mode)")
         };
@@ -431,6 +460,25 @@ impl PyroWaveEncoder {
              in pyrowave mode (session_plan::output_format must set OutputFormat::pyrowave)",
         )?;
 
+        // Ring recreate ⇒ every cached plane import belongs to textures that no longer exist. Their
+        // COM addresses can be handed back out by the allocator, so a pointer-keyed hit could return
+        // an image bound to freed memory. Flush on the generation change rather than relying on the
+        // address (or the FIFO cap) to notice.
+        if self.ring_gen != Some(share.ring_gen) {
+            if self.ring_gen.is_some() {
+                tracing::info!(
+                    from = ?self.ring_gen,
+                    to = share.ring_gen,
+                    cached = self.y_images.len() + self.cbcr_images.len(),
+                    "pyrowave: capturer recreated its ring — flushing stale plane imports"
+                );
+            }
+            for (_, img) in self.y_images.drain(..).chain(self.cbcr_images.drain(..)) {
+                pw::pyrowave_image_destroy(img);
+            }
+            self.ring_gen = Some(share.ring_gen);
+        }
+
         // Import the fence whenever this encoder has no timeline yet — the first frame, OR a fresh
         // encoder after a client mode-switch rebuild (the capturer passes the persistent handle on
         // every frame precisely so a rebuilt encoder can re-import it).
@@ -465,7 +513,7 @@ impl PyroWaveEncoder {
         };
         let pw_dev = self.pw_dev;
         let y_img = {
-            let key = d3d.texture.as_raw() as isize;
+            let key = (d3d.texture.as_raw() as isize, w, h);
             let tex = &d3d.texture;
             Self::cached_plane(
                 &mut self.y_images,
@@ -474,7 +522,7 @@ impl PyroWaveEncoder {
             )?
         };
         let cbcr_img = {
-            let key = share.cbcr.as_raw() as isize;
+            let key = (share.cbcr.as_raw() as isize, cw, ch);
             let tex = &share.cbcr;
             Self::cached_plane(
                 &mut self.cbcr_images,
@@ -976,6 +1024,9 @@ mod tests {
                     cbcr: cbcr_tex,
                     fence_handle: Some(fence_handle.0 as isize),
                     fence_value: 1,
+                    // One synthetic ring for the whole case: a constant generation exercises the
+                    // steady-state cache-hit path (a changing one would flush every frame).
+                    ring_gen: 1,
                 }),
             }),
             cursor: None,

crates/pf-encode/src/lib.rs

@@ -209,6 +209,11 @@ impl Encoder for TrackedEncoder {
     fn set_wire_chunking(&mut self, shard_payload: usize) {
         self.inner.set_wire_chunking(shard_payload)
     }
+    // Forwarded for the same reason as `set_wire_chunking` above — an unforwarded default here
+    // would silently leave the in-place backends pipelining past the capturer's ring.
+    fn set_input_ring_depth(&mut self, depth: usize) {
+        self.inner.set_input_ring_depth(depth)
+    }
     fn poll(&mut self) -> Result<Option<EncodedFrame>> {
         self.inner.poll()
     }

crates/pf-frame/src/dxgi.rs

@@ -52,6 +52,12 @@ pub struct PyroFrameShare {
     /// The fence value the capturer signalled after THIS frame's convert. The encoder's Vulkan
     /// acquire waits on it, so the wavelet read is ordered after the D3D11 CSC.
     pub fence_value: u64,
+    /// The capturer's ring generation, bumped every time it recreates its texture ring. The
+    /// PyroWave encoder caches its plane imports keyed on the texture's COM address, which carries
+    /// no reference — after a recreate those addresses can be recycled by the allocator, so a
+    /// cached import may describe a texture that no longer exists. The encoder flushes its import
+    /// cache whenever this changes, making cache identity independent of allocator behaviour.
+    pub ring_gen: u32,
 }
 
 /// A GPU-resident captured texture (the Windows zero-copy path: NVENC/AMF/QSV encode it in place;

crates/punktfunk-host/src/native/stream.rs

@@ -1477,6 +1477,9 @@ pub(super) fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDispl
                         if let Some(c) = plan.wire_chunk {
                             new_enc.set_wire_chunking(c);
                         }
+                        // (`max_depth` is computed later in the iteration — read the capturer
+                        // directly so an ABR rebuild re-establishes the bound immediately.)
+                        new_enc.set_input_ring_depth(capturer.pipeline_depth().max(1));
                         enc = new_enc;
                         bitrate_kbps = new_kbps;
                         live_bitrate.store(new_kbps, Ordering::Relaxed);
@@ -2265,6 +2268,9 @@ fn try_inplace_resize(
     if let Some(c) = plan.wire_chunk {
         new_enc.set_wire_chunking(c);
     }
+    // Re-report the capturer's ring depth: in-place backends bound async pipelining by it, and a
+    // rebuilt encoder starts with it unset.
+    new_enc.set_input_ring_depth(capturer.pipeline_depth().max(1));
     *enc = new_enc;
     *frame = new_frame;
     *interval = std::time::Duration::from_secs_f64(1.0 / effective_hz.max(1) as f64);
@@ -2579,6 +2585,10 @@ fn build_pipeline(
     if let Some(c) = plan.wire_chunk {
         enc.set_wire_chunking(c);
     }
+    // Tell in-place backends (Windows direct-NVENC) how deep they may pipeline against the
+    // capturer's texture ring — without it they use only the env/pool cap and can encode a texture
+    // the capturer has already rotated and overwritten.
+    enc.set_input_ring_depth(capturer.pipeline_depth().max(1));
     // Post-open cross-check: the Welcome already committed `chroma_format` from the pre-open probe, so
     // warn loudly if the encoder actually opened a different chroma than negotiated (the in-band SPS is
     // authoritative for the decoder, but a mismatch means the probe and the live open disagreed).