From c2666ef5eec54fb69511d60e037665589ba7cfb9 Mon Sep 17 00:00:00 2001 From: Patrick Date: Tue, 18 Aug 2026 21:41:18 +0900 Subject: [PATCH] Nudge non-monotonic source DTS instead of crashing the mux Sloppy source files (e.g. web remuxes with 1/1000 audio time bases) can carry duplicate or backwards DTS in their audio/subtitle streams. The copy path in _pump_source_streams muxed those packets verbatim, and the mp4 muxer hard-fails with 'non monotonically increasing dts', killing the whole job partway through. Track the last muxed DTS per output stream and nudge offending packets forward by one tick, mirroring what ffmpeg's CLI muxer does and what _clamp_pts_monotonic already does for the video stream. Co-Authored-By: Claude Fable 5 --- jasna/media/video_encoder.py | 24 ++++++++++++++++++++++-- tests/test_video_encoder_unit.py | 11 +++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/jasna/media/video_encoder.py b/jasna/media/video_encoder.py index f6b66489..815c660b 100644 --- a/jasna/media/video_encoder.py +++ b/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 diff --git a/tests/test_video_encoder_unit.py b/tests/test_video_encoder_unit.py index 0ad334fd..8776cbd3 100644 --- a/tests/test_video_encoder_unit.py +++ b/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