fix(encode): bound NVENC async pipelining by the capturer's texture ring

The Windows direct-NVENC backend registers and encodes the capturer's textures
IN PLACE (no CopyResource), so how deep it may pipeline is a property of the
CAPTURER, not of the encoder. It was bounded only by `async_inflight_cap()` —
`PUNKTFUNK_NVENC_ASYNC_DEPTH`, default 4, clamped to the output-bitstream pool
— which consults nothing about the capturer, while the comment at the
backpressure loop claimed it "keep[s] in-flight depth within the capturer's
texture ring". It never did.

The IDD-push capturer rotates `OUT_RING = 3` per delivered frame with no regard
for encode completion (its own invariant note says OUT_RING(3) > max
pipeline_depth(2)). With the default async depth of 4 the encoder can therefore
still be reading a texture the capturer has already handed out again and
overwritten: torn or mixed frames. It is visual corruption rather than UB, so it
fails silently and intermittently — the worst shape to diagnose from a field
report.

Adds `Encoder::set_input_ring_depth`, reported from `Capturer::pipeline_depth`,
and bounds the async backpressure loop by `min(async_inflight_cap(), depth)`.
For IDD-push that yields 2, matching the capturer's stated contract; backends
that copy their input, or are synchronous, ignore it.

Wired at ALL THREE encoder-creation sites (initial open, stall/resize rebuild,
ABR rebuild) and forwarded through `TrackedEncoder` — this crate has a
documented trap where an unforwarded defaulted trait method silently no-ops
through that wrapper, which has already bitten the direct-NVENC work once and
the wire-chunking probe once.

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

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/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/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/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).