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

.github/workflows/bench-integration.yml

@@ -0,0 +1,186 @@
+name: Integration Benchmark
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+
+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-pr:
+    if: github.event_name == 'pull_request'
+    name: Integration Benchmark (PR)
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: write
+    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
+
+      - name: Compare against baseline
+        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
+
+  bench-integration-push:
+    if: github.event_name == 'push'
+    name: Integration Benchmark (push)
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      deployments: write
+    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
+
+      - name: Store baseline
+        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

.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,51 +38,11 @@ impl StanzaHandler for MessageHandler {
             }
         };
 
-        // Single cache lookup: get or create the lane (lock + queue + worker).
+        // Single-flight: get_with_by_ref guarantees exactly one init runs per key,
+        // preventing duplicate workers for the same chat (TOCTOU race).
         let lane = client
             .chat_lanes
-            .get_with_by_ref(&chat_jid, async {
-                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);
-
-                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
-                                );
-                            }
-                        }
-                    }))
-                    .detach();
-
-                ChatLane {
-                    enqueue_lock: Arc::new(async_lock::Mutex::new(())),
-                    queue_tx: tx,
-                }
-            })
+            .get_with_by_ref(&chat_jid, async { create_chat_lane(&client) })
             .await;
 
         // Lock serializes enqueue order for this chat
@@ -97,3 +57,47 @@ impl StanzaHandler for MessageHandler {
         true
     }
 }
+
+/// Construct a ChatLane with a spawned worker task. Extracted to keep the
+/// init closure passed to `get_with_by_ref` small.
+fn create_chat_lane(client: &Arc<Client>) -> ChatLane {
+    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);
+
+    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
+                    );
+                }
+            }
+        }))
+        .detach();
+
+    ChatLane {
+        enqueue_lock: Arc::new(async_lock::Mutex::new(())),
+        queue_tx: tx,
+    }
+}