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
83 changes: 83 additions & 0 deletions src/features/groups.rs
Original file line number Diff line number Diff line change
Expand Up @@ -415,6 +415,17 @@ impl<'a> Groups<'a> {
pub async fn leave(&self, jid: &Jid) -> Result<(), anyhow::Error> {
self.client.execute(LeaveGroupIq::new(jid)).await?;
self.client.get_group_cache().await.invalidate(jid).await;
// Drop the persisted blob too: we're no longer in the group, so a stale
// phash from it would only force a needless full re-query if ever read.
if let Err(e) = self
.client
.persistence_manager
.backend()
.delete_group_metadata(&jid.to_string())
.await
{
log::warn!("Failed to delete persisted group metadata for {jid}: {e}");
}
Ok(())
}

Expand Down Expand Up @@ -446,7 +457,11 @@ impl<'a> Groups<'a> {
.filter(|r| r.is_ok())
.map(|r| (&r.jid, r.phone_number.as_ref())),
);
self.client.persist_group_metadata(jid, &info).await;
group_cache.insert(jid.clone(), Arc::new(info)).await;
} else {
// Cache expired: can't patch in place, so drop the now-stale blob.
self.client.invalidate_persisted_group_metadata(jid).await;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
}
Ok(result)
Expand All @@ -471,7 +486,11 @@ impl<'a> Groups<'a> {
if let Some(info) = group_cache.get(jid).await {
let mut info = Arc::unwrap_or_clone(info);
info.remove_participants(&accepted);
self.client.persist_group_metadata(jid, &info).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.

P2 Badge Rotate sender keys before persisting removal metadata

When a participant removal succeeds while the group cache is warm, this new awaited persistence write runs before rotate_sender_key_on_participant_remove; with SQLite or a custom backend this can block on disk/network while other outgoing sends are still allowed, so a concurrent send can reuse the old sender key after the server has accepted the removal. Move the rotation ahead of this best-effort metadata persistence (the same ordering exists in the group-notification remove path) so removed members cannot retain decryptable sender-key material during a slow metadata write.

Useful? React with 👍 / 👎.

group_cache.insert(jid.clone(), Arc::new(info)).await;
} else {
// Cache expired: can't patch in place, so drop the now-stale blob.
self.client.invalidate_persisted_group_metadata(jid).await;
}
self.client
.rotate_sender_key_on_participant_remove(&jid.to_string(), &accepted)
Expand Down Expand Up @@ -994,6 +1013,38 @@ impl Client {
pub fn groups(&self) -> Groups<'_> {
Groups::new(self)
}

/// Re-serialize and persist a group's metadata after a local membership change
/// so the phash fast-path stays consistent: the in-memory cache expires after
/// ~1h, after which a stale persisted blob would force a needless full re-query
/// (or be compared against the server as an out-of-date phash). Shared by the
/// participant-mutation API and the inbound group-notification handler.
pub(crate) async fn persist_group_metadata(&self, jid: &Jid, info: &GroupInfo) {
let backend = self.persistence_manager.backend();
match serde_json::to_vec(info) {
Ok(blob) => {
if let Err(e) = backend.put_group_metadata(&jid.to_string(), &blob).await {
log::warn!("Failed to persist group metadata for {jid}: {e}");
}
}
Err(e) => log::warn!("Failed to serialize group metadata for {jid}: {e}"),
}
}

/// Drop the persisted group metadata on a membership change we can't patch in
/// place (the in-memory cache had already expired), so the next query re-fetches
/// fresh instead of comparing a now-stale phash. Without this, persisting only on
/// a cache hit would miss the exact post-expiry case this fix targets.
pub(crate) async fn invalidate_persisted_group_metadata(&self, jid: &Jid) {
if let Err(e) = self
.persistence_manager
.backend()
.delete_group_metadata(&jid.to_string())
.await
{
log::warn!("Failed to invalidate persisted group metadata for {jid}: {e}");
}
}
}

/// Extract the invite code from any supported invite URL format.
Expand Down Expand Up @@ -1171,6 +1222,38 @@ mod tests {
assert_eq!(a.participants.len(), 2);
}

#[tokio::test]
async fn invalidate_persisted_group_metadata_drops_blob() {
// The cache-miss branch of add/remove/leave relies on this to drop a now-stale
// persisted blob so the next query re-fetches fresh instead of sending a stale phash.
let client = crate::test_utils::create_test_client().await;
let backend = client.persistence_manager.backend();
let group_jid: Jid = "123456789@g.us".parse().unwrap();

backend
.put_group_metadata(&group_jid.to_string(), b"stale-blob")
.await
.unwrap();
assert!(
backend
.get_group_metadata(&group_jid.to_string())
.await
.unwrap()
.is_some()
);

client.invalidate_persisted_group_metadata(&group_jid).await;

assert!(
backend
.get_group_metadata(&group_jid.to_string())
.await
.unwrap()
.is_none(),
"invalidation must delete the persisted blob"
);
}

// Protocol-level tests (node building, parsing, validation) are in wacore/src/iq/groups.rs

#[test]
Expand Down
26 changes: 26 additions & 0 deletions src/handlers/notification.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1380,6 +1380,9 @@ async fn handle_group_notification(client: &Arc<Client>, node: Arc<OwnedNodeRef>
.iter()
.map(|p| (&p.jid, p.phone_number.as_ref())),
);
client
.persist_group_metadata(&notification.group_jid, &info)
.await;
Comment on lines +1383 to +1385

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

P2 Badge Persist membership changes when the cache has expired

When a participant notification arrives after the in-memory group cache has expired (the default TTL is one hour) but a persisted metadata blob still exists, this block is skipped entirely, so the new persist_group_metadata path never updates the stored participant list/phash. The same cache-hit gating exists in the add/remove API paths, which leaves the exact stale persisted blob this change is meant to avoid until a later full query_info refreshes it; consider loading/mutating the persisted GroupInfo or invalidating/deleting it on cache miss.

Useful? React with 👍 / 👎.

group_cache
.insert(notification.group_jid.clone(), Arc::new(info))
.await;
Expand All @@ -1388,6 +1391,16 @@ async fn handle_group_notification(client: &Arc<Client>, node: Arc<OwnedNodeRef>
"Patched group cache for {}: added {} participants",
notification.group_jid.observe(), participants.len()
);
} else {
// Cache expired: can't patch in place, so drop the now-stale blob.
debug!(
target: "Client/Group",
"Group cache expired for {}: invalidating persisted metadata (add)",
notification.group_jid.observe()
);
client
.invalidate_persisted_group_metadata(&notification.group_jid)
.await;
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
GroupNotificationAction::Remove { participants, .. } => {
Expand All @@ -1396,6 +1409,9 @@ async fn handle_group_notification(client: &Arc<Client>, node: Arc<OwnedNodeRef>
if let Some(info) = group_cache.get(&notification.group_jid).await {
let mut info = Arc::unwrap_or_clone(info);
info.remove_participants(&users);
client
.persist_group_metadata(&notification.group_jid, &info)
.await;
group_cache
.insert(notification.group_jid.clone(), Arc::new(info))
.await;
Expand All @@ -1404,6 +1420,16 @@ async fn handle_group_notification(client: &Arc<Client>, node: Arc<OwnedNodeRef>
"Patched group cache for {}: removed {} participants",
notification.group_jid.observe(), participants.len()
);
} else {
// Cache expired: can't patch in place, so drop the now-stale blob.
debug!(
target: "Client/Group",
"Group cache expired for {}: invalidating persisted metadata (remove)",
notification.group_jid.observe()
);
client
.invalidate_persisted_group_metadata(&notification.group_jid)
.await;
}
client
.rotate_sender_key_on_participant_remove(
Expand Down
26 changes: 26 additions & 0 deletions storages/sqlite-storage/src/sqlite_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2538,6 +2538,28 @@ impl ProtocolStore for SqliteStore {
Ok(())
}

async fn delete_group_metadata(&self, group_jid: &str) -> Result<()> {
let pool = self.pool.clone();
let device_id = self.device_id;
let group_jid = group_jid.to_string();
tokio::task::spawn_blocking(move || -> Result<()> {
let mut conn = pool
.get()
.map_err(|e| StoreError::Connection(Box::new(e)))?;
diesel::delete(
group_metadata::table
.filter(group_metadata::group_jid.eq(&group_jid))
.filter(group_metadata::device_id.eq(device_id)),
)
.execute(&mut conn)
.map_err(|e| StoreError::Database(Box::new(e)))?;
Ok(())
})
.await
.map_err(|e| StoreError::Database(Box::new(e)))??;
Ok(())
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.

async fn get_tc_token(&self, jid: &str) -> Result<Option<TcTokenEntry>> {
let pool = self.pool.clone();
let device_id = self.device_id;
Expand Down Expand Up @@ -3648,6 +3670,10 @@ mod tests {
store.get_group_metadata(jid).await.unwrap().as_deref(),
Some(&b"blob-v2"[..])
);

// Delete drops the blob so the next query re-fetches in full.
store.delete_group_metadata(jid).await.unwrap();
assert!(store.get_group_metadata(jid).await.unwrap().is_none());
}

#[tokio::test]
Expand Down
8 changes: 8 additions & 0 deletions wacore/src/store/in_memory.rs
Original file line number Diff line number Diff line change
Expand Up @@ -517,6 +517,11 @@ impl ProtocolStore for InMemoryBackend {
Ok(())
}

async fn delete_group_metadata(&self, group_jid: &str) -> Result<()> {
self.state.lock().await.group_metadata.remove(group_jid);
Ok(())
}

// --- TcToken Storage ---

async fn get_tc_token(&self, jid: &str) -> Result<Option<TcTokenEntry>> {
Expand Down Expand Up @@ -730,6 +735,9 @@ mod tests {
backend.get_group_metadata(jid).await.unwrap().as_deref(),
Some(&b"blob-v2"[..])
);
// Delete drops the blob so the next query re-fetches in full.
backend.delete_group_metadata(jid).await.unwrap();
assert!(backend.get_group_metadata(jid).await.unwrap().is_none());
}

#[tokio::test]
Expand Down
7 changes: 7 additions & 0 deletions wacore/src/store/traits.rs
Original file line number Diff line number Diff line change
Expand Up @@ -398,6 +398,13 @@ pub trait ProtocolStore: Send + Sync {
Ok(())
}

/// Remove the persisted group metadata blob for `group_jid` (e.g. on leave),
/// so the next query re-fetches in full instead of comparing a stale phash.
/// No-op by default.
async fn delete_group_metadata(&self, _group_jid: &str) -> Result<()> {
Ok(())
}

// --- TcToken Storage ---

/// Get a trusted contact token for a JID (stored under LID).
Expand Down
Loading