recording: preserve incoming timestamps by default (#993)

Update the recorder to use the new get_packet API that carries
timestamps.

We have timestamps coming in from packets, so don't attempt to re-write
them which could cause playback-rate irregularities.

Also do a bit of cleanup around legacy code:

* Remove unused env vars
* Remove now-unneeded timestamp rewriting helpers in the recorder
* Remove some other unused code, eg `recording_coordinator.put`

---------

Signed-off-by: Josh Allmann <joshua.allmann@gmail.com>

Author: j0sh

src/scope/cloud/livepeer_app.py

@@ -466,7 +466,9 @@ async def _media_output_loop(
             # TODO make this blocking; we busy-wait a LOT
             frame_item = None
             if record_node_id is not None:
-                frame_item = frame_processor.sink_manager.recording.get(record_node_id)
+                frame_item = frame_processor.sink_manager.recording.get_packet(
+                    record_node_id
+                )
                 if frame_item is None:
                     await asyncio.sleep(0.01)  # no frame yet, wait a bit
                     continue

src/scope/server/recording.py

@@ -1,17 +1,15 @@
 """Recording-related utility functions for cleanup and download handling."""
 
-import fractions
 import logging
 import os
 import shutil
 import tempfile
 import threading
-import time
 from pathlib import Path
 
 from aiortc import MediaStreamTrack
 from aiortc.contrib.media import MediaRecorder, MediaRelay
-from aiortc.mediastreams import VIDEO_CLOCK_RATE, VIDEO_TIME_BASE
+from av import VideoFrame
 
 logger = logging.getLogger(__name__)
 
@@ -21,118 +19,28 @@
     "download": "scope_download_",
 }
 
-# Environment variables
-RECORDING_ENABLED = os.getenv("RECORDING_ENABLED", "false").lower() == "true"
-RECORDING_STARTUP_CLEANUP_ENABLED = (
-    os.getenv("RECORDING_STARTUP_CLEANUP_ENABLED", "true").lower() == "true"
-)
-
 RECORDING_MAX_FPS = 30.0  # Must match MediaRecorder's hardcoded rate=30
 
 
-class TimestampNormalizingTrack(MediaStreamTrack):
-    """Wraps a track and assigns wall-clock timestamps starting from 0.
-
-    Uses monotonic wall-clock time to compute PTS so that the recorded
-    MP4 plays back at real-time speed regardless of the source track's
-    own PTS cadence.  This is critical for cloud-relay recordings where
-    frames may arrive slower than the source track's nominal rate (e.g.
-    CloudTrack stamps every frame at 1/30 s intervals even when network
-    round-trips deliver them at 10-15 FPS).
-
-    Important: We must create a copy of the frame rather than modifying it
-    in place, because the relay shares frame objects across all subscribers.
-    Modifying in place would affect the WebRTC sender and cause encoding errors.
-    """
-
-    def __init__(self, source_track: MediaStreamTrack):
-        super().__init__()
-        self.kind = source_track.kind
-        self._source = source_track
-        self._start_time: float | None = None
-        self._last_frame_time: float | None = None
-        self._min_frame_interval = 1.0 / RECORDING_MAX_FPS
-
-    async def recv(self):
-        import av
-
-        while True:
-            frame = await self._source.recv()
-
-            # Frame rate limiting - skip frames arriving faster than MAX_RECORDING_FPS
-            current_time = time.monotonic()
-            if self._last_frame_time is not None:
-                elapsed = current_time - self._last_frame_time
-                if elapsed < self._min_frame_interval:
-                    continue  # Skip this frame
-            self._last_frame_time = current_time
-
-            if self._start_time is None:
-                self._start_time = current_time
-
-            # Create a new frame with wall-clock-based timestamp.
-            # Pad to even dimensions — libx264 requires width and height divisible by 2.
-            arr = frame.to_ndarray(format="rgb24")
-            h, w = arr.shape[:2]
-            pad_w = w % 2
-            pad_h = h % 2
-            if pad_w or pad_h:
-                import numpy as np
-
-                arr = np.pad(arr, ((0, pad_h), (0, pad_w), (0, 0)), mode="edge")
-            new_frame = av.VideoFrame.from_ndarray(arr, format="rgb24")
-            new_frame.pts = int((current_time - self._start_time) * VIDEO_CLOCK_RATE)
-            new_frame.time_base = VIDEO_TIME_BASE
-            return new_frame
-
-    def stop(self):
-        self._source.stop()
-        super().stop()
-
-
-class AudioTimestampNormalizingTrack(MediaStreamTrack):
-    """Wraps an audio track and assigns wall-clock timestamps starting from 0.
-
-    Analogous to TimestampNormalizingTrack but for AudioFrame objects.
-    Uses wall-clock time for PTS to stay in sync with the video track's
-    wall-clock timestamps.  Unlike video, audio frames are not rate-limited
-    here because the source AudioProcessingTrack already paces at 20ms
-    intervals.
-    """
-
-    kind = "audio"
-
-    def __init__(self, source_track: MediaStreamTrack):
-        super().__init__()
-        self._source = source_track
-        self._start_time: float | None = None
-
-    async def recv(self):
-        from av import AudioFrame as AvAudioFrame
-
-        frame = await self._source.recv()
-
-        current_time = time.monotonic()
-        if self._start_time is None:
-            self._start_time = current_time
+def ensure_even_video_frame(frame: VideoFrame) -> VideoFrame:
+    """Pad odd-dimension video frames so encoders like libx264 accept them."""
+    pts = frame.pts
+    time_base = frame.time_base
+    arr = frame.to_ndarray(format="rgb24")
+    h, w = arr.shape[:2]
+    pad_w = w % 2
+    pad_h = h % 2
+    if not (pad_w or pad_h):
+        return frame
 
-        # Create a copy with wall-clock PTS (relay shares frame objects,
-        # so we must not mutate in place).
-        new_frame = AvAudioFrame(
-            format=frame.format.name,
-            layout=frame.layout.name,
-            samples=frame.samples,
-        )
-        new_frame.sample_rate = frame.sample_rate
-        new_frame.pts = int((current_time - self._start_time) * frame.sample_rate)
-        new_frame.time_base = fractions.Fraction(1, frame.sample_rate)
-        for i, plane in enumerate(frame.planes):
-            new_frame.planes[i].update(bytes(plane))
-        return new_frame
+    import numpy as np
 
-    def stop(self):
-        self._source.stop()
-        super().stop()
+    padded = np.pad(arr, ((0, pad_h), (0, pad_w), (0, 0)), mode="edge")
+    even_frame = VideoFrame.from_ndarray(padded, format="rgb24")
+    even_frame.pts = pts
+    if time_base is not None:
+        even_frame.time_base = time_base
+    return even_frame
 
 
 class RecordingManager:
@@ -182,42 +90,34 @@ def _stop_track_safe(track: MediaStreamTrack | None) -> None:
                 logger.warning(f"Error stopping recording track: {e}")
 
     def _create_recording_track(self) -> MediaStreamTrack | None:
-        """Create a video recording track.
-
-        Returns None if no video track is configured.  The track is wrapped
-        in TimestampNormalizingTrack to ensure frame timestamps start from 0
-        for each new recording.
-        """
+        """Create a video recording track, preserving source timestamps."""
         if self.video_track is None:
             return None
         if self.relay:
-            relay_track = self.relay.subscribe(self.video_track)
-            return TimestampNormalizingTrack(relay_track)
-        else:
-            logger.warning("No relay available for recording, using track directly")
-            return TimestampNormalizingTrack(self.video_track)
+            return self.relay.subscribe(self.video_track)
+        logger.warning("No relay available for recording, using track directly")
+        return self.video_track
 
     def _create_audio_recording_track(self) -> MediaStreamTrack | None:
-        """Create an audio recording track.
-
-        Returns None if no audio track is configured.
-        """
+        """Create an audio recording track, preserving source timestamps."""
         if self.audio_track is None:
             return None
         if self.audio_relay:
-            relay_track = self.audio_relay.subscribe(self.audio_track)
-            return AudioTimestampNormalizingTrack(relay_track)
-        else:
-            logger.warning(
-                "No audio relay available for recording, using track directly"
-            )
-            return AudioTimestampNormalizingTrack(self.audio_track)
+            return self.audio_relay.subscribe(self.audio_track)
+        logger.warning("No audio relay available for recording, using track directly")
+        return self.audio_track
 
     def _create_media_recorder(self, file_path: str) -> MediaRecorder:
         """Create a MediaRecorder instance with standard settings."""
         return MediaRecorder(
             file_path,
             format="mp4",
+            options={
+                # force timestamps to start at zero
+                "use_editlist": "0",
+                # allows playback before file is fully loaded, eg over http
+                "movflags": "+faststart",
+            },
         )
 
     async def start_recording(self):
@@ -450,12 +350,6 @@ def cleanup_recording_files():
     Clean up all recording files from previous sessions.
     This handles cases where the process crashed and files weren't cleaned up.
     """
-    if not RECORDING_STARTUP_CLEANUP_ENABLED:
-        logger.info(
-            "Recording startup cleanup disabled via RECORDING_STARTUP_CLEANUP_ENABLED"
-        )
-        return
-
     temp_dir = Path(tempfile.gettempdir())
     if not temp_dir.exists():
         return

src/scope/server/recording_coordinator.py

@@ -7,18 +7,21 @@
 import logging
 import queue
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
-import torch
+from .media_packets import VideoPacket, ensure_video_packet
 
-from .media_packets import ensure_video_packet
+if TYPE_CHECKING:
+    from .recording import RecordingManager
+    from .tracks import QueueVideoTrack
 
 logger = logging.getLogger(__name__)
 
 
 @dataclass
 class _RecordingEntry:
-    manager: object  # RecordingManager
-    track: object  # QueueVideoTrack
+    manager: "RecordingManager"
+    track: "QueueVideoTrack"
     stopped_file: str | None = None  # File path after stop (before download)
 
 
@@ -89,35 +92,21 @@ def _drain_queue(q: queue.Queue) -> int:
             except queue.Empty:
                 return dropped
 
-    def get(self, record_node_id: str) -> torch.Tensor | None:
-        """Read a frame from a record node's output queue."""
+    def get_packet(self, record_node_id: str) -> VideoPacket | None:
+        """Read a packet from a record node's output queue."""
         rec_q = self._record_queues.get(record_node_id)
         if rec_q is None:
             return None
         try:
-            frame = ensure_video_packet(rec_q.get_nowait()).tensor
+            packet = ensure_video_packet(rec_q.get_nowait())
+            frame = packet.tensor
             frame = frame.squeeze(0)
             if frame.is_cuda:
                 frame = frame.cpu()
-            return frame
+            return VideoPacket(tensor=frame, timestamp=packet.timestamp)
         except queue.Empty:
             return None
 
-    def put(self, record_node_id: str, frame: torch.Tensor) -> bool:
-        """Write a frame into a record node's queue (cloud mode).
-
-        Returns True if the frame was enqueued, False if the queue is
-        missing or full.
-        """
-        rec_q = self._record_queues.get(record_node_id)
-        if rec_q is None:
-            return False
-        try:
-            rec_q.put_nowait(frame)
-            return True
-        except queue.Full:
-            return False
-
     # ------------------------------------------------------------------
     # Recording lifecycle
     # ------------------------------------------------------------------

src/scope/server/sink_manager.py

@@ -10,12 +10,13 @@
 import queue
 import threading
 import time
+from fractions import Fraction
 from typing import TYPE_CHECKING, Any
 
 import numpy as np
 import torch
 
-from .media_packets import VideoPacket, ensure_video_packet
+from .media_packets import MediaTimestamp, VideoPacket, ensure_video_packet
 from .recording_coordinator import RecordingCoordinator
 
 if TYPE_CHECKING:
@@ -436,26 +437,30 @@ def _per_node_sink_loop(self, node_id: str, sink_type: str) -> None:
     # Recording
     # ------------------------------------------------------------------
 
-    def get_from_record(self, record_node_id: str):
-        """Read a frame from a record node's output queue."""
-        return self._recording.get(record_node_id)
-
     def put_to_record(self, node_id: str, frame) -> None:
         """Convert a VideoFrame to tensor and put it into a record node's queue."""
-        import torch
-
         rec_q = self._recording._record_queues.get(node_id)
         if rec_q is None:
             return
         try:
             frame_np = frame.to_ndarray(format="rgb24")
             t = torch.as_tensor(frame_np, dtype=torch.uint8).unsqueeze(0)
+            timestamp = MediaTimestamp()
+            if (
+                getattr(frame, "pts", None) is not None
+                and getattr(frame, "time_base", None) is not None
+            ):
+                timestamp = MediaTimestamp(
+                    pts=frame.pts,
+                    time_base=Fraction(frame.time_base),
+                )
+            packet = VideoPacket(tensor=t, timestamp=timestamp)
             try:
-                rec_q.put_nowait(t)
+                rec_q.put_nowait(packet)
             except queue.Full:
                 try:
                     rec_q.get_nowait()
-                    rec_q.put_nowait(t)
+                    rec_q.put_nowait(packet)
                 except queue.Empty:
                     pass
         except Exception as e:

src/scope/server/tracks.py

@@ -16,6 +16,7 @@
 
 from .media_packets import VideoPacket, ensure_video_packet
 from .pipeline_manager import PipelineManager
+from .recording import ensure_even_video_frame
 
 if TYPE_CHECKING:
     from .frame_processor import FrameProcessor
@@ -135,8 +136,8 @@ async def recv(self) -> VideoFrame:
             if frame_squeezed.is_cuda:
                 frame_squeezed = frame_squeezed.cpu()
 
-            video_frame = VideoFrame.from_ndarray(
-                frame_squeezed.numpy(), format="rgb24"
+            video_frame = ensure_even_video_frame(
+                VideoFrame.from_ndarray(frame_squeezed.numpy(), format="rgb24")
             )
             if packet.timestamp.is_valid:
                 await _pace_preserved_timestamp(self, self._pacing, packet)
@@ -181,7 +182,9 @@ async def recv(self) -> VideoFrame:
             packet = self._frame_getter(fp)
             if packet is not None:
                 packet = ensure_video_packet(packet)
-                frame = VideoFrame.from_ndarray(packet.tensor.numpy(), format="rgb24")
+                frame = ensure_even_video_frame(
+                    VideoFrame.from_ndarray(packet.tensor.numpy(), format="rgb24")
+                )
                 if packet.timestamp.is_valid:
                     await _pace_preserved_timestamp(self, self._pacing, packet)
                     frame.pts = packet.timestamp.pts