fix(rec): bounded encoder queue + drop policy + burned frame counter — the "1...23...4...56..." smeared-ticker take

slice 9 made the DURATION right but content still hiccuped; aggregates (301/300, uniform file PTS) could not see it. Measured root cause: FfmpegEncoder.SubmitFrameAsync BLOCKED on WriteAsync(8.3MB)+FlushAsync when ffmpeg lagged the pipe, and the burst while-loop re-wrote that same stale composite per crossed slot — frozen runs. OBS shape (derivative, wrapped pre-1.0): the encoder queue in libobs/obs-encoder.c — encoder thread never couples back into the video thread; overflow = dropped data, never a frozen producer. https://github.com/obsproject/obs-studio/blob/master/libobs/obs-encoder.c - FfmpegEncoder: SubmitFrameAsync is now an enqueue (ArrayPool copy) into a bounded Channel (cap 120) drained by its own task; drop-newest + count when full; StopAsync flushes the queue then EOF (TryComplete). IFfmpegEncoder.DroppedFrames. - FramePump: ONE fresh composite per iteration (burst loop deleted); worst-submit stat, stall logger (>2x interval names the stage), dropped/stalls in stats. - Burned-in 6-digit dot-matrix frame counter (white box, bottom-right) on every composite — the clock-independent judge replacing the WSL ticker: +1/frame, jumps = counted drops. - ONE new test Backpressure_QueueOverflow_DropsFrames_AndNeverBlocks (slow-sink fake: submit never blocks, drops counted, stop flushes exactly submitted-minus-dropped). - Full suite 290 tests, 289 pass — sole failure the pre-existing compositor pixel test. - Docs same-commit: ai.md slice 10 (+ encoder/stop-note corrections), MyMistakes point 8, HANDOFF. Audio untouched (queued follow-up); web overlay still frozen pending timing closure.
2026-09-10 09:32:55 -07:00
parent bd396e488c
commit c45cbc93b7
8 changed files with 414 additions and 79 deletions
@@ -1,4 +1,6 @@
+using System.Buffers;
 using System.Diagnostics;
+using System.Threading.Channels;
 using ytLive.Helpers;
 using ytLive.Models;

@@ -30,6 +32,21 @@ public sealed class FfmpegEncoder : IFfmpegEncoder
    private Task? _stderrLoop;
    private bool _stopRequested;

+    // Pipe decoupling (slice 10, 2026-09-10): a bounded pending-frame queue drained
+    // by its own task (the OBS video-thread → encoder-queue model — the encoder's
+    // thread never couples back into the video thread; libobs obs-encoder.c). The
+    // pump's SubmitFrameAsync now ENQUEUES (copying into a pooled buffer) instead of
+    // blocking on WriteAsync when ffmpeg lags the pipe; overflow DROPS the newest
+    // frame (skip-newest) and counts it. BoundedChannelFullMode.Wait + TryWrite gives
+    // exactly that: when full, TryWrite returns false and the caller drops the item.
+    private const int QueueCapacity = 120; // ~2s at the tier's 60fps
+    private readonly Channel<PendingWrite> _frames = Channel.CreateBounded<PendingWrite>(
+        new BoundedChannelOptions(QueueCapacity) { SingleReader = true });
+    private Task? _drainTask;
+    private int _droppedBackpressure;
+
+    private readonly record struct PendingWrite(byte[] Buffer, int Length);
+
    public FfmpegEncoder(
        IFfmpegLocator locator,
        Func<IEncoderProcess>? processFactory = null)
@@ -40,6 +57,13 @@ public sealed class FfmpegEncoder : IFfmpegEncoder

    public bool IsRunning { get; private set; }

+    /// <summary>Frames dropped by the bounded queue because the drain task couldn't
+    /// keep the pipe fed (the encoder lagging the real-time capture rate). The pump
+    /// reports this in its stats; the burned-in frame counter in the recording
+    /// shows the identical jumps. Distinct from <see cref="StreamHealth.DroppedFrames"/>
+    /// (ffmpeg's own progress-derived estimate).</summary>
+    public int DroppedFrames => Volatile.Read(ref _droppedBackpressure);
+
    public async Task StartAsync(EncoderOptions options, CancellationToken cancellationToken = default)
    {
        if (options == null) throw new ArgumentNullException(nameof(options));
@@ -93,26 +117,40 @@ public sealed class FfmpegEncoder : IFfmpegEncoder

        _stopRequested = false;
        _stderrLoop = RunStderrLoopAsync(process);
+        StartDrainLoop(process);
    }

    /// <summary>
-    /// Write one raw BGRA frame to ffmpeg's stdin. Serialized internally; callers
-    /// (the compositor pump) may race freely. Frames are written as-is — the caller
-    /// paces to capture rate (the compositor's job, ship step 5).
+    /// Enqueue one raw BGRA frame for ffmpeg's stdin (slice 10). The caller races
+    /// freely (the compositor pump); this never blocks on the pipe. The pixels are
+    /// copied into a pooled buffer first because — unlike the old direct write — the
+    /// write happens later on the drain thread, so the caller may (and does) recycle
+    /// its scratch buffer the moment this returns. When the bounded queue is full the
+    /// NEWEST frame is dropped and counted (freshness over coverage; the encoder is
+    /// already behind, so the stale-content failure mode never encodes).
    /// </summary>
-    public async Task SubmitFrameAsync(VideoFrame frame, CancellationToken cancellationToken = default)
+    public Task SubmitFrameAsync(VideoFrame frame, CancellationToken cancellationToken = default)
    {
        if (frame == null) throw new ArgumentNullException(nameof(frame));
-        IEncoderProcess? process;
        lock (_gate)
        {
            if (!IsRunning) throw new InvalidOperationException("The encoder is not running.");
-            process = _process;
        }

-        var bytes = frame.BgraPixels;
-        await process!.StandardInput.WriteAsync(bytes, cancellationToken).ConfigureAwait(false);
-        await process.StandardInput.FlushAsync(cancellationToken).ConfigureAwait(false);
+        if (cancellationToken.IsCancellationRequested)
+            return Task.FromCanceled(cancellationToken);
+
+        var pooled = ArrayPool<byte>.Shared.Rent(frame.BgraPixels.Length);
+        Buffer.BlockCopy(frame.BgraPixels, 0, pooled, 0, frame.BgraPixels.Length);
+        if (_frames.Writer.TryWrite(new PendingWrite(pooled, frame.BgraPixels.Length)))
+            return Task.CompletedTask;
+
+        // Queue full (encoder lagging) or the channel completed during stop — drop the
+        // newest and count it. The caller never blocks; the encoder catches up or the
+        // session ends with a (reported) short gap instead of a frozen stall.
+        ArrayPool<byte>.Shared.Return(pooled);
+        Interlocked.Increment(ref _droppedBackpressure);
+        return Task.CompletedTask;
    }

    public async Task StopAsync(CancellationToken cancellationToken = default)
@@ -127,13 +165,17 @@ public sealed class FfmpegEncoder : IFfmpegEncoder
            loop = _stderrLoop;
        }

+        // Flush every queued frame, then EOF: TryComplete lets the drain task write
+        // the buffered frames, close stdin → ffmpeg finalizes and exits by itself
+        // (the "did not exit after EOF" kill below is only the backstop).
        try
        {
-            process!.StandardInput.Dispose(); // EOF → ffmpeg finalizes + exits
+            _frames.Writer.TryComplete();
+            if (_drainTask != null) await _drainTask.ConfigureAwait(false);
        }
        catch (Exception ex)
        {
-            AppLog.Write(ex, "FFmpeg encoder: closing stdin failed");
+            AppLog.Write(ex, "FFmpeg encoder: drain loop faulted during stop");
        }

        try
@@ -178,6 +220,7 @@ public sealed class FfmpegEncoder : IFfmpegEncoder

    public void Dispose()
    {
+        _frames.Writer.TryComplete(); // unblock a stuck drain loop alongside the kill
        lock (_gate)
        {
            if (!IsRunning) return;
@@ -188,6 +231,54 @@ public sealed class FfmpegEncoder : IFfmpegEncoder
        }
    }

+    /// <summary>
+    /// The drain task (slice 10): owns every stdin write, so pipe backpressure — the
+    /// pump's old stall — lives HERE, on a thread the pump never touches. Frames are
+    /// written exactly as queued (the pooled array through its recorded length —
+    /// ArrayPool may return a larger buffer), returned to the pool after the write
+    /// copies into the pipe, and stdin is closed (EOF) once the queue drains.
+    /// </summary>
+    private void StartDrainLoop(IEncoderProcess process)
+    {
+        _drainTask = Task.Run(async () =>
+        {
+            try
+            {
+                try
+                {
+                    while (await _frames.Reader.WaitToReadAsync().ConfigureAwait(false))
+                    {
+                        while (_frames.Reader.TryRead(out var pending))
+                        {
+                            await process.StandardInput.WriteAsync(pending.Buffer, 0, pending.Length)
+                                .ConfigureAwait(false);
+                            await process.StandardInput.FlushAsync().ConfigureAwait(false);
+                            ArrayPool<byte>.Shared.Return(pending.Buffer);
+                        }
+                    }
+                }
+                catch (ChannelClosedException)
+                {
+                    // channel completed while waiting — fall through to EOF
+                }
+                catch (Exception ex)
+                {
+                    AppLog.Write(ex, "FFmpeg encoder: drain loop write faulted");
+                }
+                finally
+                {
+                    // EOF on every path — a byte that sits in the pool when the pipe
+                    // breaks is garbage anyway, and ffmpeg MUST see the close to finalize.
+                    process.StandardInput.Dispose();
+                }
+            }
+            catch (Exception ex)
+            {
+                AppLog.Write(ex, "FFmpeg encoder: drain loop faulted");
+            }
+        });
+    }
+
    private async Task<string> ProbeEncoderAsync(string ffmpegPath, CancellationToken cancellationToken)
    {
        try