From d4a8124fa4509f6d0471f2d6df089e035a03ac20 Mon Sep 17 00:00:00 2001 From: Avik Sethia Date: Tue, 11 Aug 2026 07:50:45 -0700 Subject: [PATCH] Record every frame of a batched emit and downmix stereo audio MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A model that emits media in multi-frame batches hands on_chunk a whole generation window at once. The feed queue held four frames and a full queue abandoned the rest of the chunk, so only the head of every burst was recorded: a session recorded as a ~7x time-lapse of itself. The queue now absorbs a two-second burst on the recording grid — the emit thread pays out the same media in real time in the connection pacers right after, which is when the encoder drains it — and an overflowing slot drops alone instead of taking the rest of the chunk with it. Stereo (2, M) audio flattened channel-after-channel into the jitter buffer, so recordings carried alternating blocks of one channel instead of the mix. The buffer now reduces tracks through to_int16_mono, the same reduction the live transport applies. Co-Authored-By: Claude Fable 5 Signed-off-by: Avik Sethia --- src/reactor_runtime/recording/recorder.py | 27 +++++++-- tests/unit/recording/test_recorder.py | 68 +++++++++++++++++++++++ 2 files changed, 91 insertions(+), 4 deletions(-) diff --git a/src/reactor_runtime/recording/recorder.py b/src/reactor_runtime/recording/recorder.py index a26b1d9a..8bbb0b96 100644 --- a/src/reactor_runtime/recording/recorder.py +++ b/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 diff --git a/tests/unit/recording/test_recorder.py b/tests/unit/recording/test_recorder.py index f2b73f6a..b71ca2c9 100644 --- a/tests/unit/recording/test_recorder.py +++ b/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