feat(time_sync): implement time synchronization between client and server

alvr/client_core/src/connection.rs

@@ -9,14 +9,15 @@ use crate::{
 };
 use alvr_common::{
     ALVR_VERSION, AnyhowToCon, ConResult, ConnectionError, ConnectionState, LifecycleState,
-    dbg_connection, debug, error, info,
+    TimeSyncEstimator, dbg_connection, debug, error, info, monotonic_ns,
     parking_lot::{Condvar, Mutex, RwLock},
     wait_rwlock, warn,
 };
 use alvr_packets::{
     AUDIO, ClientConnectionResult, ClientControlPacket, ClientStatistics, ConnectionAcceptedInfo,
-    HAPTICS, Haptics, STATISTICS, ServerControlPacket, StreamConfigPacket, TRACKING, TrackingData,
-    VIDEO, VideoPacketHeader, VideoStreamingCapabilities, VideoStreamingCapabilitiesExt,
+    HAPTICS, Haptics, STATISTICS, ServerControlPacket, StreamConfigPacket, TRACKING, TimeSyncPacket,
+    TrackingData, VIDEO, VideoPacketHeader, VideoStreamingCapabilities,
+    VideoStreamingCapabilitiesExt,
 };
 use alvr_session::{SocketProtocol, settings_schema::Switch};
 use alvr_sockets::{
@@ -50,6 +51,7 @@ const SOCKET_INIT_RETRY_INTERVAL: Duration = Duration::from_millis(500);
 const CONNECTION_RETRY_INTERVAL: Duration = Duration::from_secs(1);
 const HANDSHAKE_ACTION_TIMEOUT: Duration = Duration::from_secs(2);
 const STREAMING_RECV_TIMEOUT: Duration = Duration::from_millis(500);
+const TIME_SYNC_INTERVAL: Duration = Duration::from_secs(15);
 
 const MAX_UNREAD_PACKETS: usize = 10; // Applies per stream
 const VIDEO_FRAME_METADATA_HISTORY_SIZE: usize = 128;
@@ -64,6 +66,7 @@ pub struct ConnectionContext {
     pub tracking_sender: Mutex<Option<StreamSender<TrackingData>>>,
     pub statistics_sender: Mutex<Option<StreamSender<ClientStatistics>>>,
     pub statistics_manager: Mutex<Option<StatisticsManager>>,
+    pub time_sync_estimator: Mutex<TimeSyncEstimator>,
     pub decoder_callback: Mutex<Option<Box<DecoderCallback>>>,
     pub video_frame_metadata_queue: Mutex<VecDeque<(Duration, VideoFrameMetadata)>>,
     pub max_prediction: RwLock<Duration>,
@@ -426,6 +429,7 @@ fn connection_pipeline(
         let disconnect_notif = Arc::clone(&disconnect_notif);
         move || {
             let mut keepalive_deadline = Instant::now();
+            let mut timesync_deadline = Instant::now();
 
             #[cfg(target_os = "android")]
             let mut battery_deadline = Instant::now();
@@ -449,6 +453,22 @@ fn connection_pipeline(
                     keepalive_deadline = Instant::now() + KEEPALIVE_INTERVAL;
                 }
 
+                if Instant::now() > timesync_deadline
+                    && let Some(sender) = &mut *ctx.control_sender.lock()
+                {
+                    let offset = ctx.time_sync_estimator.lock().get_offset();
+                    sender
+                        .send(&ClientControlPacket::TimeSync(TimeSyncPacket {
+                            client_send_ns: monotonic_ns(),
+                            server_recv_ns: 0,
+                            server_send_ns: 0,
+                            client_to_server_offset_ns: offset,
+                        }))
+                        .ok();
+
+                    timesync_deadline = Instant::now() + TIME_SYNC_INTERVAL;
+                }
+
                 #[cfg(target_os = "android")]
                 if Instant::now() > battery_deadline {
                     let (gauge_value, is_plugged) = alvr_system_info::get_battery_status();
@@ -502,6 +522,48 @@ fn connection_pipeline(
                         error!("Unexpected StartStream paceket");
                     }
                     Ok(ServerControlPacket::KeepAlive) => (),
+                    Ok(ServerControlPacket::TimeSync(packet)) => {
+                        let client_recv_ns = monotonic_ns();
+                        let mut estimator = ctx.time_sync_estimator.lock();
+
+                        let t1 = packet.client_send_ns as i128;
+                        let t2 = packet.server_recv_ns as i128;
+                        let t3 = packet.server_send_ns as i128;
+                        let t4 = client_recv_ns as i128;
+
+                        let raw_sample_offset = (((t2 - t1) + (t3 - t4)) / 2) as i64;
+                        let rtt_ns = (t4 - t1).saturating_sub(t3 - t2);
+
+                        // Observe drift BEFORE applying correction
+                        if let Some(predicted_offset) = estimator.predict_offset_at(client_recv_ns) {
+                            let drift_ns = raw_sample_offset - predicted_offset;
+                            let drift_ppm = (drift_ns as f64 / 15e9) * 1e6;
+
+                            info!(
+                                "[TIME_SYNC 15s] RTT: {:.2}ms | Measured: {:+.3}ms | Model Pred: {:+.3}ms | Drift over 15s: {:+.3}ms ({:+.1} ppm) | Learned Skew: {:+.1} ppm",
+                                rtt_ns as f64 / 1e6,
+                                raw_sample_offset as f64 / 1e6,
+                                predicted_offset as f64 / 1e6,
+                                drift_ns as f64 / 1e6,
+                                drift_ppm,
+                                estimator.get_skew_ppm(),
+                            );
+                        } else {
+                            info!(
+                                "[TIME_SYNC Init] Baseline offset established: {:+.3}ms | RTT: {:.2}ms",
+                                raw_sample_offset as f64 / 1e6,
+                                rtt_ns as f64 / 1e6,
+                            );
+                        }
+
+                        // Apply correction
+                        estimator.push_sample(
+                            packet.client_send_ns,
+                            packet.server_recv_ns,
+                            packet.server_send_ns,
+                            client_recv_ns,
+                        );
+                    }
                     Ok(
                         ServerControlPacket::Reserved(_) | ServerControlPacket::ReservedBuffer(_),
                     ) => {}

alvr/client_core/src/lib.rs

@@ -257,6 +257,17 @@ impl ClientCoreContext {
         }
     }
 
+    pub fn client_time_to_server_ns(&self, client_time_ns: u64) -> Option<u64> {
+        self.connection_context
+            .time_sync_estimator
+            .lock()
+            .client_to_server_ns(client_time_ns)
+    }
+
+    pub fn get_time_sync_offset(&self) -> Option<i64> {
+        self.connection_context.time_sync_estimator.lock().get_offset()
+    }
+
     /// The callback should return true if the frame was successfully submitted to the decoder
     pub fn set_decoder_input_callback(&self, callback: Box<DecoderCallback>) {
         dbg_client_core!("set_decoder_input_callback");

alvr/common/Cargo.toml

@@ -16,3 +16,6 @@ semver = { version = "1", features = ["serde"] }
 serde = { version = "1", features = ["derive"] }
 settings-schema = { git = "https://github.com/alvr-org/settings-schema-rs", rev = "676185f" }
 # settings-schema = { path = "../../../../settings-schema-rs/settings-schema" }
+
+[target.'cfg(unix)'.dependencies]
+libc = "0.2"

alvr/common/src/lib.rs

@@ -4,6 +4,7 @@ mod connection_result;
 pub mod inputs;
 mod logging;
 mod primitives;
+pub mod time;
 mod version;
 
 use parking_lot::{Condvar, Mutex, RwLockWriteGuard};
@@ -24,6 +25,7 @@ pub use inputs::*;
 pub use log::{debug, error, info, warn};
 pub use logging::*;
 pub use primitives::*;
+pub use time::*;
 pub use version::*;
 
 pub const ALVR_NAME: &str = "ALVR";

alvr/common/src/time.rs

@@ -0,0 +1,227 @@
+use std::collections::VecDeque;
+
+const WINDOW_SIZE: usize = 16;
+const EMA_ALPHA: f64 = 0.1;
+const SKEW_ALPHA: f64 = 0.05;
+const MAX_SKEW_PPM: f64 = 200.0 * 1e-6; // Clamp to +/- 200 ppm
+
+pub fn monotonic_ns() -> u64 {
+    #[cfg(unix)]
+    {
+        let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 };
+        unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) };
+        (ts.tv_sec as u64) * 1_000_000_000 + (ts.tv_nsec as u64)
+    }
+    #[cfg(windows)]
+    {
+        // On Windows: QueryPerformanceCounter
+        extern "system" {
+            fn QueryPerformanceCounter(lpPerformanceCount: *mut i64) -> i32;
+            fn QueryPerformanceFrequency(lpFrequency: *mut i64) -> i32;
+        }
+        static FREQ: std::sync::LazyLock<i64> = std::sync::LazyLock::new(|| {
+            let mut freq = 0;
+            unsafe { QueryPerformanceFrequency(&mut freq) };
+            freq
+        });
+        let mut count = 0;
+        unsafe { QueryPerformanceCounter(&mut count) };
+        ((count as u128 * 1_000_000_000) / (*FREQ as u128)) as u64
+    }
+ }
+
+
+struct Sample {
+    rtt_ns: i128,
+    offset_ns: i64,
+    client_time_ns: u64,
+}
+pub struct TimeSyncEstimator {
+    samples: VecDeque<Sample>,
+    base_client_time_ns: u64,
+    base_offset_ns: Option<f64>,
+    skew: f64,                   // ppm ratio
+    last_skew_client_time_ns: u64,
+    last_skew_offset_ns: f64,
+}
+
+impl Default for TimeSyncEstimator {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl TimeSyncEstimator {
+    pub fn new() -> Self {
+        Self {
+            samples: VecDeque::with_capacity(WINDOW_SIZE),
+            base_client_time_ns: 0,
+            base_offset_ns: None,
+            skew: 0.0,
+            last_skew_client_time_ns: 0,
+            last_skew_offset_ns: 0.0,
+        }
+    }
+
+    pub fn push_sample(&mut self, client_send_ns: u64, server_recv_ns: u64, server_send_ns: u64, client_recv_ns: u64) {
+        let rtt_ns = (client_recv_ns as i128 - client_send_ns as i128) - (server_send_ns as i128 - server_recv_ns as i128);
+        let sample_offset = (((server_recv_ns as i128 - client_send_ns as i128) + (server_send_ns as i128 - client_recv_ns as i128)) / 2) as i64;
+
+        if self.samples.len() >= WINDOW_SIZE {
+            self.samples.pop_front();
+        }
+        self.samples.push_back(Sample { rtt_ns, offset_ns: sample_offset, client_time_ns: client_send_ns });
+
+        if let Some(best) = self.samples.iter().min_by_key(|s| s.rtt_ns) {
+            match self.base_offset_ns {
+                None => {
+                    self.base_offset_ns = Some(best.offset_ns as f64);
+                    self.base_client_time_ns = best.client_time_ns;
+                    self.last_skew_client_time_ns = best.client_time_ns;
+                    self.last_skew_offset_ns = best.offset_ns as f64;
+                }
+                Some(prev_offset) => {
+                    let dt = (best.client_time_ns as f64) - (self.base_client_time_ns as f64);
+                    // Project previous offset forward by current skew rate
+                    let projected_offset = prev_offset + self.skew * dt;
+                    // Blend residual error with EMA
+                    let updated_offset = projected_offset + EMA_ALPHA * ((best.offset_ns as f64) - projected_offset);
+
+                    // Skew estimation over multi-second baseline
+                    let skew_dt = best.client_time_ns.saturating_sub(self.last_skew_client_time_ns);
+                    if skew_dt >= 2_000_000_000 {
+                        let raw_skew = ((best.offset_ns as f64) - self.last_skew_offset_ns) / (skew_dt as f64);
+                        let clamped_skew = raw_skew.clamp(-MAX_SKEW_PPM, MAX_SKEW_PPM);
+                        if self.skew == 0.0 {
+                            self.skew = clamped_skew;
+                        } else {
+                            self.skew = SKEW_ALPHA * clamped_skew + (1.0 - SKEW_ALPHA) * self.skew;
+                        }
+                        self.last_skew_client_time_ns = best.client_time_ns;
+                        self.last_skew_offset_ns = best.offset_ns as f64;
+                    }
+
+                    self.base_offset_ns = Some(updated_offset);
+                    self.base_client_time_ns = best.client_time_ns;
+                }
+            }
+        }
+    }
+
+    pub fn client_to_server_ns(&self, client_time_ns: u64) -> Option<u64> {
+        self.base_offset_ns.map(|theta_0| {
+            let dt = client_time_ns as f64 - self.base_client_time_ns as f64;
+            let current_offset = theta_0 + self.skew * dt;
+            client_time_ns.saturating_add_signed(current_offset.round() as i64)
+        })
+    }
+
+    pub fn get_offset(&self) -> Option<i64> {
+        self.base_offset_ns.map(|o| o.round() as i64)
+    }
+
+    ///
+    pub fn get_skew_ppm(&self) -> f64 {
+        self.skew * 1e6
+    }
+
+    /// Predicts what the offset should be at `client_time_ns` based on current base offset and skew
+    /// For allowing extrapolation of the offset of the server clock without having to ping the server(debugging/tooling)
+    pub fn predict_offset_at(&self, client_time_ns: u64) -> Option<i64> {
+        self.base_offset_ns.map(|theta_0| {
+            let dt = client_time_ns as f64 - self.base_client_time_ns as f64;
+            (theta_0 + self.skew * dt).round() as i64
+        })
+    }
+}
+
+#[test]
+fn test_time_sync_ground_truth() {
+    let mut estimator = TimeSyncEstimator::new();
+    let true_offset: i128 = 15_000_000_000; // Server is 15s ahead
+    let one_way_delay: i128 = 2_000_000;    // 2ms network flight time
+    let server_hold: i128 = 500_000;        // 0.5ms turnaround
+
+    for i in 0..16 {
+        let t1 = 1_000_000_000 + i * 200_000_000; // Ping every 200ms
+        let t2 = t1 + true_offset + one_way_delay;
+        let t3 = t2 + server_hold;
+        let t4 = t3 - true_offset + one_way_delay;
+
+        estimator.push_sample(t1 as u64, t2 as u64, t3 as u64, t4 as u64);
+    }
+
+    let estimated_offset = estimator.get_offset().unwrap();
+    // Must converge to within 1 microsecond of ground truth
+    assert!((estimated_offset - true_offset as i64).abs() < 1_000);
+}
+
+#[test]
+fn test_time_sync_negative_offset() {
+    let mut estimator = TimeSyncEstimator::new();
+    // Client booted 24h ago, Server booted 1h ago: offset is -23 hours
+    let true_offset: i128 = -23 * 3600 * 1_000_000_000;
+    let delay: i128 = 3_000_000; // 3ms
+
+    let t1 = 86_400_000_000_000u64; // 24 hours
+    let t2 = (t1 as i128 + true_offset + delay) as u64;
+    let t3 = t2 + 100_000;
+    let t4 = (t3 as i128 - true_offset + delay) as u64;
+
+    estimator.push_sample(t1, t2, t3, t4);
+    assert_eq!(estimator.get_offset().unwrap(), true_offset as i64);
+}
+
+#[test]
+fn test_wi_fi_spike_rejection() {
+    let mut estimator = TimeSyncEstimator::new();
+    let true_offset: i128 = 5_000_000_000;
+
+    for i in 0..20 {
+        let t1 = 10_000_000_000 + i * 200_000_000;
+        // Inject 60ms spikes on samples 5 and 12, 2ms baseline otherwise
+        let delay = if i == 5 || i == 12 { 30_000_000 } else { 1_000_000 };
+        let t2 = t1 + true_offset + delay;
+        let t3 = t2 + 200_000;
+        let t4 = t3 - true_offset + delay;
+
+        estimator.push_sample(t1 as u64, t2 as u64, t3 as u64, t4 as u64);
+    }
+
+    let diff = (estimator.get_offset().unwrap() - true_offset as i64).abs();
+    // Offset must not get pulled by the 60ms spikes (error must stay under 0.1ms)
+    assert!(diff < 100_000, "Spike leaked into estimator! Error: {diff}ns");
+}
+
+#[test]
+fn test_time_sync_skew_correction() {
+    let mut estimator = TimeSyncEstimator::new();
+    let initial_offset: i128 = 10_000_000_000; // 10s
+    let skew_ppm: f64 = 40.0; // Server ticks 40 ppm faster
+    let skew_ratio = skew_ppm * 1e-6;
+    let delay: i128 = 1_500_000; // 1.5ms one-way
+
+    // Simulate 30 pings every 500ms (15 seconds total)
+    for i in 0..30 {
+        let t1 = 1_000_000_000 + i * 500_000_000;
+        let drift = (t1 as f64 * skew_ratio) as i128;
+        let current_true_offset = initial_offset + drift;
+
+        let t2 = t1 + current_true_offset + delay;
+        let t3 = t2 + 100_000;
+        let t4 = t3 - current_true_offset + delay;
+
+        estimator.push_sample(t1 as u64, t2 as u64, t3 as u64, t4 as u64);
+    }
+
+    // Extrapolate 3 seconds into the future without any new pings
+    let future_client_time = 1_000_000_000u64 + 18 * 1_000_000_000u64;
+    let future_true_offset = initial_offset + (future_client_time as f64 * skew_ratio) as i128;
+    let true_server_time = (future_client_time as i128 + future_true_offset) as u64;
+
+    let predicted_server_time = estimator.client_to_server_ns(future_client_time as u64).unwrap();
+    let error_ns = (predicted_server_time as i128 - true_server_time as i128).abs();
+
+    // With skew correction, error stays under 50 microseconds (0.05ms) even when extrapolating!
+    assert!(error_ns < 50_000, "Error was too large: {error_ns}ns");
+}