perf: remove four per-message tasks and buffers from the send/receive round trip

.gitignore

@@ -16,3 +16,9 @@ dhat-heap*.json
 # proptest persists failure seeds next to the test on failure; these are
 # environment-local replay caches, not source.
 *.proptest-regressions
+
+# Diagnostic store snapshots tests write into the repo root
+# (PersistenceManager::create_snapshot). Generated artifacts, never sources.
+# Two spellings: the plain name and the "file:" form the sqlite URI produces.
+memdb_*.snapshot-*
+file:memdb_*.snapshot-*

src/client.rs

@@ -288,6 +288,12 @@ pub struct MemoryReport {
     pub group_distribution_lock_eviction_blocks: u64,
     pub resend_rate_limiter_chats: u64,
     // -- Unbounded collections --
+    /// Deferred acks queued for the transport-ack worker. Unbounded, and each
+    /// entry retains the full inbound node plus a flush guard, so a stalled
+    /// transport shows up here as a growing backlog.
+    pub transport_ack_queue: usize,
+    /// Delivery receipts queued for their worker, same shape as above.
+    pub delivery_receipt_queue: usize,
     pub response_waiters: usize,
     pub node_waiters: usize,
     pub pending_retries: usize,
@@ -393,6 +399,12 @@ impl std::fmt::Display for MemoryReport {
             self.resend_rate_limiter_chats
         )?;
         writeln!(f, "--- Unbounded collections ---")?;
+        writeln!(f, "  transport_ack_queue:    {}", self.transport_ack_queue)?;
+        writeln!(
+            f,
+            "  delivery_receipt_queue: {}",
+            self.delivery_receipt_queue
+        )?;
         writeln!(f, "  response_waiters:       {}", self.response_waiters)?;
         writeln!(f, "  node_waiters:           {}", self.node_waiters)?;
         writeln!(f, "  pending_retries:        {}", self.pending_retries)?;
@@ -709,9 +721,36 @@ pub(crate) struct OfflineSyncMetrics {
 
 type ResponseWaiterSender = futures::channel::oneshot::Sender<Arc<wacore_binary::OwnedNodeRef>>;
 
+/// What a pending ack/IQ entry is waiting to do once the response arrives.
+///
+/// A phash check used to be an `Iq` waiter plus a spawned task holding the
+/// receiver and a ten second timer, which is a task, a channel and a timer per
+/// outgoing message for a comparison that almost always succeeds. Carrying the
+/// expected value in the map instead lets the read loop compare it inline and
+/// spawn only on the rare mismatch.
+pub(crate) enum ResponseWaiter {
+    /// Classic request/response: hand the node to whoever is awaiting it.
+    Iq(ResponseWaiterSender),
+    /// Compare the server's `phash` against ours; act only if they differ.
+    Phash(PhashWaiter),
+}
+
+pub(crate) struct PhashWaiter {
+    pub(crate) expected: wacore_binary::CompactString,
+    pub(crate) jid: Jid,
+    pub(crate) invalidate_group_cache: bool,
+    /// Sweep epoch this waiter was registered in. Expiry is counted in sweeps
+    /// rather than seconds: a wall deadline is subject to clock jumps (see
+    /// wacore::time) and would have to be derived from an instant sampled well
+    /// before registration, while reading a fresh clock here is what the send
+    /// clock budget forbids. Surviving one full sweep is the trigger, so the
+    /// window is one keepalive tick (15 to 30 s) instead of the old fixed 10 s.
+    pub(crate) registered_epoch: u64,
+}
+
 struct ResponseWaiterEntry {
     generation: NonZeroU64,
-    sender: ResponseWaiterSender,
+    waiter: ResponseWaiter,
 }
 
 /// Map of pending IQ/ack response waiters, keyed by request id.
@@ -722,6 +761,9 @@ struct ResponseWaiterEntry {
 pub(crate) struct ResponseWaiterMap {
     entries: HashMap<String, ResponseWaiterEntry>,
     last_generation: u64,
+    /// Advanced once per sweep. Registration reads it under the lock it already
+    /// takes, so a waiter records its age without touching a clock.
+    sweep_epoch: u64,
 }
 
 impl ResponseWaiterMap {
@@ -737,14 +779,14 @@ impl ResponseWaiterMap {
     pub(crate) fn try_insert_guarded(
         &mut self,
         request_id: String,
-        sender: ResponseWaiterSender,
+        waiter: ResponseWaiter,
     ) -> Option<NonZeroU64> {
         use std::collections::hash_map::Entry;
 
         let generation = self.next_generation();
         match self.entries.entry(request_id) {
             Entry::Vacant(entry) => {
-                entry.insert(ResponseWaiterEntry { generation, sender });
+                entry.insert(ResponseWaiterEntry { generation, waiter });
                 Some(generation)
             }
             Entry::Occupied(_) => None,
@@ -754,16 +796,37 @@ impl ResponseWaiterMap {
     pub(crate) fn insert(
         &mut self,
         request_id: String,
-        sender: ResponseWaiterSender,
-    ) -> Option<ResponseWaiterSender> {
+        waiter: ResponseWaiter,
+    ) -> Option<ResponseWaiter> {
         let generation = self.next_generation();
         self.entries
-            .insert(request_id, ResponseWaiterEntry { generation, sender })
-            .map(|entry| entry.sender)
+            .insert(request_id, ResponseWaiterEntry { generation, waiter })
+            .map(|entry| entry.waiter)
     }
 
-    pub(crate) fn remove(&mut self, request_id: &str) -> Option<ResponseWaiterSender> {
-        self.entries.remove(request_id).map(|entry| entry.sender)
+    pub(crate) fn remove(&mut self, request_id: &str) -> Option<ResponseWaiter> {
+        self.entries.remove(request_id).map(|entry| entry.waiter)
+    }
+
+    /// The epoch a waiter registered now belongs to.
+    pub(crate) fn current_epoch(&self) -> u64 {
+        self.sweep_epoch
+    }
+
+    /// Drop phash waiters that lived through a whole sweep without their ack.
+    ///
+    /// Runs on the keepalive tick, before the recent-activity early return: a
+    /// connection with steady inbound traffic skips the ping entirely, and
+    /// sweeping only inside the ping would let lost acks accumulate for as long
+    /// as traffic keeps flowing. The map is also what makes keepalive treat the
+    /// connection as "IQs pending", so a stranded waiter silences pings.
+    pub(crate) fn drop_expired_phash(&mut self) {
+        let epoch = self.sweep_epoch;
+        self.entries.retain(|_, entry| match &entry.waiter {
+            ResponseWaiter::Phash(waiter) => waiter.registered_epoch >= epoch,
+            ResponseWaiter::Iq(_) => true,
+        });
+        self.sweep_epoch = self.sweep_epoch.wrapping_add(1);
     }
 
     pub(crate) fn remove_guarded(&mut self, request_id: &str, cleanup_generation: NonZeroU64) {
@@ -1058,6 +1121,16 @@ pub struct Client {
             crate::flush_scope::FlushGuard,
         )>,
     >,
+    /// Feed of the persistent transport-ack worker, mirroring
+    /// [`Self::delivery_receipt_queue`]. Deferred acks used to be one spawned
+    /// task each; the queue also gives them FIFO order, which the spawns did
+    /// not guarantee.
+    pub(crate) transport_ack_queue: std::sync::OnceLock<
+        async_channel::Sender<(
+            Arc<wacore_binary::OwnedNodeRef>,
+            crate::flush_scope::FlushGuard,
+        )>,
+    >,
     /// Contacts with active presence subscriptions that must be re-subscribed on reconnect.
     pub(crate) presence_subscriptions: Arc<Mutex<HashSet<Jid>>>,
     /// Metrics for granular offline sync logging

src/client/accessors.rs

@@ -257,6 +257,8 @@ impl Client {
             group_distribution_lock_evictions: group_distribution_locks.evictions,
             group_distribution_lock_eviction_blocks: group_distribution_locks.eviction_blocks,
             resend_rate_limiter_chats: self.resend_rate_limiter.entry_count(),
+            transport_ack_queue: self.transport_ack_queue.get().map_or(0, |tx| tx.len()),
+            delivery_receipt_queue: self.delivery_receipt_queue.get().map_or(0, |tx| tx.len()),
             response_waiters,
             node_waiters: self.node_waiter_count.load(Ordering::Relaxed),
             pending_retries: pending_retries_count,

src/client/lifecycle.rs

@@ -345,6 +345,7 @@ impl Client {
             history_sync_activity: Arc::new(crate::sync_task::HistorySyncActivity::new()),
             outbound_flush: Arc::new(crate::flush_scope::FlushScope::new()),
             delivery_receipt_queue: std::sync::OnceLock::new(),
+            transport_ack_queue: std::sync::OnceLock::new(),
             presence_subscriptions: Arc::new(Mutex::new(HashSet::new())),
             socket_ready_notifier: Arc::new(event_listener::Event::new()),
             is_ready: Arc::new(AtomicBool::new(false)),

src/client/messaging.rs

@@ -269,16 +269,51 @@ impl Client {
     /// Register a oneshot waiter for a server ack by message ID.
     /// Returns the receiver — caller sends the node separately and awaits this in background.
     /// Sync: registration is just a `std::sync::Mutex` insert (no await).
+    /// Register a waiter that receives the ack node itself.
+    ///
+    /// Used where the caller needs the response: the VoIP offer reads the relay
+    /// out of its ack. A phash check does not, which is why that path uses
+    /// [`Self::register_phash_waiter`] and pays no channel per message. Gated on
+    /// the only consumer's feature, or it is dead code in a default build.
+    #[cfg(feature = "voip-runtime")]
     pub(crate) fn register_ack_waiter(
         &self,
         message_id: &str,
     ) -> futures::channel::oneshot::Receiver<Arc<wacore_binary::OwnedNodeRef>> {
         let (tx, rx) = futures::channel::oneshot::channel();
         self.response_waiters_guard()
-            .insert(message_id.to_string(), tx);
+            .insert(message_id.to_string(), ResponseWaiter::Iq(tx));
         rx
     }
 
+    /// Register the phash the server is expected to echo for this send.
+    ///
+    /// Nothing awaits the result: the read loop compares inline when the ack
+    /// lands and only acts on a mismatch, so a send costs a map entry instead of
+    /// a task, a oneshot and a timer.
+    pub(crate) fn register_phash_waiter(
+        &self,
+        message_id: &str,
+        expected: wacore_binary::CompactString,
+        jid: Jid,
+        invalidate_group_cache: bool,
+    ) {
+        let mut waiters = self.response_waiters_guard();
+        // Stamped with the sweep epoch under the lock the insert already holds:
+        // a deadline derived from the instant the send started would already be
+        // stale here when preparation is slow, and a wall clock can jump.
+        let registered_epoch = waiters.current_epoch();
+        waiters.insert(
+            message_id.to_string(),
+            ResponseWaiter::Phash(PhashWaiter {
+                expected,
+                jid,
+                invalidate_group_cache,
+                registered_epoch,
+            }),
+        );
+    }
+
     /// Creates a normalized ChatMessageId by resolving PN to LID JIDs.
     pub(crate) async fn make_chat_message_id(&self, chat: &Jid, id: &str) -> ChatMessageId {
         // Resolve chat JID to LID if possible