From 1b252629cb507fb71cfdc14f249848df66b3e87f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:40:21 -0300 Subject: [PATCH 1/2] fix(signal): persist retry advances before wire --- src/retry.rs | 136 +++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 126 insertions(+), 10 deletions(-) diff --git a/src/retry.rs b/src/retry.rs index f70f363bb..6cc8268ba 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -12,10 +12,10 @@ use wacore::libsignal::protocol::{PreKeyBundle, PublicKey, UsePQRatchet, process use wacore::protocol::ProtocolNode; use wacore::types::jid::JidExt; use wacore_binary::JidExt as _; -use wacore_binary::builder::NodeBuilder; -use wacore_binary::{Jid, OwnedNodeRef}; #[cfg(test)] -use wacore_binary::{Node, NodeContent}; +use wacore_binary::NodeContent; +use wacore_binary::builder::NodeBuilder; +use wacore_binary::{Jid, Node, OwnedNodeRef}; use wacore_binary::{NodeContentRef, NodeRef}; /// Helper to extract bytes content from a Node (used in tests). @@ -560,12 +560,10 @@ impl Client { ) .await?; - // The Signal mutation is done; the batch-safe flush acquires the - // processing permit, and inbound processing holds that permit - // while taking this same session lock — release it first. + // The pre-wire gate can take the processing permit, whose holder + // may need this session lock. drop(_session_guard); - self.send_node(stanza).await?; - self.flush_signal_cache_batch_safe().await?; + self.send_retry_stanza(stanza).await?; } else { // DM retry: pairwise resend to the requesting device only. // Use _resolved variant: resolved_jid is already in the correct @@ -601,13 +599,18 @@ impl Client { // Same lock-ordering rule as the group branch above. drop(_session_guard); - self.send_node(stanza).await?; - self.flush_signal_cache_batch_safe().await?; + self.send_retry_stanza(stanza).await?; } Ok(()) } + async fn send_retry_stanza(&self, stanza: Node) -> Result<(), anyhow::Error> { + self.persist_signal_state_pre_wire().await?; + self.send_node(stanza).await?; + Ok(()) + } + /// Mirrors WAWebUpdateLocalSignalSession (`WAWeb/Update/LocalSignalSession.js`). /// Runs before ensureE2ESessions + sendRetry for all chat types (DM, group, /// status). Order and semantics match the WA Web implementation: @@ -1315,6 +1318,119 @@ mod tests { super::resolve_retry_chat_info(receipt, node, own_pn, own_lid) } + async fn attach_mock_noise_socket(client: &Arc) { + use crate::socket::NoiseSocket; + use crate::transport::mock::MockTransport; + use wacore::handshake::NoiseCipher; + + let key = [0u8; 32]; + let socket = NoiseSocket::new( + Arc::new(crate::runtime_impl::TokioRuntime), + Arc::new(MockTransport), + NoiseCipher::new(&key).expect("write cipher"), + NoiseCipher::new(&key).expect("read cipher"), + ); + *client.noise_socket.lock().await = Some(Arc::new(socket)); + } + + async fn seed_retry_lease( + client: &Arc, + address: &wacore::libsignal::protocol::ProtocolAddress, + durable: bool, + ) { + use wacore::libsignal::protocol::SessionRecord; + + let mut record = SessionRecord::new_fresh(); + record.reserve_sender_chain_counters(0); + client.signal_cache.put_session(address, record).await; + if !durable { + return; + } + + client.flush_signal_cache().await.expect("durable lease"); + let snapshot = client.persistence_manager.get_device_snapshot(); + let record = client + .signal_cache + .get_session(address, &*snapshot.backend) + .await + .expect("session read") + .expect("leased session"); + assert!(record.reserved_sender_chain_index() > 0); + client.signal_cache.put_session(address, record).await; + } + + #[tokio::test] + async fn retry_pre_wire_flush_failure_never_reaches_send_node() { + use std::sync::atomic::Ordering; + + let client = + crate::test_utils::create_test_client_with_name("retry_pre_wire_failure").await; + attach_mock_noise_socket(&client).await; + let address = Jid::lid_device("100000000001035".to_string(), 7).to_protocol_address(); + seed_retry_lease(&client, &address, false).await; + assert!(client.signal_cache.needs_pre_wire_flush().await); + + client.inbound_commit_batch.reset(); + client + .inbound_commit_batch + .fail_flushes + .store(true, Ordering::Release); + + let id = "RETRY_PRE_WIRE_FAILURE"; + let mut waiter = + client.wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", id)); + let result = client + .send_retry_stanza(NodeBuilder::new("message").attr("id", id).build()) + .await; + + client + .inbound_commit_batch + .fail_flushes + .store(false, Ordering::Release); + assert!(result.is_err(), "the failed durability gate must abort"); + assert!( + waiter.try_recv().expect("waiter stays live").is_none(), + "send_node must not observe a stanza before durability" + ); + assert!( + client.signal_cache.needs_pre_wire_flush().await, + "the failed reservation must remain gated" + ); + } + + #[tokio::test] + async fn retry_inside_durable_lease_skips_synchronous_full_flush() { + use std::sync::atomic::Ordering; + + let client = + crate::test_utils::create_test_client_with_name("retry_covered_by_lease").await; + attach_mock_noise_socket(&client).await; + let address = Jid::lid_device("100000000001036".to_string(), 8).to_protocol_address(); + seed_retry_lease(&client, &address, true).await; + assert!(!client.signal_cache.needs_pre_wire_flush().await); + + client.inbound_commit_batch.reset(); + client + .inbound_commit_batch + .fail_flushes + .store(true, Ordering::Release); + + let id = "RETRY_COVERED_BY_LEASE"; + let waiter = + client.wait_for_sent_node(crate::client::NodeFilter::tag("message").attr("id", id)); + let result = client + .send_retry_stanza(NodeBuilder::new("message").attr("id", id).build()) + .await; + + client + .inbound_commit_batch + .fail_flushes + .store(false, Ordering::Release); + result.expect("an existing durable lease must not synchronously flush"); + let sent = waiter.await.expect("retry stanza reached send_node"); + assert_eq!(sent.attrs().required_string("id").unwrap(), id); + } + #[tokio::test] async fn recent_message_cache_insert_and_take() { let _ = env_logger::builder().is_test(true).try_init(); From f62988997fe8ce1d85ad011283828280a57025ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:50:47 -0300 Subject: [PATCH 2/2] refactor(test): pass client directly to retry helpers --- src/retry.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/retry.rs b/src/retry.rs index 6cc8268ba..3fd93001e 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -1318,7 +1318,7 @@ mod tests { super::resolve_retry_chat_info(receipt, node, own_pn, own_lid) } - async fn attach_mock_noise_socket(client: &Arc) { + async fn attach_mock_noise_socket(client: &Client) { use crate::socket::NoiseSocket; use crate::transport::mock::MockTransport; use wacore::handshake::NoiseCipher; @@ -1334,7 +1334,7 @@ mod tests { } async fn seed_retry_lease( - client: &Arc, + client: &Client, address: &wacore::libsignal::protocol::ProtocolAddress, durable: bool, ) {