use a mapping of rtp timestamp to user timestamp on subscriber side too

Author: chenosaurus

libwebrtc/src/native/user_timestamp.rs

@@ -149,14 +149,14 @@ impl UserTimestampHandler {
         }
     }
 
-    /// Pop the next received user timestamp from the receive queue.
-    /// Returns None if the queue is empty.
+    /// Lookup the user timestamp for a given RTP timestamp (receiver side).
+    /// Returns None if no timestamp was found for this RTP timestamp.
+    /// The entry is removed from the map after a successful lookup.
     ///
-    /// Each decoded frame should call this once to get its matching
-    /// timestamp, maintaining 1:1 correspondence between received
-    /// encoded frames and decoded video frames.
-    pub fn pop_user_timestamp(&self) -> Option<i64> {
-        let ts = self.sys_handle.pop_user_timestamp();
+    /// Use the RTP timestamp from the decoded video frame to correlate
+    /// it with the user timestamp that was embedded in the encoded frame.
+    pub fn lookup_user_timestamp(&self, rtp_timestamp: u32) -> Option<i64> {
+        let ts = self.sys_handle.lookup_user_timestamp(rtp_timestamp);
         if ts >= 0 {
             Some(ts)
         } else {
@@ -190,7 +190,9 @@ pub fn create_sender_handler(
 /// Create a receiver-side user timestamp handler.
 ///
 /// This handler will extract user timestamps from received frames
-/// and make them available via `last_user_timestamp()`.
+/// and store them in a map keyed by RTP timestamp. Use
+/// `lookup_user_timestamp(rtp_timestamp)` to retrieve the user
+/// timestamp for a specific decoded frame.
 pub fn create_receiver_handler(
     peer_factory: &PeerConnectionFactory,
     store: &UserTimestampStore,

libwebrtc/src/native/video_stream.rs

@@ -96,11 +96,12 @@ struct VideoTrackObserver {
 
 impl sys_vt::VideoSink for VideoTrackObserver {
     fn on_frame(&self, frame: UniquePtr<webrtc_sys::video_frame::ffi::VideoFrame>) {
+        let rtp_timestamp = frame.timestamp();
         let user_timestamp_us = self
             .user_timestamp_handler
             .lock()
             .as_ref()
-            .and_then(|h| h.pop_user_timestamp());
+            .and_then(|h| h.lookup_user_timestamp(rtp_timestamp));
 
         let _ = self.frame_tx.send(VideoFrame {
             rotation: frame.rotation().into(),

libwebrtc/src/video_frame.rs

@@ -60,9 +60,6 @@ where
     pub rotation: VideoRotation,
     pub timestamp_us: i64, // When the frame was captured in microseconds
     /// Optional user timestamp in microseconds, if available.
-    /// This is typically a hardware or device timestamp supplied by the
-    /// application that can be propagated end-to-end through the media
-    /// pipeline.
     pub user_timestamp_us: Option<i64>,
     pub buffer: T,
 }

webrtc-sys/include/livekit/user_timestamp.h

@@ -110,10 +110,10 @@ class UserTimestampTransformer : public webrtc::FrameTransformerInterface {
   /// Get the last received user timestamp (receiver side only)
   std::optional<int64_t> last_user_timestamp() const;
 
-  /// Pop the next received user timestamp from the receive queue.
-  /// Returns the user timestamp if available, nullopt otherwise.
-  /// Each decoded frame should call this once to get its matching timestamp.
-  std::optional<int64_t> pop_user_timestamp();
+  /// Lookup the user timestamp associated with a given RTP timestamp.
+  /// Returns the user timestamp if found, nullopt otherwise.
+  /// The entry is removed from the map after lookup.
+  std::optional<int64_t> lookup_user_timestamp(uint32_t rtp_timestamp);
 
  private:
   void TransformSend(
@@ -148,11 +148,14 @@ class UserTimestampTransformer : public webrtc::FrameTransformerInterface {
   mutable webrtc::Mutex send_cache_mutex_;
   mutable int64_t last_sent_user_timestamp_{0};
 
-  // Receive-side FIFO queue: one entry per received encoded frame, popped
-  // one-to-one as decoded frames are delivered to the video sink.
-  mutable webrtc::Mutex recv_queue_mutex_;
-  mutable std::deque<int64_t> recv_queue_;
-  static constexpr size_t kMaxRecvQueueEntries = 300;
+  // Receive-side map: RTP timestamp -> user timestamp.
+  // Keyed by RTP timestamp so decoded frames can look up their user
+  // timestamp regardless of frame drops or reordering.
+  mutable webrtc::Mutex recv_map_mutex_;
+  mutable std::unordered_map<uint32_t, int64_t> recv_map_;
+  // Track insertion order for pruning old entries.
+  mutable std::deque<uint32_t> recv_map_order_;
+  static constexpr size_t kMaxRecvMapEntries = 300;
 };
 
 /// Wrapper class for Rust FFI that manages user timestamp transformers.
@@ -178,9 +181,9 @@ class UserTimestampHandler {
   /// Returns -1 if no timestamp has been received yet
   int64_t last_user_timestamp() const;
 
-  /// Pop the next received user timestamp from the receive queue.
-  /// Returns -1 if the queue is empty.
-  int64_t pop_user_timestamp() const;
+  /// Lookup the user timestamp for a given RTP timestamp (receiver side).
+  /// Returns -1 if not found.
+  int64_t lookup_user_timestamp(uint32_t rtp_timestamp) const;
 
   /// Check if a user timestamp has been received
   bool has_user_timestamp() const;

webrtc-sys/src/user_timestamp.cpp

@@ -253,13 +253,17 @@ void UserTimestampTransformer::TransformReceive(
     last_user_timestamp_.store(user_ts.value());
     has_last_user_timestamp_.store(true);
 
-    // Also push to the receive queue so decoded frames can pop 1:1
+    // Store in the receive map keyed by RTP timestamp so decoded frames
+    // can look up their user timestamp regardless of frame drops.
     {
-      webrtc::MutexLock lock(&recv_queue_mutex_);
-      if (recv_queue_.size() >= kMaxRecvQueueEntries) {
-        recv_queue_.pop_front();
+      webrtc::MutexLock lock(&recv_map_mutex_);
+      // Evict oldest entry if at capacity
+      while (recv_map_.size() >= kMaxRecvMapEntries && !recv_map_order_.empty()) {
+        recv_map_.erase(recv_map_order_.front());
+        recv_map_order_.pop_front();
       }
-      recv_queue_.push_back(user_ts.value());
+      recv_map_[rtp_timestamp] = user_ts.value();
+      recv_map_order_.push_back(rtp_timestamp);
     }
 
     // Update frame with stripped data
@@ -416,13 +420,22 @@ std::optional<int64_t> UserTimestampTransformer::last_user_timestamp()
   return last_user_timestamp_.load();
 }
 
-std::optional<int64_t> UserTimestampTransformer::pop_user_timestamp() {
-  webrtc::MutexLock lock(&recv_queue_mutex_);
-  if (recv_queue_.empty()) {
+std::optional<int64_t> UserTimestampTransformer::lookup_user_timestamp(
+    uint32_t rtp_timestamp) {
+  webrtc::MutexLock lock(&recv_map_mutex_);
+  auto it = recv_map_.find(rtp_timestamp);
+  if (it == recv_map_.end()) {
     return std::nullopt;
   }
-  int64_t ts = recv_queue_.front();
-  recv_queue_.pop_front();
+  int64_t ts = it->second;
+  recv_map_.erase(it);
+  // Remove from insertion-order tracker (linear scan is fine for bounded size)
+  for (auto oit = recv_map_order_.begin(); oit != recv_map_order_.end(); ++oit) {
+    if (*oit == rtp_timestamp) {
+      recv_map_order_.erase(oit);
+      break;
+    }
+  }
   return ts;
 }
 
@@ -461,8 +474,8 @@ int64_t UserTimestampHandler::last_user_timestamp() const {
   return ts.value_or(-1);
 }
 
-int64_t UserTimestampHandler::pop_user_timestamp() const {
-  auto ts = transformer_->pop_user_timestamp();
+int64_t UserTimestampHandler::lookup_user_timestamp(uint32_t rtp_timestamp) const {
+  auto ts = transformer_->lookup_user_timestamp(rtp_timestamp);
   return ts.value_or(-1);
 }
 

webrtc-sys/src/user_timestamp.rs

@@ -67,10 +67,9 @@ pub mod ffi {
         /// Returns -1 if no timestamp has been received yet.
         fn last_user_timestamp(self: &UserTimestampHandler) -> i64;
 
-        /// Pop the next received user timestamp from the receive queue.
-        /// Returns -1 if the queue is empty.
-        /// Each decoded frame should call this once to get its matching timestamp.
-        fn pop_user_timestamp(self: &UserTimestampHandler) -> i64;
+        /// Lookup the user timestamp for a given RTP timestamp (receiver side).
+        /// Returns -1 if not found. The entry is removed after lookup.
+        fn lookup_user_timestamp(self: &UserTimestampHandler, rtp_timestamp: u32) -> i64;
 
         /// Check if a user timestamp has been received.
         fn has_user_timestamp(self: &UserTimestampHandler) -> bool;