Nudge non-monotonic source DTS instead of crashing the mux

jasna/media/video_encoder.py

@@ -658,6 +658,7 @@ def _setup_source_streams(self, in_v) -> None:
         self._source_pipes: dict[int, tuple[str, object, object]] = {}
         self._source_backlog: deque = deque()
         self._source_iter = None
+        self._last_source_dts: dict[int, int] = {}
         if self.smart_fragment:
             return
 
@@ -919,7 +920,7 @@ def _pump_source_streams(self, upto_seconds: float | None):
                 ):
                     return
                 self._source_backlog.popleft()
-                self.dst.mux(packet)
+                self._mux_source_packet(packet)
                 continue
             in_packet = next(self._source_iter, None)
             if in_packet is None:
@@ -937,7 +938,26 @@ def _drain_source_streams(self):
                 packets.extend(out_stream.encode(rframe))
             packets.extend(out_stream.encode(None))
             for packet in packets:
-                self.dst.mux(packet)
+                self._mux_source_packet(packet)
+
+    def _mux_source_packet(self, packet):
+        # Sloppy sources (e.g. web remuxes with 1/1000 audio time bases) can
+        # carry duplicate/backwards DTS; the mp4 muxer hard-fails on them, so
+        # nudge forward like ffmpeg's CLI does instead of crashing the job.
+        if packet.dts is not None:
+            last = self._last_source_dts.get(packet.stream.index)
+            if last is not None and packet.dts <= last:
+                logger.warning(
+                    "Non-monotonic DTS %s (last %s) in source output stream %s; nudging forward",
+                    packet.dts,
+                    last,
+                    packet.stream.index,
+                )
+                packet.dts = last + 1
+                if packet.pts is not None and packet.pts < packet.dts:
+                    packet.pts = packet.dts
+            self._last_source_dts[packet.stream.index] = packet.dts
+        self.dst.mux(packet)
 
     def _clamp_pts_monotonic(self, pts: int) -> int:
         last = self._last_emitted_pts

tests/test_video_encoder_unit.py

@@ -1019,6 +1019,7 @@ def _source_encoder(self, tmp_path, packets):
         enc._source_pipes = {1: ("copy", out_a, None)}
         enc._source_backlog = deque()
         enc._source_iter = iter(packets)
+        enc._last_source_dts = {}
         enc.dst = MagicMock()
         return enc, out_a
 
@@ -1048,6 +1049,16 @@ def test_pump_respects_threshold(self, tmp_path):
         enc._pump_source_streams(None)
         assert enc.dst.mux.call_count == 3
 
+    def test_pump_nudges_non_monotonic_dts(self, tmp_path):
+        packets = [_packet(1, dts=0), _packet(1, dts=370390), _packet(1, dts=370390)]
+        enc, _ = self._source_encoder(tmp_path, packets)
+
+        enc._pump_source_streams(None)
+
+        muxed = [call.args[0] for call in enc.dst.mux.call_args_list]
+        assert [p.dts for p in muxed] == [0, 370390, 370391]
+        assert muxed[-1].pts == 370391
+
     def test_pump_without_source_streams_is_noop(self, tmp_path):
         enc = _make_encoder(tmp_path)
         enc._source_iter = None