diff --git a/src/client.rs b/src/client.rs index e10057644..b524c27ed 100644 --- a/src/client.rs +++ b/src/client.rs @@ -25,7 +25,7 @@ use crate::types::events::{ConnectFailureReason, Event}; use log::{debug, error, info, trace, warn}; -use rand::RngCore; +use rand::{Rng, RngCore}; use scopeguard; use std::collections::{HashMap, HashSet}; use wacore_binary::jid::Jid; @@ -144,6 +144,14 @@ pub struct Client { pub(crate) is_connecting: Arc, pub(crate) is_running: Arc, pub(crate) shutdown_notifier: Arc, + /// Timestamp (ms since UNIX epoch) of the last received WebSocket data. + /// Updated on every `DataReceived` transport event. + /// WA Web: `parseAndHandleStanza` → `deadSocketTimer.cancel()`. + pub(crate) last_data_received_ms: Arc, + /// Timestamp (ms since UNIX epoch) of the last sent WebSocket data. + /// Updated on every `send_node` call. + /// WA Web: `callStanza` → `deadSocketTimer.onOrBefore(deadSocketTime)`. + pub(crate) last_data_sent_ms: Arc, pub(crate) transport: Arc>>>, pub(crate) transport_events: @@ -410,6 +418,8 @@ impl Client { is_connecting: Arc::new(AtomicBool::new(false)), is_running: Arc::new(AtomicBool::new(false)), shutdown_notifier: Arc::new(Notify::new()), + last_data_received_ms: Arc::new(AtomicU64::new(0)), + last_data_sent_ms: Arc::new(AtomicU64::new(0)), transport: Arc::new(Mutex::new(None)), transport_events: Arc::new(Mutex::new(None)), @@ -684,8 +694,10 @@ impl Client { } let error_count = self.auto_reconnect_errors.fetch_add(1, Ordering::SeqCst); - let delay_secs = u64::from(error_count * 2).min(30); - let delay = Duration::from_secs(delay_secs); + // WA Web: Fibonacci backoff with 10% jitter, max 900s. + // algo: { type: "fibonacci", first: 1000, second: 1000 } + // jitter: 0.1, max: 9e5 + let delay = fibonacci_backoff(error_count); info!( "Will attempt to reconnect in {:?} (attempt {})", delay, @@ -763,6 +775,15 @@ impl Client { self.cleanup_connection_state().await; } + /// Backoff step used by [`reconnect()`] to create an offline window. + /// + /// `fibonacci_backoff(RECONNECT_BACKOFF_STEP)` determines the delay before + /// the run loop re-connects. This must be longer than the mock server's + /// chatstate TTL (`CHATSTATE_TTL_SECS=3`) so TTL-expiry tests pass. + /// + /// Sequence: fib(0)=1s, fib(1)=1s, fib(2)=2s, fib(3)=3s, **fib(4)=5s**. + pub const RECONNECT_BACKOFF_STEP: u32 = 4; + /// Drop the current connection and trigger the auto-reconnect loop. /// /// Unlike [`disconnect`], this does **not** stop the run loop. The client @@ -776,10 +797,8 @@ impl Client { /// - Testing offline message delivery pub async fn reconnect(self: &Arc) { info!("Reconnecting: dropping transport for auto-reconnect."); - // Create a deterministic offline window before the next reconnect - // attempt so reconnect-based e2e tests can reliably exercise - // queued-offline behavior before the run loop dials back in. - self.auto_reconnect_errors.store(2, Ordering::Relaxed); + self.auto_reconnect_errors + .store(Self::RECONNECT_BACKOFF_STEP, Ordering::Relaxed); if let Some(transport) = self.transport.lock().await.as_ref() { transport.disconnect().await; } @@ -815,6 +834,10 @@ impl Client { // Old workers holding the previous semaphore Arc will finish normally. *self.message_processing_semaphore.lock().unwrap() = Arc::new(tokio::sync::Semaphore::new(1)); + // Reset dead-socket timestamps so stale values from the previous + // connection don't trigger an immediate reconnect on the next one. + self.last_data_received_ms.store(0, Ordering::Relaxed); + self.last_data_sent_ms.store(0, Ordering::Relaxed); // Reset offline sync state for next connection self.offline_sync_completed.store(false, Ordering::Relaxed); self.server_has_prekeys.store(true, Ordering::Relaxed); @@ -883,6 +906,12 @@ impl Client { event_result = transport_events.recv() => { match event_result { Ok(crate::transport::TransportEvent::DataReceived(data)) => { + // Update dead-socket timer (WA Web: deadSocketTimer reset) + self.last_data_received_ms.store( + chrono::Utc::now().timestamp_millis() as u64, + Ordering::Relaxed, + ); + // Feed data into the frame decoder frame_decoder.feed(&data); @@ -2729,6 +2758,13 @@ impl Client { { pool.push(plaintext_buf); } + + // WA Web: callStanza → deadSocketTimer.onOrBefore(deadSocketTime, socketId) + self.last_data_sent_ms.store( + chrono::Utc::now().timestamp_millis() as u64, + Ordering::Relaxed, + ); + Ok(()) } @@ -2872,6 +2908,34 @@ fn build_pong(to: String, id: Option<&str>) -> wacore_binary::node::Node { builder.build() } +/// Computes a reconnect delay matching WhatsApp Web's Fibonacci backoff: +/// `{ algo: { type: "fibonacci", first: 1000, second: 1000 }, jitter: 0.1, max: 9e5 }` +/// +/// Sequence: 1s, 1s, 2s, 3s, 5s, 8s, 13s, 21s, 34s, 55s, 89s, 144s, ... capped at 900s. +/// Each value gets ±10% random jitter. +fn fibonacci_backoff(attempt: u32) -> Duration { + const MAX_MS: u64 = 900_000; // WA Web: 9e5 + + let mut a: u64 = 1000; + let mut b: u64 = 1000; + for _ in 0..attempt { + let next = a.saturating_add(b).min(MAX_MS); + a = b; + b = next; + } + let base = a.min(MAX_MS); + + // ±10% jitter (WA Web: jitter: 0.1) + let jitter_range = base / 10; + let jitter = if jitter_range > 0 { + rand::rng().random_range(0..=(jitter_range * 2)) as i64 - jitter_range as i64 + } else { + 0 + }; + let ms = (base as i64 + jitter).max(0) as u64; + Duration::from_millis(ms) +} + #[cfg(test)] mod tests { use super::*; @@ -4479,6 +4543,50 @@ mod tests { ); } + // ── fibonacci_backoff tests ──────────────────────────────────────── + + #[test] + fn test_fibonacci_backoff_sequence() { + // WA Web: first=1000, second=1000 → 1,1,2,3,5,8,13,21,34,55,89,144...s + // We test base values without jitter by checking the range (±10%). + let expected_base_ms = [1000, 1000, 2000, 3000, 5000, 8000, 13000, 21000]; + for (attempt, &base) in expected_base_ms.iter().enumerate() { + let delay = fibonacci_backoff(attempt as u32); + let ms = delay.as_millis() as u64; + let low = base - base / 10; + let high = base + base / 10; + assert!( + ms >= low && ms <= high, + "attempt {attempt}: expected {low}..={high}ms, got {ms}ms" + ); + } + } + + #[test] + fn test_fibonacci_backoff_max_900s() { + // After many attempts, should cap at 900s (±10%) + let delay = fibonacci_backoff(100); + let ms = delay.as_millis() as u64; + assert!( + ms <= 990_000, + "should never exceed 900s + 10% jitter, got {ms}ms" + ); + assert!( + ms >= 810_000, + "should be at least 900s - 10% jitter, got {ms}ms" + ); + } + + #[test] + fn test_fibonacci_backoff_first_attempt_is_1s() { + let delay = fibonacci_backoff(0); + let ms = delay.as_millis() as u64; + assert!( + (900..=1100).contains(&ms), + "first attempt should be ~1s (±10%), got {ms}ms" + ); + } + #[tokio::test] async fn test_custom_cache_config_is_respected() { use crate::cache_config::{CacheConfig, CacheEntryConfig}; diff --git a/src/keepalive.rs b/src/keepalive.rs index 19e8323f2..cd7f08874 100644 --- a/src/keepalive.rs +++ b/src/keepalive.rs @@ -5,12 +5,46 @@ use rand::Rng; use std::sync::Arc; use std::sync::atomic::Ordering; use std::time::Duration; -use wacore::iq::keepalive::KeepaliveSpec; +use wacore::iq::spec::IqSpec; -const KEEP_ALIVE_INTERVAL_MIN: Duration = Duration::from_secs(20); +/// Returns the number of milliseconds elapsed since a stored timestamp. +/// Returns `None` if the timestamp was never set (value 0). +fn ms_since(timestamp_ms: u64) -> Option { + if timestamp_ms == 0 { + return None; + } + let now = chrono::Utc::now().timestamp_millis() as u64; + Some(now.saturating_sub(timestamp_ms)) +} + +/// Checks the dead-socket condition: data was sent but nothing received +/// within `DEAD_SOCKET_TIME`. +/// +/// WA Web: `deadSocketTimer` is armed on every `callStanza` (send) and +/// cancelled on every `parseAndHandleStanza` (receive). It fires when +/// `deadSocketTime` (20 s) elapses after the last send without any receive. +fn is_dead_socket(last_sent_ms: u64, last_received_ms: u64) -> bool { + // Never sent anything yet — timer not armed. + if last_sent_ms == 0 { + return false; + } + // Received data after (or at) the last send — timer cancelled. + if last_received_ms >= last_sent_ms { + return false; + } + // Sent but no reply: check if DEAD_SOCKET_TIME has elapsed since the send. + ms_since(last_sent_ms) + .map(|elapsed| elapsed > DEAD_SOCKET_TIME.as_millis() as u64) + .unwrap_or(false) +} + +/// WA Web: `healthCheckInterval = 15` → `15 * (1 + random())` = 15–30 s. +const KEEP_ALIVE_INTERVAL_MIN: Duration = Duration::from_secs(15); const KEEP_ALIVE_INTERVAL_MAX: Duration = Duration::from_secs(30); -const KEEP_ALIVE_MAX_FAIL_TIME: Duration = Duration::from_secs(180); const KEEP_ALIVE_RESPONSE_DEADLINE: Duration = Duration::from_secs(20); +/// WA Web: `deadSocketTime = 20_000` — if no data arrives for this long +/// after a send, the socket is considered dead and forcibly closed. +const DEAD_SOCKET_TIME: Duration = Duration::from_secs(20); #[derive(Debug, PartialEq)] enum KeepaliveResult { @@ -25,7 +59,7 @@ enum KeepaliveResult { /// Classifies an IQ error into a keepalive result. /// /// Fatal errors indicate the connection is already gone — there is no point -/// waiting for the 180 s grace window. Transient errors (timeout, unexpected +/// waiting for the grace window. Transient errors (timeout, unexpected /// server response) still count as failures but allow the grace window to /// decide whether to force-reconnect. fn classify_keepalive_error(e: &IqError) -> KeepaliveResult { @@ -43,19 +77,39 @@ fn classify_keepalive_error(e: &IqError) -> KeepaliveResult { } impl Client { + /// Sends a keepalive ping and updates the server time offset from + /// the pong's `t` attribute using RTT-adjusted midpoint calculation. + /// + /// WA Web: `sendPing` → `onClockSkewUpdate(Math.round((start + rtt/2) / 1000 - serverTime))` async fn send_keepalive(&self) -> KeepaliveResult { if !self.is_connected() { return KeepaliveResult::FatalFailure; } + // WA Web: skip ping if there are pending IQs + // (`activePing || ackHandlers.length || pendingIqs.size`) + let has_pending = !self.response_waiters.lock().await.is_empty(); + if has_pending { + debug!(target: "Client/Keepalive", "Skipping ping: IQ responses pending"); + return KeepaliveResult::Ok; + } + debug!(target: "Client/Keepalive", "Sending keepalive ping"); - match self - .execute(KeepaliveSpec::with_timeout(KEEP_ALIVE_RESPONSE_DEADLINE)) - .await - { - Ok(()) => { - debug!(target: "Client/Keepalive", "Received keepalive pong"); + let start_ms = chrono::Utc::now().timestamp_millis(); + let iq = wacore::iq::keepalive::KeepaliveSpec::with_timeout(KEEP_ALIVE_RESPONSE_DEADLINE) + .build_iq(); + match self.send_iq(iq).await { + Ok(response_node) => { + let end_ms = chrono::Utc::now().timestamp_millis(); + let rtt_ms = end_ms - start_ms; + debug!(target: "Client/Keepalive", "Received keepalive pong (RTT: {rtt_ms}ms)"); + // WA Web: onClockSkewUpdate — Math.round((startTime + rtt/2) / 1000 - serverTime) + self.unified_session.update_server_time_offset_with_rtt( + &response_node, + start_ms, + rtt_ms, + ); KeepaliveResult::Ok } Err(e) => { @@ -67,7 +121,6 @@ impl Client { } pub(crate) async fn keepalive_loop(self: Arc) { - let mut last_success = chrono::Utc::now(); let mut error_count = 0u32; loop { @@ -83,13 +136,43 @@ impl Client { return; } + // Dead-socket check (WA Web: deadSocketTimer → softCloseSocket). + // Armed on send, cancelled on receive. Fires when data was sent + // but no reply arrived within DEAD_SOCKET_TIME. + let last_sent = self.last_data_sent_ms.load(Ordering::Relaxed); + let last_recv = self.last_data_received_ms.load(Ordering::Relaxed); + if is_dead_socket(last_sent, last_recv) { + let elapsed = ms_since(last_sent).unwrap_or(0); + warn!( + target: "Client/Keepalive", + "No data received for {:.1}s after send (dead socket), forcing reconnect.", + elapsed as f64 / 1000.0 + ); + self.reconnect_immediately().await; + return; + } + + // WA Web: maybeScheduleHealthCheck — only send ping when idle. + // If we recently received data, the connection is proven alive; + // skip the ping and reschedule (same as WA Web rescheduling the + // healthCheckTimer after activity). + if let Some(since_recv) = ms_since(last_recv) + && since_recv < KEEP_ALIVE_INTERVAL_MIN.as_millis() as u64 + { + // Connection alive — reset error state, skip ping. + if error_count > 0 { + debug!(target: "Client/Keepalive", "Keepalive restored (recent activity)."); + error_count = 0; + } + continue; + } + match self.send_keepalive().await { KeepaliveResult::Ok => { if error_count > 0 { debug!(target: "Client/Keepalive", "Keepalive restored after {error_count} failure(s)."); } error_count = 0; - last_success = chrono::Utc::now(); } KeepaliveResult::FatalFailure => { debug!(target: "Client/Keepalive", "Fatal keepalive failure, exiting loop."); @@ -98,16 +181,6 @@ impl Client { KeepaliveResult::TransientFailure => { error_count += 1; warn!(target: "Client/Keepalive", "Keepalive timeout, error count: {error_count}"); - - if self.enable_auto_reconnect.load(Ordering::Relaxed) - && chrono::Utc::now().signed_duration_since(last_success) - > chrono::Duration::from_std(KEEP_ALIVE_MAX_FAIL_TIME) - .expect("KEEP_ALIVE_MAX_FAIL_TIME fits in chrono::Duration") - { - warn!(target: "Client/Keepalive", "Forcing reconnect due to keepalive failure for over {} seconds.", KEEP_ALIVE_MAX_FAIL_TIME.as_secs()); - self.reconnect_immediately().await; - return; - } } } }, @@ -188,4 +261,88 @@ mod tests { "ParseError should be transient — bad response, not a dead connection" ); } + + // ── ms_since tests ─────────────────────────────────────────────────── + + #[test] + fn test_ms_since_never_set() { + assert_eq!(ms_since(0), None, "should return None when timestamp is 0"); + } + + #[test] + fn test_ms_since_recent() { + let now_ms = chrono::Utc::now().timestamp_millis() as u64; + let elapsed = ms_since(now_ms).unwrap(); + assert!(elapsed < 100, "should be near-zero, got {elapsed}ms"); + } + + #[test] + fn test_ms_since_stale() { + let thirty_sec_ago = (chrono::Utc::now().timestamp_millis() as u64).saturating_sub(30_000); + let elapsed = ms_since(thirty_sec_ago).unwrap(); + assert!( + (29_000..=31_000).contains(&elapsed), + "should be ~30s, got {elapsed}ms" + ); + } + + // ── is_dead_socket tests ───────────────────────────────────────────── + + #[test] + fn test_dead_socket_never_sent() { + // Never sent anything → timer not armed + assert!(!is_dead_socket(0, 0)); + } + + #[test] + fn test_dead_socket_received_after_send() { + // Sent at T, received at T+1 → timer cancelled + let t = chrono::Utc::now().timestamp_millis() as u64; + assert!(!is_dead_socket(t, t + 1)); + } + + #[test] + fn test_dead_socket_sent_recently() { + // Sent just now, no reply yet but within 20s → not dead + let now = chrono::Utc::now().timestamp_millis() as u64; + assert!(!is_dead_socket(now, 0)); + } + + #[test] + fn test_dead_socket_sent_long_ago_no_reply() { + // Sent 30s ago, no reply → dead + let thirty_ago = (chrono::Utc::now().timestamp_millis() as u64).saturating_sub(30_000); + assert!(is_dead_socket(thirty_ago, 0)); + } + + #[test] + fn test_dead_socket_sent_long_ago_old_reply() { + // Sent 30s ago, last reply was 31s ago (before the send) → dead + let thirty_ago = (chrono::Utc::now().timestamp_millis() as u64).saturating_sub(30_000); + let thirty_one_ago = thirty_ago.saturating_sub(1_000); + assert!(is_dead_socket(thirty_ago, thirty_one_ago)); + } + + #[test] + fn test_dead_socket_sent_long_ago_recent_reply() { + // Sent 30s ago, last reply was 1s ago → not dead (reply cancelled timer) + let thirty_ago = (chrono::Utc::now().timestamp_millis() as u64).saturating_sub(30_000); + let one_ago = (chrono::Utc::now().timestamp_millis() as u64).saturating_sub(1_000); + assert!(!is_dead_socket(thirty_ago, one_ago)); + } + + // ── constants sanity tests ─────────────────────────────────────────── + + #[test] + fn test_keepalive_interval_matches_wa_web() { + // WA Web: healthCheckInterval = 15, formula 15*(1+random()) = 15–30s + assert_eq!(KEEP_ALIVE_INTERVAL_MIN, Duration::from_secs(15)); + assert_eq!(KEEP_ALIVE_INTERVAL_MAX, Duration::from_secs(30)); + } + + #[test] + fn test_dead_socket_time_matches_wa_web() { + // WA Web: deadSocketTime = 20_000 + assert_eq!(DEAD_SOCKET_TIME, Duration::from_secs(20)); + } } diff --git a/src/unified_session.rs b/src/unified_session.rs index d8e73da5a..f207109f2 100644 --- a/src/unified_session.rs +++ b/src/unified_session.rs @@ -55,6 +55,25 @@ impl UnifiedSessionManager { } } + /// Update server time offset using RTT-adjusted midpoint calculation. + /// + /// WA Web: `Math.round((startTime + rtt/2) / 1000 - serverTime)` + /// + /// This gives a more accurate clock skew estimate by assuming the server + /// timestamp corresponds to the midpoint of the round trip. + pub fn update_server_time_offset_with_rtt(&self, node: &Node, start_time_ms: i64, rtt_ms: i64) { + if let Some(t_str) = node.attrs.get("t").and_then(|v| v.as_str()) + && let Ok(server_time) = t_str.parse::() + && server_time > 0 + { + let midpoint_s = (start_time_ms + rtt_ms / 2) / 1000; + let offset_ms = (server_time - midpoint_s) * 1000; + self.server_time_offset_ms + .store(offset_ms, Ordering::Relaxed); + debug!(target: "UnifiedSession", "Server time offset: {}ms (RTT: {}ms)", offset_ms, rtt_ms); + } + } + pub fn calculate_session_id(&self) -> String { let offset = self.server_time_offset_ms.load(Ordering::Relaxed); UnifiedSession::calculate_id(offset) diff --git a/tests/e2e/tests/chatstate_ttl.rs b/tests/e2e/tests/chatstate_ttl.rs index 525fb9ee7..05b882ba2 100644 --- a/tests/e2e/tests/chatstate_ttl.rs +++ b/tests/e2e/tests/chatstate_ttl.rs @@ -14,11 +14,11 @@ use wacore::types::events::Event; /// /// Flow: /// 1. A and B connect -/// 2. B goes offline via reconnect() — auto_reconnect_errors=2 creates a ~4s offline window +/// 2. B goes offline via reconnect() — creates a ~5s offline window (see `RECONNECT_BACKOFF_STEP`) /// 3. A sends typing indicator to B (queued with TTL) /// 4. B auto-reconnects after the TTL expires — chatstate should be filtered out during drain /// -/// Requires mock server with CHATSTATE_TTL_SECS=3 (so TTL expires before the ~4s reconnect). +/// Requires mock server with CHATSTATE_TTL_SECS=3 (so TTL expires before the ~5s reconnect). #[tokio::test] async fn test_expired_chatstate_not_delivered() -> anyhow::Result<()> { let _ = env_logger::builder().is_test(true).try_init(); @@ -30,18 +30,18 @@ async fn test_expired_chatstate_not_delivered() -> anyhow::Result<()> { info!("B={jid_b}"); - // B goes offline — reconnect() sets auto_reconnect_errors=2, causing ~4s delay - // before auto-reconnect. With CHATSTATE_TTL_SECS=3, the chatstate expires at 3s, - // and B reconnects at ~4s, so the drain filters it out. + // B goes offline — reconnect() uses RECONNECT_BACKOFF_STEP to create a ~5s + // offline window. With CHATSTATE_TTL_SECS=3, the chatstate expires at 3s, + // and B reconnects at ~5s, so the drain filters it out. client_b.client.reconnect().await; - info!("B disconnected (will auto-reconnect in ~4s)"); + info!("B disconnected (will auto-reconnect after backoff)"); tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; // A sends typing indicator while B is offline (queued with short TTL) client_a.client.chatstate().send_composing(&jid_b).await?; info!("A sent typing indicator to offline B"); - // B auto-reconnects after ~4s. Wait for reconnect + event drain. + // B auto-reconnects after backoff. Wait for reconnect + event drain. // The expired chatstate should NOT be delivered. let result = client_b .wait_for_event(15, |e| matches!(e, Event::ChatPresence(_))) diff --git a/tests/e2e/tests/offline_receipts.rs b/tests/e2e/tests/offline_receipts.rs index 5d813163e..7076116ee 100644 --- a/tests/e2e/tests/offline_receipts.rs +++ b/tests/e2e/tests/offline_receipts.rs @@ -181,7 +181,9 @@ async fn test_deferred_delivery_receipt_on_reconnect() -> anyhow::Result<()> { let msg_id = client_a.client.send_message(jid_b.clone(), message).await?; info!("A sent message to offline B: {msg_id}"); - // A should NOT get delivery receipt yet (only sender receipt) + // A should NOT get delivery receipt yet (only sender receipt). + // Timeout must be shorter than the reconnect backoff (see RECONNECT_BACKOFF_STEP) + // so B is still offline during this window. let early_receipt = client_a .wait_for_event(3, |e| { matches!( @@ -302,7 +304,7 @@ async fn test_offline_presence_coalescing() -> anyhow::Result<()> { panic!("Expected Presence event"); } - // Try to get a second presence — should timeout (coalesced to one) + // Try to get a second presence — should timeout (coalesced to one). let second = client_b .wait_for_event(3, |e| matches!(e, Event::Presence(_))) .await;