feat(65): the clip mux — a third output on the same ffmpeg, drained off the frame path

Slice 2 of TASK 65. 602 green, 0 warnings. **Clips are still off**: no `ClipBuffer` is passed, so the
ffmpeg command line is byte-for-byte what it was before this commit, and a test asserts exactly that
(the original args are a strict prefix of the new ones).

- `FfmpegArgs` appends a fragmented-MP4 output to `pipe:1` only when `ClipBuffer` is set. One-second
  fragments (`-frag_duration 1000000`) rather than the 4 s GOP, so a clip can start inside the window
  instead of at the last keyframe. Stats already own stderr, so stdout was free.
- `Fmp4FragmentSplitter` turns that byte stream into whole fragments: `ftyp`+`moov` is the init
  segment, each `moof`+`mdat` is a fragment. It parses box headers itself, survives a box being cut
  anywhere across reads, and **resyncs instead of throwing** — a clip fault must never take the stream
  down.
- `FfmpegEncoder` reads it on its own task, never blocking the frame path. Any exception logs and
  disables clips for the session; the broadcast is untouched.
- `IEncoderProcess` gained `StandardOutputBytes` (the seam already had the text reader for probing).

Next slices: the replay layer + retention, then the Settings switch with disabled-but-remembered controls.
This commit is contained in:
2026-11-10 12:25:19 -08:00
parent b93744ec7f
commit 5066d81aab
8 changed files with 251 additions and 0 deletions
+5
View File
@@ -50,6 +50,11 @@ public sealed class EncoderOptions
/// <summary>Name of the mixer's audio pipe (default <see cref="DefaultAudioPipeName"/>).</summary>
public string AudioPipeName { get; init; } = DefaultAudioPipeName;
/// <summary>TASK 65: when set, the encoder muxes a SECOND video output into this ring (fragmented
/// MP4 on stdout) for the replay buffer. Null = clips off, and the ffmpeg command line is exactly
/// what it was before the feature existed.</summary>
public Services.ClipRing? ClipBuffer { get; init; }
/// <summary>GOP in frames = Fps × 4s — the YouTube keyframe ≤ 4s compliance bound.</summary>
public int GopSize => Fps * 4;
}
+13
View File
@@ -52,6 +52,19 @@ public static class FfmpegArgs
args.AddRange(["-f", "mp4", options.RecordPath!]);
}
// TASK 65: the replay buffer is a third mux of the same encoded frames. Fragmented MP4 on
// stdout (stats already own stderr) — one fragment per second, so a clip can start anywhere
// inside the chosen window instead of at the last 4-second keyframe.
if (options.ClipBuffer != null)
{
AddVideoTags(args, options, videoEncoder);
args.AddRange(["-c:a", "aac",
"-ar", options.AudioSampleRate.ToString(), "-ac", options.AudioChannels.ToString()]);
args.AddRange(["-movflags", "+frag_keyframe+empty_moov+default_base_moof"]);
args.AddRange(["-frag_duration", "1000000"]);
args.AddRange(["-f", "mp4", "pipe:1"]);
}
return args;
}
+23
View File
@@ -118,6 +118,11 @@ public sealed class FfmpegEncoder : IFfmpegEncoder
_stopRequested = false;
_stderrLoop = RunStderrLoopAsync(process);
StartDrainLoop(process);
// TASK 65: the clip mux is a side channel. It reads its own thread, never blocks the frame
// path, and a fault disables clips instead of touching the stream.
if (options.ClipBuffer is { } clipRing)
_ = Task.Run(() => ReadClipOutputAsync(process, clipRing));
}
/// <summary>
@@ -239,6 +244,24 @@ public sealed class FfmpegEncoder : IFfmpegEncoder
/// 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 static async Task ReadClipOutputAsync(IEncoderProcess process, Services.ClipRing ring)
{
var splitter = new Fmp4FragmentSplitter(ring);
var buffer = new byte[64 * 1024];
try
{
var stdout = process.StandardOutputBytes;
int read;
while ((read = await stdout.ReadAsync(buffer).ConfigureAwait(false)) > 0)
splitter.Write(buffer.AsSpan(0, read));
}
catch (Exception ex)
{
// Clips stop; the broadcast does not.
AppLog.Write(ex, "Clip buffer reader failed — replay clips disabled for this session");
}
}
private void StartDrainLoop(IEncoderProcess process)
{
_drainTask = Task.Run(async () =>
+2
View File
@@ -21,6 +21,8 @@ public sealed class FfmpegEncoderProcess : IEncoderProcess
}
public Stream StandardInput => _process.StandardInput.BaseStream;
public Stream StandardOutputBytes => _process.StandardOutput.BaseStream;
public TextReader StandardOutput => _process.StandardOutput;
public TextReader StandardError => _process.StandardError;
+102
View File
@@ -0,0 +1,102 @@
using System;
using System.Collections.Generic;
using System.IO;
using System.Text;
namespace ytLive.Services.Encoder;
/// <summary>
/// TASK 65: turns the fragmented-MP4 byte stream ffmpeg writes to stdout into a playable ring.
///
/// <para>MP4 boxes are length-prefixed, so this needs no muxer cooperation: <c>ftyp</c>+<c>moov</c> are
/// the initialisation segment, and each <c>moof</c>+<c>mdat</c> pair is one fragment. The ring only ever
/// sees whole fragments, which is what makes a snapshot playable and a trim safe.</para>
///
/// <para>Deliberately forgiving: an unparseable header clears the buffer and waits for the next one. A
/// clip that fails must never take the encoder down with it — the stream is running through the same
/// process.</para>
/// </summary>
public sealed class Fmp4FragmentSplitter
{
private readonly ClipRing _ring;
private readonly List<byte> _pending = new(capacity: 64 * 1024);
private readonly MemoryStream _init = new();
private readonly MemoryStream _fragment = new();
private bool _inFragment;
public Fmp4FragmentSplitter(ClipRing ring) => _ring = ring;
public void Write(ReadOnlySpan<byte> data)
{
foreach (var b in data) _pending.Add(b);
Drain();
}
private void Drain()
{
while (true)
{
if (_pending.Count < 8) return;
long size = ((long)_pending[0] << 24) | ((long)_pending[1] << 16) | ((long)_pending[2] << 8) | _pending[3];
var type = Encoding.ASCII.GetString(_pending.GetRange(4, 4).ToArray());
var header = 8;
if (size == 1)
{
if (_pending.Count < 16) return;
size = 0;
for (var i = 8; i < 16; i++) size = (size << 8) | _pending[i];
header = 16;
}
if (size < header || size > int.MaxValue)
{
// Not a box boundary any more; drop what we have and resync on the next write.
_pending.Clear();
return;
}
if (_pending.Count < size) return;
var box = _pending.GetRange(0, (int)size).ToArray();
_pending.RemoveRange(0, (int)size);
Route(type, box);
}
}
private void Route(string type, byte[] box)
{
switch (type)
{
case "ftyp":
_init.Write(box);
break;
case "moov":
_init.Write(box);
_ring.SetInit(_init.ToArray()); // init is complete once moov lands
break;
case "moof":
FlushFragment();
_fragment.Write(box);
_inFragment = true;
break;
case "mdat":
if (!_inFragment) return; // mdat without a moof is not replayable
_fragment.Write(box);
FlushFragment();
break;
default:
if (_inFragment) _fragment.Write(box);
break;
}
}
private void FlushFragment()
{
if (_inFragment && _fragment.Length > 0)
_ring.Append(_fragment.ToArray());
_fragment.SetLength(0);
_inFragment = false;
}
}
+4
View File
@@ -17,6 +17,10 @@ public interface IEncoderProcess : IDisposable
Stream StandardInput { get; }
TextReader StandardOutput { get; }
/// <summary>TASK 65: the same stdout as raw bytes — the replay-buffer mux writes fragmented MP4
/// here while <c>-stats</c> keeps stderr for itself.</summary>
Stream StandardOutputBytes { get; }
TextReader StandardError { get; }
bool HasExited { get; }