fix: isolate ADV discovery and hosted account anchors

agent_docs/adv_account_identity.md

@@ -0,0 +1,46 @@
+# ADV account anchors and hosted devices
+
+The ADV account anchor and the Signal device address are different identities.
+For a `user:99@hosted` device the anchor is `user@s.whatsapp.net`; for
+`user:99@hosted.lid` it is `user@lid`. The device session/identity retains its
+original hosted namespace. `adv_account_jid` is restricted to selecting the
+anchor; do not replace general JID normalization or Signal address rendering.
+
+Static reference: whatspec commit
+`ae3880925bacd23863cc30ec480c9f82bfd869f1`, schema `4.5.0`, WA
+`2.3000.1052559453`, 676 bundles, set hash
+`d6ab7ae601eda281cfac570a68de1d787adde8f4d778f0bb994416f661a21abf`.
+The `bundle-store` asset `bundles-2.3000.1052559453-<set hash>.tar.xz` has
+SHA256 `2958d99a3d84141fa8479f1f991e06246e68c063ca8f618a8d5d361287387f62`.
+The archive and all 676 bundle hashes/sizes were checked against the pinned
+lock. Complete modules were extracted using an AST, without executing them.
+
+Relevant original byte offsets (start inclusive, end exclusive):
+
+| Bundle | Module | Offsets |
+| --- | --- | --- |
+| `0RwvWNH8_qZ.js` | `WAWebAdvSignatureApi` | 767477–775940 |
+| `0RwvWNH8_qZ.js` | `WAWebHandleAdvDeviceNotificationUtils` | 731532–733850 |
+| `331d6f8883e8cd73ff267508b8e5889920b0219c7fd88325ee56ee22eefdcb00.js` | `WAWebWidFactory` | 545429–549200 |
+| same | `WAWebWid` | 535329–545312 |
+| same | `WAWebSignalAddress` | 3107699–3108968 |
+
+`validateADVwithIdentityKey` reads the original device identity separately
+from the anchor selected with `asUserWidOrThrow`. That helper maps Hosted to
+the ordinary PN family and HostedLid to LID. SignalAddress preserves hosted
+device addressing; Wid rejects a hosted primary with device 0. This is static
+Web evidence, not a claim about the server's response to malformed IQs.
+
+Account fences must select the same anchor as verification. Multi-lock
+acquisition groups by account family, then user/device, retaining the original
+server as a tie-breaker. Plain lexical server ordering would acquire a hosted
+LID device before its ordinary LID anchor, opposite to inbound processing.
+The comparator is only for Signal locks; phash and device-list ordering retain
+their existing rules.
+
+The paired own-account fallback remains trusted only for an exact account
+family/user match. Embedded certificate keys cannot replace an existing
+anchor, and both signatures still use the original device's hosted semantics.
+Cold bootstrap retains bounded singleflight, generation cancellation and owned
+durable publication. A hosted USync list follows its existing separately
+authenticated policy and cannot replace our paired primary identity.

src/client/accessors.rs

@@ -645,7 +645,7 @@ impl Client {
             transport_ack_queue: self
                 .inbound_queues
                 .get()
-                .and_then(|q| q.ack_queue.get())
+                .and_then(|q| q.ack_sender())
                 .map_or(0, |tx| tx.len()),
             delivery_receipt_queue: self.delivery_receipt_queue.get().map_or(0, |tx| tx.len()),
             response_waiters,

src/client/adapters.rs

@@ -70,7 +70,7 @@ impl Client {
         if jid.device == 0 {
             return None;
         }
-        let primary = jid.with_device(0).to_protocol_address();
+        let primary = wacore::types::jid::adv_account_jid(jid).to_protocol_address();
         Some(
             self.session_lock_for(primary.as_str())
                 .await

src/client/lid_pn.rs

@@ -1144,8 +1144,8 @@ impl Client {
             // must not publish new trust while this iteration retains old
             // session material. Use the send path's order for the whole set.
             let mut keys = vec![
-                from.with_device(0),
-                to.with_device(0),
+                wacore::types::jid::adv_account_jid(from),
+                wacore::types::jid::adv_account_jid(to),
                 pn_jid.clone(),
                 lid_jid.clone(),
             ];

src/client/node_io.rs

@@ -1168,7 +1168,7 @@ impl Client {
         drop(node);
         let tx = self
             .inbound_queues()
-            .ack_queue
+            .ack_queue()
             .get_or_init(|| self.start_transport_ack_worker());
         let mut lease = crate::inbound_queue::reserve(&self.inbound_queues().ack, bytes.capacity());
         for _ in 0..crate::inbound_queue::SATURATION_RETRY_TURNS {
@@ -2924,6 +2924,54 @@ impl Client {
 
 #[cfg(test)]
 mod tests {
+    /// Compare this diagnostic on the exact base and head with identical flags.
+    /// It constructs futures without polling them or opening a network session.
+    #[tokio::test]
+    #[ignore = "layout diagnostic: run on base and head with --ignored --nocapture"]
+    async fn audit_connected_retention_future_layouts() {
+        use std::mem::{size_of, size_of_val};
+        let client = crate::test_utils::create_test_client().await;
+        let shutdown = client.connection_shutdown_signal();
+        let node = crate::test_utils::node_to_owned_ref(&NodeBuilder::new("iq").build());
+        println!(
+            "QueueBudgets: {} bytes",
+            size_of::<crate::inbound_queue::QueueBudgets>()
+        );
+        println!(
+            "read_messages_loop: {} bytes",
+            size_of_val(&client.read_messages_loop())
+        );
+        println!(
+            "read_messages_loop_inner: {} bytes",
+            size_of_val(&client.read_messages_loop_inner())
+        );
+        println!(
+            "process_node_scoped: {} bytes",
+            size_of_val(&client.process_node_scoped(node.clone(), &shutdown, false))
+        );
+        println!(
+            "dispatch_node_scoped: {} bytes",
+            size_of_val(&client.dispatch_node_scoped(node, &shutdown, Some(StanzaTag::Iq), 0))
+        );
+        let reader_client = client.clone();
+        let reader = async move {
+            let connection = reader_client.connect().await?;
+            connection.read_until_disconnected().await;
+            Ok::<(), ConnectError>(())
+        };
+        println!(
+            "connected fixture reader task: {} bytes",
+            size_of_val(&reader)
+        );
+        drop(reader);
+        let primary = Jid::pn("12025550189");
+        println!(
+            "ensure_adv_account_identity: {} bytes",
+            size_of_val(&client.ensure_adv_account_identity(&primary))
+        );
+        client.signal_shutdown_sync();
+    }
+
     #[cfg(feature = "mobile-core")]
     use crate::test_utils::primary_startup_client;
 
@@ -3428,7 +3476,7 @@ mod tests {
         use wacore_binary::builder::NodeBuilder;
         let (client, _transport) = create_iq_test_client().await;
         let (tx, rx) = async_channel::bounded(1);
-        assert!(client.inbound_queues().ack_queue.set(tx).is_ok());
+        assert!(client.inbound_queues().ack_queue().set(tx).is_ok());
         let receipt = || {
             node_to_owned_ref(
                 &NodeBuilder::new("receipt")
@@ -3519,7 +3567,7 @@ mod tests {
         use wacore_binary::builder::NodeBuilder;
         let (client, _transport) = create_iq_test_client().await;
         let (tx, rx) = async_channel::bounded(1);
-        assert!(client.inbound_queues().ack_queue.set(tx).is_ok());
+        assert!(client.inbound_queues().ack_queue().set(tx).is_ok());
         let old = client.connection_shutdown_signal();
         client.reset_connection_shutdown();
         let receipt = node_to_owned_ref(
@@ -3576,7 +3624,7 @@ mod tests {
         use wacore_binary::builder::NodeBuilder;
         let (client, _transport) = create_iq_test_client().await;
         let (tx, rx) = async_channel::bounded(1);
-        assert!(client.inbound_queues().ack_queue.set(tx).is_ok());
+        assert!(client.inbound_queues().ack_queue().set(tx).is_ok());
         let receipt = || {
             node_to_owned_ref(
                 &NodeBuilder::new("receipt")

src/inbound_queue.rs

@@ -122,8 +122,16 @@ pub(crate) async fn yield_for_capacity(runtime: &dyn wacore::runtime::Runtime) {
     }
 }
 
+/// ACK delivery and cold ADV discovery are optional on a connected client.
+/// Share their lazy attachment so unused services do not enlarge its controller.
+#[derive(Default)]
+struct QueueTasks {
+    ack_queue: std::sync::OnceLock<async_channel::Sender<Box<QueuedAck>>>,
+    adv_bootstrap: std::sync::OnceLock<Arc<crate::prekeys::AdvBootstrapRegistry>>,
+}
+
 pub(crate) struct QueueBudgets {
-    pub ack_queue: std::sync::OnceLock<async_channel::Sender<Box<QueuedAck>>>,
+    tasks: std::sync::OnceLock<Box<QueueTasks>>,
     pub chat: Arc<ByteBudget>,
     pub ack: Arc<ByteBudget>,
     pub work: Arc<crate::flush_scope::FlushScope>,
@@ -133,7 +141,7 @@ pub(crate) struct QueueBudgets {
 impl Default for QueueBudgets {
     fn default() -> Self {
         Self {
-            ack_queue: std::sync::OnceLock::new(),
+            tasks: std::sync::OnceLock::new(),
             chat: Arc::new(ByteBudget::new(GLOBAL_CHAT_BYTES)),
             ack: Arc::new(ByteBudget::new(ACK_BYTES)),
             work: Arc::default(),
@@ -143,6 +151,25 @@ impl Default for QueueBudgets {
     }
 }
 impl QueueBudgets {
+    /// Attach the ACK queue only when a stanza actually needs deferred delivery.
+    pub(crate) fn ack_queue(&self) -> &std::sync::OnceLock<async_channel::Sender<Box<QueuedAck>>> {
+        &self.tasks.get_or_init(Box::default).ack_queue
+    }
+
+    /// Inspect an existing ACK worker without allocating optional task state.
+    pub(crate) fn ack_sender(&self) -> Option<&async_channel::Sender<Box<QueuedAck>>> {
+        self.tasks.get().and_then(|tasks| tasks.ack_queue.get())
+    }
+
+    /// Attach account-flight coordination only when ADV discovery needs it.
+    pub(crate) fn adv_bootstrap(&self) -> &Arc<crate::prekeys::AdvBootstrapRegistry> {
+        self.tasks
+            .get_or_init(Box::default)
+            .adv_bootstrap
+            .get_or_init(Arc::default)
+    }
+
+    /// Count queue controllers, live chat budgets and retained ADV flights, including empty map capacity.
     pub(crate) fn memory_stats(&self) -> (wacore::stats::CollectionStats, u64, u64) {
         use wacore::stats::HeapSize;
         let entries = self.per_chat.lock().unwrap_or_else(|p| p.into_inner());
@@ -162,7 +189,14 @@ impl QueueBudgets {
             // Controller, global counters and work scope are counted once.
             + size_of::<Self>()
             + 2 * (size_of::<ByteBudget>() + 2 * size_of::<usize>())
-            + size_of::<crate::flush_scope::FlushScope>() + 2 * size_of::<usize>();
+            + size_of::<crate::flush_scope::FlushScope>() + 2 * size_of::<usize>()
+            + self.tasks.get().map_or(0, |tasks| {
+                size_of::<QueueTasks>()
+                    + tasks
+                        .adv_bootstrap
+                        .get()
+                        .map_or(0, |registry| registry.retained_bytes())
+            });
         (
             wacore::stats::CollectionStats::new(entries.len() as u64, registry_bytes as u64),
             GLOBAL_CHAT_BYTES.saturating_sub(self.chat.available()) as u64,
@@ -350,6 +384,38 @@ impl crate::client::Client {
 
 #[cfg(test)]
 mod tests {
+    /// Optional workers must not add inline words or allocate during stats reads.
+    #[test]
+    fn cold_queue_controller_keeps_its_pre_adv_footprint() {
+        // The pre-ADV controller had one ACK slot, four Arc pointers and a
+        // draining flag padded to pointer alignment. Keep that representation.
+        let legacy = size_of::<std::sync::OnceLock<async_channel::Sender<Box<QueuedAck>>>>()
+            + 5 * size_of::<usize>();
+        assert_eq!(size_of::<QueueBudgets>(), legacy);
+        let budgets = QueueBudgets::default();
+        let before = budgets.memory_stats().0.bytes;
+        assert!(budgets.ack_sender().is_none());
+        assert_eq!(budgets.memory_stats().0.bytes, before);
+        assert!(budgets.tasks.get().is_none());
+    }
+
+    /// Count the shared sidecar once, then count ADV ownership when it is used.
+    #[test]
+    fn optional_queue_task_allocations_remain_visible() {
+        let budgets = QueueBudgets::default();
+        let cold = budgets.memory_stats().0.bytes;
+        let (tx, _rx) = async_channel::bounded(1);
+        assert!(budgets.ack_queue().set(tx).is_ok());
+        let ack = budgets.memory_stats().0.bytes;
+        assert_eq!(ack - cold, size_of::<QueueTasks>() as u64);
+        let registry = budgets.adv_bootstrap();
+        assert_eq!(
+            budgets.memory_stats().0.bytes - ack,
+            registry.retained_bytes() as u64
+        );
+        assert!(budgets.ack_sender().is_some());
+    }
+
     use super::*;
     use crate::client::{ChatLane, QueuedChatMessage};
     use async_lock::Mutex;

src/message/durability.rs

@@ -648,7 +648,7 @@ mod tests {
             client
                 .inbound_queues
                 .get()
-                .and_then(|q| q.ack_queue.get())
+                .and_then(|q| q.ack_sender())
                 .is_none(),
             "a successful subset hook must not acknowledge the retained group"
         );
@@ -666,7 +666,7 @@ mod tests {
                 || client
                     .inbound_queues
                     .get()
-                    .and_then(|q| q.ack_queue.get())
+                    .and_then(|q| q.ack_sender())
                     .is_some(),
             "only the complete replay may schedule the stanza acknowledgement"
         );