Fix seek A/V desync, seek latency and segment-boundary media loss

src/Session.cpp

@@ -28,6 +28,7 @@
 
 #include <algorithm>
 #include <cassert>
+#include <optional>
 
 using namespace adaptive;
 using namespace PLAYLIST;
@@ -935,6 +936,14 @@ bool SESSION::CSession::SeekTime(double seekTime, bool& isError)
 
   // NOTE: It is assumed that the streams are ordered by video type (on m_streams), so we will seek first the video stream
   // to get the closest sample PTS to the requested seek time, then we will use to seek/align the other streams
+
+  // PTS of the video sample actually delivered (reader/native domain). Once known, audio
+  // streams are aligned directly to it instead of via the manifest timing: Kodi synchronises
+  // on the emitted PTS, and audio/video share the source PTS clock, so this avoids the drift
+  // between the audio and video manifest timelines that would otherwise leave audio starting
+  // seconds after the picture on a seek.
+  std::optional<uint64_t> videoReaderPts;
+
   for (auto& stream : m_streams)
   {
     ISampleReader* streamReader{stream->GetReader()};
@@ -967,7 +976,38 @@ bool SESSION::CSession::SeekTime(double seekTime, bool& isError)
       }
     }
 
-    if (!SeekReader(*stream, seekTimePts))
+    // Align the audio to the video sample actually delivered (co-timed in the reader PTS
+    // domain Kodi uses for A/V sync), rather than seeking it independently through its own
+    // manifest timing which drifts from the video timeline deep into the recording. This also
+    // covers resume-from-position, where Kodi issues the seek before the readers are started.
+    // hasAdStream excludes fMP4 audio muxed into the video stream (it has no own segment buffer
+    // and is seeked through the video reader); such a stream keeps its existing behaviour.
+    const ISampleReader::Type readerType{streamReader->GetType()};
+    const bool alignToVideoPts = videoReaderPts.has_value() && hasAdStream &&
+                                 stream->m_info.GetStreamType() == INPUTSTREAM_TYPE_AUDIO &&
+                                 (readerType == ISampleReader::Type::TS ||
+                                  readerType == ISampleReader::Type::ADTS ||
+                                  readerType == ISampleReader::Type::FMP4);
+
+    bool seekOk;
+    if (alignToVideoPts)
+    {
+      // SeekAdStream above reset the audio segment to its start, so a single forward scan
+      // lands straight on the video PTS whether or not the reader was already running.
+      const bool wasStarted = streamReader->IsStarted();
+      seekOk = streamReader->TimeSeekReaderPts(*videoReaderPts);
+
+      // Reproduce Start()'s first-start bookkeeping when the reader was not running yet
+      // (resume-from-position seeks before the first DemuxRead started the readers).
+      if (seekOk && !wasStarted && streamReader->GetInformation(stream->m_info))
+        m_changed = true;
+    }
+    else
+    {
+      seekOk = SeekReader(*stream, seekTimePts);
+    }
+
+    if (!seekOk)
     {
       streamReader->Reset(true);
 
@@ -988,6 +1028,25 @@ bool SESSION::CSession::SeekTime(double seekTime, bool& isError)
       LOG::Log(LOGINFO, "Seek time %0.1lf for stream: %i continues at %0.1lf (PTS: %llu)", seekTime,
                streamReader->GetStreamId(), destTimeSecs, streamReader->PTS());
 
+      // Report the residual A/V offset for streams aligned to the video PTS, so a bad landing
+      // (e.g. the audio segment did not contain the target and the reader snapped to a segment
+      // boundary) is immediately visible in the log instead of only as audible desync.
+      if (alignToVideoPts)
+      {
+        const int64_t avDeltaPts{static_cast<int64_t>(streamReader->PTS()) -
+                                 static_cast<int64_t>(*videoReaderPts)};
+        // 0.5s: comfortably above one audio frame, below one segment - a larger delta means
+        // the co-timing did not land and playback will still be out of sync.
+        constexpr int64_t avDeltaWarnPts{STREAM_TIME_BASE / 2};
+        const int64_t avDeltaAbs{avDeltaPts < 0 ? -avDeltaPts : avDeltaPts};
+
+        LOG::Log(avDeltaAbs > avDeltaWarnPts ? LOGWARNING : LOGINFO,
+                 "Seek A/V align: audio stream %i landed at PTS %llu, target (video) PTS %llu, "
+                 "delta %lld (%0.3lfs)",
+                 streamReader->GetStreamId(), streamReader->PTS(), *videoReaderPts, avDeltaPts,
+                 static_cast<double>(avDeltaPts) / STREAM_TIME_BASE);
+      }
+
       // We replace the seek time PTS initially requested with the PTS of the video sample found
       // in order to search the audio/subtitle sample packet more accurately.
       // f.e. in the case of MP4 the video packet has very few sample sync points
@@ -997,6 +1056,8 @@ bool SESSION::CSession::SeekTime(double seekTime, bool& isError)
       if (stream->m_info.GetStreamType() == INPUTSTREAM_TYPE_VIDEO)
       {
         seekTime = destTimeSecs;
+        // Remember the native PTS delivered so the audio streams can be co-timed with it.
+        videoReaderPts = streamReader->PTS();
 
         if (seekTimeCorrected != destTimePts)
         {

src/common/AdaptiveStream.cpp

@@ -776,6 +776,44 @@ bool adaptive::AdaptiveStream::ensureSegment()
   if (m_segBuffers.IsEmpty() || (m_segBuffers.Front().BufferSize() != 0 &&
                                  segment_read_pos_ >= m_segBuffers.Front().BufferSize()))
   {
+    // BufferSize() is how much of the segment has been downloaded so far, not how large the segment
+    // is. When the reader catches up with an ongoing download the read position reaches that end
+    // while the segment is far from complete - dropping it here discards everything that had not
+    // arrived yet, and reading silently continues in the *next* segment: the delivered stream loses
+    // seconds of media without any read or seek ever failing. Wait for the download to deliver more
+    // data (or to finish) before concluding that the segment has been consumed.
+    if (!m_segBuffers.IsEmpty())
+    {
+      // Poll rather than wait indefinitely. The download thread changes the buffer state and
+      // notifies without holding mutexRW, so a notification issued between our predicate check and
+      // the wait is lost and we would never wake up - and downloads can also be paused, in which
+      // case no notification is coming at all. Re-evaluating on a timeout covers both, and the
+      // overall bound guarantees that this can never hang the demuxer thread: on expiry we fall
+      // through to the previous behaviour rather than block playback.
+      constexpr auto pollInterval{std::chrono::milliseconds(50)};
+      constexpr auto maxWait{std::chrono::seconds(5)};
+
+      std::unique_lock<std::mutex> lckrw(thread_data_->mutexRW);
+      const SegmentBuffer& frontBuffer = m_segBuffers.Front();
+      const auto waitUntil{std::chrono::steady_clock::now() + maxWait};
+
+      while (segment_read_pos_ >= frontBuffer.BufferSize() &&
+             (frontBuffer.State() == BufferState::QUEUED ||
+              frontBuffer.State() == BufferState::DOWNLOADING) &&
+             thread_data_->State() == THREADDATA::ThState::RUNNING &&
+             std::chrono::steady_clock::now() < waitUntil)
+      {
+        thread_data_->cvRW.wait_for(lckrw, pollInterval);
+      }
+    }
+
+    // Re-check: the wait above may have made more data available, so the segment is not consumed
+    if (!m_segBuffers.IsEmpty() && m_segBuffers.Front().BufferSize() != 0 &&
+        segment_read_pos_ < m_segBuffers.Front().BufferSize())
+    {
+      return true;
+    }
+
     if (!m_segBuffers.IsEmpty() && m_segBuffers.Front().State() == BufferState::DOWNLOADING)
     {
       // Although the reading position has reached the end, the segment status may not yet be updated
@@ -1001,6 +1039,12 @@ uint32_t adaptive::AdaptiveStream::read(void* buffer, uint32_t bytesToRead)
 
     SegmentBuffer& currSegBuffer = m_segBuffers.Front();
 
+    // The subtraction below is unsigned: a read position past the downloaded end would wrap to a
+    // huge count and read beyond the buffer. Equality is legitimate (nothing available yet, the
+    // wait below may still deliver data), only a position beyond it is not.
+    if (segment_read_pos_ > currSegBuffer.BufferSize())
+      return 0;
+
     size_t avail = currSegBuffer.BufferSize() - segment_read_pos_;
 
     {
@@ -1019,14 +1063,20 @@ uint32_t adaptive::AdaptiveStream::read(void* buffer, uint32_t bytesToRead)
     if (avail > bytesToRead)
       avail = bytesToRead;
 
+    if (avail == 0)
+      return 0;
+
+    // Deliver what is available rather than requiring the full amount: this is a partial read and
+    // the caller is expected to ask again for the remainder. Advancing the positions for bytes that
+    // are then neither copied nor reported discards the tail of every segment - and the read error
+    // it produces sends the caller into its IO error recovery, which re-anchors on the *next*
+    // segment and drops everything the demuxer had not yet emitted from the current one.
+    currSegBuffer.CopyBufferTo(buffer, segment_read_pos_, avail);
+
     segment_read_pos_ += avail;
     absolute_position_ += avail;
 
-    if (avail == bytesToRead)
-    {
-      currSegBuffer.CopyBufferTo(buffer, segment_read_pos_ - avail, avail);
-      return static_cast<uint32_t>(avail);
-    }
+    return static_cast<uint32_t>(avail);
   }
 
   return 0;
@@ -1139,7 +1189,15 @@ bool adaptive::AdaptiveStream::seek(uint64_t const pos, bool& isEos)
     }
   }
 
-  segment_read_pos_ = static_cast<size_t>(pos - (absolute_position_ - segment_read_pos_));
+  // Positions are absolute over the whole stream while only the current segment is buffered, so a
+  // position before that segment cannot be served. Reject it explicitly: the subtraction below is
+  // unsigned, so letting it through wraps around to a huge offset that the clamp then turns into
+  // "end of the current segment" - a silent jump to a completely different point in the timeline.
+  const uint64_t segStartPos{absolute_position_ - segment_read_pos_};
+  if (pos < segStartPos)
+    return false;
+
+  segment_read_pos_ = static_cast<size_t>(pos - segStartPos);
 
   if (segment_read_pos_ > currSegBuffer.BufferSize())
   {

src/common/Segment.cpp

@@ -140,12 +140,23 @@ const CSegment* PLAYLIST::CSegContainer::FindByPTSOrNext(uint64_t pts) const
 {
   for (const CSegment& seg : m_segments)
   {
-    if (seg.startPTS_ <= pts && pts <= seg.m_endPts)
+    // m_endPts is the exclusive end of the segment (== start of the next one for
+    // contiguous timelines). Use a half-open range [startPTS_, m_endPts) so that a
+    // pts landing exactly on a segment boundary selects the segment that *starts*
+    // there, not the one that ends there. Otherwise aligning audio to a video seek
+    // that lands on a segment-start keyframe (an exact boundary) picks the previous
+    // audio segment, starting audio a full segment ahead of the picture.
+    if (seg.startPTS_ <= pts && pts < seg.m_endPts)
       return &seg;
 
     if (seg.startPTS_ > pts)
       return &seg;
   }
+
+  // pts is at/after the end of the last segment: fall back to it (seek to stream end)
+  if (!m_segments.empty() && pts == m_segments.back().m_endPts)
+    return &m_segments.back();
+
   return nullptr;
 }
 

src/demuxers/TSReader.cpp

@@ -77,7 +77,14 @@ TSReader::~TSReader()
 
 bool TSReader::ReadAV(uint64_t pos, unsigned char * data, size_t len)
 {
-  m_stream->Seek(pos);
+  // The seek result must be honoured: a position that is no longer reachable (before the start of
+  // the segment currently buffered) leaves the stream clamped to the end of that segment. Reading
+  // there and reporting success hands the demuxer data from an entirely different point in the
+  // timeline than the position it asked for - the parser then continues in the *next* segment and
+  // the delivered stream loses everything in between.
+  if (AP4_FAILED(m_stream->Seek(pos)))
+    return false;
+
   return AP4_SUCCEEDED(m_stream->Read(data, static_cast<AP4_Size>(len)));
 }
 
@@ -217,21 +224,44 @@ bool TSReader::SeekTime(uint64_t timeInTs)
       break;
     }
 
-  uint64_t lastRecovery(static_cast<uint64_t>(m_startPos));
-  while (m_pkt.pts == PTS_UNSET || static_cast<uint64_t>(m_pkt.pts) < timeInTs)
+  if (hasVideo)
   {
-    uint64_t thisFrameStart(m_AVContext->GetRecoveryPos());
-    if (!ReadPacket())
-      return false;
-    if (!hasVideo || m_pkt.recoveryPoint || thisFrameStart >= m_startPos)
+    // Stop at the first recovery point (keyframe), which is the one that starts the segment
+    // AdaptiveStream::seek_time already selected as containing the requested time.
+    //
+    // Scanning on to the first keyframe at/after timeInTs - the previous behaviour - cannot land
+    // inside the current segment at all: this content carries a single keyframe per segment, at its
+    // start (verified on a sample segment: one random_access_indicator, at the first video packet,
+    // followed by a 250 frame / 10 second GOP). The scan therefore always ran into the *next*
+    // segment, downloading the whole remainder of the current one and overshooting the requested
+    // time by up to a full segment.
+    //
+    // Bounded by timeInTs so a stream that never flags a recovery point degrades to the previous
+    // behaviour instead of scanning to EOS.
+    while (m_pkt.pts == PTS_UNSET ||
+           (!m_pkt.recoveryPoint && static_cast<uint64_t>(m_pkt.pts) < timeInTs))
     {
-      lastRecovery = thisFrameStart;
-      if (static_cast<uint64_t>(m_pkt.pts) >= timeInTs)
-        break;
+      if (!ReadPacket())
+        return false;
+    }
+  }
+  else
+  {
+    // Audio-only TS has no keyframes, so land as close to the requested time as possible.
+    while (m_pkt.pts == PTS_UNSET || static_cast<uint64_t>(m_pkt.pts) < timeInTs)
+    {
+      if (!ReadPacket())
+        return false;
     }
   }
-  m_AVContext->GoPosition(lastRecovery, true);
 
+  // Do not reposition to the recovery position afterwards. The packet we want is already read and
+  // CTSSampleReader::TimeSeek hands this m_pkt to Kodi as the first sample, while the packets that
+  // follow it are queued in the elementary stream buffers and are drained in order. Seeking the
+  // AVContext back to GetRecoveryPos() instead loses them: the MPEG-TS parser reads ahead, so that
+  // position lies *behind* the packet just delivered, and reading resumed a full GOP later - the
+  // video stream got a hole of seconds right after every seek while audio kept feeding, which
+  // starves Kodi's video player into a stillframe and a decoder reset.
   return true;
 }
 

src/samplereader/SampleReader.h

@@ -95,6 +95,24 @@ class ATTR_DLL_LOCAL ISampleReader
    */
   virtual bool TimeSeek(uint64_t pts) = 0;
 
+  /*!
+   * \brief Seek the reader to an absolute reader PTS - the value domain returned by
+   *        PTS() (native to this reader), not the manifest timing that TimeSeek() expects.
+   *        Used after a seek to co-time a secondary stream (audio) with the video sample
+   *        actually delivered: Kodi synchronises on the emitted PTS, and the audio/video
+   *        share the source PTS clock, so aligning directly to the video's reader PTS avoids
+   *        the drift between the audio and video manifest timelines that a manifest-domain
+   *        seek (TimeSeek) reintroduces via GetPTSDiff().
+   * \param pts The absolute reader PTS to align to.
+   * \return True if has success, otherwise false.
+   */
+  virtual bool TimeSeekReaderPts(uint64_t pts)
+  {
+    // Default: remove the manifest compensation so TimeSeek lands on the given reader PTS.
+    const int64_t manifestPts{static_cast<int64_t>(pts) - GetPTSDiff()};
+    return TimeSeek(manifestPts < 0 ? 0 : static_cast<uint64_t>(manifestPts));
+  }
+
   virtual void SetPTSOffset(uint64_t offset) = 0;
   virtual int64_t GetPTSDiff() const = 0;