fix(goLive): tail-drop emitter clock — always show the freshest frame, never lag

The Node token-bucket pacer used HEAD-drop (emit frames in arrival order,
drop newer ones when over budget). Under the encoder's ~330fps burst (ffmpeg
-re does not reliably throttle YouTube-DASH webm), the viewer was watching
frames ~10s behind live → frozen / 'patah-patah' video while audio (not
rate-limited) played current = desync.

Replace it with a steady setInterval emission clock at videoFps: each tick
emits exactly ONE frame — the NEWEST buffered one — and discards everything
older (tail-drop). At most one frame is ever held, so no backlog and no lag;
the emit clock (not the encoder rate) defines playback speed. Keyframes are
never superseded so the decoder keeps getting IDRs. Audio stays in sync.

services/discord-gateway/src/goLive/Demuxer.ts

@@ -375,16 +375,18 @@ export async function demux(
   // NALs and flush one frame per slice, prepending the parameter sets that
   // precede it, and timestamp it as ONE frame at the video frame rate.
 
-  // PACING: a token-bucket rate limiter enforces 1x video output at the
-  // encoder's frame rate, independent of how fast ffmpeg produces frames.
-  // ffmpeg `-re` was tried here (see spawn args) but does NOT reliably
-  // throttle a multi-stage live pipe (encoder x264 -> NUT -> demuxer): in
-  // production the demuxer still emitted ~240fps while the sender consumed
-  // 30fps, building a 100k+ frame backlog → video frozen on the oldest
-  // buffered frame while audio (tiny, jitter-buffer recoverable) stayed
-  // smooth. A Node-level limiter is deterministic and never stalls the
-  // ffmpeg process, so audio on fd3 keeps flowing. Surplus video frames are
-  // DROPPED (not buffered) so the sender always emits the newest frame.
+  // EMISSION CLOCK: a steady setInterval at the video frame period is the
+  // authoritative real-time clock for OUTPUT. Whatever the encoder's actual
+  // production rate (it bursts ~330fps in production because ffmpeg `-re`
+  // does not reliably throttle YouTube-DASH webm and may read a local file
+  // instantly), we always emit exactly ONE frame per tick — the NEWEST one
+  // we have buffered — and discard everything older. This is TAIL-DROP: the
+  // viewer always sees the freshest picture, so video and audio stay in sync
+  // and the in-flight buffer can never grow (we keep at most one frame). The
+  // previous token-bucket design used HEAD-DROP (emit in arrival order,
+  // dropping later frames) which, under the encoder burst, left the viewer
+  // watching frames ~10s behind live → frozen / "patah-patah" video while
+  // audio (not rate-limited) played current → desync. Tail-drop fixes that.
   let videoBuf = Buffer.alloc(0);
   let frameCount = 0;
   let pendingNals: Buffer[] = [];
@@ -395,28 +397,13 @@ export async function demux(
   // setting); it drives both RTP timestamp advance and pacing.
   const videoFps =
     opts.frameRate ?? (vInfo.framerate_num / vInfo.framerate_den || 30);
-  // token-bucket: capacity = 1s of frames; refill one token per 1000/fps ms.
-  const tokenIntervalMs = 1000 / videoFps;
-  let tokens = videoFps; // start with 1s of credit
-  let lastToken = Date.now();
-  let droppedFrames = 0;
+  const emitIntervalMs = 1000 / videoFps;
 
-  const tryConsume = (): boolean => {
-    const now = Date.now();
-    const elapsed = now - lastToken;
-    if (elapsed >= tokenIntervalMs) {
-      tokens = Math.min(
-        videoFps,
-        tokens + Math.floor(elapsed / tokenIntervalMs),
-      );
-      lastToken = now - (elapsed % tokenIntervalMs);
-    }
-    if (tokens >= 1) {
-      tokens -= 1;
-      return true;
-    }
-    return false;
-  };
+  // The single frame we will emit on the next tick. Only the newest survives;
+  // keyframes are never superseded so the decoder always gets its IDRs.
+  let latestAu: Buffer | null = null;
+  let latestIsKey = false;
+  let skippedFrames = 0; // frames superseded before they could be shown
 
   const flushAccessUnit = () => {
     if (pendingNals.length === 0) return;
@@ -430,18 +417,28 @@ export async function demux(
     pendingNals = [];
     pendingHasSlice = false;
     pendingIsKey = false;
-    // Drop surplus frames: if no token is available, discard this AU so the
-    // sender never emits a stale (old) frame. Keyframes are forced through
-    // even when over budget so the decoder always has a fresh IDR to recover.
-    if (!isKey && !tryConsume()) {
-      droppedFrames++;
-      if (droppedFrames === 1 || droppedFrames % 300 === 0) {
-        console.log(
-          `[goLive:Demuxer] drop=${droppedFrames} (pacing ${videoFps}fps; encoder burst)`,
-        );
-      }
-      return;
+    // Tail-drop: keep only the newest frame. A keyframe always wins (never
+    // superseded by later non-keys in the same tick window) so the decoder
+    // keeps getting IDRs; a non-key only replaces a pending non-key.
+    if (isKey) {
+      latestAu = au;
+      latestIsKey = true;
+    } else if (latestAu === null || latestIsKey) {
+      latestAu = au;
+      latestIsKey = false;
+    } else {
+      skippedFrames++;
     }
+  };
+
+  // Steady real-time emission clock. Emits the freshest buffered frame once
+  // per tick (≈ videoFps); never buffers more than one, so no backlog and
+  // no lag. This — not the encoder rate — defines playback speed.
+  const emitTick = () => {
+    if (latestAu === null) return;
+    const au = latestAu;
+    const isKey = latestIsKey;
+    latestAu = null;
     vPipe.write({
       data: au,
       // One frame at videoFps: duration=1 in a 1/fps timebase →
@@ -458,11 +455,13 @@ export async function demux(
     frameCount++;
     if (frameCount === 1 || frameCount % 30 === 0) {
       console.log(
-        `[goLive:Demuxer] frames=${frameCount} last=${au.length}B key=${isKey}`,
+        `[goLive:Demuxer] frames=${frameCount} last=${au.length}B key=${isKey} skipped=${skippedFrames}`,
       );
     }
   };
 
+  const emitTimer = setInterval(emitTick, emitIntervalMs);
+
   if (proc.stdout) {
     proc.stdout.on("data", (chunk: Buffer) => {
       videoBuf = Buffer.concat([videoBuf, chunk]);
@@ -540,17 +539,21 @@ export async function demux(
     });
     proc.stdout.on("end", () => {
       flushAccessUnit();
+      emitTick(); // flush the final frame if any
+      clearInterval(emitTimer);
       vPipe.end();
       aPipe.end();
     });
   }
 
   proc.on("close", () => {
+    clearInterval(emitTimer);
     vPipe.end();
     aPipe.end();
   });
 
   const close = () => {
+    clearInterval(emitTimer);
     proc.kill("SIGTERM");
     vPipe.end();
     aPipe.end();