improve: use receive timestamp for rtp capture

Author: yeoleobun

src/bin/sipflow.rs

@@ -224,20 +224,24 @@ mod tests {
         let mut packets = Vec::new(); // should use Vec<(i32, u64, Vec<u8>)>
         let payload = vec![0x7F; 160]; // Silence
 
-        // 12 bytes dummy header
+        // 12 bytes RTP header
         let mut header = vec![0u8; 12];
+        header[0] = 0x80; // RTP v2
         header[1] = 0; // PCMU
 
         let mut p1 = header.clone();
+        p1[4..8].copy_from_slice(&1000u32.to_be_bytes());
         p1.extend_from_slice(&payload);
         packets.push((0, 1000u64, p1));
 
         let mut p2 = header.clone();
+        p2[4..8].copy_from_slice(&1000u32.to_be_bytes());
         p2.extend_from_slice(&payload);
         packets.push((1, 1000u64, p2));
 
         // Next 20ms
         let mut p3 = header.clone();
+        p3[4..8].copy_from_slice(&1160u32.to_be_bytes());
         p3.extend_from_slice(&payload);
         packets.push((0, 1160u64, p3));
 

src/media/bridge.rs

@@ -41,6 +41,7 @@
 //! 3. The bridge's WebRTC side connects to the WebRTC client
 //! 4. The bridge's RTP side connects to the SIP/RTP endpoint
 
+use crate::media::ReceiveTimestampClock;
 use crate::media::recorder::{Leg as RecLeg, Recorder};
 use crate::media::transcoder::{RtpTiming, Transcoder, rewrite_dtmf_duration};
 use anyhow::Result;
@@ -340,7 +341,8 @@ pub struct BridgePeer {
     /// When true, the bridge skips writing to the recorder (recording paused).
     recording_paused: Arc<AtomicBool>,
     /// Non-blocking RTP capture tee for SipFlow.
-    sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample)>>,
+    sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample, u64)>>,
+    receive_clock: ReceiveTimestampClock,
     dtmf_sink: Arc<parking_lot::RwLock<Option<BridgeDtmfSink>>>,
     /// Audio sender channels for forwarding — fast-path aliases
     caller_send: Arc<AsyncMutex<Option<MediaSender>>>,
@@ -411,6 +413,7 @@ impl BridgePeer {
             recorder: None,
             recording_paused: Arc::new(AtomicBool::new(false)),
             sipflow_tx: None,
+            receive_clock: ReceiveTimestampClock::new(),
             dtmf_sink: Arc::new(parking_lot::RwLock::new(None)),
             caller_send: Arc::new(AsyncMutex::new(None)),
             callee_send: Arc::new(AsyncMutex::new(None)),
@@ -1314,6 +1317,7 @@ impl BridgePeer {
         let recorder = self.recorder.clone();
         let recording_paused = self.recording_paused.clone();
         let sipflow_tx = self.sipflow_tx.clone();
+        let receive_clock = self.receive_clock.clone();
         let dtmf_sink = Arc::clone(&self.dtmf_sink);
         let caller_to_callee_transcoder = Arc::clone(&self.caller_to_callee_transcoder);
         let caller_to_callee_timing = Arc::clone(&self.caller_to_callee_timing);
@@ -1432,6 +1436,7 @@ impl BridgePeer {
                                         if !is_video { recorder.clone() } else { None },
                                         if !is_video { Some(RecLeg::A) } else { None },
                                         if !is_video { sipflow_tx.clone() } else { None },
+                                        receive_clock.clone(),
                                         recording_paused.clone(),
                                         Arc::clone(&dtmf_sink),
                                         Some(Arc::clone(&caller_to_callee_transcoder)),
@@ -1556,6 +1561,7 @@ impl BridgePeer {
                                         if !is_video { recorder.clone() } else { None },
                                         if !is_video { Some(RecLeg::B) } else { None },
                                         if !is_video { sipflow_tx.clone() } else { None },
+                                        receive_clock.clone(),
                                         recording_paused.clone(),
                                         Arc::clone(&dtmf_sink),
                                         Some(Arc::clone(&callee_to_caller_transcoder)),
@@ -1603,7 +1609,8 @@ impl BridgePeer {
         leg_stats: Arc<LegStats>,
         recorder: Option<Arc<parking_lot::RwLock<Option<Recorder>>>>,
         recorder_leg: Option<RecLeg>,
-        sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample)>>,
+        sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample, u64)>>,
+        receive_clock: ReceiveTimestampClock,
         recording_paused: Arc<AtomicBool>,
         dtmf_sink: Arc<parking_lot::RwLock<Option<BridgeDtmfSink>>>,
         transcoder: Option<Arc<parking_lot::Mutex<Option<Transcoder>>>>,
@@ -1661,6 +1668,7 @@ impl BridgePeer {
                     match sample_result {
                         Ok(sample) => {
                             packet_count += 1;
+                            let received_at_micros = receive_clock.now_micros();
                             if !is_video {
                                 if let (Some(rec), Some(leg)) = (&recorder, recorder_leg)
                                     && !recording_paused.load(std::sync::atomic::Ordering::Relaxed)
@@ -1677,7 +1685,7 @@ impl BridgePeer {
                                 }
 
                                 if let (Some(tx), Some(leg)) = (&sipflow_tx, recorder_leg) {
-                                    let _ = tx.try_send((leg, sample.clone()));
+                                    let _ = tx.try_send((leg, sample.clone(), received_at_micros));
                                 }
                             }
                             if !is_video {
@@ -2005,7 +2013,8 @@ impl BridgePeer {
         leg_stats: Arc<LegStats>,
         recorder: Option<Arc<parking_lot::RwLock<Option<Recorder>>>>,
         recorder_leg: Option<RecLeg>,
-        sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample)>>,
+        sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample, u64)>>,
+        receive_clock: ReceiveTimestampClock,
         recording_paused: Arc<AtomicBool>,
         dtmf_sink: Arc<parking_lot::RwLock<Option<BridgeDtmfSink>>>,
         transcoder: Option<Arc<parking_lot::Mutex<Option<Transcoder>>>>,
@@ -2025,6 +2034,7 @@ impl BridgePeer {
                 recorder,
                 recorder_leg,
                 sipflow_tx,
+                receive_clock,
                 recording_paused,
                 dtmf_sink,
                 transcoder,
@@ -2102,7 +2112,7 @@ pub struct BridgePeerBuilder {
     ice_servers: Vec<IceServer>,
     recorder: Option<Arc<parking_lot::RwLock<Option<Recorder>>>>,
     recording_paused: Arc<AtomicBool>,
-    sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample)>>,
+    sipflow_tx: Option<mpsc::Sender<(RecLeg, MediaSample, u64)>>,
     cname: Option<String>,
 }
 
@@ -2251,7 +2261,7 @@ impl BridgePeerBuilder {
 
     pub fn with_sipflow_capture(
         mut self,
-        sipflow_tx: mpsc::Sender<(RecLeg, MediaSample)>,
+        sipflow_tx: mpsc::Sender<(RecLeg, MediaSample, u64)>,
     ) -> Self {
         self.sipflow_tx = Some(sipflow_tx);
         self
@@ -3632,6 +3642,7 @@ mod tests {
                     None,                             // no recorder
                     None,                             // no recorder leg
                     None,                             // no sipflow capture
+                    ReceiveTimestampClock::new(),
                     Arc::new(AtomicBool::new(false)), // not paused
                     ds,
                     tr,
@@ -3736,6 +3747,7 @@ mod tests {
                     None,
                     None,
                     None,
+                    ReceiveTimestampClock::new(),
                     Arc::new(AtomicBool::new(false)),
                     ds,
                     tr,
@@ -4051,6 +4063,7 @@ mod tests {
                     None,
                     None,
                     None,
+                    ReceiveTimestampClock::new(),
                     Arc::new(AtomicBool::new(false)),
                     ds,
                     None, // no transcoder

src/media/forwarding_track.rs

@@ -1,6 +1,6 @@
 use crate::media::negotiate::NegotiatedLegProfile;
 use crate::media::transcoder::{RtpTiming, Transcoder, rewrite_dtmf_duration};
-use crate::media::{Track, recorder::Leg};
+use crate::media::{ReceiveTimestampClock, Track, recorder::Leg};
 use anyhow::{Result, anyhow};
 use async_trait::async_trait;
 use parking_lot::Mutex;
@@ -43,7 +43,8 @@ pub struct ForwardingTrack {
     audio_timing: Mutex<Option<RtpTiming>>,
     dtmf_timing: Mutex<Option<RtpTiming>>,
     recorder_tx: Option<mpsc::Sender<(Leg, MediaSample)>>,
-    sipflow_tx: Option<mpsc::Sender<(Leg, MediaSample)>>,
+    sipflow_tx: Option<mpsc::Sender<(Leg, MediaSample, u64)>>,
+    receive_clock: ReceiveTimestampClock,
     recorder_leg: Leg,
     dtmf_mapping: Mutex<Option<DtmfMapping>>,
 }
@@ -60,7 +61,7 @@ impl ForwardingTrack {
         track_id: String,
         inner: Arc<dyn MediaStreamTrack>,
         recorder_tx: Option<mpsc::Sender<(Leg, MediaSample)>>,
-        sipflow_tx: Option<mpsc::Sender<(Leg, MediaSample)>>,
+        sipflow_tx: Option<mpsc::Sender<(Leg, MediaSample, u64)>>,
         recorder_leg: Leg,
         ingress_profile: NegotiatedLegProfile,
         egress_profile: NegotiatedLegProfile,
@@ -78,6 +79,7 @@ impl ForwardingTrack {
             dtmf_timing: Mutex::new(None),
             recorder_tx,
             sipflow_tx,
+            receive_clock: ReceiveTimestampClock::new(),
             recorder_leg,
             dtmf_mapping: Mutex::new(None),
         }
@@ -250,14 +252,15 @@ impl MediaStreamTrack for ForwardingTrack {
             let audio_mapping = self.audio_mapping.lock().clone();
             let dtmf_mapping = self.dtmf_mapping.lock().clone();
             let sample = self.inner.recv().await?;
+            let received_at_micros = self.receive_clock.now_micros();
 
             if let Some(tx) = &self.recorder_tx {
                 let _ = tx.try_send((self.recorder_leg, sample.clone()));
             }
 
             // SipFlow RTP recording: non-blocking tee, drops if consumer falls behind.
             if let Some(tx) = &self.sipflow_tx {
-                let _ = tx.try_send((self.recorder_leg, sample.clone()));
+                let _ = tx.try_send((self.recorder_leg, sample.clone(), received_at_micros));
             }
 
             if let MediaSample::Audio(ref frame) = sample {
@@ -471,7 +474,7 @@ mod tests {
     /// blocking the hot path and without interfering with the recorder_tx.
     #[tokio::test]
     async fn sipflow_tx_receives_sample() {
-        let (sf_tx, mut sf_rx) = mpsc::channel::<(Leg, MediaSample)>(256);
+        let (sf_tx, mut sf_rx) = mpsc::channel::<(Leg, MediaSample, u64)>(256);
         let sample = audio_sample(0 /* PCMU */);
         let track = OneShotTrack::new(sample.clone());
 
@@ -494,16 +497,18 @@ mod tests {
         assert!(matches!(result, MediaSample::Audio(_)));
 
         // sipflow channel must also have received the sample.
-        let (leg, _sf_sample) = sf_rx.try_recv().expect("sample must be in sipflow channel");
+        let (leg, _sf_sample, received_at_micros) =
+            sf_rx.try_recv().expect("sample must be in sipflow channel");
         assert_eq!(leg, Leg::A);
+        assert!(received_at_micros > 0);
     }
 
     /// Both recorder_tx AND sipflow_tx can be active simultaneously; each
     /// must receive its own copy of the sample.
     #[tokio::test]
     async fn both_recorder_and_sipflow_receive_sample() {
         let (rec_tx, mut rec_rx) = mpsc::channel::<(Leg, MediaSample)>(256);
-        let (sf_tx, mut sf_rx) = mpsc::channel::<(Leg, MediaSample)>(256);
+        let (sf_tx, mut sf_rx) = mpsc::channel::<(Leg, MediaSample, u64)>(256);
         let sample = audio_sample(0 /* PCMU */);
         let track = OneShotTrack::new(sample.clone());
 
@@ -527,16 +532,18 @@ mod tests {
         let (rec_leg, _) = rec_rx
             .try_recv()
             .expect("recorder channel must have sample");
-        let (sf_leg, _) = sf_rx.try_recv().expect("sipflow channel must have sample");
+        let (sf_leg, _, received_at_micros) =
+            sf_rx.try_recv().expect("sipflow channel must have sample");
         assert_eq!(rec_leg, Leg::B);
         assert_eq!(sf_leg, Leg::B);
+        assert!(received_at_micros > 0);
     }
 
     #[tokio::test]
     async fn sipflow_full_channel_does_not_block() {
-        let (sf_tx, _sf_rx) = mpsc::channel::<(Leg, MediaSample)>(1);
+        let (sf_tx, _sf_rx) = mpsc::channel::<(Leg, MediaSample, u64)>(1);
 
-        let _ = sf_tx.try_send((Leg::A, audio_sample(0)));
+        let _ = sf_tx.try_send((Leg::A, audio_sample(0), 1));
 
         let track = OneShotTrack::new(audio_sample(0));
         let ft = ForwardingTrack::new(

src/media/mod.rs

@@ -50,6 +50,37 @@ pub fn get_timestamp() -> u64 {
         .as_millis() as u64
 }
 
+#[derive(Debug, Clone)]
+pub struct ReceiveTimestampClock {
+    base_instant: std::time::Instant,
+    base_epoch_micros: u64,
+}
+
+impl ReceiveTimestampClock {
+    pub fn new() -> Self {
+        let base_epoch_micros = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|duration| duration.as_micros() as u64)
+            .unwrap_or_default();
+
+        Self {
+            base_instant: std::time::Instant::now(),
+            base_epoch_micros,
+        }
+    }
+
+    pub fn now_micros(&self) -> u64 {
+        self.base_epoch_micros
+            .saturating_add(self.base_instant.elapsed().as_micros() as u64)
+    }
+}
+
+impl Default for ReceiveTimestampClock {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct AudioFrameTiming {
     pcm_sample_rate: u32,

src/proxy/proxy_call/sip_session.rs

@@ -77,6 +77,7 @@ use tracing::{debug, error, info, warn};
 type SipFlowRtpCaptureTx = mpsc::Sender<(
     crate::media::recorder::Leg,
     rustrtc::media::frame::MediaSample,
+    u64,
 )>;
 
 mod conference;
@@ -4581,19 +4582,20 @@ impl SipSession {
         let (tx, mut rx) = mpsc::channel::<(
             crate::media::recorder::Leg,
             rustrtc::media::frame::MediaSample,
+            u64,
         )>(crate::media::forwarding_track::ForwardingTrack::DEFAULT_SIPFLOW_CHANNEL_CAPACITY);
 
         let call_id = self.context.session_id.clone();
         crate::utils::spawn(async move {
             use crate::sipflow::{SipFlowItem, SipFlowMsgType};
 
-            while let Some((leg, sample)) = rx.recv().await {
+            while let Some((leg, sample, received_at_micros)) = rx.recv().await {
                 if let rustrtc::media::frame::MediaSample::Audio(ref frame) = sample
                     && let Some(ref rtp_packet) = frame.raw_packet
                     && let Ok(rtp_bytes) = rtp_packet.marshal()
                 {
                     let item = SipFlowItem {
-                        timestamp: frame.rtp_timestamp as u64,
+                        timestamp: received_at_micros,
                         seq: frame.sequence_number.unwrap_or(0) as u64,
                         msg_type: SipFlowMsgType::Rtp,
                         src_addr: format!("{leg:?}"),
@@ -4947,6 +4949,7 @@ impl SipSession {
             tokio::sync::mpsc::Sender<(
                 crate::media::recorder::Leg,
                 rustrtc::media::frame::MediaSample,
+                u64,
             )>,
         >,
         session_id: &str,