fix: take a jump back in a camera's clock out of the timestamps at capture refs #4242

When a camera's clock steps back by under 10 s (an NTP sync, a clock reset),
the packets after the step carry dts values behind the ones before it. The
video store patched each one to last_dts + last_duration and logged "non
increasing dts, fixing." for every packet until the event ended, thousands
per event. Splitting the event at the step doesn't help on its own: in
passthrough an event has to start on a keyframe, and the walk back for one
crosses the step again.

Handle it the way ffmpeg's demuxer does, where packets enter the pipeline:
the Monitor keeps each stream's last dts and one offset shared by all the
camera's streams. When a stream steps back by more than 1 s, the offset moves
so the stream continues one frame after where it was, and every packet gets
the offset from then on. Monitor::Capture applies it before the stream
socket, the packet queue or anything else sees the packet, and Pause() resets
it when the capture session ends. The packet queue, the event's choice of
first keyframe and the video store then see one continuous timeline: nothing
is split, flagged or patched downstream, and no video is dropped. Steps under
1 s are reordering jitter, left to the video store's reorder queue; steps of
10 s or more keep FfmpegCamera's existing handling.

tests/zm_monitor_dts.cpp covers a continuous stream, jitter, the 8.9 s and
1.048 s steps, audio following the offset video set, a missing duration and
packets without a dts.

Alternative to #5212 and #5068.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/zm_monitor.cpp

@@ -3047,6 +3047,47 @@ bool Monitor::AudioLevelWanted(SystemTimePoint now) const {
       static_cast<uint32_t>(std::chrono::system_clock::to_time_t(now)));
 }
 
+// Keeps a camera's timestamps going forward across a jump back in its clock (an
+// NTP sync or a clock reset), the way ffmpeg's demuxer does with its timestamp
+// offset: when a stream's dts steps back by more than kDtsStepBack, the offset
+// shared by all the camera's streams moves so that the stream carries on one
+// frame after where it was, and every packet gets the offset from then on. The
+// queue, the event's choice of first keyframe and the video store then see one
+// continuous timeline, with nothing to split or patch (refs #4242). Smaller
+// steps are reordering jitter, left to the video store's reorder queue; steps of
+// 10 s or more never get here, FfmpegCamera drops them.
+int64_t Monitor::KeepDtsContinuous(AVPacket *packet, AVRational time_base) {
+  if (packet->dts == AV_NOPTS_VALUE) return dts_offset_;
+
+  int64_t dts = av_rescale_q(packet->dts, time_base, AV_TIME_BASE_Q) + dts_offset_;
+  auto last = last_dts_.find(packet->stream_index);
+  if (last != last_dts_.end()) {
+    int64_t step = dts - last->second;
+    if (step < -kDtsStepBack) {
+      // One frame on from the last packet: this packet's duration, else the
+      // stream's last step.
+      int64_t frame = (packet->duration > 0) ? av_rescale_q(packet->duration, time_base, AV_TIME_BASE_Q)
+                                             : last_dts_step_[packet->stream_index];
+      if (frame <= 0) frame = 1;
+      int64_t shift = last->second + frame - dts;
+      dts_offset_ += shift;
+      dts += shift;
+      Info("%s: stream %d timestamps jumped back %.3f seconds, continuing them from where they were",
+           name.c_str(), packet->stream_index, -step / 1000000.0);
+    } else if (step > 0) {
+      last_dts_step_[packet->stream_index] = step;
+    }
+  }
+  last_dts_[packet->stream_index] = dts;
+
+  if (dts_offset_) {
+    int64_t shift = av_rescale_q(dts_offset_, AV_TIME_BASE_Q, time_base);
+    packet->dts += shift;
+    if (packet->pts != AV_NOPTS_VALUE) packet->pts += shift;
+  }
+  return dts_offset_;
+}
+
 int Monitor::Capture() {
   if (!shared_data->capturing) {
     Debug(1, "Not capturing");
@@ -3093,6 +3134,13 @@ int Monitor::Capture() {
     // Don't want to do analysis on it, but we won't due to signal
     return -1;
   } else if (captureResult > 0) {
+    // Before anything else sees the packet: a jump back in the camera's clock is
+    // taken out of the timestamps here, so the queue, the event and the video
+    // store all see one continuous timeline (refs #4242).
+    if (packet->stream) {
+      int64_t offset = KeepDtsContinuous(packet->packet.get(), packet->stream->time_base);
+      if (offset and (packet->packet->pts != AV_NOPTS_VALUE)) packet->pts += offset;
+    }
     shared_data->signal = true;   // Assume if getting packets that we are getting something useful. CheckSignalPoints can correct this later.
     // If we captured, let's assume signal, Decode will detect further
     if (decoding == DECODING_NONE) {
@@ -4544,6 +4592,10 @@ int Monitor::Pause() {
   }
 
   packetqueue.clear();
+  // The next capture session's timestamps start over.
+  dts_offset_ = 0;
+  last_dts_.clear();
+  last_dts_step_.clear();
   return 1;
 } // end int Monitor::Pause()
 

src/zm_monitor.h

@@ -37,6 +37,7 @@
 
 #include <atomic>
 #include <list>
+#include <map>
 #include <memory>
 #include <sys/time.h>
 #include <vector>
@@ -778,6 +779,17 @@ class Monitor : public std::enable_shared_from_this<Monitor> {
 
   VideoStore          *videoStore;
   PacketQueue      packetqueue;
+  // Stream timestamps kept going forward across a jump back in the camera's
+  // clock: an offset (microseconds) added to every packet's dts and pts, and
+  // each stream's last dts with the offset applied and its last step.
+  int64_t dts_offset_ = 0;
+  std::map<int, int64_t> last_dts_;
+  std::map<int, int64_t> last_dts_step_;
+  // A step back of more than this in a stream's dts is a jump in the camera's
+  // clock, not reordering jitter.
+  static constexpr int64_t kDtsStepBack = 1000000;
+  // Moves the packet onto the continuous timeline; returns the offset applied.
+  int64_t KeepDtsContinuous(AVPacket *packet, AVRational time_base);
   std::unique_ptr<PollThread> Poller;
   std::list<ZMPacketLock> decoder_queue;
   packetqueue_iterator  *analysis_it;

tests/CMakeLists.txt

@@ -17,6 +17,7 @@ set(TEST_SOURCES
   zm_db_contention_live.cpp
   zm_db_logging_live.cpp
   zm_db_schema.cpp
+  zm_monitor_dts.cpp
   zm_box.cpp
   zm_buffer.cpp
   zm_audio_detector.cpp

tests/zm_monitor_dts.cpp

@@ -0,0 +1,138 @@
+/*
+ * This file is part of the ZoneMinder Project. See AUTHORS file for Copyright information
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the
+ * Free Software Foundation; either version 2 of the License, or (at your
+ * option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "zm_catch2.h"
+
+#include "zm_monitor.h"
+
+#include <vector>
+
+// Monitor::KeepDtsContinuous(): a camera whose clock steps back (refs #4242).
+// Once the step is taken out at capture, every packet after it continues from
+// where the stream was, for video and audio alike.
+
+namespace {
+
+void EnsureConfig() {
+  if (!config.font_file_location) config.font_file_location = "";
+  if (!config.event_close_mode) config.event_close_mode = "idle";
+}
+
+// Monitor keeps the state; this exposes the method and the offset to the tests.
+class DtsContinuity : public Monitor {
+ public:
+  DtsContinuity() : Monitor() {}
+  int64_t Apply(AVPacket *packet, AVRational time_base) { return KeepDtsContinuous(packet, time_base); }
+  Microseconds Offset() const { return Microseconds(dts_offset_); }
+};
+
+constexpr AVRational kVideoTb = {1, 90000};  // RTP/MPEG-TS video
+constexpr AVRational kAudioTb = {1, 8000};
+constexpr int kVideo = 0;
+constexpr int kAudio = 1;
+
+struct Pkt {
+  av_packet_ptr p{av_packet_alloc()};
+  Pkt(int stream, int64_t dts, int64_t duration) {
+    p->stream_index = stream;
+    p->dts = dts;
+    p->pts = dts;
+    p->duration = duration;
+  }
+};
+
+// Feed dts values (in the stream's time base) through and return what comes out.
+std::vector<int64_t> Run(DtsContinuity &c, int stream, AVRational tb, const std::vector<int64_t> &dts,
+                         int64_t duration) {
+  std::vector<int64_t> out;
+  for (int64_t d : dts) {
+    Pkt pkt(stream, d, duration);
+    c.Apply(pkt.p.get(), tb);
+    REQUIRE(pkt.p->pts == pkt.p->dts);
+    out.push_back(pkt.p->dts);
+  }
+  return out;
+}
+
+}  // namespace
+
+TEST_CASE("Monitor dts continuity: a continuous stream is left alone", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  std::vector<int64_t> in = {0, 3000, 6000, 9000, 12000};
+  CHECK(Run(c, kVideo, kVideoTb, in, 3000) == in);
+  CHECK(c.Offset() == Microseconds(0));
+}
+
+TEST_CASE("Monitor dts continuity: jitter under the threshold is left alone", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  // 100 ms back is reordering jitter: the video store's reorder queue sorts it.
+  std::vector<int64_t> in = {0, 3000, 6000, 6000 - 9000, 9000};
+  CHECK(Run(c, kVideo, kVideoTb, in, 3000) == in);
+  CHECK(c.Offset() == Microseconds(0));
+}
+
+TEST_CASE("Monitor dts continuity: a step back continues from where the stream was", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  // 8.933 s back (803970 ticks at 90 kHz), the step in the #4242 reproduction.
+  std::vector<int64_t> out = Run(c, kVideo, kVideoTb, {900000, 903000, 906000, 906000 - 803970, 906000 - 803970 + 3000}, 3000);
+  CHECK(out == std::vector<int64_t>({900000, 903000, 906000, 909000, 912000}));
+  CHECK(c.Offset() > Microseconds(8900000));
+}
+
+TEST_CASE("Monitor dts continuity: the 1.048 s step from the report is taken out", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  std::vector<int64_t> out = Run(c, kVideo, kVideoTb, {200000, 203000, 203000 - 94320, 203000 - 94320 + 3000}, 3000);
+  CHECK(out == std::vector<int64_t>({200000, 203000, 206000, 209000}));
+}
+
+TEST_CASE("Monitor dts continuity: audio follows the offset video set", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  // Both clocks step back 8 s together. Video sees it first and moves the
+  // shared offset; audio, stepped back by the same amount, carries on.
+  Run(c, kVideo, kVideoTb, {900000, 903000}, 3000);
+  Run(c, kAudio, kAudioTb, {80000, 80160}, 160);
+  std::vector<int64_t> video = Run(c, kVideo, kVideoTb, {903000 - 720000 + 3000}, 3000);
+  CHECK(video == std::vector<int64_t>({906000}));
+  std::vector<int64_t> audio = Run(c, kAudio, kAudioTb, {80320 - 64000, 80480 - 64000}, 160);
+  // Continuous and increasing, within a video frame of where it would have been.
+  CHECK(audio[1] - audio[0] == 160);
+  CHECK(audio[0] > 80160);
+  CHECK(audio[0] < 80320 + 8000 / 30);
+}
+
+TEST_CASE("Monitor dts continuity: without a duration the last step stands in", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  std::vector<int64_t> out = Run(c, kVideo, kVideoTb, {900000, 903600, 907200, 107200}, 0);
+  CHECK(out == std::vector<int64_t>({900000, 903600, 907200, 910800}));
+}
+
+TEST_CASE("Monitor dts continuity: packets without a dts pass through", "[Monitor][dts]") {
+  EnsureConfig();
+  DtsContinuity c;
+  Pkt pkt(kVideo, AV_NOPTS_VALUE, 3000);
+  pkt.p->pts = AV_NOPTS_VALUE;
+  CHECK(c.Apply(pkt.p.get(), kVideoTb) == 0);
+  CHECK(pkt.p->dts == AV_NOPTS_VALUE);
+  CHECK(pkt.p->pts == AV_NOPTS_VALUE);
+}
+