Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::sync::Mutex<HashSet<String>>>,

/// Track retry attempts per message to prevent infinite retry loops.
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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
Expand Down
6 changes: 6 additions & 0 deletions src/client/device_registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
8 changes: 8 additions & 0 deletions src/handlers/ib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,14 @@ async fn handle_ib_impl(client: Arc<Client>, 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 {
client_clone.flush_pending_device_sync().await;
}))
Comment on lines +170 to +174

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Flush pending device sync after offline queue fully settles

This schedules exactly one flush_pending_device_sync() call 2 seconds after the offline end marker, but unknown-device entries are enqueued from handle_unknown_device_sync while offline messages are still being processed. Under large offline backlogs, message handlers can continue adding users after this one-shot flush runs, leaving those users stuck in PendingDeviceSync with no later trigger to flush them, so device lists never refresh and retries can keep failing for those senders.

Useful? React with 👍 / 👎.

.detach();
Comment on lines +164 to +175

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Wait for the offline pipeline to drain before flushing.

This is fired off the terminal <offline> marker, but a few lines above the file already treats “offline delivery ended” as a separate barrier via wait_for_offline_delivery_end(). Running flush_pending_device_sync() in a detached task here can miss senders that are still queued for message processing, and reconnect cleanup can clear the pending set before the task runs.

Suggested fix
                 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 {
-                        client_clone.flush_pending_device_sync().await;
-                    }))
-                    .detach();
+                client.wait_for_offline_delivery_end().await;
+                if !client.is_shutting_down() {
+                    client.flush_pending_device_sync().await;
+                }
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/handlers/ib.rs` around lines 164 - 170, The detached task running
client_clone.flush_pending_device_sync() can run before queued senders finish;
change the flow to wait for the offline delivery barrier first (use
wait_for_offline_delivery_end() where the offline marker is handled) and then
run flush_pending_device_sync() so it executes after the barrier—either call
client.flush_pending_device_sync().await directly or spawn it and await the
JoinHandle instead of .detach(); keep using Arc::clone(&client) and
client.runtime.spawn if you must offload, but ensure you await completion so
pending senders aren’t dropped by reconnect cleanup.

}
"thread_metadata" => {
// Present in some sessions; safe to ignore for now until feature implemented.
Expand Down
1 change: 1 addition & 0 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
49 changes: 45 additions & 4 deletions src/message.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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() {
Expand All @@ -1092,6 +1112,26 @@ impl Client {
Ok(())
}

/// WA Web: online → `syncDeviceListJob`, offline → `OfflinePendingDeviceCache`.
async fn handle_unknown_device_sync(self: &Arc<Self>, info: &MessageInfo) {
let user_jid = info.source.sender.to_non_ad();

if info.offline.is_some() {
log::debug!("Queueing {} for pending device sync (offline)", user_jid);
self.pending_device_sync.add(user_jid).await;
} else {
log::debug!("Triggering immediate device sync for {}", user_jid);
let client = Arc::clone(self);
self.runtime
.spawn(Box::pin(async move {
if let Err(e) = client.get_user_devices(&[user_jid]).await {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Force-refresh device list for unknown-device recovery

Calling get_user_devices here does not guarantee a server sync when the sender already has a cached/DB device record, because get_user_devices returns registry hits without fetching from network (src/usync.rs, early get_devices_from_registry fast-path). In the exact unknown-device scenario this change targets (known user adds a new companion), the stale record usually exists and this call becomes a no-op, so the new device is never learned and subsequent messages keep failing with retries.

Useful? React with 👍 / 👎.

log::warn!("Immediate device sync failed: {e:?}");
}
}))
.detach();
Comment on lines +1117 to +1136

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== handle_unknown_device_sync =="
sed -n '1115,1135p' src/message.rs

echo
echo "== get_user_devices / flush_pending_device_sync definitions =="
rg -nP --type rust -C8 '^\s*(pub(?:\([^)]*\))?\s+)?async\s+fn\s+(get_user_devices|flush_pending_device_sync)\b'

echo
echo "== LID->PN normalization around device queries =="
rg -n --type rust -C4 'get_phone_number|phone_jid_for_lid_user|sender_alt|is_lid\(|to_non_ad\(|pending_device_sync'

Repository: jlucaso1/whatsapp-rust

Length of output: 50379


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== handle_unknown_device_sync function signature and context ==="
sed -n '1100,1140p' src/message.rs | head -50

echo
echo "=== Client fields and lid_pn_cache access ==="
rg -n 'struct Client|pub.*lid_pn_cache' src/client.rs | head -20

echo
echo "=== Check if sender_alt is available in handle_unknown_device_sync (MessageInfo struct) ==="
rg -n 'pub struct MessageInfo|pub sender_alt' --type rust wacore/ | grep -A2 MessageInfo

echo
echo "=== Examples of sender_alt usage for LID→PN conversion elsewhere ==="
rg -n 'sender_alt' src/message.rs | head -10

Repository: jlucaso1/whatsapp-rust

Length of output: 2628


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== MessageInfo struct with sender_alt field ==="
sed -n '132,150p' wacore/src/types/message.rs

echo
echo "=== Verify sender_alt is populated in group messages with LID senders ==="
sed -n '170,195p' wacore/src/messages.rs

echo
echo "=== cache_lid_pn_from_message implementation ==="
rg -n 'cache_lid_pn_from_message' src/message.rs -A 10 | head -20

Repository: jlucaso1/whatsapp-rust

Length of output: 2736


Use sender_alt to derive the device-query JID for LID senders.

info.source.sender.to_non_ad() preserves @lid identifiers, and this value is passed directly to both pending_device_sync.add() and get_user_devices(). For group messages with LID-mode addressing, info.source.sender_alt contains the corresponding phone-number JID and is reliably populated. Without using sender_alt here, LID senders will continuously refresh against their cached @lid identity and never update companion device lists.

Derive the query JID as:

  • If sender_alt.is_some(), prefer it (the alternate form for device queries)
  • Otherwise, fall back to sender.to_non_ad() (for non-group or PN-mode groups)

This matches the pattern used elsewhere in the codebase (e.g., wacore/src/send.rs:1010–1018) where LID→PN normalization is the caller's responsibility before issuing device queries.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/message.rs` around lines 1117 - 1133, The code currently derives user_jid
via info.source.sender.to_non_ad(), which preserves `@lid` and causes LID senders
to never refresh companion device lists; change the query JID derivation to
prefer info.source.sender_alt when present and fall back to
info.source.sender.to_non_ad() otherwise, then use that normalized JID for
pending_device_sync.add(), invalidate_device_cache(&query_jid.user), and
get_user_devices(&[query_jid]) so LID→PN normalization matches other call sites
(see functions/methods: pending_device_sync.add, invalidate_device_cache,
get_user_devices, info.source.sender_alt, info.source.sender.to_non_ad()).

}
}

async fn handle_decrypted_plaintext(
self: Arc<Self>,
enc_type: &str,
Expand Down Expand Up @@ -3833,6 +3873,7 @@ mod tests {
verified_name: None,
device_sent_meta: None,
ephemeral_expiration: None,
offline: None,
}
}

Expand Down
1 change: 1 addition & 0 deletions src/pdo.rs
Original file line number Diff line number Diff line change
Expand Up @@ -414,6 +414,7 @@ impl Client {
verified_name: None,
device_sent_meta: None,
ephemeral_expiration: None,
offline: None,
})
}

Expand Down
29 changes: 29 additions & 0 deletions src/pending_device_sync.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
//! 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<HashSet<Jid>>,
}

impl PendingDeviceSync {
pub(crate) fn new() -> Self {
Self {
pending: async_lock::Mutex::new(HashSet::new()),
}
}

pub(crate) async fn add(&self, jid: Jid) {
self.pending.lock().await.insert(jid);
}

pub(crate) async fn take_all(&self) -> Vec<Jid> {
self.pending.lock().await.drain().collect()
}

pub(crate) async fn clear(&self) {
self.pending.lock().await.clear();
}
}
23 changes: 23 additions & 0 deletions src/usync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,29 @@ 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());

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: {e:?}");
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
}
}

#[cfg(test)]
Expand Down
3 changes: 3 additions & 0 deletions wacore/src/messages.rs
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,8 @@ pub fn parse_message_info(
source.chat.agent = 0;
}

let offline = attrs.optional_string("offline").map(|_| true);

Ok(MessageInfo {
source,
id,
Expand All @@ -259,6 +261,7 @@ pub fn parse_message_info(
.optional_string("edit")
.map(|s| EditAttribute::from(s.to_string()))
.unwrap_or_default(),
offline,
..Default::default()
})
}
19 changes: 18 additions & 1 deletion wacore/src/protocol/retry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -89,7 +97,8 @@ pub fn extract_registration_id_from_node(node: &Node) -> Option<u32> {
/// 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
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

Expand Down Expand Up @@ -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!(
Expand Down
2 changes: 2 additions & 0 deletions wacore/src/types/message.rs
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,8 @@ pub struct MessageInfo {
pub device_sent_meta: Option<DeviceSentMeta>,
/// Ephemeral duration in seconds, extracted from `contextInfo.expiration`.
pub ephemeral_expiration: Option<u32>,
/// Stanza `offline` attribute. `Some` = offline delivery, `None` = online.
pub offline: Option<bool>,

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick | 🔵 Trivial

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verify whether `MessageInfo.offline` is actually used as a tri-state anywhere.
rg -n --type rust '\boffline\s*:\s*(Some\(true\)|Some\(false\)|None)' .
rg -n --type rust '\.offline\b' .

Repository: jlucaso1/whatsapp-rust

Length of output: 216


Model offline as a two-state field instead of Option<bool>.

Current usage only creates Some(true) or None and checks presence via .is_some(); the Some(false) state is unreachable dead code. Use bool is_offline or a small enum to clarify the contract and remove offline: None boilerplate from fallback constructors.

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@wacore/src/types/message.rs` around lines 149 - 150, The field `offline` in
the message struct is currently an Option<bool> but is only ever set to
Some(true) or None and callers use .is_some(); change the field to a two-state
representation (prefer `is_offline: bool` or a small enum like
`DeliveryState::{Online, Offline}`) to remove unreachable `Some(false)` and
clarify intent, then update all constructors/fallbacks that set `offline` to use
the new default (false or Online) and replace all `.offline.is_some()` checks
with the new boolean or enum pattern matches; ensure the struct definition in
message.rs and every usage site (constructors, deserializers, and conditionals)
are updated accordingly.

}

impl MessageInfo {
Expand Down
Loading