perf(handlers): shrink the inbound non-message stanza path; inbound_stanza bench

Cargo.toml

@@ -345,6 +345,15 @@ name = "client_dm_send"
 harness = false
 required-features = ["bench-harness"]
 
+# The inbound counterpart: the non-message stanzas (`<receipt>`, `<presence>`,
+# `<notification>`, `<ack>`) the read loop handles without a decrypt. Its
+# regression signal is bytes per stanza, not wall time; the target's own header
+# says why.
+[[bench]]
+name = "inbound_stanza"
+harness = false
+required-features = ["bench-harness"]
+
 # The `Cache<Jid, _>` lookup shape `chat_lanes` runs per inbound message. No
 # `required-features`: it reaches only the public cache, so it builds (and is
 # regression-gated) in a plain `cargo bench`.

benches/inbound_stanza.rs

@@ -0,0 +1,186 @@
+//! Client-level inbound non-message stanzas: `<receipt>`, `<presence>`,
+//! `<notification>` and `<ack>`, from the decoded stanza to the dispatched
+//! event.
+//!
+//! `client_receive` covers the `<message>` path, where a Signal decrypt
+//! dominates everything around it. These four are the rest of the read loop —
+//! no crypto, so what they measure *is* the machinery: classification, the
+//! handler's boxed future, the parse, and the event bus. On an offline drain
+//! they are also the bulk of the queue.
+//!
+//! **The metric to read here is bytes and allocations per stanza, not wall
+//! time.** Every row's body is a few hundred nanoseconds of branch-heavy work
+//! against a fixture that shares a runtime with an ack worker, so timings move
+//! more between runs than any change worth making moves them; the allocation
+//! columns (`divan::AllocProfiler` is wired in below) are exact. The regression
+//! this target exists to catch is a per-stanza heap block growing back — an
+//! `.await` on an unboxed handler arm re-inflates the notification future the
+//! way it was before PR "perf(handlers)".
+//!
+//! Every stanza runs twice: once with nothing subscribed to its event kind and
+//! once with a subscriber. That is not redundancy — the subscriber is what
+//! decides how much of the pipeline runs. `has_handler_for` gates the receipt
+//! parse, steers `processes_inline`, and is what makes an `Arc<Event>` exist at
+//! all.
+//!
+//! What it does not cover, stated once so no number here is over-read:
+//!
+//! - **The read loop's task spawn.** Stanzas enter at `process_node`, which is
+//!   what the spawned task calls; the spawn itself is not measured.
+//! - **The socket write.** The transport `<ack>` a `<receipt>` or
+//!   `<notification>` owes is marshalled and noise-encrypted by a worker; the
+//!   sink transport drops the frame.
+//! - **Stanza construction.** Each row builds its stanza once, before
+//!   sampling, and re-submits it. Nothing on these paths dedupes, so a repeat
+//!   is the same work as a fresh one.
+
+use divan::black_box;
+use std::sync::Arc;
+use std::sync::OnceLock;
+use whatsapp_rust::bench_support::ReceiveHarness;
+
+fn main() {
+    divan::main();
+}
+
+/// Byte and allocation counts per row. The point of the target: see the module
+/// header.
+#[global_allocator]
+static ALLOC: divan::AllocProfiler = divan::AllocProfiler::system();
+
+/// Few, large samples, matching `client_receive`: the fixture's flush worker
+/// fires on its own 25 ms clock and a coalesced flush that lands mid-sample is
+/// amortised over a long sample rather than moving a short one.
+const SAMPLE_COUNT: u32 = 20;
+const SAMPLE_SIZE: u32 = 50;
+
+/// A reconnect drains its offline queue in one go. 500 is the size at which the
+/// per-item cost of a drain is clearly separated from the fixed cost of
+/// entering it.
+const BURST: usize = 500;
+
+/// One harness for every row, so no row pays for a second fixture. Unlike the
+/// message rows these stanzas carry no ratchet state, so sharing cannot cross
+/// their streams.
+fn harness() -> &'static ReceiveHarness {
+    static HARNESS: OnceLock<ReceiveHarness> = OnceLock::new();
+    HARNESS.get_or_init(ReceiveHarness::new)
+}
+
+/// Submit one stanza per iteration, asserting afterwards that the subscriber
+/// saw exactly the events it should have: `expected_events` per iteration, so a
+/// stanza silently dropped by a parse failure (which is fast) cannot pass for
+/// a processed one.
+fn bench_one(
+    bencher: divan::Bencher,
+    node: Arc<wacore_binary::OwnedNodeRef>,
+    subscribed: bool,
+    expected_events: u64,
+) {
+    let harness = harness();
+    let subscription = subscribed.then(|| harness.subscribe_stanza_events());
+    let before = harness.stanza_events();
+    let mut submitted = 0u64;
+    bencher.bench_local(|| {
+        submitted += 1;
+        harness.process_nowait(black_box(Arc::clone(&node)));
+    });
+    let delivered = harness.stanza_events() - before;
+    // Dropping it before the flush keeps the assertion below about what this
+    // row submitted.
+    drop(subscription);
+    harness.flush();
+    assert_eq!(delivered, submitted * expected_events);
+}
+
+/// Same, for the drain shape: `BURST` stanzas under one `block_on`.
+fn bench_burst(
+    bencher: divan::Bencher,
+    node: Arc<wacore_binary::OwnedNodeRef>,
+    expected_events: u64,
+) {
+    let harness = harness();
+    let burst: Vec<_> = std::iter::repeat_n(node, BURST).collect();
+    let subscription = harness.subscribe_stanza_events();
+    let before = harness.stanza_events();
+    let mut submitted = 0u64;
+    bencher.bench_local(|| {
+        submitted += BURST as u64;
+        harness.process_burst(black_box(&burst));
+    });
+    let delivered = harness.stanza_events() - before;
+    drop(subscription);
+    harness.flush();
+    assert_eq!(delivered, submitted * expected_events);
+}
+
+/// A delivery `<receipt>` nobody is listening for: the read-loop shape, where
+/// the whole point is that the parse stops at the subscriber gate.
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn receipt(bencher: divan::Bencher) {
+    bench_one(bencher, harness().receipt_stanza(), false, 0);
+}
+
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn receipt_subscribed(bencher: divan::Bencher) {
+    bench_one(bencher, harness().receipt_stanza(), true, 1);
+}
+
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn presence(bencher: divan::Bencher) {
+    bench_one(bencher, harness().presence_stanza(), false, 0);
+}
+
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn presence_subscribed(bencher: divan::Bencher) {
+    bench_one(bencher, harness().presence_stanza(), true, 1);
+}
+
+/// The row the boxed-arm regression shows up in: the notification handler's
+/// future is sized for whichever arm the compiler keeps inline, whatever type
+/// this stanza actually carries.
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn notification(bencher: divan::Bencher) {
+    bench_one(bencher, harness().notification_stanza(), false, 0);
+}
+
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn notification_subscribed(bencher: divan::Bencher) {
+    bench_one(bencher, harness().notification_stanza(), true, 1);
+}
+
+/// A server `<ack>` no waiter is parked on — the shape every fire-and-forget
+/// send draws back, and the cheapest stanza the client handles.
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn ack(bencher: divan::Bencher) {
+    bench_one(bencher, harness().ack_stanza(), false, 0);
+}
+
+#[divan::bench(sample_count = SAMPLE_COUNT, sample_size = SAMPLE_SIZE)]
+fn ack_subscribed(bencher: divan::Bencher) {
+    bench_one(bencher, harness().ack_stanza(), true, 1);
+}
+
+// The drain rows all run subscribed: a reconnect that drains a queue into a
+// consumer is the case where the per-item cost is actually paid, and the
+// no-subscriber shape is already covered per stanza above. Divide every column
+// by `BURST` for the per-item figure.
+#[divan::bench(sample_count = 10, sample_size = 1)]
+fn burst_receipt(bencher: divan::Bencher) {
+    bench_burst(bencher, harness().receipt_stanza(), 1);
+}
+
+#[divan::bench(sample_count = 10, sample_size = 1)]
+fn burst_presence(bencher: divan::Bencher) {
+    bench_burst(bencher, harness().presence_stanza(), 1);
+}
+
+#[divan::bench(sample_count = 10, sample_size = 1)]
+fn burst_notification(bencher: divan::Bencher) {
+    bench_burst(bencher, harness().notification_stanza(), 1);
+}
+
+#[divan::bench(sample_count = 10, sample_size = 1)]
+fn burst_ack(bencher: divan::Bencher) {
+    bench_burst(bencher, harness().ack_stanza(), 1);
+}

src/bench_support.rs

@@ -752,6 +752,42 @@ impl wacore::types::events::EventHandler for MessageCounter {
     }
 }
 
+/// Counts the non-message stanza events the inbound benchmarks produce, so a
+/// "with a subscriber" row can prove the event actually reached a handler
+/// rather than being dropped by the interest bitmask.
+///
+/// Kept unsubscribed by the fixture: the same benchmarks also measure the
+/// no-subscriber shape, which is what a bot that only wants `Messages` runs,
+/// and that shape only exists while nothing is interested in these kinds.
+struct StanzaEventCounter {
+    delivered: portable_atomic::AtomicU64,
+}
+
+/// The kinds [`ReceiveHarness`]'s non-message stanzas turn into: a `<receipt>`,
+/// a `<presence>`, a `<notification type="picture">` and an `<ack>`.
+const STANZA_EVENT_KINDS: &[wacore::types::events::EventKind] = &[
+    wacore::types::events::EventKind::Receipt,
+    wacore::types::events::EventKind::Presence,
+    wacore::types::events::EventKind::PictureUpdate,
+    wacore::types::events::EventKind::ServerAck,
+];
+
+impl wacore::types::events::EventHandler for StanzaEventCounter {
+    fn handle_event(&self, event: Arc<wacore::types::events::Event>) {
+        use wacore::types::events::Event;
+        if matches!(
+            &*event,
+            Event::Receipt(_) | Event::Presence(_) | Event::PictureUpdate(_) | Event::ServerAck(_)
+        ) {
+            self.delivered.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    fn interest(&self) -> wacore::types::events::EventInterest {
+        wacore::types::events::EventInterest::of(STANZA_EVENT_KINDS)
+    }
+}
+
 /// A logged-in client receiving from one peer it already shares a session and
 /// a sender key with: the steady state every message after the first arrives
 /// in.
@@ -775,6 +811,7 @@ pub struct ReceiveHarness {
     group: Jid,
     group_sender_key: wacore::libsignal::protocol::SenderKeyName,
     counter: Arc<MessageCounter>,
+    stanza_counter: Arc<StanzaEventCounter>,
     /// Dropping it would unsubscribe the counter.
     _subscription: wacore::types::events::Subscription,
     next_id: portable_atomic::AtomicU64,
@@ -799,6 +836,7 @@ impl ReceiveHarness {
             group: fixture.group,
             group_sender_key: fixture.group_sender_key,
             counter: fixture.counter,
+            stanza_counter: fixture.stanza_counter,
             _subscription: fixture.subscription,
             next_id: portable_atomic::AtomicU64::new(0),
         }
@@ -902,6 +940,119 @@ impl ReceiveHarness {
         self.counter.delivered.load(Ordering::Relaxed)
     }
 
+    /// Subscribe the non-message stanza counter for as long as the returned
+    /// [`Subscription`](wacore::types::events::Subscription) is held.
+    ///
+    /// The inbound benchmarks run each stanza both ways, because the
+    /// subscriber is what decides how much of the pipeline runs at all:
+    /// `has_handler_for` gates the receipt parse and steers `processes_inline`,
+    /// and without one an event is never materialised. Neither shape is the
+    /// "real" one — a bot that only consumes `Messages` runs the first, a
+    /// client that mirrors presence and receipts runs the second.
+    pub fn subscribe_stanza_events(&self) -> wacore::types::events::Subscription {
+        self.client.subscribe_handler(
+            Arc::clone(&self.stanza_counter) as Arc<dyn wacore::types::events::EventHandler>
+        )
+    }
+
+    /// How many of the four non-message stanza events reached the subscriber.
+    pub fn stanza_events(&self) -> u64 {
+        self.stanza_counter.delivered.load(Ordering::Relaxed)
+    }
+
+    /// A `<receipt>` from the peer: the delivery acknowledgement one of our
+    /// outgoing DMs draws back, in the simple (non-aggregated) shape.
+    pub fn receipt_stanza(&self) -> Arc<wacore_binary::OwnedNodeRef> {
+        let node = wacore_binary::builder::NodeBuilder::new("receipt")
+            .attr("from", self.peer_jid.to_string())
+            .attr("id", self.next_message_id())
+            .attr("t", wacore::time::now_secs().to_string())
+            .build();
+        decoded(&node)
+    }
+
+    /// A `<presence>` from the peer. The one inbound stanza the server does
+    /// not expect an `<ack>` for, which is what makes it the control against
+    /// the acked kinds.
+    pub fn presence_stanza(&self) -> Arc<wacore_binary::OwnedNodeRef> {
+        let node = wacore_binary::builder::NodeBuilder::new("presence")
+            .attr("from", self.peer_jid.to_string())
+            .attr("type", "available")
+            .build();
+        decoded(&node)
+    }
+
+    /// A `<notification type="picture">`, the cheapest notification the client
+    /// models: one synchronous arm and a single-`String` payload. Anything the
+    /// row reports beyond that belongs to the dispatch machinery around the
+    /// arm, not to the arm.
+    pub fn notification_stanza(&self) -> Arc<wacore_binary::OwnedNodeRef> {
+        let jid = self.peer_jid.to_string();
+        let set = wacore_binary::builder::NodeBuilder::new("set")
+            .attr("jid", jid.clone())
+            .attr("id", "1700000000")
+            .build();
+        let node = wacore_binary::builder::NodeBuilder::new("notification")
+            .attr("from", jid)
+            .attr("type", "picture")
+            .attr("id", self.next_message_id())
+            .attr("t", wacore::time::now_secs().to_string())
+            .children([set])
+            .build();
+        decoded(&node)
+    }
+
+    /// A server `<ack>` for an outgoing stanza no waiter is parked on — the
+    /// shape every fire-and-forget send (receipts, acks, presence) draws back.
+    pub fn ack_stanza(&self) -> Arc<wacore_binary::OwnedNodeRef> {
+        let node = wacore_binary::builder::NodeBuilder::new("ack")
+            .attr("from", self.peer_jid.to_string())
+            .attr("class", "message")
+            .attr("id", self.next_message_id())
+            .attr("t", wacore::time::now_secs().to_string())
+            .build();
+        decoded(&node)
+    }
+
+    /// Run one stanza through the real `Client::process_node` — classification,
+    /// the handler, the event — and stop there.
+    ///
+    /// Unlike [`Self::receive`] this does *not* flush the outbound scope. The
+    /// transport `<ack>` a `<receipt>` or `<notification>` owes is handed to a
+    /// persistent worker that only runs while something else awaits, so
+    /// flushing it per stanza would fold a socket write into a measurement of
+    /// the inbound path. [`Self::flush`] settles it between benchmarks
+    /// instead.
+    pub fn process_nowait(&self, node: Arc<wacore_binary::OwnedNodeRef>) {
+        self.runtime
+            .block_on(Arc::clone(&self.client).process_node(node));
+    }
+
+    /// Run a whole batch of stanzas under one `block_on`, as an offline drain
+    /// does: the read loop hands the queue over without yielding to anything
+    /// else between items, so the per-item cost includes whatever the ack
+    /// worker gets to do in the gaps.
+    pub fn process_burst(&self, nodes: &[Arc<wacore_binary::OwnedNodeRef>]) {
+        self.runtime.block_on(async {
+            for node in nodes {
+                Arc::clone(&self.client)
+                    .process_node(Arc::clone(node))
+                    .await;
+            }
+        });
+    }
+
+    /// Let the outbound work the processed stanzas queued (their transport
+    /// `<ack>`s) finish, so it cannot leak into the next benchmark.
+    pub fn flush(&self) {
+        self.runtime.block_on(async {
+            self.client
+                .outbound_flush
+                .flush(&*self.client.runtime, std::time::Duration::from_secs(5))
+                .await;
+        });
+    }
+
     /// A fresh stanza id per built message, so dedup never sees a repeat.
     fn next_message_id(&self) -> String {
         let n = self.next_id.fetch_add(1, Ordering::Relaxed);
@@ -935,6 +1086,7 @@ struct ReceiveFixture {
     group: Jid,
     group_sender_key: wacore::libsignal::protocol::SenderKeyName,
     counter: Arc<MessageCounter>,
+    stanza_counter: Arc<StanzaEventCounter>,
     subscription: wacore::types::events::Subscription,
 }
 
@@ -1015,6 +1167,9 @@ async fn build_receive_fixture() -> ReceiveFixture {
         group,
         group_sender_key,
         counter,
+        stanza_counter: Arc::new(StanzaEventCounter {
+            delivered: portable_atomic::AtomicU64::new(0),
+        }),
         subscription,
     }
 }