diff --git a/src/client.rs b/src/client.rs index 4747f6856..76a833958 100644 --- a/src/client.rs +++ b/src/client.rs @@ -355,6 +355,8 @@ pub struct Client { pub(crate) sender_key_device_cache: crate::sender_key_device_cache::SenderKeyDeviceCache, + pub(crate) pending_device_sync: crate::pending_device_sync::PendingDeviceSync, + pub(crate) pending_retries: Arc>>, /// Track retry attempts per message to prevent infinite retry loops. @@ -660,6 +662,8 @@ impl Client { &cache_config.sender_key_devices_cache, ), + pending_device_sync: crate::pending_device_sync::PendingDeviceSync::new(), + pending_retries: Arc::new(std::sync::Mutex::new(HashSet::new())), message_retry_counts: cache_config.message_retry_counts.build_with_ttl(), @@ -1229,6 +1233,7 @@ impl Client { // 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); + self.pending_device_sync.clear().await; // Reset offline sync state for next connection self.offline_sync_completed.store(false, Ordering::Relaxed); self.offline_sync_metrics diff --git a/src/client/device_registry.rs b/src/client/device_registry.rs index 0b63269ed..ec8846824 100644 --- a/src/client/device_registry.rs +++ b/src/client/device_registry.rs @@ -96,6 +96,12 @@ impl Client { .collect() } + /// WA Web: `isFromKnownDevice(author)` — local check only, no network. + pub(crate) async fn is_from_known_device(&self, sender: &wacore_binary::jid::Jid) -> bool { + let device_id = sender.device as u32; + self.has_device(&sender.user, device_id).await + } + /// Check if a device exists for a user. /// Returns true for device_id 0 (primary device always exists). pub(crate) async fn has_device(&self, user: &str, device_id: u32) -> bool { diff --git a/src/handlers/ib.rs b/src/handlers/ib.rs index 04376e270..5266a1b40 100644 --- a/src/handlers/ib.rs +++ b/src/handlers/ib.rs @@ -160,6 +160,19 @@ async fn handle_ib_impl(client: Arc, node: &Node) { debug!(target: "Client/OfflineSync", "Offline sync completed, received {} items", count); client.complete_offline_sync(count); + + let client_clone = Arc::clone(&client); + client + .runtime + .spawn(Box::pin(async move { + // WA Web: OFFLINE_DEVICE_SYNC_DELAY = 2000ms + client_clone + .runtime + .sleep(std::time::Duration::from_secs(2)) + .await; + client_clone.flush_pending_device_sync().await; + })) + .detach(); } "thread_metadata" => { // Present in some sessions; safe to ignore for now until feature implemented. diff --git a/src/lib.rs b/src/lib.rs index 6f754336e..0bf2109da 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod portable_cache; pub mod cache_config; pub use cache_config::{CacheConfig, CacheEntryConfig, CacheStores}; pub mod cache_store; +pub(crate) mod pending_device_sync; pub(crate) mod sender_key_device_cache; pub use cache_store::CacheStore; pub mod http; diff --git a/src/message.rs b/src/message.rs index c375c3b83..274d13031 100644 --- a/src/message.rs +++ b/src/message.rs @@ -1025,6 +1025,17 @@ impl Client { match decrypt_result { Ok(padded_plaintext) => { + // WA Web: isFromKnownDevice() in preProcessMsg + if !self.is_from_known_device(&info.source.sender).await { + warn!( + "[msg:{}] Unknown device {}, triggering device sync", + info.id, info.source.sender + ); + self.handle_unknown_device_sync(info).await; + self.spawn_retry_receipt(info, RetryReason::UnknownCompanionNoPrekey); + continue; + } + if let Err(e) = self .clone() .handle_decrypted_plaintext( @@ -1058,15 +1069,24 @@ impl Client { continue; } - // No sender key for this group/sender — the SKDM was never received - // (sender thinks we have it from a previous status/session). - // Send retry receipt to ask sender to re-distribute SKDM. + let is_unknown_device = !self.is_from_known_device(&info.source.sender).await; + let retry_reason = if is_unknown_device { + RetryReason::UnknownCompanionNoPrekey + } else { + RetryReason::NoSession + }; + warn!( "No sender key state for group message [msg:{}] from {}: {}. Sending retry receipt.", info.id, info.source.sender, msg ); + + if is_unknown_device { + self.handle_unknown_device_sync(info).await; + } + self.dispatch_undecryptable_event(info, decrypt_fail_mode); - self.spawn_retry_receipt(info, RetryReason::NoSession); + self.spawn_retry_receipt(info, retry_reason); } Err(e) => { if info.is_expired_status() { @@ -1092,6 +1112,31 @@ impl Client { Ok(()) } + /// WA Web: online → `syncDeviceListJob`, offline → `OfflinePendingDeviceCache`. + async fn handle_unknown_device_sync(self: &Arc, info: &MessageInfo) { + let user_jid = info.source.sender.to_non_ad(); + + // Dedup: skip if we already have a sync pending/in-flight for this user + if !self.pending_device_sync.add(user_jid.clone()).await { + return; + } + + if info.is_offline { + log::debug!("Queueing {} for pending device sync (offline)", user_jid); + } else { + log::debug!("Triggering immediate device sync for {}", user_jid); + let client = Arc::clone(self); + self.runtime + .spawn(Box::pin(async move { + client.invalidate_device_cache(&user_jid.user).await; + if let Err(e) = client.get_user_devices(&[user_jid]).await { + log::warn!("Immediate device sync failed: {e:?}"); + } + })) + .detach(); + } + } + async fn handle_decrypted_plaintext( self: Arc, enc_type: &str, @@ -3833,6 +3878,7 @@ mod tests { verified_name: None, device_sent_meta: None, ephemeral_expiration: None, + is_offline: false, } } diff --git a/src/pdo.rs b/src/pdo.rs index 331d8fd6d..164c1a24c 100644 --- a/src/pdo.rs +++ b/src/pdo.rs @@ -414,6 +414,7 @@ impl Client { verified_name: None, device_sent_meta: None, ephemeral_expiration: None, + is_offline: false, }) } diff --git a/src/pending_device_sync.rs b/src/pending_device_sync.rs new file mode 100644 index 000000000..6759b30e4 --- /dev/null +++ b/src/pending_device_sync.rs @@ -0,0 +1,30 @@ +//! Batches unknown-device users during offline sync for deferred usync. +//! WA Web: `OfflinePendingDeviceCache` + `doPendingDeviceSync()`. + +use std::collections::HashSet; +use wacore_binary::jid::Jid; + +pub(crate) struct PendingDeviceSync { + pending: async_lock::Mutex>, +} + +impl PendingDeviceSync { + pub(crate) fn new() -> Self { + Self { + pending: async_lock::Mutex::new(HashSet::new()), + } + } + + /// Insert a user. Returns `true` if newly inserted, `false` if already present. + pub(crate) async fn add(&self, jid: Jid) -> bool { + self.pending.lock().await.insert(jid) + } + + pub(crate) async fn take_all(&self) -> Vec { + self.pending.lock().await.drain().collect() + } + + pub(crate) async fn clear(&self) { + self.pending.lock().await.clear(); + } +} diff --git a/src/retry.rs b/src/retry.rs index 6575dbae8..023f45a13 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -48,11 +48,6 @@ fn extract_registration_id_from_node(node: &Node) -> Option { /// We refuse to resend if the requester has already retried this many times. const MAX_RETRY_COUNT: u8 = 5; -/// Minimum retry count before we include keys in retry receipts. -/// WhatsApp Web only includes keys when retryCount >= 2, giving the first -/// retry a chance to succeed without key exchange overhead. -const MIN_RETRY_COUNT_FOR_KEYS: u8 = 2; - /// Minimum retry count before we start tracking base keys. /// WhatsApp Web saves base key on retry 2, checks on retry > 2. const MIN_RETRY_FOR_BASE_KEY_CHECK: u8 = 2; @@ -717,14 +712,7 @@ impl Client { .bytes(registration_id_bytes) .build(); - // WhatsApp Web only includes keys when retryCount >= 2. - // First retry gives the sender a chance to resend without full key exchange. - // - // WA Web includes keys at retryCount >= MIN_RETRY_COUNT_FOR_KEYS. - // Optimization for NoSession: include keys on retry#1 to reduce round-trips - // for skmsg-only failures where the sender needs our prekeys for SKDM. - let include_keys_early = reason == RetryReason::NoSession; - let keys_node = if retry_count >= MIN_RETRY_COUNT_FOR_KEYS || include_keys_early { + let keys_node = if wacore::protocol::retry::should_include_keys(retry_count, reason) { let device_store = self.persistence_manager.get_device_arc().await; let device_guard = device_store.read().await; @@ -1590,8 +1578,8 @@ mod tests { for (retry_count, reason, should_include_keys, description) in test_cases { // Replicate the logic from send_retry_receipt - let include_keys_early = reason == RetryReason::NoSession; - let would_include_keys = retry_count >= MIN_RETRY_COUNT_FOR_KEYS || include_keys_early; + let would_include_keys = + wacore::protocol::retry::should_include_keys(retry_count, reason); assert_eq!( would_include_keys, should_include_keys, @@ -1642,10 +1630,8 @@ mod tests { RetryReason::NoSession }; - // Apply the optimization logic - let include_keys_early = reason == RetryReason::NoSession; let would_include_keys = - retry_count >= MIN_RETRY_COUNT_FOR_KEYS || include_keys_early; + wacore::protocol::retry::should_include_keys(retry_count, reason); if would_include_keys { keys_included.fetch_add(1, Ordering::SeqCst); @@ -1701,8 +1687,7 @@ mod tests { let reason = RetryReason::NoSession; // With optimization, we include keys on retry#1 - let include_keys_early = reason == RetryReason::NoSession; - let would_include_keys = retry_count >= MIN_RETRY_COUNT_FOR_KEYS || include_keys_early; + let would_include_keys = wacore::protocol::retry::should_include_keys(retry_count, reason); assert!( would_include_keys, diff --git a/src/usync.rs b/src/usync.rs index 9a0c5c940..b32e17ab4 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -190,6 +190,40 @@ impl Client { ); Ok(()) } + + /// WA Web: `doPendingDeviceSync()` — flush batched unknown-device users. + pub(crate) async fn flush_pending_device_sync(&self) { + let pending = self.pending_device_sync.take_all().await; + if pending.is_empty() { + return; + } + + debug!("Flushing pending device sync for {} users", pending.len()); + + // Invalidate stale records so get_user_devices hits the network + for jid in &pending { + self.invalidate_device_cache(&jid.user).await; + } + + match self.get_user_devices(&pending).await { + Ok(devices) => { + debug!( + "Pending device sync completed: {} devices across {} users", + devices.len(), + pending.len() + ); + } + Err(e) => { + warn!( + "Pending device sync failed, re-enqueueing {} users: {e:?}", + pending.len() + ); + for jid in pending { + self.pending_device_sync.add(jid).await; + } + } + } + } } #[cfg(test)] diff --git a/wacore/src/messages.rs b/wacore/src/messages.rs index 0b10d7887..1d4b5a172 100644 --- a/wacore/src/messages.rs +++ b/wacore/src/messages.rs @@ -244,6 +244,8 @@ pub fn parse_message_info( source.chat.agent = 0; } + let is_offline = attrs.optional_string("offline").is_some(); + Ok(MessageInfo { source, id, @@ -259,6 +261,7 @@ pub fn parse_message_info( .optional_string("edit") .map(|s| EditAttribute::from(s.to_string())) .unwrap_or_default(), + is_offline, ..Default::default() }) } diff --git a/wacore/src/protocol/retry.rs b/wacore/src/protocol/retry.rs index 785100658..3fd129ff0 100644 --- a/wacore/src/protocol/retry.rs +++ b/wacore/src/protocol/retry.rs @@ -47,6 +47,14 @@ pub enum RetryReason { InvalidSession = 8, /// Invalid message key InvalidMsgKey = 9, + /// Bad broadcast ephemeral setting + BadBroadcastEphemeralSetting = 10, + /// Unknown companion device, not in our device list + UnknownCompanionNoPrekey = 11, + /// ADV signature or device identity failure + AdvFailure = 12, + /// Status revoke delay exceeded + StatusRevokeDelay = 13, } /// Helper to extract bytes content from a Node. @@ -89,7 +97,8 @@ pub fn extract_registration_id_from_node(node: &Node) -> Option { /// keys are included on retry #1 for `NoSession` errors to reduce round-trips /// for skmsg-only message failures. pub fn should_include_keys(retry_count: u8, reason: RetryReason) -> bool { - let include_keys_early = reason == RetryReason::NoSession; + let include_keys_early = + reason == RetryReason::NoSession || reason == RetryReason::UnknownCompanionNoPrekey; retry_count >= MIN_RETRY_COUNT_FOR_KEYS || include_keys_early } @@ -192,6 +201,14 @@ mod tests { ); } + #[test] + fn should_include_keys_unknown_companion_retry_1() { + assert!( + should_include_keys(1, RetryReason::UnknownCompanionNoPrekey), + "UnknownCompanionNoPrekey at retry#1 should include keys" + ); + } + #[test] fn should_include_keys_invalid_message_retry_1() { assert!( diff --git a/wacore/src/types/message.rs b/wacore/src/types/message.rs index 83a3498a2..89b5fc444 100644 --- a/wacore/src/types/message.rs +++ b/wacore/src/types/message.rs @@ -146,6 +146,8 @@ pub struct MessageInfo { pub device_sent_meta: Option, /// Ephemeral duration in seconds, extracted from `contextInfo.expiration`. pub ephemeral_expiration: Option, + /// Whether this message was delivered during offline sync. + pub is_offline: bool, } impl MessageInfo {