perf: integration benchmarks + allocation optimizations (-23% total, -43% connect allocs)

.github/workflows/bench-integration.yml

@@ -0,0 +1,119 @@
+name: Integration Benchmark
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+
+permissions:
+  contents: write
+  pull-requests: write
+  deployments: write
+
+concurrency:
+  group: bench-integration-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+env:
+  CARGO_TERM_COLOR: always
+  PROTOC_VERSION: "3.25.3"
+  SCCACHE_GHA_ENABLED: "true"
+  RUSTC_WRAPPER: "sccache"
+
+jobs:
+  bench-integration:
+    name: Integration Benchmark
+    runs-on: ubuntu-latest
+    services:
+      mock-server:
+        image: ghcr.io/whiskeysockets-devtools/bartender:latest
+        credentials:
+          username: ${{ github.actor }}
+          password: ${{ secrets.BARTENDER_GHCR_TOKEN }}
+        ports:
+          - 8080:8080
+        env:
+          CHATSTATE_TTL_SECS: "3"
+        options: --log-driver none
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: nightly-2026-04-05
+
+      - name: Install protoc
+        uses: taiki-e/install-action@v2
+        with:
+          tool: protoc@${{ env.PROTOC_VERSION }}
+
+      - name: Setup sccache
+        uses: mozilla-actions/sccache-action@v0.0.9
+
+      - name: Cache Rust registry
+        uses: Swatinem/rust-cache@v2
+        with:
+          prefix-key: ${{ runner.os }}-cargo-bench-int
+          cache-targets: "false"
+
+      - name: Wait for mock server
+        run: |
+          for i in $(seq 1 30); do
+            if curl -sk https://localhost:8080/ > /dev/null 2>&1; then
+              echo "Mock server is ready"
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "Mock server failed to become ready"
+          exit 1
+
+      - name: Run integration benchmarks
+        env:
+          MOCK_SERVER_URL: "wss://127.0.0.1:8080/ws/chat"
+          RUST_LOG: warn
+        run: |
+          cargo run -p bench-integration --release \
+            > bench_integration_results.json \
+            2> bench_integration.log
+          cat bench_integration.log >&2
+          echo "Results:"
+          cat bench_integration_results.json
+
+      - name: Upload benchmark artifacts
+        uses: actions/upload-artifact@v4
+        with:
+          name: bench-integration-results
+          path: |
+            bench_integration_results.json
+            bench_integration.log
+          retention-days: 30
+
+      # Push to main: store baseline
+      - name: Store baseline (push to main)
+        if: github.event_name == 'push'
+        uses: benchmark-action/github-action-benchmark@v1
+        with:
+          name: "whatsapp-rust integration benchmarks"
+          tool: "customSmallerIsBetter"
+          output-file-path: bench_integration_results.json
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          auto-push: true
+          benchmark-data-dir-path: dev/bench-integration
+          max-items-in-chart: 100
+          summary-always: true
+
+      # PR: compare against baseline
+      - name: Compare against baseline (pull request)
+        if: github.event_name == 'pull_request'
+        uses: benchmark-action/github-action-benchmark@v1
+        with:
+          name: "whatsapp-rust integration benchmarks"
+          tool: "customSmallerIsBetter"
+          output-file-path: bench_integration_results.json
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          auto-push: false
+          save-data-file: false
+          benchmark-data-dir-path: dev/bench-integration
+          summary-always: true

.gitignore

@@ -10,3 +10,4 @@ docs
 .claude
 __pycache__
 .codex
+dhat-heap.json

Cargo.toml

@@ -14,6 +14,7 @@ members = [
     ".",
     "http_clients/ureq-client",
     "storages/sqlite-storage",
+    "tests/bench-integration",
     "tests/e2e",
     "transports/tokio-transport",
     "wacore",
@@ -161,3 +162,9 @@ inherits = "release"
 lto = "thin"
 debug = 1
 strip = false
+
+# Profiling: optimized with debug symbols for DHAT / heaptrack / perf
+[profile.profiling]
+inherits = "release"
+debug = 1
+strip = false

src/appstate_sync.rs

@@ -59,7 +59,7 @@ mod tests {
         async fn store_prekey(&self, _: u32, _: &[u8], _: bool) -> StoreResult<()> {
             Ok(())
         }
-        async fn load_prekey(&self, _: u32) -> StoreResult<Option<Vec<u8>>> {
+        async fn load_prekey(&self, _: u32) -> StoreResult<Option<bytes::Bytes>> {
             Ok(None)
         }
         async fn remove_prekey(&self, _: u32) -> StoreResult<()> {

src/client.rs

@@ -893,9 +893,8 @@ impl Client {
     /// [`send_node`](Client::send_node) for normal stanza sending.
     pub async fn send_raw_bytes(&self, plaintext: Vec<u8>) -> Result<(), ClientError> {
         let noise_socket = self.get_noise_socket().await?;
-        let encrypted_buf = Vec::with_capacity(plaintext.len() + 32);
         noise_socket
-            .encrypt_and_send(plaintext, encrypted_buf)
+            .encrypt_and_send(bytes::Bytes::from(plaintext))
             .await?;
         self.last_data_sent_ms
             .store(wacore::time::now_millis().max(0) as u64, Ordering::Relaxed);

src/features/signal.rs

@@ -151,7 +151,7 @@ impl<'a> Signal<'a> {
                 wacore::send::create_sender_key_distribution_message_for_group(
                     &mut adapter.sender_key_store,
                     group_jid,
-                    &own_jid,
+                    &sender_addr,
                 )
                 .await?,
             )
@@ -162,7 +162,7 @@ impl<'a> Signal<'a> {
         let ciphertext = wacore::send::encrypt_group_message(
             &mut adapter.sender_key_store,
             group_jid,
-            &own_jid,
+            &sender_addr,
             plaintext,
             &mut rng,
         )

src/handlers/message.rs

@@ -38,52 +38,57 @@ impl StanzaHandler for MessageHandler {
             }
         };
 
-        // Single cache lookup: get or create the lane (lock + queue + worker).
-        let lane = client
-            .chat_lanes
-            .get_with_by_ref(&chat_jid, async {
-                let (tx, rx) = async_channel::unbounded::<Arc<wacore_binary::OwnedNodeRef>>();
+        // Fast path: cache hit avoids building the init future entirely.
+        // This shrinks the handler's async state machine because the large
+        // worker-spawning closure isn't captured across the .await.
+        let lane = if let Some(existing) = client.chat_lanes.get(&chat_jid).await {
+            existing
+        } else {
+            let (tx, rx) = async_channel::unbounded::<Arc<wacore_binary::OwnedNodeRef>>();
 
-                let client_for_worker = client.clone();
-                let spawn_generation = client
-                    .connection_generation
-                    .load(std::sync::atomic::Ordering::Acquire);
+            let client_for_worker = client.clone();
+            let spawn_generation = client
+                .connection_generation
+                .load(std::sync::atomic::Ordering::Acquire);
 
-                client
-                    .runtime
-                    .spawn(Box::pin(async move {
-                        while let Ok(msg_node) = rx.recv().await {
-                            if client_for_worker
-                                .connection_generation
-                                .load(std::sync::atomic::Ordering::Acquire)
-                                != spawn_generation
-                            {
-                                log::debug!(target: "MessageQueue", "Stale worker exiting; remaining messages will be redelivered by server");
-                                break;
-                            }
-                            let start = wacore::time::now_millis() as u64;
-                            let client = client_for_worker.clone();
-                            Box::pin(client.handle_incoming_message(msg_node)).await;
-                            let elapsed =
-                                (wacore::time::now_millis() as u64).saturating_sub(start);
-                            if elapsed > MAX_MESSAGE_DELAY_MS {
-                                warn!(
-                                    target: "MessageQueue",
-                                    "Message processing took {:.1}s (MAX_MESSAGE_DELAY is {}s)",
-                                    elapsed as f64 / 1000.0,
-                                    MAX_MESSAGE_DELAY_MS / 1000
-                                );
-                            }
+            client
+                .runtime
+                .spawn(Box::pin(async move {
+                    while let Ok(msg_node) = rx.recv().await {
+                        if client_for_worker
+                            .connection_generation
+                            .load(std::sync::atomic::Ordering::Acquire)
+                            != spawn_generation
+                        {
+                            log::debug!(target: "MessageQueue", "Stale worker exiting; remaining messages will be redelivered by server");
+                            break;
                         }
-                    }))
-                    .detach();
+                        let start = wacore::time::now_millis() as u64;
+                        let client = client_for_worker.clone();
+                        Box::pin(client.handle_incoming_message(msg_node)).await;
+                        let elapsed = (wacore::time::now_millis() as u64).saturating_sub(start);
+                        if elapsed > MAX_MESSAGE_DELAY_MS {
+                            warn!(
+                                target: "MessageQueue",
+                                "Message processing took {:.1}s (MAX_MESSAGE_DELAY is {}s)",
+                                elapsed as f64 / 1000.0,
+                                MAX_MESSAGE_DELAY_MS / 1000
+                            );
+                        }
+                    }
+                }))
+                .detach();
 
-                ChatLane {
-                    enqueue_lock: Arc::new(async_lock::Mutex::new(())),
-                    queue_tx: tx,
-                }
-            })
-            .await;
+            let lane = ChatLane {
+                enqueue_lock: Arc::new(async_lock::Mutex::new(())),
+                queue_tx: tx,
+            };
+            client
+                .chat_lanes
+                .insert(chat_jid.clone(), lane.clone())
+                .await;
+            lane
+        };
 
         // Lock serializes enqueue order for this chat
         let _guard = lane.enqueue_lock.lock().await;

src/handshake.rs

@@ -71,7 +71,7 @@ pub async fn do_handshake(
     // First message includes the WA connection header (with optional edge routing)
     let framed = wacore::framing::encode_frame(&client_hello_bytes, Some(&header))
         .map_err(HandshakeError::Transport)?;
-    transport.send(framed).await?;
+    transport.send(bytes::Bytes::from(framed)).await?;
 
     // Wait for server response frame
     let resp_frame = loop {
@@ -113,7 +113,7 @@ pub async fn do_handshake(
     // Subsequent messages don't need the header
     let framed = wacore::framing::encode_frame(&client_finish_bytes, None)
         .map_err(HandshakeError::Transport)?;
-    transport.send(framed).await?;
+    transport.send(bytes::Bytes::from(framed)).await?;
 
     let (write_key, read_key) = handshake_state.finish()?;
     info!("Handshake complete, switching to encrypted communication");