Record every frame of a batched emit and downmix stereo audio

src/reactor_runtime/recording/recorder.py

@@ -36,6 +36,7 @@
 from reactor_runtime.log import get_logger
 from reactor_runtime.recording.chunk_encoder import ChunkEncoder
 from reactor_runtime.recording.markers import MarkerBookkeeper
+from reactor_runtime.transport.webrtc.frames import to_int16_mono
 
 logger = get_logger(__name__)
 
@@ -45,6 +46,15 @@
 # model that runs slower or faster than real time still records at true duration.
 RECORDING_FPS = 30
 
+# A model may hand the fan-out several seconds of media in one emit (a batching
+# model produces a whole generation window at a time), and `on_chunk` queues the
+# entire resampled burst before the emit thread moves on to the connection
+# pacers — which is where it pays out the same media in real time, giving the
+# encoder that long to drain. The queue therefore has to hold the largest burst
+# a single emit produces on the recording grid; two seconds covers every
+# current model with margin while bounding memory to that many frames.
+_FEED_QUEUE_MAX_FRAMES = 2 * RECORDING_FPS
+
 _INIT_FILENAME = "init.mp4"
 # Written into a recording's directory once it is finished, so its final segment
 # (which has no successor to prove it closed) is recognised as fetchable.
@@ -205,7 +215,9 @@ def __init__(
         self._started = False
         self._disabled = False
 
-        self._feed_queue: queue.Queue[_FeedItem | None] = queue.Queue(maxsize=4)
+        self._feed_queue: queue.Queue[_FeedItem | None] = queue.Queue(
+            maxsize=_FEED_QUEUE_MAX_FRAMES
+        )
         self._feed_thread: threading.Thread | None = None
         self._feed_stop = threading.Event()
         self._watch_thread: threading.Thread | None = None
@@ -463,13 +475,16 @@ def on_chunk(self, chunk: MediaChunk) -> None:
             try:
                 self._feed_queue.put_nowait((video_data, audio_data))
             except queue.Full:
+                # Drop this grid slot (its audio was already pulled, so the
+                # tracks stay aligned) and keep going: the encoder drains
+                # concurrently, so later slots in the same chunk may still fit.
                 self._dropped_frames += 1
                 if self._dropped_frames == 1 or self._dropped_frames % 300 == 0:
                     logger.warning(
                         "recorder feed queue full; dropping a frame to keep the model unblocked",
                         dropped_total=self._dropped_frames,
                     )
-                break
+                continue
             fed += 1
         if fed:
             markers.advance(fed / RECORDING_FPS)
@@ -558,14 +573,18 @@ def _buffer_audio(self, bundle: MediaBundle) -> None:
 
         The whole chunk's audio is buffered once; :meth:`_take_audio` then pulls a
         grid slot's worth per recorded frame, so the audio DTS tracks the video
-        PTS regardless of how the chunk's frames map onto the grid.
+        PTS regardless of how the chunk's frames map onto the grid. The track is
+        reduced through :func:`to_int16_mono` — the same reduction the live
+        transport applies — so multi-channel audio is mixed down and float
+        samples are scaled, and the buffer always holds the encoder's mono
+        int16 form.
         """
         if self._audio_track is None:
             return
         track = bundle.get_track(self._audio_track)
         if track is None or track.data.size == 0:
             return
-        flat = np.ascontiguousarray(track.data, dtype=np.int16).reshape(-1)
+        flat = np.ascontiguousarray(to_int16_mono(track.data))
         self._audio_jitter_buf.append(flat)
         self._audio_buffered_samples += int(flat.size)
         cap = self._audio_sample_rate

tests/unit/recording/test_recorder.py

@@ -309,6 +309,74 @@ def _av_bundle(width: int, height: int) -> MediaBundle:
     )
 
 
+def _batched_av_bundle(n_frames: int) -> MediaBundle:
+    """A bundle carrying a whole emit burst: *n_frames* of video plus stereo audio."""
+    frames = np.zeros((n_frames, 8, 8, 3), dtype=np.uint8)
+    video = TrackInfo(
+        name="main_video", kind=TrackKind.VIDEO, rate=float(n_frames), direction=TrackDirection.OUT
+    )
+    audio = TrackInfo(
+        name="main_audio", kind=TrackKind.AUDIO, rate=48_000.0, direction=TrackDirection.OUT
+    )
+    left = np.full((1, 48_000), 100, dtype=np.int16)
+    right = np.full((1, 48_000), 300, dtype=np.int16)
+    samples = np.concatenate([left, right], axis=0)
+    return MediaBundle(
+        tracks={
+            "main_video": TrackData(info=video, data=frames),
+            "main_audio": TrackData(info=audio, data=samples),
+        }
+    )
+
+
+def _park_feed_worker(recorder: Recorder) -> None:
+    """Stop the feed worker so `on_chunk`'s queueing is observable."""
+    recorder._feed_stop.set()
+    feed_thread = recorder._feed_thread
+    assert feed_thread is not None
+    feed_thread.join(timeout=2.0)
+
+
+def test_a_batched_emit_queues_every_grid_frame(tmp_path: Path) -> None:
+    # A batching model hands the fan-out a whole second of media in one emit.
+    # Every grid frame of that burst has to fit the feed queue up front: the
+    # emit thread only pays out real time in the connection pacers *after*
+    # `on_chunk` returns, so a queue smaller than the burst loses the excess
+    # before the encoder ever gets a chance to drain it.
+    recorder = Recorder(RecordingConfig(enabled=True, recording_dir=str(tmp_path)))
+    recorder.start(_SID)
+    try:
+        _park_feed_worker(recorder)
+
+        recorder.on_chunk(MediaChunk(bundle=_batched_av_bundle(24), fps=24.0, n_frames=24))
+
+        assert recorder._dropped_frames == 0
+        assert recorder._feed_queue.qsize() == 30
+    finally:
+        recorder.stop()
+
+
+def test_stereo_audio_is_downmixed_to_mono(tmp_path: Path) -> None:
+    # A `(2, M)` stereo track mixes down per sample, the same way the live
+    # transport does. Flattening it channel-after-channel instead would fill
+    # the first grid slots entirely with left-channel samples.
+    recorder = Recorder(RecordingConfig(enabled=True, recording_dir=str(tmp_path)))
+    recorder.start(_SID)
+    try:
+        _park_feed_worker(recorder)
+
+        recorder.on_chunk(MediaChunk(bundle=_batched_av_bundle(24), fps=24.0, n_frames=24))
+
+        item = recorder._feed_queue.get_nowait()
+        assert item is not None
+        _video, audio = item
+        assert audio is not None
+        assert audio.dtype == np.int16
+        assert np.all(audio == 200)
+    finally:
+        recorder.stop()
+
+
 def test_a_saturated_feed_queue_drops_a_frame_and_keeps_recording(tmp_path: Path) -> None:
     # An encoder that falls behind costs frames, never the session. The queue is
     # the only thing between the model thread and the encoder, so a full one makes