From af47f1743c8a8c46e24bfa947effaef75bebf04c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 14:21:57 +0000 Subject: [PATCH 1/8] bench(integration): port integration benchmarks to CodSpeed (simulation + memory) Replace the custom `[[bin]]` integration benchmark (timing + a hand-rolled counting allocator, emitting `customSmallerIsBetter` JSON for github-action-benchmark) with a divan harness run under `cargo codspeed`, the same model the unit benches already use. The benches keep the real async client and the external bartender mock server; only the measurement layer changes. CodSpeed's memory instrument supersedes the custom counting allocator and adds deterministic memory metrics, simulation instruction counts, and flamegraph profiling on the dashboard. - New harness `tests/bench-integration/benches/integration.rs` with a shared multi-thread tokio runtime and a connected-and-warmed client pair, both built once outside the measured region. Benches: connect_to_ready, send_message, send_and_receive, reconnect. The single + x20 coverage is preserved via `#[divan::bench(args = [1, 20])]` looping the op inside the measured region. - `bench-integration` becomes bench-only: drop the `[[bin]]`, the `dhat` dependency / `dhat-heap` feature / cargo-shear ignore, and the now-unused `anyhow`/`wacore`/`log`/`serde`/`serde_json` deps; add `divan` and the `[[bench]]` target. Delete `src/main.rs` and `src/counting_alloc.rs`. - Add a separate `integration-benchmarks` job to `codspeed.yml` carrying the bartender service container, the mock-server wait, a `-p bench-integration` build, and `MOCK_SERVER_URL`/`RUST_LOG` on the run step. The unit-bench job is left untouched so it stays fast and mock-server-free. Delete the old `bench-integration.yml`. End-to-end runs are CI-only (they need the GHCR bartender image); local validation covers compile, clippy, and `cargo codspeed build`. --- .github/workflows/bench-integration.yml | 186 --------- .github/workflows/codspeed.yml | 66 ++++ Cargo.lock | 7 +- tests/bench-integration/Cargo.toml | 21 +- .../bench-integration/benches/integration.rs | 181 +++++++++ tests/bench-integration/src/counting_alloc.rs | 63 ---- tests/bench-integration/src/main.rs | 356 ------------------ 7 files changed, 253 insertions(+), 627 deletions(-) delete mode 100644 .github/workflows/bench-integration.yml create mode 100644 tests/bench-integration/benches/integration.rs delete mode 100644 tests/bench-integration/src/counting_alloc.rs delete mode 100644 tests/bench-integration/src/main.rs diff --git a/.github/workflows/bench-integration.yml b/.github/workflows/bench-integration.yml deleted file mode 100644 index 8a7883a74..000000000 --- a/.github/workflows/bench-integration.yml +++ /dev/null @@ -1,186 +0,0 @@ -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-06-16 - - - 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-06-16 - - - 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 diff --git a/.github/workflows/codspeed.yml b/.github/workflows/codspeed.yml index ce161f12c..44e80c587 100644 --- a/.github/workflows/codspeed.yml +++ b/.github/workflows/codspeed.yml @@ -60,3 +60,69 @@ jobs: with: mode: simulation,memory run: cargo codspeed run + + # Integration benches drive the real async client against the bartender mock + # server, so they need the service container + MOCK_SERVER_URL. Kept as a + # separate job so the unit-bench job above stays fast and mock-server-free. + integration-benchmarks: + name: Run CodSpeed integration benchmarks + runs-on: ubuntu-latest + permissions: + contents: read + id-token: write # for OpenID Connect authentication with CodSpeed + 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 + with: + persist-credentials: false + + - uses: dtolnay/rust-toolchain@master + with: + toolchain: nightly-2026-06-16 + + - name: Install tools (protoc, cargo-codspeed) + uses: taiki-e/install-action@v2 + with: + tool: protoc@${{ env.PROTOC_VERSION }},cargo-codspeed@4.7.0 + + - name: Cache Rust registry + uses: Swatinem/rust-cache@v2 + with: + prefix-key: ${{ runner.os }}-cargo-codspeed-integration + 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 + + # simulation and memory share the same instrumented build, so one + # build covers both instruments. + - name: Build the benchmark targets + run: cargo codspeed build -m simulation -m memory -p bench-integration + + - name: Run the benchmarks + uses: CodSpeedHQ/action@v4 + env: + MOCK_SERVER_URL: "wss://127.0.0.1:8080/ws/chat" + RUST_LOG: warn + with: + mode: simulation,memory + run: cargo codspeed run diff --git a/Cargo.lock b/Cargo.lock index 6f2a88420..d6cf9221f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -191,15 +191,10 @@ checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" name = "bench-integration" version = "0.0.0" dependencies = [ - "anyhow", - "dhat", + "codspeed-divan-compat", "e2e-tests", "env_logger", - "log", - "serde", - "serde_json", "tokio", - "wacore", "whatsapp-rust", ] diff --git a/tests/bench-integration/Cargo.toml b/tests/bench-integration/Cargo.toml index 0fa538451..13d1e0db1 100644 --- a/tests/bench-integration/Cargo.toml +++ b/tests/bench-integration/Cargo.toml @@ -4,26 +4,11 @@ version = "0.0.0" edition = "2024" publish = false -[package.metadata.cargo-shear] -ignored = ["dhat"] - -[[bin]] -name = "bench-integration" -path = "src/main.rs" - -[features] -dhat-heap = ["dep:dhat"] - [dependencies] -anyhow = { workspace = true } -dhat = { version = "0.3", optional = true } +divan = { workspace = true } e2e-tests = { path = "../e2e" } env_logger = { workspace = true } -log = { workspace = true } -serde = { workspace = true } -serde_json = { workspace = true, features = ["std"] } tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] } -wacore = { path = "../../wacore" } whatsapp-rust = { path = "../..", default-features = false, features = [ "danger-skip-tls-verify", "debug-diagnostics", @@ -33,5 +18,9 @@ whatsapp-rust = { path = "../..", default-features = false, features = [ "signal", ] } +[[bench]] +name = "integration" +harness = false + [lints] workspace = true diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs new file mode 100644 index 000000000..497062fb7 --- /dev/null +++ b/tests/bench-integration/benches/integration.rs @@ -0,0 +1,181 @@ +//! Integration benchmarks driven through the real async client against the +//! bartender mock server. Ported from a custom timing/allocation binary to +//! divan so CodSpeed records deterministic memory metrics (replacing the old +//! counting allocator), simulation instruction counts, and flamegraphs. +//! +//! These cannot run without the mock server (`MOCK_SERVER_URL`), so they only +//! execute in CI under `cargo codspeed run`. + +// Large `--all-features` async fns (tracing + tracing-pii) need a deeper +// recursion limit; matches `src/lib.rs` and the e2e crate. +#![recursion_limit = "512"] + +use std::sync::OnceLock; +use std::sync::atomic::{AtomicU64, Ordering}; + +use e2e_tests::{TestClient, text_msg}; +use tokio::sync::Mutex; +use whatsapp_rust::Jid; + +fn main() { + divan::main(); +} + +// A single multi-thread runtime shared by every bench. Building one per +// iteration would charge thread-pool startup syscalls to the measured region, +// and the real client expects a multi-thread scheduler. +static RT: OnceLock = OnceLock::new(); + +fn rt() -> &'static tokio::runtime::Runtime { + RT.get_or_init(|| { + // Best-effort logger init, mirroring the old binary; ignore errors so a + // second bench module call is harmless. + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("warn")) + .try_init() + .ok(); + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("build multi-thread bench runtime") + }) +} + +/// A connected, session-warmed client pair reused across all send/receive +/// iterations. Connecting + warming a pair is far more expensive than the +/// measured send, so it is done once (unmeasured) and shared. `wait_for_text` +/// needs `&mut`, hence the `Mutex`. +struct Pair { + a: TestClient, + b: TestClient, + jid_b: Jid, +} + +static PAIR: OnceLock> = OnceLock::new(); + +/// Monotonic counter producing a unique body per measured iteration. The mock +/// server dedupes identical message bodies, so reusing one would let later +/// iterations short-circuit and stop measuring the real send path. +static COUNTER: AtomicU64 = AtomicU64::new(0); + +fn unique_body(tag: &str) -> String { + format!("bench-{tag}-{}", COUNTER.fetch_add(1, Ordering::Relaxed)) +} + +/// Connect two clients and warm their Signal session with one throwaway +/// round-trip, so measured sends exercise steady state (plain `msg`), not the +/// first-message pre-key path. +fn pair() -> &'static Mutex { + PAIR.get_or_init(|| { + rt().block_on(async { + let a = TestClient::connect("bench_pair_a") + .await + .expect("connect client a"); + let mut b = TestClient::connect("bench_pair_b") + .await + .expect("connect client b"); + let jid_b = b.jid().await; + + let warm = unique_body("warmup"); + a.client + .send_message(jid_b.clone(), text_msg(&warm)) + .await + .expect("send warmup message"); + b.wait_for_text(&warm, 30).await.expect("receive warmup"); + + Mutex::new(Pair { a, b, jid_b }) + }) + }) +} + +// x20 coverage: divan's `args` runs the bench once per value and reports each +// separately, so one fn yields both the single-op result (n=1) and the 20-op +// batch (n=20) — the loop runs inside the measured region. This preserves the +// original single + x20 signal without a duplicate fn; the batch total stands +// in for the old hand-divided amortized number. +const BATCH_SIZES: [u64; 2] = [1, 20]; + +/// Client creation through Connected (ready). The cheap disconnect is inside +/// the measured region on purpose: the connect handshake dominates its cost, +/// and tearing the client down each iteration stops sessions leaking across +/// iterations on the mock server. +#[divan::bench] +fn connect_to_ready() { + rt().block_on(async { + let c = TestClient::connect("bench_connect") + .await + .expect("connect client"); + c.disconnect().await; + }); +} + +/// Sending a single DM (sender side only, matching the original — no +/// `wait_for_text` here). Covers protobuf encode, Signal encrypt, node marshal +/// and the WebSocket write. +#[divan::bench(args = BATCH_SIZES)] +fn send_message(n: u64) { + let pair = pair(); + rt().block_on(async { + let guard = pair.lock().await; + for _ in 0..n { + let body = unique_body("send"); + guard + .a + .client + .send_message(guard.jid_b.clone(), text_msg(&body)) + .await + .expect("send message"); + } + }); +} + +/// Full send + receive round-trip on the warmed pair. +#[divan::bench(args = BATCH_SIZES)] +fn send_and_receive(n: u64) { + let pair = pair(); + rt().block_on(async { + let mut guard = pair.lock().await; + for _ in 0..n { + let body = unique_body("recv"); + guard + .a + .client + .send_message(guard.jid_b.clone(), text_msg(&body)) + .await + .expect("send message"); + guard + .b + .wait_for_text(&body, 30) + .await + .expect("receive text"); + } + }); +} + +/// A reconnect cycle (disconnect -> reconnect -> ready). The client is created +/// once outside the measured region; only `reconnect_and_wait` is measured. +#[divan::bench] +fn reconnect(bencher: divan::Bencher) { + let client: &'static Mutex = { + static RECONNECT: OnceLock> = OnceLock::new(); + RECONNECT.get_or_init(|| { + rt().block_on(async { + Mutex::new( + TestClient::connect("bench_reconn") + .await + .expect("connect reconnect client"), + ) + }) + }) + }; + + bencher.bench_local(|| { + rt().block_on(async { + client + .lock() + .await + .reconnect_and_wait() + .await + .expect("reconnect and wait"); + }); + }); +} diff --git a/tests/bench-integration/src/counting_alloc.rs b/tests/bench-integration/src/counting_alloc.rs deleted file mode 100644 index 74d683e27..000000000 --- a/tests/bench-integration/src/counting_alloc.rs +++ /dev/null @@ -1,63 +0,0 @@ -use std::alloc::{GlobalAlloc, Layout, System}; -use std::sync::atomic::{AtomicU64, Ordering}; - -static ALLOC_COUNT: AtomicU64 = AtomicU64::new(0); -static ALLOC_BYTES: AtomicU64 = AtomicU64::new(0); - -pub struct CountingAlloc; - -unsafe impl GlobalAlloc for CountingAlloc { - unsafe fn alloc(&self, layout: Layout) -> *mut u8 { - ALLOC_COUNT.fetch_add(1, Ordering::Relaxed); - ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed); - unsafe { System.alloc(layout) } - } - - unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { - unsafe { System.dealloc(ptr, layout) } - } - - unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { - // Track the size difference but not as a new allocation - if new_size > layout.size() { - ALLOC_BYTES.fetch_add((new_size - layout.size()) as u64, Ordering::Relaxed); - } - unsafe { System.realloc(ptr, layout, new_size) } - } - - unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { - ALLOC_COUNT.fetch_add(1, Ordering::Relaxed); - ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed); - unsafe { System.alloc_zeroed(layout) } - } -} - -#[derive(Debug, Clone, Copy)] -pub struct AllocSnapshot { - alloc_count: u64, - alloc_bytes: u64, -} - -impl AllocSnapshot { - pub fn now() -> Self { - Self { - alloc_count: ALLOC_COUNT.load(Ordering::Relaxed), - alloc_bytes: ALLOC_BYTES.load(Ordering::Relaxed), - } - } -} - -#[derive(Debug, Clone, Copy)] -pub struct AllocDelta { - pub alloc_count: u64, - pub alloc_bytes: u64, -} - -impl AllocDelta { - pub fn between(before: AllocSnapshot, after: AllocSnapshot) -> Self { - Self { - alloc_count: after.alloc_count.saturating_sub(before.alloc_count), - alloc_bytes: after.alloc_bytes.saturating_sub(before.alloc_bytes), - } - } -} diff --git a/tests/bench-integration/src/main.rs b/tests/bench-integration/src/main.rs deleted file mode 100644 index accc47613..000000000 --- a/tests/bench-integration/src/main.rs +++ /dev/null @@ -1,356 +0,0 @@ -// See the matching note in lib.rs: large async fns need a deeper recursion limit -// when the `--all-features` paths (tracing + tracing-pii) combine. -#![recursion_limit = "512"] - -#[cfg(not(feature = "dhat-heap"))] -mod counting_alloc; - -#[cfg(feature = "dhat-heap")] -#[global_allocator] -static GLOBAL: dhat::Alloc = dhat::Alloc; - -#[cfg(not(feature = "dhat-heap"))] -#[global_allocator] -static GLOBAL: counting_alloc::CountingAlloc = counting_alloc::CountingAlloc; - -use wacore::time::Instant; - -use e2e_tests::{TestClient, text_msg}; -use serde::Serialize; - -// --------------------------------------------------------------------------- -// Measurement helpers -// --------------------------------------------------------------------------- - -/// Snapshot-based measurement: runs `f`, returns wall time + alloc delta. -#[cfg(not(feature = "dhat-heap"))] -async fn measure( - mut f: F, -) -> anyhow::Result<(T, std::time::Duration, counting_alloc::AllocDelta)> -where - F: AsyncFnMut() -> anyhow::Result, -{ - let before = counting_alloc::AllocSnapshot::now(); - let t0 = Instant::now(); - let result = f().await?; - let elapsed = t0.elapsed(); - let delta = counting_alloc::AllocDelta::between(before, counting_alloc::AllocSnapshot::now()); - Ok((result, elapsed, delta)) -} - -/// DHAT mode: just runs `f` and returns wall time (DHAT captures everything). -#[cfg(feature = "dhat-heap")] -async fn measure(mut f: F) -> anyhow::Result<(T, std::time::Duration)> -where - F: AsyncFnMut() -> anyhow::Result, -{ - let t0 = Instant::now(); - let result = f().await?; - let elapsed = t0.elapsed(); - Ok((result, elapsed)) -} - -// --------------------------------------------------------------------------- -// Results collection -// --------------------------------------------------------------------------- - -#[derive(Serialize)] -struct BenchEntry { - name: String, - unit: String, - value: u64, -} - -struct BenchResults(Vec); - -impl BenchResults { - fn new() -> Self { - Self(Vec::new()) - } - - fn record(&mut self, scenario: &str, metric: &str, unit: &str, value: u64) { - self.0.push(BenchEntry { - name: format!("integration::{scenario}::{metric}"), - unit: unit.to_string(), - value, - }); - } - - fn record_wall(&mut self, scenario: &str, elapsed: std::time::Duration) { - self.record( - scenario, - "wall_ms", - "milliseconds", - elapsed.as_millis() as u64, - ); - } - - #[cfg(not(feature = "dhat-heap"))] - fn record_measured( - &mut self, - scenario: &str, - elapsed: std::time::Duration, - delta: &counting_alloc::AllocDelta, - ) { - self.record(scenario, "alloc_count", "allocations", delta.alloc_count); - self.record(scenario, "alloc_bytes", "bytes", delta.alloc_bytes); - self.record_wall(scenario, elapsed); - } - - #[cfg(not(feature = "dhat-heap"))] - fn record_measured_amortized( - &mut self, - scenario: &str, - n: u64, - elapsed: std::time::Duration, - delta: &counting_alloc::AllocDelta, - ) { - self.record( - scenario, - "alloc_count", - "allocations", - delta.alloc_count / n, - ); - self.record(scenario, "alloc_bytes", "bytes", delta.alloc_bytes / n); - self.record( - scenario, - "wall_ms", - "milliseconds", - elapsed.as_millis() as u64 / n, - ); - } -} - -// --------------------------------------------------------------------------- -// Scenarios -// --------------------------------------------------------------------------- - -/// Measure allocations from client creation through Connected (ready). -async fn bench_connect_to_ready(results: &mut BenchResults) -> anyhow::Result<()> { - let m = measure(async || TestClient::connect("bench_connect").await).await?; - - #[cfg(not(feature = "dhat-heap"))] - { - let (client, elapsed, delta) = m; - results.record_measured("connect_to_ready", elapsed, &delta); - client.disconnect().await; - } - #[cfg(feature = "dhat-heap")] - { - let (client, elapsed) = m; - results.record_wall("connect_to_ready", elapsed); - client.disconnect().await; - } - Ok(()) -} - -/// Measure allocations for sending a single DM (sender side only). -/// -/// Both clients are connected before measurement starts. -/// We track the `send_message` call which covers: protobuf encoding, -/// Signal encrypt, node marshal, and WebSocket write. -async fn bench_send_message(results: &mut BenchResults) -> anyhow::Result<()> { - let client_a = TestClient::connect("bench_send_a").await?; - let mut client_b = TestClient::connect("bench_send_b").await?; - let jid_b = client_b.jid().await; - - // Warm up: establish Signal session with a throwaway message - client_a - .client - .send_message(jid_b.clone(), text_msg("warmup-send")) - .await?; - client_b.wait_for_text("warmup-send", 30).await?; - - // -- Single send -- - let m = measure(async || { - Ok(client_a - .client - .send_message(jid_b.clone(), text_msg("bench-send-single")) - .await?) - }) - .await?; - - #[cfg(not(feature = "dhat-heap"))] - { - let (_, elapsed, delta) = m; - results.record_measured("send_message", elapsed, &delta); - } - #[cfg(feature = "dhat-heap")] - { - let (_, elapsed) = m; - results.record_wall("send_message", elapsed); - } - - client_b.wait_for_text("bench-send-single", 30).await?; - - // -- Amortized: send N messages -- - const N: u64 = 20; - let send_texts: Vec = (0..N).map(|i| format!("bench-send-{i}")).collect(); - let m = measure(async || { - for text in &send_texts { - client_a - .client - .send_message(jid_b.clone(), text_msg(text)) - .await?; - } - Ok(()) - }) - .await?; - - #[cfg(not(feature = "dhat-heap"))] - { - let (_, elapsed, delta) = m; - results.record_measured_amortized("send_message_x20_amortized", N, elapsed, &delta); - } - #[cfg(feature = "dhat-heap")] - { - let (_, elapsed) = m; - results.record( - "send_message_x20_amortized", - "wall_ms", - "milliseconds", - elapsed.as_millis() as u64 / N, - ); - } - - for text in &send_texts { - client_b.wait_for_text(text, 30).await?; - } - - client_a.disconnect().await; - client_b.disconnect().await; - Ok(()) -} - -/// Measure allocations for a full send+receive round-trip. -/// -/// Both clients are connected and session is warm before measurement. -async fn bench_receive_message(results: &mut BenchResults) -> anyhow::Result<()> { - let client_a = TestClient::connect("bench_recv_a").await?; - let mut client_b = TestClient::connect("bench_recv_b").await?; - let jid_b = client_b.jid().await; - - // Warm up Signal session - client_a - .client - .send_message(jid_b.clone(), text_msg("warmup-recv")) - .await?; - client_b.wait_for_text("warmup-recv", 30).await?; - - // -- Single send+receive -- - let m = measure(async || { - client_a - .client - .send_message(jid_b.clone(), text_msg("bench-recv-single")) - .await?; - client_b.wait_for_text("bench-recv-single", 30).await?; - Ok(()) - }) - .await?; - - #[cfg(not(feature = "dhat-heap"))] - { - let (_, elapsed, delta) = m; - results.record_measured("send_and_receive_message", elapsed, &delta); - } - #[cfg(feature = "dhat-heap")] - { - let (_, elapsed) = m; - results.record_wall("send_and_receive_message", elapsed); - } - - // -- Amortized N round-trips -- - const N: u64 = 20; - let m = measure(async || { - for i in 0..N { - let text = format!("bench-recv-{i}"); - client_a - .client - .send_message(jid_b.clone(), text_msg(&text)) - .await?; - client_b.wait_for_text(&text, 30).await?; - } - Ok(()) - }) - .await?; - - #[cfg(not(feature = "dhat-heap"))] - { - let (_, elapsed, delta) = m; - results.record_measured_amortized("send_and_receive_x20_amortized", N, elapsed, &delta); - } - #[cfg(feature = "dhat-heap")] - { - let (_, elapsed) = m; - results.record( - "send_and_receive_x20_amortized", - "wall_ms", - "milliseconds", - elapsed.as_millis() as u64 / N, - ); - } - - client_a.disconnect().await; - client_b.disconnect().await; - Ok(()) -} - -/// Measure allocations for a reconnect cycle (disconnect -> reconnect -> ready). -async fn bench_reconnect(results: &mut BenchResults) -> anyhow::Result<()> { - let mut client = TestClient::connect("bench_reconn").await?; - - let m = measure(async || { - client.reconnect_and_wait().await?; - Ok(()) - }) - .await?; - - #[cfg(not(feature = "dhat-heap"))] - { - let (_, elapsed, delta) = m; - results.record_measured("reconnect", elapsed, &delta); - } - #[cfg(feature = "dhat-heap")] - { - let (_, elapsed) = m; - results.record_wall("reconnect", elapsed); - } - - client.disconnect().await; - Ok(()) -} - -// --------------------------------------------------------------------------- -// Main -// --------------------------------------------------------------------------- - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("warn")) - .try_init() - .ok(); - - #[cfg(feature = "dhat-heap")] - let _profiler = dhat::Profiler::new_heap(); - - let mut results = BenchResults::new(); - - eprintln!("--- bench: connect_to_ready ---"); - bench_connect_to_ready(&mut results).await?; - - eprintln!("--- bench: send_message ---"); - bench_send_message(&mut results).await?; - - eprintln!("--- bench: receive_message ---"); - bench_receive_message(&mut results).await?; - - eprintln!("--- bench: reconnect ---"); - bench_reconnect(&mut results).await?; - - // Output customSmallerIsBetter JSON to stdout - let json = serde_json::to_string_pretty(&results.0)?; - println!("{json}"); - - eprintln!("--- done: {} metrics collected ---", results.0.len()); - Ok(()) -} From d881a473a96431df55a7d0771c4eb8ceb3b4136d Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 14:55:42 +0000 Subject: [PATCH 2/8] fix(bench): select integration package on run and move setup out of measured region The integration job's `cargo codspeed run` discovered no benchmarks ("No benchmarks found for the simulation mode") because tests/bench-integration is not a workspace default-member, so a bare run ignores it. Select it explicitly with `-p bench-integration`, matching the build step. Also convert connect_to_ready, send_message and send_and_receive to the Bencher setup-outside pattern (like reconnect already does) so runtime construction and the pair connect + Signal warmup run before the measured closure rather than being charged to the first iteration. --- .github/workflows/codspeed.yml | 5 +- .../bench-integration/benches/integration.rs | 94 +++++++++++-------- 2 files changed, 57 insertions(+), 42 deletions(-) diff --git a/.github/workflows/codspeed.yml b/.github/workflows/codspeed.yml index 44e80c587..adfe92937 100644 --- a/.github/workflows/codspeed.yml +++ b/.github/workflows/codspeed.yml @@ -118,6 +118,9 @@ jobs: - name: Build the benchmark targets run: cargo codspeed build -m simulation -m memory -p bench-integration + # `-p bench-integration` is required: `tests/bench-integration` is not a + # workspace default-member, so a bare `cargo codspeed run` discovers no + # benchmarks for it ("No benchmarks found for the simulation mode"). - name: Run the benchmarks uses: CodSpeedHQ/action@v4 env: @@ -125,4 +128,4 @@ jobs: RUST_LOG: warn with: mode: simulation,memory - run: cargo codspeed run + run: cargo codspeed run -p bench-integration diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs index 497062fb7..a85b6a967 100644 --- a/tests/bench-integration/benches/integration.rs +++ b/tests/bench-integration/benches/integration.rs @@ -94,60 +94,72 @@ fn pair() -> &'static Mutex { // in for the old hand-divided amortized number. const BATCH_SIZES: [u64; 2] = [1, 20]; -/// Client creation through Connected (ready). The cheap disconnect is inside -/// the measured region on purpose: the connect handshake dominates its cost, -/// and tearing the client down each iteration stops sessions leaking across -/// iterations on the mock server. +/// Client creation through Connected (ready). The runtime is initialized +/// outside the measured closure so first-call thread-pool/logger startup is not +/// charged to the result. The cheap disconnect stays inside on purpose: the +/// connect handshake dominates its cost, and tearing the client down each +/// iteration stops sessions leaking across iterations on the mock server. #[divan::bench] -fn connect_to_ready() { - rt().block_on(async { - let c = TestClient::connect("bench_connect") - .await - .expect("connect client"); - c.disconnect().await; +fn connect_to_ready(bencher: divan::Bencher) { + let rt = rt(); + bencher.bench_local(|| { + rt.block_on(async { + let c = TestClient::connect("bench_connect") + .await + .expect("connect client"); + c.disconnect().await; + }); }); } /// Sending a single DM (sender side only, matching the original — no /// `wait_for_text` here). Covers protobuf encode, Signal encrypt, node marshal -/// and the WebSocket write. +/// and the WebSocket write. The runtime and warmed pair are initialized outside +/// the measured closure so connect + Signal warmup are not charged to the send. #[divan::bench(args = BATCH_SIZES)] -fn send_message(n: u64) { +fn send_message(bencher: divan::Bencher, n: u64) { + let rt = rt(); let pair = pair(); - rt().block_on(async { - let guard = pair.lock().await; - for _ in 0..n { - let body = unique_body("send"); - guard - .a - .client - .send_message(guard.jid_b.clone(), text_msg(&body)) - .await - .expect("send message"); - } + bencher.bench_local(|| { + rt.block_on(async { + let guard = pair.lock().await; + for _ in 0..n { + let body = unique_body("send"); + guard + .a + .client + .send_message(guard.jid_b.clone(), text_msg(&body)) + .await + .expect("send message"); + } + }); }); } -/// Full send + receive round-trip on the warmed pair. +/// Full send + receive round-trip on the warmed pair. Runtime and pair are +/// initialized outside the measured closure (see `send_message`). #[divan::bench(args = BATCH_SIZES)] -fn send_and_receive(n: u64) { +fn send_and_receive(bencher: divan::Bencher, n: u64) { + let rt = rt(); let pair = pair(); - rt().block_on(async { - let mut guard = pair.lock().await; - for _ in 0..n { - let body = unique_body("recv"); - guard - .a - .client - .send_message(guard.jid_b.clone(), text_msg(&body)) - .await - .expect("send message"); - guard - .b - .wait_for_text(&body, 30) - .await - .expect("receive text"); - } + bencher.bench_local(|| { + rt.block_on(async { + let mut guard = pair.lock().await; + for _ in 0..n { + let body = unique_body("recv"); + guard + .a + .client + .send_message(guard.jid_b.clone(), text_msg(&body)) + .await + .expect("send message"); + guard + .b + .wait_for_text(&body, 30) + .await + .expect("receive text"); + } + }); }); } From a5cb9e6f22112df64665860827c172b80b3380ed Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 15:05:34 +0000 Subject: [PATCH 3/8] fix(bench): isolate send and round-trip benches with separate pairs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit send_message is sender-only and never drains client b, so its delivered messages accumulate in b's unbounded event channel. With a single shared pair, send_and_receive reused it and its measured wait_for_text had to discard that backlog before seeing its own message — leaking send_message's iteration count (and divan's run order) into the round-trip result. Give the sender-only and round-trip benches their own warmed pairs so each bench's measurement is independent of the other. --- .../bench-integration/benches/integration.rs | 64 +++++++++++-------- 1 file changed, 38 insertions(+), 26 deletions(-) diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs index a85b6a967..0508a8b8f 100644 --- a/tests/bench-integration/benches/integration.rs +++ b/tests/bench-integration/benches/integration.rs @@ -40,17 +40,23 @@ fn rt() -> &'static tokio::runtime::Runtime { }) } -/// A connected, session-warmed client pair reused across all send/receive -/// iterations. Connecting + warming a pair is far more expensive than the -/// measured send, so it is done once (unmeasured) and shared. `wait_for_text` -/// needs `&mut`, hence the `Mutex`. +/// A connected, session-warmed client pair. Connecting + warming a pair is far +/// more expensive than the measured send, so it is done once (unmeasured) and +/// reused across iterations. `wait_for_text` needs `&mut`, hence the `Mutex`. struct Pair { a: TestClient, b: TestClient, jid_b: Jid, } -static PAIR: OnceLock> = OnceLock::new(); +// Separate pairs for the sender-only and round-trip benches. `send_message` +// never drains client `b`, so its delivered messages accumulate in `b`'s +// unbounded event channel; sharing one pair would force `send_and_receive`'s +// measured `wait_for_text` to discard that backlog first, leaking the send +// bench's iteration count into the round-trip result. Dedicated pairs keep each +// bench's measurement independent of the other (and of divan's run order). +static PAIR_SEND: OnceLock> = OnceLock::new(); +static PAIR_RECV: OnceLock> = OnceLock::new(); /// Monotonic counter producing a unique body per measured iteration. The mock /// server dedupes identical message bodies, so reusing one would let later @@ -64,27 +70,33 @@ fn unique_body(tag: &str) -> String { /// Connect two clients and warm their Signal session with one throwaway /// round-trip, so measured sends exercise steady state (plain `msg`), not the /// first-message pre-key path. -fn pair() -> &'static Mutex { - PAIR.get_or_init(|| { - rt().block_on(async { - let a = TestClient::connect("bench_pair_a") - .await - .expect("connect client a"); - let mut b = TestClient::connect("bench_pair_b") - .await - .expect("connect client b"); - let jid_b = b.jid().await; +fn connect_warmed_pair(prefix_a: &str, prefix_b: &str) -> Mutex { + rt().block_on(async { + let a = TestClient::connect(prefix_a) + .await + .expect("connect client a"); + let mut b = TestClient::connect(prefix_b) + .await + .expect("connect client b"); + let jid_b = b.jid().await; + + let warm = unique_body("warmup"); + a.client + .send_message(jid_b.clone(), text_msg(&warm)) + .await + .expect("send warmup message"); + b.wait_for_text(&warm, 30).await.expect("receive warmup"); + + Mutex::new(Pair { a, b, jid_b }) + }) +} - let warm = unique_body("warmup"); - a.client - .send_message(jid_b.clone(), text_msg(&warm)) - .await - .expect("send warmup message"); - b.wait_for_text(&warm, 30).await.expect("receive warmup"); +fn pair_send() -> &'static Mutex { + PAIR_SEND.get_or_init(|| connect_warmed_pair("bench_send_a", "bench_send_b")) +} - Mutex::new(Pair { a, b, jid_b }) - }) - }) +fn pair_recv() -> &'static Mutex { + PAIR_RECV.get_or_init(|| connect_warmed_pair("bench_recv_a", "bench_recv_b")) } // x20 coverage: divan's `args` runs the bench once per value and reports each @@ -119,7 +131,7 @@ fn connect_to_ready(bencher: divan::Bencher) { #[divan::bench(args = BATCH_SIZES)] fn send_message(bencher: divan::Bencher, n: u64) { let rt = rt(); - let pair = pair(); + let pair = pair_send(); bencher.bench_local(|| { rt.block_on(async { let guard = pair.lock().await; @@ -141,7 +153,7 @@ fn send_message(bencher: divan::Bencher, n: u64) { #[divan::bench(args = BATCH_SIZES)] fn send_and_receive(bencher: divan::Bencher, n: u64) { let rt = rt(); - let pair = pair(); + let pair = pair_recv(); bencher.bench_local(|| { rt.block_on(async { let mut guard = pair.lock().await; From 6c4affc8051eb2931d87d903677249ab6dcd1962 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 15:16:24 +0000 Subject: [PATCH 4/8] fix(bench): drain the sender-only pair's receiver to bound retained events send_message never reads its recipient (client b), so b's unbounded event channel retained every delivered message and grew across samples, letting CodSpeed memory/simulation numbers depend on how many sends ran earlier. Spawn a background task that drains the send pair's receiver (left off the round-trip pair, which consumes b via wait_for_text) so b stays bounded without adding work to the measured send. --- .../bench-integration/benches/integration.rs | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs index 0508a8b8f..609d30e22 100644 --- a/tests/bench-integration/benches/integration.rs +++ b/tests/bench-integration/benches/integration.rs @@ -50,11 +50,11 @@ struct Pair { } // Separate pairs for the sender-only and round-trip benches. `send_message` -// never drains client `b`, so its delivered messages accumulate in `b`'s -// unbounded event channel; sharing one pair would force `send_and_receive`'s -// measured `wait_for_text` to discard that backlog first, leaking the send +// never reads client `b`, so sharing one pair would force `send_and_receive`'s +// measured `wait_for_text` to discard the send backlog first, leaking the send // bench's iteration count into the round-trip result. Dedicated pairs keep each -// bench's measurement independent of the other (and of divan's run order). +// bench independent of the other (and of divan's run order); the send pair also +// drains `b` in the background so its queue stays bounded (see below). static PAIR_SEND: OnceLock> = OnceLock::new(); static PAIR_RECV: OnceLock> = OnceLock::new(); @@ -70,7 +70,14 @@ fn unique_body(tag: &str) -> String { /// Connect two clients and warm their Signal session with one throwaway /// round-trip, so measured sends exercise steady state (plain `msg`), not the /// first-message pre-key path. -fn connect_warmed_pair(prefix_a: &str, prefix_b: &str) -> Mutex { +/// +/// `drain_b` spawns a background task that keeps draining `b`'s event channel. +/// The sender-only `send_message` bench never reads `b`, so without this its +/// unbounded channel would retain every delivered message and grow across +/// samples, making CodSpeed memory/simulation numbers depend on how many sends +/// ran earlier. The round-trip pair must leave it off — `send_and_receive` +/// consumes `b` itself via `wait_for_text`. +fn connect_warmed_pair(prefix_a: &str, prefix_b: &str, drain_b: bool) -> Mutex { rt().block_on(async { let a = TestClient::connect(prefix_a) .await @@ -87,16 +94,22 @@ fn connect_warmed_pair(prefix_a: &str, prefix_b: &str) -> Mutex { .expect("send warmup message"); b.wait_for_text(&warm, 30).await.expect("receive warmup"); + // Started after warmup so it can't steal the warmup receipt above. + if drain_b { + let rx = b.event_rx.clone(); + tokio::spawn(async move { while rx.recv().await.is_ok() {} }); + } + Mutex::new(Pair { a, b, jid_b }) }) } fn pair_send() -> &'static Mutex { - PAIR_SEND.get_or_init(|| connect_warmed_pair("bench_send_a", "bench_send_b")) + PAIR_SEND.get_or_init(|| connect_warmed_pair("bench_send_a", "bench_send_b", true)) } fn pair_recv() -> &'static Mutex { - PAIR_RECV.get_or_init(|| connect_warmed_pair("bench_recv_a", "bench_recv_b")) + PAIR_RECV.get_or_init(|| connect_warmed_pair("bench_recv_a", "bench_recv_b", false)) } // x20 coverage: divan's `args` runs the bench once per value and reports each From a79a75e22f0f8532ca4da5fb1e03b13b33e78997 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 15:29:07 +0000 Subject: [PATCH 5/8] fix(bench): also drain the sender's receipt channel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The earlier drain only covered client b's incoming messages, but a also accrues an Event::Receipt for every delivered DM in its own unread channel — the same unbounded growth across samples, present on both pairs. Drain a's channel in the background as well (always, since no bench reads it); b stays conditional, as the round-trip pair consumes it via wait_for_text. --- .../bench-integration/benches/integration.rs | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs index 609d30e22..abf4c555e 100644 --- a/tests/bench-integration/benches/integration.rs +++ b/tests/bench-integration/benches/integration.rs @@ -71,12 +71,13 @@ fn unique_body(tag: &str) -> String { /// round-trip, so measured sends exercise steady state (plain `msg`), not the /// first-message pre-key path. /// -/// `drain_b` spawns a background task that keeps draining `b`'s event channel. -/// The sender-only `send_message` bench never reads `b`, so without this its -/// unbounded channel would retain every delivered message and grow across -/// samples, making CodSpeed memory/simulation numbers depend on how many sends -/// ran earlier. The round-trip pair must leave it off — `send_and_receive` -/// consumes `b` itself via `wait_for_text`. +/// No bench reads the clients' event channels, so left alone they grow without +/// bound across divan samples — delivery receipts pile up on `a`, delivered +/// messages on `b` — making CodSpeed memory/simulation numbers depend on how +/// many sends ran earlier. A background task drains each unread channel: `a` +/// always (only receipts land there), `b` only when `drain_b` is set. The +/// round-trip pair leaves `drain_b` off, since `send_and_receive` consumes `b` +/// itself via `wait_for_text`. fn connect_warmed_pair(prefix_a: &str, prefix_b: &str, drain_b: bool) -> Mutex { rt().block_on(async { let a = TestClient::connect(prefix_a) @@ -94,7 +95,12 @@ fn connect_warmed_pair(prefix_a: &str, prefix_b: &str, drain_b: bool) -> Mutex

Date: Thu, 18 Jun 2026 15:41:00 +0000 Subject: [PATCH 6/8] fix(bench): build unique messages in unmeasured setup The measured send closures generated each unique body with format! and text_msg(), charging two harness String allocations per message to the memory-mode results. Move that generation into divan's with_inputs setup (unmeasured) and send the pre-built message by value, so the reported allocations track the send / round-trip path rather than harness overhead. --- .../bench-integration/benches/integration.rs | 79 +++++++++++-------- 1 file changed, 48 insertions(+), 31 deletions(-) diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs index abf4c555e..802a55aaa 100644 --- a/tests/bench-integration/benches/integration.rs +++ b/tests/bench-integration/benches/integration.rs @@ -151,20 +151,27 @@ fn connect_to_ready(bencher: divan::Bencher) { fn send_message(bencher: divan::Bencher, n: u64) { let rt = rt(); let pair = pair_send(); - bencher.bench_local(|| { - rt.block_on(async { - let guard = pair.lock().await; - for _ in 0..n { - let body = unique_body("send"); - guard - .a - .client - .send_message(guard.jid_b.clone(), text_msg(&body)) - .await - .expect("send message"); - } + bencher + // Build the unique messages in unmeasured setup so the body `format!` + // and protobuf `String` allocations aren't charged to the send path. + .with_inputs(|| { + (0..n) + .map(|_| text_msg(&unique_body("send"))) + .collect::>() + }) + .bench_local_values(|msgs| { + rt.block_on(async { + let guard = pair.lock().await; + for msg in msgs { + guard + .a + .client + .send_message(guard.jid_b.clone(), msg) + .await + .expect("send message"); + } + }); }); - }); } /// Full send + receive round-trip on the warmed pair. Runtime and pair are @@ -173,25 +180,35 @@ fn send_message(bencher: divan::Bencher, n: u64) { fn send_and_receive(bencher: divan::Bencher, n: u64) { let rt = rt(); let pair = pair_recv(); - bencher.bench_local(|| { - rt.block_on(async { - let mut guard = pair.lock().await; - for _ in 0..n { - let body = unique_body("recv"); - guard - .a - .client - .send_message(guard.jid_b.clone(), text_msg(&body)) - .await - .expect("send message"); - guard - .b - .wait_for_text(&body, 30) - .await - .expect("receive text"); - } + bencher + // Messages are built in unmeasured setup (see `send_message`); the body + // is kept alongside to match the delivered text in `wait_for_text`. + .with_inputs(|| { + (0..n) + .map(|_| { + let body = unique_body("recv"); + (text_msg(&body), body) + }) + .collect::>() + }) + .bench_local_values(|items| { + rt.block_on(async { + let mut guard = pair.lock().await; + for (msg, body) in items { + guard + .a + .client + .send_message(guard.jid_b.clone(), msg) + .await + .expect("send message"); + guard + .b + .wait_for_text(&body, 30) + .await + .expect("receive text"); + } + }); }); - }); } /// A reconnect cycle (disconnect -> reconnect -> ready). The client is created From f2e490c303221bbe4b0c8fe5111b88700f8d1d16 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 15:55:01 +0000 Subject: [PATCH 7/8] fix(bench): pre-drain the reconnect client outside the measured region MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reconnect_and_wait() drains its event channel via try_recv at the start of the call, inside the measured closure — so a sample paid to clear the late init-IQ responses left buffered by the previous reconnect, making the result depend on prior-sample traffic. Drain the channel in divan's with_inputs setup (unmeasured) so each measured reconnect starts from an empty queue. --- .../bench-integration/benches/integration.rs | 32 +++++++++++++------ 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/tests/bench-integration/benches/integration.rs b/tests/bench-integration/benches/integration.rs index 802a55aaa..fa26d6514 100644 --- a/tests/bench-integration/benches/integration.rs +++ b/tests/bench-integration/benches/integration.rs @@ -215,10 +215,11 @@ fn send_and_receive(bencher: divan::Bencher, n: u64) { /// once outside the measured region; only `reconnect_and_wait` is measured. #[divan::bench] fn reconnect(bencher: divan::Bencher) { + let rt = rt(); let client: &'static Mutex = { static RECONNECT: OnceLock> = OnceLock::new(); RECONNECT.get_or_init(|| { - rt().block_on(async { + rt.block_on(async { Mutex::new( TestClient::connect("bench_reconn") .await @@ -228,14 +229,25 @@ fn reconnect(bencher: divan::Bencher) { }) }; - bencher.bench_local(|| { - rt().block_on(async { - client - .lock() - .await - .reconnect_and_wait() - .await - .expect("reconnect and wait"); + bencher + // Drain events buffered since the previous reconnect (late init-IQ + // responses) in unmeasured setup, so `reconnect_and_wait`'s own + // start-of-call `try_recv` drain doesn't charge the prior sample's + // leftovers to this one. + .with_inputs(|| { + rt.block_on(async { + let guard = client.lock().await; + while guard.event_rx.try_recv().is_ok() {} + }); + }) + .bench_local_values(|()| { + rt.block_on(async { + client + .lock() + .await + .reconnect_and_wait() + .await + .expect("reconnect and wait"); + }); }); - }); } From f220ef720cec8871a6f5402bdec37dd8b719c433 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 18 Jun 2026 16:17:28 +0000 Subject: [PATCH 8/8] ci(codspeed): cap the integration benchmark job runtime The integration benches drive the real client (connect/reconnect handshakes) against the mock server under Valgrind, so the run step is inherently slow and variable. Add a 30-minute job timeout so a genuinely stuck run fails fast with a clear signal instead of hanging CI; a healthy run finishes well under it even on slow runners. --- .github/workflows/codspeed.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/codspeed.yml b/.github/workflows/codspeed.yml index adfe92937..eb3ba6ec4 100644 --- a/.github/workflows/codspeed.yml +++ b/.github/workflows/codspeed.yml @@ -67,6 +67,10 @@ jobs: integration-benchmarks: name: Run CodSpeed integration benchmarks runs-on: ubuntu-latest + # These benches drive the real client (connect/reconnect handshakes) under + # Valgrind, so they are slow; cap the job so a stuck run can't hang CI + # indefinitely. A healthy run finishes well under this even on slow runners. + timeout-minutes: 30 permissions: contents: read id-token: write # for OpenID Connect authentication with CodSpeed