Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
2 changes: 1 addition & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ cargo test -p e2e-tests # requires mock server running

## Critical Conventions

- **State**: Never modify Device state directly. Use `DeviceCommand` + `PersistenceManager::process_command()`. Read via `get_device_snapshot()`.
- **State**: Never modify Device state directly (not even in tests — a write-lock mutation bypasses the cached snapshot). Use `DeviceCommand` + `PersistenceManager::process_command()` (or `modify_device` internally). Read via `get_device_snapshot()` — it returns a cached `Arc<Device>` (sync, refcount-cheap, safe to call per message); borrow fields from the held snapshot instead of cloning them. `get_device_arc()` is only for store adapters that need `&mut Device` trait access.
- **Async**: All I/O uses Tokio. Wrap blocking I/O (`ureq`) and heavy CPU work in `tokio::task::spawn_blocking`.
- **Concurrency**: `session_locks` serializes per-sender Signal encrypt/decrypt. `message_enqueue_locks` serializes per-chat incoming message processing. Outgoing sends are not per-chat locked (matches WA Web).
- **Errors**: `thiserror` for typed errors, `anyhow` for multi-failure functions. No `.unwrap()` outside tests.
Expand Down
2 changes: 1 addition & 1 deletion agent_docs/feature_implementation.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ When adding a new feature, follow this flow that mirrors WhatsApp Web behavior w

4. **Keep state changes behind the PersistenceManager**
- Use `DeviceCommand` + `PersistenceManager::process_command()` for mutations
- Use `get_device_snapshot()` for read access
- Use `get_device_snapshot()` for read access — sync, returns a cached `Arc<Device>` (refcount bump, no Device clone, no lock); hold it and borrow fields rather than cloning them

5. **Confirm concurrency requirements**
- Network I/O stays async
Expand Down
10 changes: 5 additions & 5 deletions src/bot.rs
Original file line number Diff line number Diff line change
Expand Up @@ -974,7 +974,7 @@ mod tests {

let client = bot.client();
let persistence_manager = client.persistence_manager();
let device = persistence_manager.get_device_snapshot().await;
let device = persistence_manager.get_device_snapshot();

// Verify the device props were overridden
assert_eq!(device.device_props.os, Some(custom_os));
Expand All @@ -1001,7 +1001,7 @@ mod tests {

let client = bot.client();
let persistence_manager = client.persistence_manager();
let device = persistence_manager.get_device_snapshot().await;
let device = persistence_manager.get_device_snapshot();

// Verify only OS was overridden, version should be default
assert_eq!(device.device_props.os, Some(custom_os));
Expand Down Expand Up @@ -1037,7 +1037,7 @@ mod tests {

let client = bot.client();
let persistence_manager = client.persistence_manager();
let device = persistence_manager.get_device_snapshot().await;
let device = persistence_manager.get_device_snapshot();

// Verify only version was overridden, OS should be default ("rust")
assert_eq!(device.device_props.version, Some(custom_version));
Expand Down Expand Up @@ -1069,7 +1069,7 @@ mod tests {

let client = bot.client();
let persistence_manager = client.persistence_manager();
let device = persistence_manager.get_device_snapshot().await;
let device = persistence_manager.get_device_snapshot();

// Verify platform type was set to Chrome
assert_eq!(
Expand Down Expand Up @@ -1119,7 +1119,7 @@ mod tests {

let client = bot.client();
let persistence_manager = client.persistence_manager();
let device = persistence_manager.get_device_snapshot().await;
let device = persistence_manager.get_device_snapshot();

// Verify all device props were overridden
assert_eq!(device.device_props.os, Some(custom_os));
Expand Down
39 changes: 14 additions & 25 deletions src/client/accessors.rs
Original file line number Diff line number Diff line change
Expand Up @@ -114,38 +114,27 @@ impl Client {
self.persistence_manager.clone()
}

pub async fn get_push_name(&self) -> String {
// The owned returns below are the only clones left: the snapshot read
// itself is an Arc refcount bump (no lock against writers). Callers that
// only need a borrow can hold `persistence_manager().get_device_snapshot()`
// and read fields directly.
pub fn get_push_name(&self) -> String {
self.persistence_manager
.get_device_arc()
.await
.read()
.await
.get_device_snapshot()
.push_name
.clone()
}

pub async fn get_pn(&self) -> Option<Jid> {
self.persistence_manager
.get_device_arc()
.await
.read()
.await
.pn
.clone()
pub fn get_pn(&self) -> Option<Jid> {
self.persistence_manager.get_device_snapshot().pn.clone()
}

pub async fn get_lid(&self) -> Option<Jid> {
self.persistence_manager
.get_device_arc()
.await
.read()
.await
.lid
.clone()
pub fn get_lid(&self) -> Option<Jid> {
self.persistence_manager.get_device_snapshot().lid.clone()
}

pub(crate) async fn require_pn(&self) -> Result<Jid> {
self.get_pn().await.ok_or(ClientError::NotLoggedIn.into())
pub(crate) fn require_pn(&self) -> Result<Jid> {
self.get_pn().ok_or(ClientError::NotLoggedIn.into())
}

/// Resolve our own JID for a group, respecting its addressing mode.
Expand All @@ -156,7 +145,7 @@ impl Client {
&self,
group_jid: &Jid,
) -> Result<Jid, anyhow::Error> {
let device_snapshot = self.persistence_manager.get_device_snapshot().await;
let device_snapshot = self.persistence_manager.get_device_snapshot();
let own_pn = device_snapshot
.pn
.clone()
Expand All @@ -178,7 +167,7 @@ impl Client {
}

pub(crate) async fn update_push_name_and_notify(self: &Arc<Self>, new_name: String) {
let device_snapshot = self.persistence_manager.get_device_snapshot().await;
let device_snapshot = self.persistence_manager.get_device_snapshot();
let old_name = device_snapshot.push_name.clone();

if old_name == new_name {
Expand Down
6 changes: 3 additions & 3 deletions src/client/app_state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -579,7 +579,7 @@ impl Client {
if raw_key_ids.is_empty() {
return Ok(());
}
let device_snapshot = self.persistence_manager.get_device_snapshot().await;
let device_snapshot = self.persistence_manager.get_device_snapshot();
let own_jid = match device_snapshot.pn.clone() {
Some(j) => j,
None => {
Expand All @@ -605,7 +605,7 @@ impl Client {
self.send_message_impl(
own_jid,
&msg,
Some(self.generate_message_id().await),
Some(self.generate_message_id()),
true,
false,
None,
Expand Down Expand Up @@ -721,7 +721,7 @@ impl Client {
let new_name = new_name.clone();
let bus = self.core.event_bus.clone();

let snapshot = self.persistence_manager.get_device_snapshot().await;
let snapshot = self.persistence_manager.get_device_snapshot();
let old = snapshot.push_name.clone();
if old != new_name {
debug!(target: "Client/AppState", "Persisting push name from app state mutation: '{}' (old='{}')", new_name, old);
Expand Down
10 changes: 4 additions & 6 deletions src/client/device_registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2059,11 +2059,10 @@ mod tests {
.rotate_sender_key_on_participant_remove(group, &["271060335329480"])
.await;

let device_arc = client.persistence_manager.get_device_arc().await;
let device = device_arc.read().await;
let device_snapshot = client.persistence_manager.get_device_snapshot();
let key = client
.signal_cache
.get_sender_key(&sk_name, &*device.backend)
.get_sender_key(&sk_name, &*device_snapshot.backend)
.await
.unwrap();
assert!(
Expand Down Expand Up @@ -2114,11 +2113,10 @@ mod tests {
.rotate_sender_key_on_participant_remove(group, &["271060335329480"])
.await;

let device_arc = client.persistence_manager.get_device_arc().await;
let device = device_arc.read().await;
let device_snapshot = client.persistence_manager.get_device_snapshot();
let key = client
.signal_cache
.get_sender_key(&sk_name, &*device.backend)
.get_sender_key(&sk_name, &*device_snapshot.backend)
.await
.unwrap();
assert!(
Expand Down
9 changes: 1 addition & 8 deletions src/client/iq_ops.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ impl Client {
let stored_hash = self
.persistence_manager
.get_device_snapshot()
.await
.props_hash
.clone();

Expand Down Expand Up @@ -192,13 +191,7 @@ impl Client {
if override_.is_empty() {
return;
}
if self
.persistence_manager
.get_device_snapshot()
.await
.pn
.is_some()
{
if self.persistence_manager.get_device_snapshot().pn.is_some() {
warn!(
target: "Client/DeviceProps",
"set_device_props called after pairing — stored but not sent on the wire"
Expand Down
6 changes: 3 additions & 3 deletions src/client/lifecycle.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ impl Client {
let mut unique_id_bytes = [0u8; 2];
rand::make_rng::<rand::rngs::StdRng>().fill_bytes(&mut unique_id_bytes);

let device_snapshot = persistence_manager.get_device_snapshot().await;
let device_snapshot = persistence_manager.get_device_snapshot();
let core = wacore::client::CoreClient::new(device_snapshot.core.clone());

let (tx, rx) = async_channel::bounded(32);
Expand Down Expand Up @@ -271,7 +271,7 @@ impl Client {
// Tag the session-root span with our own (pseudonymous) account id so
// connection-lifecycle traces are attributable per account.
#[cfg(feature = "tracing")]
if let Some(lid) = self.get_lid().await {
if let Some(lid) = self.get_lid() {
tracing::Span::current().record("account", tracing::field::display(lid.observe()));
}
while self.is_running.load(Ordering::Relaxed) {
Expand Down Expand Up @@ -459,7 +459,7 @@ impl Client {
self.enable_auto_reconnect.store(false, Ordering::Relaxed);

if self.is_connected()
&& let Ok(jid) = self.require_pn().await
&& let Ok(jid) = self.require_pn()
&& let Err(e) = self.execute(RemoveCompanionDeviceSpec::new(&jid)).await
{
warn!("Failed to send logout IQ: {e}");
Expand Down
7 changes: 3 additions & 4 deletions src/client/messaging.rs
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ impl Client {
.to_string(),
)
} else {
if self.get_pn().await.is_none() {
if self.get_pn().is_none() {
return Err(anyhow::Error::from(ClientError::NotLoggedIn));
}
None
Expand Down Expand Up @@ -143,7 +143,6 @@ impl Client {
self.get_own_jid_for_group(&to).await?.to_non_ad()
} else {
self.get_pn()
.await
.ok_or_else(|| anyhow::Error::from(ClientError::NotLoggedIn))?
.to_non_ad()
};
Expand Down Expand Up @@ -185,7 +184,7 @@ impl Client {
server_id: u64,
reaction: &str,
) -> Result<(), anyhow::Error> {
let request_id = self.generate_message_id().await;
let request_id = self.generate_message_id();

let stanza = NodeBuilder::new("message")
.attr("to", to)
Expand Down Expand Up @@ -236,7 +235,7 @@ impl Client {
if id.is_empty() {
return;
}
let device_snapshot = self.persistence_manager.get_device_snapshot().await;
let device_snapshot = self.persistence_manager.get_device_snapshot();
if let Some(own_jid) = &device_snapshot.pn {
// Single source of truth for the wire mapping (ReceiptType::Sent is a derived
// incoming-only state and is never sent by us).
Expand Down
15 changes: 7 additions & 8 deletions src/client/node_io.rs
Original file line number Diff line number Diff line change
Expand Up @@ -456,7 +456,7 @@ impl Client {
if !self.is_connected() {
return Err(ClientError::NotConnected);
}
let own_pn = self.get_pn().await;
let own_pn = self.get_pn();
let buf = match encode_ack_bytes(node, own_pn.as_ref()) {
Ok(Some(buf)) => buf,
Ok(None) => return Ok(()),
Expand All @@ -473,7 +473,7 @@ impl Client {
/// in a single flushed task.
pub(crate) async fn send_transport_ack(&self, info: &crate::types::message::MessageInfo) {
let source = message_ack_source_node(info);
let own_pn = self.get_pn().await;
let own_pn = self.get_pn();
match encode_ack_bytes(&source.as_node_ref(), own_pn.as_ref()) {
Ok(Some(buf)) => {
if let Err(e) = self.send_raw_bytes(buf).await
Expand Down Expand Up @@ -507,7 +507,7 @@ impl Client {
self: &Arc<Self>,
node: &wacore_binary::NodeRef<'_>,
) {
let own_pn = self.get_pn().await;
let own_pn = self.get_pn();
let buf = match encode_ack_bytes(node, own_pn.as_ref()) {
Ok(Some(b)) => b,
Ok(None) => return,
Expand Down Expand Up @@ -580,7 +580,7 @@ impl Client {
// on Device snapshot + write lock).
if let Some(lid) = lid_from_server {
let device_snapshot =
client_clone.persistence_manager.get_device_snapshot().await;
client_clone.persistence_manager.get_device_snapshot();
if device_snapshot.lid.as_ref() != Some(&lid) {
debug!("Updating LID from server to '{}'", lid.observe());
client_clone
Expand All @@ -598,7 +598,6 @@ impl Client {
let already_paired = client_clone
.persistence_manager
.get_device_snapshot()
.await
.pn
.is_some();
if already_paired {
Expand Down Expand Up @@ -626,7 +625,7 @@ impl Client {

// Check if we need initial app state sync (empty pushname indicates fresh pairing
// where pushname will come from app state sync's setting_pushName mutation)
let device_snapshot = client_clone.persistence_manager.get_device_snapshot().await;
let device_snapshot = client_clone.persistence_manager.get_device_snapshot();
let needs_pushname_from_sync = device_snapshot.push_name.is_empty();
if needs_pushname_from_sync {
debug!("Push name is empty - will be set from app state sync (setting_pushName)");
Expand Down Expand Up @@ -805,7 +804,7 @@ impl Client {
}
// Matches WhatsApp Web's $16(): check if SettingPushName was synced.
// If push_name is still empty after 180s, critical sync failed.
let push_name = timeout_client.get_push_name().await;
let push_name = timeout_client.get_push_name();
if push_name.is_empty() {
warn!(
target: "Client/AppState",
Expand Down Expand Up @@ -888,7 +887,7 @@ impl Client {
} else {
// === Reconnection path ===
// Pushname is already known, send presence and Connected immediately.
let device_snapshot = client_clone.persistence_manager.get_device_snapshot().await;
let device_snapshot = client_clone.persistence_manager.get_device_snapshot();
if !device_snapshot.push_name.is_empty() {
if let Err(e) = client_clone.presence().set_available().await {
warn!("Failed to send initial presence: {e:?}");
Expand Down
4 changes: 2 additions & 2 deletions src/client/sender_keys.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ impl Client {
exclude_own_devices: bool,
) -> Result<()> {
let snapshot = if exclude_own_devices {
Some(self.persistence_manager.get_device_snapshot().await)
Some(self.persistence_manager.get_device_snapshot())
} else {
None
};
Expand Down Expand Up @@ -115,7 +115,7 @@ impl Client {

use wacore::libsignal::store::sender_key_name::SenderKeyName;
use wacore::types::jid::JidExt;
let snapshot = self.persistence_manager.get_device_snapshot().await;
let snapshot = self.persistence_manager.get_device_snapshot();
for own_jid in snapshot.lid.iter().chain(snapshot.pn.iter()) {
let sk_name =
SenderKeyName::from_parts(group_jid, own_jid.to_protocol_address().as_str());
Expand Down
Loading
Loading