diff --git a/Cargo.lock b/Cargo.lock index 0a8fadc..d1f83af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2283,7 +2283,7 @@ dependencies = [ [[package]] name = "rustqueue-bench" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "clap", @@ -2295,7 +2295,7 @@ dependencies = [ [[package]] name = "rustqueue-console" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "axum", @@ -2323,7 +2323,7 @@ dependencies = [ [[package]] name = "rustqueue-discovery" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "axum", @@ -2346,7 +2346,7 @@ dependencies = [ [[package]] name = "rustqueue-operator" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "axum", @@ -2373,7 +2373,7 @@ dependencies = [ [[package]] name = "rustqueue-protocol" -version = "0.8.1" +version = "0.8.2" dependencies = [ "bytes", "serde", @@ -2383,7 +2383,7 @@ dependencies = [ [[package]] name = "rustqueue-proxy" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "axum", @@ -2404,7 +2404,7 @@ dependencies = [ [[package]] name = "rustqueue-queue" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "bytes", @@ -2426,7 +2426,7 @@ dependencies = [ [[package]] name = "rustqueue-server" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "async-compression", @@ -2462,7 +2462,7 @@ dependencies = [ [[package]] name = "rustqueue-storage" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "crc32c", @@ -2475,14 +2475,14 @@ dependencies = [ [[package]] name = "rustqueue-telemetry" -version = "0.8.1" +version = "0.8.2" dependencies = [ "serde", ] [[package]] name = "rustqueuectl" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "clap", diff --git a/Cargo.toml b/Cargo.toml index 2f08dd4..9e96ce3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ members = [ exclude = ["fuzz"] [workspace.package] -version = "0.8.1" +version = "0.8.2" edition = "2021" license = "Apache-2.0" rust-version = "1.88" diff --git a/Makefile b/Makefile index 7cc485c..4c63bd4 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ .PHONY: test check fmt clippy rustfmt-component clippy-component release-bin image image-from-dist operator-release-bin operator-image console-ui-build console-ui-check kodo-replay kodo-gateway-acceptance \ helm-lint helm-template k8s-acceptance k8s-console-management-acceptance k8s-multi-acceptance up down compat compat-go compat-python \ - fuzz-smoke benchmark release-gate + fuzz-smoke benchmark benchmark-qualify release-gate RUST_IMAGE := rust:1.88-bookworm CARGO_CACHE := rustqueue-cargo-registry @@ -128,5 +128,8 @@ fuzz-smoke: benchmark: ./scripts/benchmark-compare.sh +benchmark-qualify: + ./scripts/benchmark-qualify.sh + release-gate: ./scripts/release-gate.sh diff --git a/README.md b/README.md index a520e34..251a328 100644 --- a/README.md +++ b/README.md @@ -10,14 +10,14 @@ [Architecture](docs/architecture/share-nothing-v7.md) · [Kubernetes operations](docs/operations/kubernetes.md) · [Console operations](docs/operations/console.md) · -[v0.8.1 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.1) +[v0.8.2 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2) -RustQueue 0.8.1 is a Kubernetes-native, NSQ V2-compatible message queue for +RustQueue 0.8.2 is a Kubernetes-native, NSQ V2-compatible message queue for trusted internal networks. It is written in Rust and uses a deliberately simple share-nothing model: each Broker owns one durable RWO PVC, while Kubernetes provides scheduling, rollout and discovery. -> Current release: [v0.8.1](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.1). +> Current release: [v0.8.2](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2). > RustQueue is a production candidate for workloads that accept single-PVC > durability and at-least-once delivery. It does not replicate messages between > Brokers and is not an HA replacement for a replicated log. @@ -43,53 +43,56 @@ messages stored on that Broker are lost. Configure disk pressure protection, monitor the exported metrics, and choose PVC/storage failure policies that fit your workload before deploying to production. -## What's new in 0.8.1 - -- **Truthful end-to-end benchmarks.** `rustqueue-bench` now starts durable, - isolated consumers before publishing, counts unique deliveries and - duplicates, reports publish and receive throughput separately, and fails if - the requested messages do not arrive before the drain deadline. -- **Delivery-state correctness.** Generation tokens reject stale `FIN`, `REQ` - and `TOUCH` commands after redelivery. Initial leases cover buffered writes, - and disconnects no longer release messages while their durable channel - operation is still pending. -- **Crash-safe DLQ and management operations.** Dead-letter transfers are - serialized as one durable transaction, recover without replaying a completed - copy, and respect Topic/Channel fences during concurrent administrative - changes. -- **Cancellation-safe storage.** Payload and recovery-index workers retain - their guards and byte budgets until blocking I/O actually finishes. - Corruption marks storage unhealthy before a response can escape, while - retired Topics are reclaimed after their final reader drains. -- **Bounded control planes.** AUTH responses, compiled authorization regexes, - Broker management bodies, Kodo Stats aggregation and proxy error bodies all - have explicit node-wide limits and timeouts. Invalid semaphore or timer - configurations fail during startup instead of panicking later. - -The patch keeps disk format v7 and the NSQ/Kodo compatibility contract from -0.8.0. See the -[v0.8.1 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.1) -for the complete fix and validation record. - -## Download 0.8.1 +## What's new in 0.8.2 + +- **NSQ-aligned no-Channel durability.** A Topic with no durable Channel now + persists its unrouted start position and normal GC cannot cross it. The first + durable Channel receives every acknowledged publish from that interval, even + when creation happens after the bootstrap window or a Broker restart. +- **Direct-Broker preflight.** Reproducible OrbStack tooling compares the exact + `v0.8.1` tag with one candidate commit using fresh volumes, fixed + 2 vCPU / 2 GiB limits and alternating paired runs. RustQueue 0.8.2 completed + short correctness and regression preflights but does not claim completion of + the optional 60-run performance qualification. +- **Bounded Channel coalescing.** The durable Channel worker now keeps + collecting `FIN` and `REQ` requests throughout its existing bounded 1 ms + window instead of committing at the first transient queue gap. The + 64-request ceiling, channel WAL `fsync` boundary and at-least-once contract + are unchanged. The mechanism and short preflight are not a formal throughput + guarantee. +- **Reliable benchmark shutdown and warmup.** The benchmark preserves a + partially read NSQ frame while closing consumers, and a consumer warmup is + fully drained before measurement. Missing, duplicate or non-drained delivery + remains a hard failure. +- **Explicit regression policy.** Raw write and end-to-end sustainable + throughput fail only when a one-sided paired 95% bootstrap interval is + wholly below `0.95`. Fixed-rate PUB ACK p99 and comparable fixed-rate peak + RSS fail only when the interval is wholly above `1.10`. + +The patch keeps disk format v7 and remains wire-compatible with the NSQ/Kodo +contract from 0.8.1. See the +[v0.8.2 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2) +for the validation boundaries. + +## Download 0.8.2 Every release contains native Linux binaries, the Console UI, source, the Helm Chart and a checksum manifest: | Asset | Contents | | --- | --- | -| `rustqueue-0.8.1-linux-x86_64.tar.gz` | Linux x86_64 binaries, Console UI and example configuration | -| `rustqueue-0.8.1-linux-aarch64.tar.gz` | Linux ARM64 binaries, Console UI and example configuration | -| `rustqueue-0.8.1-source.tar.gz` | Source archive for the tagged commit | -| `rustqueue-0.8.1.tgz` | Helm Chart | -| `SHA256SUMS-0.8.1` | SHA-256 checksums for every downloadable artifact | +| `rustqueue-0.8.2-linux-x86_64.tar.gz` | Linux x86_64 binaries, Console UI and example configuration | +| `rustqueue-0.8.2-linux-aarch64.tar.gz` | Linux ARM64 binaries, Console UI and example configuration | +| `rustqueue-0.8.2-source.tar.gz` | Source archive for the tagged commit | +| `rustqueue-0.8.2.tgz` | Helm Chart | +| `SHA256SUMS-0.8.2` | SHA-256 checksums for every downloadable artifact | ```sh arch="$(uname -m)" -curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.1/rustqueue-0.8.1-linux-${arch}.tar.gz" -curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.1/SHA256SUMS-0.8.1" -sha256sum --check --ignore-missing SHA256SUMS-0.8.1 -tar -xzf "rustqueue-0.8.1-linux-${arch}.tar.gz" +curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.2/rustqueue-0.8.2-linux-${arch}.tar.gz" +curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.2/SHA256SUMS-0.8.2" +sha256sum --check --ignore-missing SHA256SUMS-0.8.2 +tar -xzf "rustqueue-0.8.2-linux-${arch}.tar.gz" ``` ## Architecture @@ -124,11 +127,15 @@ operator -> eligible nodes -> StatefulSet ordinal + retained RWO PVC - Delivery is at least once. A restart redelivers messages without durable FIN. - The broker PVC is the only copy; permanent PVC loss loses its messages. - Topics and channels are broker-local. Lookup consumers union all owners. -- Messages are retained for 90 seconds before a channel exists. This covers one - official Go client default 60-second lookup poll plus its 30% jitter and lets - `SUB` catch a newly selected owner without a normal-path miss. The Kodo - profile forces 180 seconds so one failed lookup request still gets a second - discovery opportunity before data can age out. +- Normal GC retains every message accepted while no durable Channel exists, + across restart and without a bootstrap timeout. The first durable Channel + starts at that persisted unrouted boundary. Deleting the last durable Channel + starts a new boundary at the current Topic tail; ephemeral Channels do not + clear it. +- Once a durable Channel exists, a later Channel can still bootstrap from the + last 90 seconds. This covers one official Go client default 60-second lookup + poll plus its 30% jitter. The Kodo profile forces 180 seconds so one failed + lookup request still gets a second discovery opportunity. - The stable v7 single-message limit is 100 MiB. The conservative defaults remain 20 MiB per message and 64 MiB per MPUB body; the opt-in Kodo profile raises them to 100 MiB and 128 MiB respectively. @@ -223,7 +230,7 @@ kubectl label node worker-1 rustqueue.io/eligible=true helm upgrade --install rustqueue deploy/helm/rustqueue \ --namespace rustqueue --create-namespace \ - --set queue.image=registry.example/rustqueue:0.8.1 \ + --set queue.image=registry.example/rustqueue:0.8.2 \ --set queue.storageClassName=ssd-rwo ``` @@ -460,7 +467,7 @@ test-only direct Pod placement; production anti-affinity is unchanged. A unit fixture covers discovery indexing for 500 brokers. No 500-broker deployment or load test is part of the functional gate. -The v0.8.1 CI/CD workflow publishes a Release only after the non-Kubernetes +The v0.8.2 CI/CD workflow publishes a Release only after the non-Kubernetes release gate, both native Linux builds, packaging and checksum verification succeed. The v0.8.0 Kodo compatibility baseline additionally passed the unmodified Kodo source replay, an exact 104,857,500-byte `PUB`/`DPUB` with one diff --git a/benchmarks/Dockerfile.qualify b/benchmarks/Dockerfile.qualify new file mode 100644 index 0000000..f201273 --- /dev/null +++ b/benchmarks/Dockerfile.qualify @@ -0,0 +1,16 @@ +FROM debian:bookworm-slim AS runtime +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates curl && \ + rm -rf /var/lib/apt/lists/* && \ + mkdir -p /data /var/lib/rustqueue && chown -R 65532:65532 /data /var/lib/rustqueue + +FROM runtime AS broker +COPY rustqueued /usr/local/bin/rustqueued +USER 65532:65532 +WORKDIR /var/lib/rustqueue +EXPOSE 4150 4151 +ENTRYPOINT ["/usr/local/bin/rustqueued"] + +FROM runtime AS tools +COPY rustqueue-bench /usr/local/bin/rustqueue-bench +COPY rustqueue-qualify /usr/local/bin/rustqueue-qualify +USER 65532:65532 diff --git a/benchmarks/qualifications/README.md b/benchmarks/qualifications/README.md new file mode 100644 index 0000000..cbbb3dd --- /dev/null +++ b/benchmarks/qualifications/README.md @@ -0,0 +1,25 @@ +# Broker qualification evidence + +`scripts/benchmark-qualify.sh` is the optional RustQueue Broker performance +qualification. Run it on OrbStack from a committed candidate: + +```sh +make benchmark-qualify +``` + +The default protocol compares the exact `v0.8.1` tag with `HEAD`, uses fresh +Docker volumes, fixes Broker and load-generator containers at 2 vCPU / 2 GiB, +and runs all three cases as 10 alternating pairs. A full run writes the +reviewable artifact to `v0.8.2-orbstack.json`. Consumer cases must drain +completely and have a fixed 1,800-second timeout so the v0.8.1 durable `FIN` +baseline is not rejected merely for exceeding a short operational timeout. + +The RustQueue 0.8.2 release does not make this optional 60-run artifact a +release metadata requirement. Short development preflights can detect hard +correctness failures and obvious regressions, but they do not substantiate a +formal performance claim. + +Per-run benchmark JSON, stderr, RSS samples and the evaluator input stay under +the ignored `benchmarks/results/` directory. Development runs may shorten the +timings or select cases with environment variables, but the script refuses to +publish those results into this directory. diff --git a/console-ui/package.json b/console-ui/package.json index 425c50f..a5b4f4b 100644 --- a/console-ui/package.json +++ b/console-ui/package.json @@ -1,7 +1,7 @@ { "name": "rustqueue-console-ui", "private": true, - "version": "0.8.1", + "version": "0.8.2", "packageManager": "pnpm@11.9.0", "type": "module", "scripts": { diff --git a/crates/bench/src/bin/rustqueue-qualify.rs b/crates/bench/src/bin/rustqueue-qualify.rs new file mode 100644 index 0000000..33c9700 --- /dev/null +++ b/crates/bench/src/bin/rustqueue-qualify.rs @@ -0,0 +1,611 @@ +use anyhow::{bail, Context}; +use clap::Parser; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::BTreeMap; +use std::fs; +use std::path::PathBuf; + +const SCENARIOS: [&str; 3] = ["raw_write", "sustainable", "low_load_latency"]; + +#[derive(Parser, Debug)] +#[command( + name = "rustqueue-qualify", + about = "Evaluate paired RustQueue Broker qualification runs" +)] +struct Args { + #[arg(long)] + input: PathBuf, +} + +#[derive(Debug, Deserialize, Serialize)] +struct QualificationInput { + schema_version: u32, + release: String, + generated_at_utc: String, + baseline: Revision, + candidate: Revision, + environment: Value, + protocol: Protocol, + runs: Vec, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Revision { + revision: String, + commit: String, + image_id: String, + binary_sha256: String, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Protocol { + pairs: usize, + warmup_seconds: u64, + measurement_seconds: u64, + drain_timeout_seconds: u64, + alternating_order: String, + bootstrap_iterations: usize, + bootstrap_seed: u64, + throughput_regression_ratio: f64, + latency_rss_regression_ratio: f64, + scenarios: Value, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +enum Variant { + Baseline, + Candidate, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Run { + #[serde(rename = "case")] + scenario: String, + pair: usize, + sequence: usize, + position_in_pair: usize, + variant: Variant, + commit: String, + benchmark_exit_code: i32, + metrics: Metrics, +} + +#[derive(Debug, Deserialize, Serialize)] +struct Metrics { + messages: u64, + received_unique_messages: u64, + duplicate_messages: u64, + missing_messages: u64, + delivery_verified: bool, + delivery_complete: bool, + drain_timed_out: bool, + final_channel_depth: Option, + final_in_flight: Option, + final_deferred: Option, + publish_messages_per_second: f64, + receive_messages_per_second: Option, + pub_ack_p99_us: u64, + rss_peak_bytes: u64, + broker_profile: BrokerProfile, +} + +#[derive(Debug, Deserialize, Serialize)] +struct BrokerProfile { + publish_group_commits: u64, + publish_group_requests: u64, + publish_group_max_requests: u64, + channel_group_commits: u64, + channel_group_requests: u64, + channel_group_max_requests: u64, + channel_fsync_count: u64, + channel_fsync_sum_seconds: f64, + channel_group_wait_count: u64, + channel_group_wait_sum_seconds: f64, + consumer_fetch_batches: u64, + consumer_fetch_messages: u64, + aggregate_channel_depth: u64, + aggregate_channel_in_flight: u64, + aggregate_channel_deferred: u64, +} + +#[derive(Debug, Serialize)] +struct QualificationEvidence { + schema_version: u32, + release: String, + generated_at_utc: String, + baseline: Revision, + candidate: Revision, + environment: Value, + protocol: Protocol, + runs: Vec, + statistics: Vec, + verdict: Verdict, +} + +#[derive(Debug, Serialize)] +struct Statistic { + #[serde(rename = "case")] + scenario: &'static str, + metric: &'static str, + direction: &'static str, + estimator: &'static str, + candidate_over_baseline: f64, + bootstrap_p05: f64, + bootstrap_p95: f64, + one_sided_95_bound: Bound, + regression_threshold: f64, + regression: bool, + statistically_significant_improvement: bool, + bootstrap_seed: u64, +} + +#[derive(Debug, Serialize)] +struct Bound { + kind: &'static str, + value: f64, +} + +#[derive(Debug, Serialize)] +struct Verdict { + status: &'static str, + hard_failures: Vec, + regressions: Vec, +} + +#[derive(Clone, Copy)] +enum Metric { + PublishThroughput, + ReceiveThroughput, + PubAckP99, + PeakRss, +} + +#[derive(Clone, Copy)] +enum Direction { + HigherIsBetter, + LowerIsBetter, +} + +struct Gate { + scenario: &'static str, + metric: Metric, + direction: Direction, +} + +fn main() -> anyhow::Result<()> { + let args = Args::parse(); + let document = fs::read(&args.input) + .with_context(|| format!("read qualification input {}", args.input.display()))?; + let input: QualificationInput = serde_json::from_slice(&document) + .with_context(|| format!("parse qualification input {}", args.input.display()))?; + let evidence = evaluate(input)?; + let passed = evidence.verdict.status == "pass"; + println!("{}", serde_json::to_string_pretty(&evidence)?); + if !passed { + bail!("Broker qualification failed"); + } + Ok(()) +} + +fn evaluate(input: QualificationInput) -> anyhow::Result { + validate_input(&input)?; + let enabled_scenarios = scenario_names(&input.protocol)?; + let gates = [ + Gate { + scenario: "raw_write", + metric: Metric::PublishThroughput, + direction: Direction::HigherIsBetter, + }, + Gate { + scenario: "sustainable", + metric: Metric::ReceiveThroughput, + direction: Direction::HigherIsBetter, + }, + Gate { + scenario: "low_load_latency", + metric: Metric::PubAckP99, + direction: Direction::LowerIsBetter, + }, + Gate { + scenario: "low_load_latency", + metric: Metric::PeakRss, + direction: Direction::LowerIsBetter, + }, + ]; + let mut statistics = Vec::with_capacity(gates.len()); + for gate in gates + .into_iter() + .filter(|gate| enabled_scenarios.contains(&gate.scenario)) + { + statistics.push(evaluate_gate(&input, &gate)?); + } + let regressions = statistics + .iter() + .filter(|statistic| statistic.regression) + .map(|statistic| format!("{}:{}", statistic.scenario, statistic.metric)) + .collect::>(); + let status = if regressions.is_empty() { + "pass" + } else { + "fail" + }; + + Ok(QualificationEvidence { + schema_version: input.schema_version, + release: input.release, + generated_at_utc: input.generated_at_utc, + baseline: input.baseline, + candidate: input.candidate, + environment: input.environment, + protocol: input.protocol, + runs: input.runs, + statistics, + verdict: Verdict { + status, + hard_failures: Vec::new(), + regressions, + }, + }) +} + +fn validate_input(input: &QualificationInput) -> anyhow::Result<()> { + if input.schema_version != 1 { + bail!( + "unsupported qualification schema version {}", + input.schema_version + ); + } + if input.protocol.pairs == 0 { + bail!("qualification requires at least one pair"); + } + if input.protocol.warmup_seconds == 0 + || input.protocol.measurement_seconds == 0 + || input.protocol.drain_timeout_seconds == 0 + { + bail!("qualification timing values must be greater than zero"); + } + if input.protocol.bootstrap_iterations == 0 { + bail!("bootstrap-iterations must be greater than zero"); + } + if input.protocol.alternating_order != "AB_then_BA" { + bail!("alternating-order must be AB_then_BA"); + } + if !(0.0..1.0).contains(&input.protocol.throughput_regression_ratio) { + bail!("throughput-regression-ratio must be between zero and one"); + } + if input.protocol.latency_rss_regression_ratio <= 1.0 { + bail!("latency-rss-regression-ratio must be greater than one"); + } + + let scenarios = scenario_names(&input.protocol)?; + let expected_runs = scenarios.len() * input.protocol.pairs * 2; + if input.runs.len() != expected_runs { + bail!( + "qualification has {} runs, expected {expected_runs}", + input.runs.len() + ); + } + for scenario in scenarios { + for pair in 1..=input.protocol.pairs { + let mut paired = input + .runs + .iter() + .filter(|run| run.scenario == scenario && run.pair == pair) + .collect::>(); + paired.sort_by_key(|run| run.position_in_pair); + if paired.len() != 2 { + bail!("{scenario} pair {pair} must contain exactly two runs"); + } + let expected = if pair % 2 == 1 { + [Variant::Baseline, Variant::Candidate] + } else { + [Variant::Candidate, Variant::Baseline] + }; + for (index, run) in paired.into_iter().enumerate() { + if run.position_in_pair != index + 1 || run.variant != expected[index] { + bail!("{scenario} pair {pair} violates the AB_then_BA order contract"); + } + validate_run(input, run)?; + } + } + } + Ok(()) +} + +fn scenario_names(protocol: &Protocol) -> anyhow::Result> { + let scenarios = protocol + .scenarios + .as_array() + .context("protocol scenarios must be an array")?; + if scenarios.is_empty() { + bail!("protocol must enable at least one qualification case"); + } + let mut names = Vec::with_capacity(scenarios.len()); + for scenario in scenarios { + let name = scenario + .get("name") + .and_then(Value::as_str) + .context("each protocol scenario must have a string name")?; + if !SCENARIOS.contains(&name) { + bail!("unknown protocol case {name}"); + } + if names.contains(&name) { + bail!("protocol case {name} is duplicated"); + } + names.push(name); + } + Ok(names) +} + +fn validate_run(input: &QualificationInput, run: &Run) -> anyhow::Result<()> { + if !SCENARIOS.contains(&run.scenario.as_str()) { + bail!("unknown qualification case {}", run.scenario); + } + let expected_commit = match run.variant { + Variant::Baseline => &input.baseline.commit, + Variant::Candidate => &input.candidate.commit, + }; + if &run.commit != expected_commit { + bail!( + "{} pair {} has commit {}, expected {}", + run.scenario, + run.pair, + run.commit, + expected_commit + ); + } + if run.benchmark_exit_code != 0 { + bail!( + "{} pair {} {:?} benchmark exited with {}", + run.scenario, + run.pair, + run.variant, + run.benchmark_exit_code + ); + } + let metrics = &run.metrics; + if metrics.messages == 0 + || !positive_finite(metrics.publish_messages_per_second) + || metrics.pub_ack_p99_us == 0 + || metrics.rss_peak_bytes == 0 + { + bail!( + "{} pair {} {:?} contains an invalid core metric", + run.scenario, + run.pair, + run.variant + ); + } + if run.scenario == "raw_write" { + if metrics.delivery_verified { + bail!("raw_write must not start consumers"); + } + } else { + if !metrics.delivery_verified + || !metrics.delivery_complete + || metrics.drain_timed_out + || metrics.received_unique_messages != metrics.messages + || metrics.missing_messages != 0 + || metrics.duplicate_messages != 0 + || metrics.final_channel_depth != Some(0) + || metrics.final_in_flight != Some(0) + || metrics.final_deferred != Some(0) + || metrics.broker_profile.aggregate_channel_depth != 0 + || metrics.broker_profile.aggregate_channel_in_flight != 0 + || metrics.broker_profile.aggregate_channel_deferred != 0 + { + bail!( + "{} pair {} {:?} failed the complete-delivery contract", + run.scenario, + run.pair, + run.variant + ); + } + if !metrics + .receive_messages_per_second + .is_some_and(positive_finite) + { + bail!( + "{} pair {} {:?} has invalid receive throughput", + run.scenario, + run.pair, + run.variant + ); + } + } + Ok(()) +} + +fn positive_finite(value: f64) -> bool { + value.is_finite() && value > 0.0 +} + +fn evaluate_gate(input: &QualificationInput, gate: &Gate) -> anyhow::Result { + let mut pairs = BTreeMap::, Option)>::new(); + for run in input + .runs + .iter() + .filter(|run| run.scenario == gate.scenario) + { + let value = metric_value(&run.metrics, gate.metric)?; + let values = pairs.entry(run.pair).or_default(); + match run.variant { + Variant::Baseline => values.0 = Some(value), + Variant::Candidate => values.1 = Some(value), + } + } + let ratios = pairs + .into_iter() + .map(|(pair, (baseline, candidate))| { + let baseline = baseline.context(format!("pair {pair} is missing baseline"))?; + let candidate = candidate.context(format!("pair {pair} is missing candidate"))?; + if !positive_finite(baseline) || !positive_finite(candidate) { + bail!("pair {pair} contains a non-positive metric"); + } + Ok(candidate / baseline) + }) + .collect::>>()?; + let seed = input.protocol.bootstrap_seed ^ stable_hash(gate.scenario, metric_name(gate.metric)); + let mut distribution = + bootstrap_geometric_means(&ratios, input.protocol.bootstrap_iterations, seed); + distribution.sort_by(f64::total_cmp); + let point = geometric_mean(&ratios); + let p05 = quantile(&distribution, 0.05); + let p95 = quantile(&distribution, 0.95); + let (bound, threshold, regression, improvement) = match gate.direction { + Direction::HigherIsBetter => ( + Bound { + kind: "upper", + value: p95, + }, + input.protocol.throughput_regression_ratio, + p95 < input.protocol.throughput_regression_ratio, + p05 > 1.0, + ), + Direction::LowerIsBetter => ( + Bound { + kind: "lower", + value: p05, + }, + input.protocol.latency_rss_regression_ratio, + p05 > input.protocol.latency_rss_regression_ratio, + p95 < 1.0, + ), + }; + Ok(Statistic { + scenario: gate.scenario, + metric: metric_name(gate.metric), + direction: match gate.direction { + Direction::HigherIsBetter => "higher_is_better", + Direction::LowerIsBetter => "lower_is_better", + }, + estimator: "geometric_mean_of_paired_ratios", + candidate_over_baseline: point, + bootstrap_p05: p05, + bootstrap_p95: p95, + one_sided_95_bound: bound, + regression_threshold: threshold, + regression, + statistically_significant_improvement: improvement, + bootstrap_seed: seed, + }) +} + +fn metric_value(metrics: &Metrics, metric: Metric) -> anyhow::Result { + let value = match metric { + Metric::PublishThroughput => metrics.publish_messages_per_second, + Metric::ReceiveThroughput => metrics + .receive_messages_per_second + .context("receive throughput is missing")?, + Metric::PubAckP99 => metrics.pub_ack_p99_us as f64, + Metric::PeakRss => metrics.rss_peak_bytes as f64, + }; + Ok(value) +} + +fn metric_name(metric: Metric) -> &'static str { + match metric { + Metric::PublishThroughput => "publish_messages_per_second", + Metric::ReceiveThroughput => "receive_messages_per_second", + Metric::PubAckP99 => "pub_ack_p99_us", + Metric::PeakRss => "rss_peak_bytes", + } +} + +fn geometric_mean(values: &[f64]) -> f64 { + (values.iter().map(|value| value.ln()).sum::() / values.len() as f64).exp() +} + +fn bootstrap_geometric_means(values: &[f64], iterations: usize, seed: u64) -> Vec { + let mut rng = SplitMix64::new(seed); + let mut samples = Vec::with_capacity(iterations); + for _ in 0..iterations { + let mut log_sum = 0.0; + for _ in 0..values.len() { + log_sum += values[rng.index(values.len())].ln(); + } + samples.push((log_sum / values.len() as f64).exp()); + } + samples +} + +fn quantile(sorted: &[f64], probability: f64) -> f64 { + let position = probability * (sorted.len() - 1) as f64; + let lower = position.floor() as usize; + let upper = position.ceil() as usize; + if lower == upper { + sorted[lower] + } else { + let fraction = position - lower as f64; + sorted[lower] * (1.0 - fraction) + sorted[upper] * fraction + } +} + +fn stable_hash(scenario: &str, metric: &str) -> u64 { + let mut hash = 0xcbf2_9ce4_8422_2325u64; + for byte in scenario.bytes().chain([b':']).chain(metric.bytes()) { + hash ^= u64::from(byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + hash +} + +struct SplitMix64 { + state: u64, +} + +impl SplitMix64 { + fn new(seed: u64) -> Self { + Self { state: seed } + } + + fn next(&mut self) -> u64 { + self.state = self.state.wrapping_add(0x9e37_79b9_7f4a_7c15); + let mut value = self.state; + value = (value ^ (value >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9); + value = (value ^ (value >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb); + value ^ (value >> 31) + } + + fn index(&mut self, upper: usize) -> usize { + ((u128::from(self.next()) * upper as u128) >> 64) as usize + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn paired_bootstrap_is_deterministic() { + let values = [0.97, 1.01, 1.03, 0.99]; + assert_eq!( + bootstrap_geometric_means(&values, 100, 802), + bootstrap_geometric_means(&values, 100, 802) + ); + } + + #[test] + fn clear_regressions_cross_the_one_sided_bounds() { + let mut throughput = bootstrap_geometric_means(&[0.90; 10], 1_000, 802); + throughput.sort_by(f64::total_cmp); + assert!(quantile(&throughput, 0.95) < 0.95); + + let mut latency = bootstrap_geometric_means(&[1.20; 10], 1_000, 802); + latency.sort_by(f64::total_cmp); + assert!(quantile(&latency, 0.05) > 1.10); + } + + #[test] + fn noise_straddling_a_threshold_does_not_fail_the_gate() { + let mut values = bootstrap_geometric_means( + &[0.80, 0.90, 0.94, 0.97, 1.0, 1.01, 1.02, 1.03, 1.04, 1.05], + 10_000, + 802, + ); + values.sort_by(f64::total_cmp); + assert!(quantile(&values, 0.95) >= 0.95); + } +} diff --git a/crates/bench/src/consumer.rs b/crates/bench/src/consumer.rs index 6570312..adf3695 100644 --- a/crates/bench/src/consumer.rs +++ b/crates/bench/src/consumer.rs @@ -3,7 +3,7 @@ use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::sync::{mpsc, watch, Notify}; use tokio::task::JoinSet; @@ -241,16 +241,17 @@ async fn consume_worker( mut stop: watch::Receiver, progress: Arc, ) -> anyhow::Result<()> { - let mut stream = TcpStream::connect(&address) + let stream = TcpStream::connect(&address) .await .with_context(|| format!("connect consumer to {address}"))?; stream.set_nodelay(true)?; - stream.write_all(b" V2").await?; - stream + let (mut reader, mut writer) = stream.into_split(); + writer.write_all(b" V2").await?; + writer .write_all(format!("SUB {topic} {channel}\n").as_bytes()) .await?; - wait_for_ok(&mut stream).await?; - stream + wait_for_ok(&mut reader, &mut writer).await?; + writer .write_all(format!("RDY {RDY_COUNT}\n").as_bytes()) .await?; ready.send(()).await.ok(); @@ -258,60 +259,76 @@ async fn consume_worker( let mut remaining_rdy = RDY_COUNT; loop { - tokio::select! { - changed = stop.changed() => { - if changed.is_err() || *stop.borrow() { - return close_consumer(&mut stream).await; + let (stopping, frame) = { + let frame = read_frame(&mut reader); + tokio::pin!(frame); + tokio::select! { + changed = stop.changed() => { + if changed.is_err() || *stop.borrow() { + writer.write_all(b"CLS\n").await?; + writer.flush().await?; + (true, frame.await) + } else { + (false, frame.await) + } } + frame = &mut frame => (false, frame), } - frame = read_frame(&mut stream) => { - let (frame_type, response) = match frame { - Ok(frame) => frame, - Err(_) if *stop.borrow() => return Ok(()), - Err(error) => return Err(error), - }; - match frame_type { - 0 if response == b"_heartbeat_" => stream.write_all(b"NOP\n").await?, - 2 if response.len() >= 26 => { - let id: [u8; 16] = response[10..26] - .try_into() - .expect("message frame ID length was checked"); - let numeric_id = parse_message_id(&id)?; - stream.write_all(b"FIN ").await?; - stream.write_all(&id).await?; - stream.write_all(b"\n").await?; - remaining_rdy = remaining_rdy.saturating_sub(1); - if remaining_rdy <= RDY_REFILL_AT { - stream - .write_all(format!("RDY {RDY_COUNT}\n").as_bytes()) - .await?; - remaining_rdy = RDY_COUNT; - } - progress.observe(numeric_id); - } - 1 => anyhow::bail!( - "consumer error: {}", - String::from_utf8_lossy(&response) - ), - _ => {} + }; + let (frame_type, response) = match frame { + Ok(frame) => frame, + Err(_) if stopping => return Ok(()), + Err(error) => return Err(error), + }; + if stopping { + return close_consumer(&mut reader, &mut writer, Some((frame_type, response))).await; + } + match frame_type { + 0 if response == b"_heartbeat_" => writer.write_all(b"NOP\n").await?, + 2 if response.len() >= 26 => { + let id: [u8; 16] = response[10..26] + .try_into() + .expect("message frame ID length was checked"); + let numeric_id = parse_message_id(&id)?; + writer.write_all(b"FIN ").await?; + writer.write_all(&id).await?; + writer.write_all(b"\n").await?; + remaining_rdy = remaining_rdy.saturating_sub(1); + if remaining_rdy <= RDY_REFILL_AT { + writer + .write_all(format!("RDY {RDY_COUNT}\n").as_bytes()) + .await?; + remaining_rdy = RDY_COUNT; } + progress.observe(numeric_id); } + 1 => anyhow::bail!("consumer error: {}", String::from_utf8_lossy(&response)), + _ => {} } } } -async fn close_consumer(stream: &mut TcpStream) -> anyhow::Result<()> { - stream.write_all(b"CLS\n").await?; - stream.flush().await?; +async fn close_consumer( + reader: &mut R, + writer: &mut W, + mut pending: Option<(i32, Vec)>, +) -> anyhow::Result<()> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ loop { - let (frame_type, response) = read_frame(stream).await?; + let (frame_type, response) = match pending.take() { + Some(frame) => frame, + None => read_frame(reader).await?, + }; match frame_type { 0 if response == b"CLOSE_WAIT" => return Ok(()), - 0 if response == b"_heartbeat_" => stream.write_all(b"NOP\n").await?, + 0 if response == b"_heartbeat_" => writer.write_all(b"NOP\n").await?, 2 if response.len() >= 26 => { - stream.write_all(b"FIN ").await?; - stream.write_all(&response[10..26]).await?; - stream.write_all(b"\n").await?; + writer.write_all(b"FIN ").await?; + writer.write_all(&response[10..26]).await?; + writer.write_all(b"\n").await?; } 1 => anyhow::bail!( "consumer close failed: {}", @@ -327,11 +344,15 @@ fn parse_message_id(id: &[u8; 16]) -> anyhow::Result { u64::from_str_radix(id, 16).context("message ID is not hexadecimal") } -async fn wait_for_ok(stream: &mut TcpStream) -> anyhow::Result<()> { +async fn wait_for_ok(reader: &mut R, writer: &mut W) -> anyhow::Result<()> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ loop { - let (frame_type, response) = read_frame(stream).await?; + let (frame_type, response) = read_frame(reader).await?; if frame_type == 0 && response == b"_heartbeat_" { - stream.write_all(b"NOP\n").await?; + writer.write_all(b"NOP\n").await?; } else if frame_type == 0 && response == b"OK" { return Ok(()); } else if frame_type == 1 { @@ -340,20 +361,26 @@ async fn wait_for_ok(stream: &mut TcpStream) -> anyhow::Result<()> { } } -async fn read_frame(stream: &mut TcpStream) -> anyhow::Result<(i32, Vec)> { - let size = stream.read_u32().await? as usize; +async fn read_frame(reader: &mut R) -> anyhow::Result<(i32, Vec)> +where + R: AsyncRead + Unpin, +{ + let size = reader.read_u32().await? as usize; if !(4..=16 * 1024 * 1024).contains(&size) { anyhow::bail!("server returned invalid frame size {size}"); } - let frame_type = stream.read_i32().await?; + let frame_type = reader.read_i32().await?; let mut response = vec![0; size - 4]; - stream.read_exact(&mut response).await?; + reader.read_exact(&mut response).await?; Ok((frame_type, response)) } #[cfg(test)] mod tests { use super::*; + use tokio::io::{AsyncBufReadExt, BufReader}; + use tokio::net::TcpListener; + use tokio::sync::oneshot; #[test] fn progress_counts_unique_deliveries_and_duplicates() { @@ -392,4 +419,92 @@ mod tests { assert!(!result.complete); assert_eq!(result.snapshot.unique, 0); } + + #[tokio::test] + async fn stopping_during_a_partial_frame_preserves_the_frame_boundary() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let (partial_tx, partial_rx) = oneshot::channel(); + let (continue_tx, continue_rx) = oneshot::channel(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let (reader, mut writer) = socket.into_split(); + let mut reader = BufReader::new(reader); + let mut magic = [0; 4]; + reader.read_exact(&mut magic).await.unwrap(); + assert_eq!(&magic, b" V2"); + + let mut command = String::new(); + reader.read_line(&mut command).await.unwrap(); + assert!(command.starts_with("SUB ")); + writer.write_all(&test_frame(0, b"OK")).await.unwrap(); + writer.flush().await.unwrap(); + + command.clear(); + reader.read_line(&mut command).await.unwrap(); + assert!(command.starts_with("RDY ")); + + let message = test_message_frame(1, &[b'x'; 1024]); + writer.write_all(&message[..20]).await.unwrap(); + writer.flush().await.unwrap(); + partial_tx.send(()).unwrap(); + continue_rx.await.unwrap(); + writer.write_all(&message[20..]).await.unwrap(); + writer.flush().await.unwrap(); + + let mut saw_close = false; + let mut saw_finish = false; + while !saw_close || !saw_finish { + command.clear(); + reader.read_line(&mut command).await.unwrap(); + saw_close |= command == "CLS\n"; + saw_finish |= command == "FIN 0000000000000001\n"; + } + writer + .write_all(&test_frame(0, b"CLOSE_WAIT")) + .await + .unwrap(); + writer.flush().await.unwrap(); + }); + + let (stop, stop_rx) = watch::channel(false); + let (ready, mut ready_rx) = mpsc::channel(1); + let client = tokio::spawn(consume_worker( + address.to_string(), + "events".into(), + "workers".into(), + ready, + stop_rx, + Arc::new(ConsumerProgress::default()), + )); + ready_rx.recv().await.unwrap(); + partial_rx.await.unwrap(); + stop.send(true).unwrap(); + tokio::task::yield_now().await; + continue_tx.send(()).unwrap(); + + tokio::time::timeout(Duration::from_secs(2), client) + .await + .unwrap() + .unwrap() + .unwrap(); + server.await.unwrap(); + } + + fn test_frame(frame_type: i32, body: &[u8]) -> Vec { + let mut frame = Vec::with_capacity(body.len() + 8); + frame.extend_from_slice(&((body.len() + 4) as u32).to_be_bytes()); + frame.extend_from_slice(&frame_type.to_be_bytes()); + frame.extend_from_slice(body); + frame + } + + fn test_message_frame(id: u64, body: &[u8]) -> Vec { + let mut message = Vec::with_capacity(body.len() + 26); + message.extend_from_slice(&0i64.to_be_bytes()); + message.extend_from_slice(&1u16.to_be_bytes()); + message.extend_from_slice(format!("{id:016x}").as_bytes()); + message.extend_from_slice(body); + test_frame(2, &message) + } } diff --git a/crates/bench/src/main.rs b/crates/bench/src/main.rs index f7a964a..ba863cc 100644 --- a/crates/bench/src/main.rs +++ b/crates/bench/src/main.rs @@ -117,21 +117,22 @@ async fn main() -> anyhow::Result<()> { } if args.warmup_seconds > 0 { let warmup_topic = isolated_topic(&args.topic, "warmup"); - let mut warmup_group = if args.consumers == 0 { - None - } else { - let progress = Arc::new(ConsumerProgress::default()); - let channel = benchmark_name("warmup-channel"); - Some( - start_consumers( - &args.address, - &warmup_topic, - &channel, - args.consumers, - progress, + let warmup_progress = (args.consumers > 0).then(|| Arc::new(ConsumerProgress::default())); + let mut warmup_group = match &warmup_progress { + None => None, + Some(progress) => { + let channel = benchmark_name("warmup-channel"); + Some( + start_consumers( + &args.address, + &warmup_topic, + &channel, + args.consumers, + Arc::clone(progress), + ) + .await?, ) - .await?, - ) + } }; let warmup = Arc::new(Mutex::new(Histogram::::new_with_max(60_000_000, 3)?)); let warmup_result = if let Some(group) = warmup_group.as_mut() { @@ -158,8 +159,27 @@ async fn main() -> anyhow::Result<()> { .await }; if let Some(group) = warmup_group.take() { + let published = match warmup_result { + Ok(published) => published, + Err(error) => { + let _ = group.stop().await; + return Err(error); + } + }; + let progress = warmup_progress + .as_ref() + .expect("warmup consumers have delivery progress"); + let delivery = progress + .wait_for(published, Duration::from_secs(args.drain_timeout_seconds)) + .await; let stop_result = group.stop().await; - warmup_result?; + require_complete_delivery( + true, + delivery.complete, + delivery.snapshot.unique, + delivery.snapshot.duplicates(), + published, + )?; stop_result?; } else { warmup_result?; @@ -351,6 +371,7 @@ async fn main() -> anyhow::Result<()> { report.delivery_verified, report.delivery_complete, report.received_unique_messages, + report.duplicate_messages, report.messages, )?; Ok(()) @@ -360,8 +381,14 @@ fn require_complete_delivery( verified: bool, complete: bool, received: u64, + duplicates: u64, published: u64, ) -> anyhow::Result<()> { + if verified && duplicates > 0 { + anyhow::bail!( + "delivery verification failed: observed {duplicates} unexpected duplicate deliveries" + ); + } if verified && !complete { anyhow::bail!( "delivery verification failed: received {received} unique messages out of {published} before the drain timeout" @@ -404,8 +431,13 @@ mod tests { #[test] fn incomplete_verified_delivery_fails_the_benchmark() { - assert!(require_complete_delivery(true, false, 99, 100).is_err()); - assert!(require_complete_delivery(true, true, 100, 100).is_ok()); - assert!(require_complete_delivery(false, false, 0, 100).is_ok()); + assert!(require_complete_delivery(true, false, 99, 0, 100).is_err()); + assert!(require_complete_delivery(true, true, 100, 0, 100).is_ok()); + assert!(require_complete_delivery(false, false, 0, 0, 100).is_ok()); + } + + #[test] + fn unexpected_duplicate_delivery_fails_the_benchmark() { + assert!(require_complete_delivery(true, true, 100, 1, 100).is_err()); } } diff --git a/crates/queue/src/broker/channel_commit.rs b/crates/queue/src/broker/channel_commit.rs index 2fa4e47..2b2ec8e 100644 --- a/crates/queue/src/broker/channel_commit.rs +++ b/crates/queue/src/broker/channel_commit.rs @@ -9,7 +9,7 @@ use std::time::{Duration, Instant}; use tokio::sync::{mpsc, oneshot}; const QUEUE_CAPACITY: usize = 1024; -const MAX_GROUP_REQUESTS: usize = 64; +const MAX_GROUP_REQUESTS: usize = 256; const COALESCE_DELAY: Duration = Duration::from_millis(1); pub(super) enum ChannelOperation { @@ -297,6 +297,7 @@ async fn collect_group( first: ChannelRequest, receiver: &mut mpsc::Receiver, ) -> Vec { + let deadline = tokio::time::Instant::now() + COALESCE_DELAY; let mut requests = vec![first]; loop { if requests.len() >= MAX_GROUP_REQUESTS { @@ -305,13 +306,12 @@ async fn collect_group( let next = match receiver.try_recv() { Ok(request) => Some(request), Err(mpsc::error::TryRecvError::Disconnected) => None, - Err(mpsc::error::TryRecvError::Empty) if requests.len() == 1 => { - tokio::time::timeout(COALESCE_DELAY, receiver.recv()) + Err(mpsc::error::TryRecvError::Empty) => { + tokio::time::timeout_at(deadline, receiver.recv()) .await .ok() .flatten() } - Err(mpsc::error::TryRecvError::Empty) => None, }; let Some(next) = next else { return requests; @@ -352,7 +352,7 @@ mod tests { #[tokio::test] async fn group_size_is_bounded() { - let (sender, mut receiver) = mpsc::channel(128); + let (sender, mut receiver) = mpsc::channel(MAX_GROUP_REQUESTS + 1); for _ in 0..=MAX_GROUP_REQUESTS { sender.send(request()).await.unwrap(); } @@ -363,4 +363,19 @@ mod tests { ); assert!(receiver.try_recv().is_ok()); } + + #[tokio::test] + async fn group_collects_requests_arriving_anytime_within_the_window() { + let (sender, mut receiver) = mpsc::channel(8); + sender.send(request()).await.unwrap(); + sender.send(request()).await.unwrap(); + let late_sender = tokio::spawn(async move { + tokio::task::yield_now().await; + sender.send(request()).await.unwrap(); + }); + + let first = receiver.recv().await.unwrap(); + assert_eq!(collect_group(first, &mut receiver).await.len(), 3); + late_sender.await.unwrap(); + } } diff --git a/crates/queue/src/broker_tests.rs b/crates/queue/src/broker_tests.rs index 68caeaf..d43118a 100644 --- a/crates/queue/src/broker_tests.rs +++ b/crates/queue/src/broker_tests.rs @@ -1167,10 +1167,21 @@ async fn bounded_gc_rotates_across_topics() { }) .unwrap(); for topic in ["alpha", "beta", "gamma"] { - broker + broker.create_channel(topic, "workers").await.unwrap(); + let id = broker .publish(topic, vec![b"expired".to_vec()], Duration::ZERO) .await - .unwrap(); + .unwrap()[0]; + assert_eq!( + broker + .next_message(topic, "workers", None) + .await + .unwrap() + .unwrap() + .id, + id + ); + broker.finish(topic, "workers", id).await.unwrap(); } assert_eq!(broker.compact_some(1).await.unwrap(), 1); diff --git a/crates/queue/src/channel.rs b/crates/queue/src/channel.rs index 8637b7c..b9ba2e0 100644 --- a/crates/queue/src/channel.rs +++ b/crates/queue/src/channel.rs @@ -54,6 +54,7 @@ pub(crate) struct ChannelCheckpoint { } struct InFlight { + id: u64, deadline: Instant, token: u64, } @@ -309,6 +310,7 @@ impl ChannelState { self.in_flight.insert( position, InFlight { + id, deadline: Instant::now() + timeout, token, }, @@ -484,12 +486,11 @@ impl ChannelState { } fn remove_in_flight(&mut self, position: u64) -> bool { - let removed = self.in_flight.remove(&position).is_some(); - if removed { - self.in_flight_ids - .retain(|_, candidate| *candidate != position); - } - removed + let Some(delivery) = self.in_flight.remove(&position) else { + return false; + }; + self.in_flight_ids.remove(&delivery.id); + true } } @@ -541,6 +542,21 @@ mod tests { assert_eq!(attempts, 1); } + #[test] + fn removing_one_delivery_preserves_other_in_flight_id_lookups() { + let mut channel = ChannelState::new("workers".into(), 0, false, 16); + channel.reserve(1, 10, Duration::from_secs(30)); + channel.reserve(2, 20, Duration::from_secs(30)); + + channel.apply(&ChannelCommand::Finish { + position: 1, + message_id: 10, + }); + + assert_eq!(channel.in_flight_position(10), None); + assert_eq!(channel.in_flight_position(20), Some(2)); + } + #[test] fn in_flight_expiry_uses_a_monotonic_deadline() { let mut channel = ChannelState::new("workers".into(), 0, false, 16); diff --git a/crates/queue/src/channel_store.rs b/crates/queue/src/channel_store.rs index 226ef0e..2d2c993 100644 --- a/crates/queue/src/channel_store.rs +++ b/crates/queue/src/channel_store.rs @@ -11,8 +11,8 @@ const CHECKPOINT_MAGIC: &[u8; 4] = b"RCC7"; const HEADER_LEN: usize = 12; const MAX_COMMAND_BYTES: usize = 1024 * 1024; const MAX_CHECKPOINT_BYTES: u64 = 64 * 1024 * 1024; -const MAX_RECOVERY_COMMANDS: usize = 2_048; -const CHECKPOINT_INTERVAL: usize = 1024; +const CHECKPOINT_INTERVAL: usize = 8 * 1024; +const MAX_RECOVERY_COMMANDS: usize = CHECKPOINT_INTERVAL * 2; pub(crate) struct ChannelStore { directory: PathBuf, @@ -546,6 +546,29 @@ mod tests { )); } + #[test] + fn recovery_accepts_a_commit_group_past_the_checkpoint_interval() { + let root = tempdir().unwrap(); + let state = ChannelState::new("workers".into(), 0, false, MAX_RECOVERY_COMMANDS); + let mut store = ChannelStore::create(root.path(), &state).unwrap(); + for position in 1..=(CHECKPOINT_INTERVAL as u64 + 1) { + store + .append_buffered(&ChannelCommand::Finish { + position, + message_id: position, + }) + .unwrap(); + } + store.sync().unwrap(); + drop(store); + + let checkpoint = root + .path() + .join(format!("{}.checkpoint", hex::encode("workers"))); + let (recovered, _) = ChannelStore::open(&checkpoint, MAX_RECOVERY_COMMANDS).unwrap(); + assert_eq!(recovered.ack_floor_position, CHECKPOINT_INTERVAL as u64 + 1); + } + #[test] fn refuses_a_checksum_corrupt_complete_wal_tail() { let root = tempdir().unwrap(); diff --git a/crates/queue/src/metadata.rs b/crates/queue/src/metadata.rs index e87aca5..f5f65f6 100644 --- a/crates/queue/src/metadata.rs +++ b/crates/queue/src/metadata.rs @@ -22,6 +22,8 @@ pub(crate) struct TopicManifest { #[serde(default)] pub deleted: bool, pub next_position: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub unrouted_from_position: Option, } pub(crate) fn load_optional(path: &Path) -> io::Result> { diff --git a/crates/queue/src/topic.rs b/crates/queue/src/topic.rs index 14af659..130ee7c 100644 --- a/crates/queue/src/topic.rs +++ b/crates/queue/src/topic.rs @@ -67,6 +67,7 @@ impl TopicHandle { MAX_CHANNELS_PER_TOPIC ))); } + topic.reconcile_unrouted_boundary()?; Ok(Arc::new(Self { state: Mutex::new(topic), wake, @@ -89,6 +90,7 @@ impl TopicHandle { paused: false, deleted: false, next_position: 1, + unrouted_from_position: Some(1), }; store_atomic(&directory.join("manifest"), &manifest)?; let log = SegmentLog::open_with_feature_level( @@ -137,6 +139,14 @@ impl Topic { "topic manifest is not an active v7 topic".into(), )); } + if manifest + .unrouted_from_position + .is_some_and(|position| position == 0 || position > manifest.next_position) + { + return Err(BrokerError::InvalidRecord( + "topic manifest has an invalid unrouted position".into(), + )); + } let expected_directory = hex::encode(manifest.name.as_bytes()); if validate_name(&manifest.name).is_err() || directory.file_name().and_then(|name| name.to_str()) @@ -244,6 +254,11 @@ impl Topic { if self.manifest.deleted { return Err(BrokerError::TopicNotFound); } + if !self.has_durable_channels() && self.manifest.unrouted_from_position.is_none() { + return Err(BrokerError::InvalidRecord( + "topic without a durable channel is missing its retention boundary".into(), + )); + } let timestamp_ns = self .messages .last_timestamp_ns() @@ -323,6 +338,54 @@ impl Topic { pub fn last_position(&self) -> u64 { self.manifest.next_position.saturating_sub(1) } + + fn has_durable_channels(&self) -> bool { + self.channels + .values() + .any(|channel| !channel.state.ephemeral) + } + + fn earliest_retained_position(&self) -> u64 { + self.messages + .first_position() + .unwrap_or(self.manifest.next_position) + } + + fn unrouted_start_position(&self) -> Result { + self.manifest.unrouted_from_position.ok_or_else(|| { + BrokerError::InvalidRecord( + "topic without a durable channel is missing its retention boundary".into(), + ) + }) + } + + fn set_unrouted_from_position(&mut self, position: Option) -> Result<(), BrokerError> { + if self.manifest.unrouted_from_position == position { + return Ok(()); + } + let mut manifest = self.manifest.clone(); + manifest.unrouted_from_position = position; + store_atomic(&self.manifest_path, &manifest)?; + self.manifest = manifest; + Ok(()) + } + + fn reconcile_unrouted_boundary(&mut self) -> Result<(), BrokerError> { + let position = if self.has_durable_channels() { + None + } else { + let earliest = self.earliest_retained_position(); + Some( + self.manifest + .unrouted_from_position + .unwrap_or(earliest) + .max(earliest) + .min(self.manifest.next_position), + ) + }; + self.set_unrouted_from_position(position) + } + pub fn set_paused(&mut self, paused: bool) -> Result<(), BrokerError> { self.manifest.paused = paused; store_atomic(&self.manifest_path, &self.manifest)?; @@ -344,11 +407,14 @@ impl Topic { return Err(BrokerError::ChannelLimit); } let ephemeral = name.ends_with("#ephemeral"); + let first_durable = !ephemeral && !self.has_durable_channels(); let barrier = if ephemeral { // NSQ ephemeral channels only observe messages published while at // least one consumer keeps the channel alive. Re-creating an // ephemeral channel must therefore start at the current tail. self.last_position() + } else if first_durable { + self.unrouted_start_position()?.saturating_sub(1) } else { let cutoff = now_ns() .saturating_sub(bootstrap_retention.as_nanos().min(i64::MAX as u128) as i64); @@ -365,6 +431,11 @@ impl Topic { &state, )?) }; + if first_durable { + // The checkpoint reaches disk before the Topic releases this + // boundary, so recovery always has either the Channel or the hold. + self.set_unrouted_from_position(None)?; + } self.channels.insert( name.into(), ChannelRuntime { @@ -377,6 +448,23 @@ impl Topic { } pub fn delete_channel(&mut self, name: &str) -> Result<(), BrokerError> { + let state = &self + .channels + .get(name) + .ok_or(BrokerError::ChannelNotFound)? + .state; + let deleting_last_durable = !state.ephemeral + && self + .channels + .values() + .filter(|channel| !channel.state.ephemeral) + .count() + == 1; + if deleting_last_durable { + // Establish the new hold before removing the final checkpoint. + // After a crash, either the Channel recovers or this boundary does. + self.set_unrouted_from_position(Some(self.manifest.next_position))?; + } let channel = self .channels .remove(name) diff --git a/crates/queue/src/topic/index.rs b/crates/queue/src/topic/index.rs index 355fcf4..c822655 100644 --- a/crates/queue/src/topic/index.rs +++ b/crates/queue/src/topic/index.rs @@ -227,6 +227,13 @@ impl MessageIndex { .or_else(|| self.sealed.back().map(|segment| segment.last_position)) } + pub(crate) fn first_position(&self) -> Option { + self.sealed + .front() + .map(|segment| segment.first_position) + .or_else(|| self.active.front().map(|message| message.position)) + } + pub(crate) fn last_timestamp_ns(&self) -> Option { self.active .back() diff --git a/crates/queue/src/topic/maintenance.rs b/crates/queue/src/topic/maintenance.rs index ddf405f..b42f2d8 100644 --- a/crates/queue/src/topic/maintenance.rs +++ b/crates/queue/src/topic/maintenance.rs @@ -16,6 +16,7 @@ impl Topic { pub fn empty_topic(&mut self) -> Result<(), BrokerError> { let through = self.last_position(); + let has_durable_channels = self.has_durable_channels(); let channels: Vec<_> = self.channels.keys().cloned().collect(); for channel in channels { self.persist_channel( @@ -25,6 +26,9 @@ impl Topic { }, )?; } + if !has_durable_channels { + self.set_unrouted_from_position(Some(self.manifest.next_position))?; + } Ok(()) } @@ -110,6 +114,13 @@ impl Topic { }; store_atomic(&self.manifest_path, &self.manifest)?; eviction::write_intent(audit_directory, &report)?; + if !self.has_durable_channels() { + let retained_from = self + .unrouted_start_position()? + .max(through_position.saturating_add(1)) + .min(self.manifest.next_position); + self.set_unrouted_from_position(Some(retained_from))?; + } let channels: Vec<_> = self.channels.keys().cloned().collect(); for channel in channels { self.persist_channel(&channel, ChannelCommand::Evict { through_position })?; @@ -143,8 +154,11 @@ impl Topic { .values() .filter(|channel| !channel.state.ephemeral) .map(|channel| channel.state.ack_floor_position.saturating_add(1)) - .min() - .unwrap_or(self.manifest.next_position); + .min(); + let channel_from = match channel_from { + Some(position) => position, + None => self.unrouted_start_position()?, + }; let in_flight_from = self .channels .values() diff --git a/crates/queue/tests/recovery.rs b/crates/queue/tests/recovery.rs index fa64e7a..6738d3d 100644 --- a/crates/queue/tests/recovery.rs +++ b/crates/queue/tests/recovery.rs @@ -143,6 +143,107 @@ async fn channels_fan_out_and_ephemeral_channels_do_not_survive_restart() { ); } +#[tokio::test] +async fn messages_without_a_durable_channel_survive_gc_and_a_legacy_restart() { + let root = tempfile::tempdir().unwrap(); + let mut cfg = config(root.path()); + cfg.bootstrap_retention = Duration::ZERO; + let broker = Broker::open(cfg.clone()).unwrap(); + let ids = broker + .publish( + "events", + vec![vec![1; 80], vec![2; 80], vec![3; 80]], + Duration::ZERO, + ) + .await + .unwrap(); + broker.flush().await.unwrap(); + + assert_eq!(broker.compact().await.unwrap(), 0); + assert_eq!(broker.stats().topics[0].message_count, ids.len() as u64); + drop(broker); + + let manifest_path = root + .path() + .join("topics") + .join(hex::encode("events")) + .join("manifest"); + let mut manifest: serde_json::Value = + serde_json::from_slice(&std::fs::read(&manifest_path).unwrap()).unwrap(); + assert!(manifest + .as_object_mut() + .unwrap() + .remove("unrouted_from_position") + .is_some()); + std::fs::write(&manifest_path, serde_json::to_vec(&manifest).unwrap()).unwrap(); + + let broker = Broker::open(cfg).unwrap(); + assert_eq!(broker.compact().await.unwrap(), 0); + broker.create_channel("events", "workers").await.unwrap(); + let delivered = broker + .fetch_batch("events", "workers", 8, 1024, Duration::ZERO, None) + .await + .unwrap(); + assert_eq!( + delivered + .iter() + .map(|message| message.id) + .collect::>(), + ids + ); +} + +#[tokio::test] +async fn messages_after_the_last_durable_channel_is_deleted_survive_gc_and_restart() { + let root = tempfile::tempdir().unwrap(); + let mut cfg = config(root.path()); + cfg.bootstrap_retention = Duration::ZERO; + let broker = Broker::open(cfg.clone()).unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + let old = broker + .publish("events", vec![vec![1; 160]], Duration::ZERO) + .await + .unwrap()[0]; + let delivered = broker + .next_message("events", "workers", None) + .await + .unwrap() + .unwrap(); + assert_eq!(delivered.id, old); + broker.finish("events", "workers", old).await.unwrap(); + broker.delete_channel("events", "workers").await.unwrap(); + + let fresh = broker + .publish( + "events", + vec![vec![2; 80], vec![3; 80], vec![4; 80]], + Duration::ZERO, + ) + .await + .unwrap(); + broker.flush().await.unwrap(); + drop(broker); + + let broker = Broker::open(cfg).unwrap(); + assert!(broker.compact().await.unwrap() > 0); + assert_eq!(broker.stats().topics[0].message_count, fresh.len() as u64); + broker + .create_channel("events", "replacement") + .await + .unwrap(); + let delivered = broker + .fetch_batch("events", "replacement", 8, 1024, Duration::ZERO, None) + .await + .unwrap(); + assert_eq!( + delivered + .iter() + .map(|message| message.id) + .collect::>(), + fresh + ); +} + #[tokio::test] async fn recreated_ephemeral_channel_starts_at_the_current_tail() { let root = tempfile::tempdir().unwrap(); diff --git a/deploy/helm/rustqueue/Chart.yaml b/deploy/helm/rustqueue/Chart.yaml index 29d31e7..5750021 100644 --- a/deploy/helm/rustqueue/Chart.yaml +++ b/deploy/helm/rustqueue/Chart.yaml @@ -2,8 +2,8 @@ apiVersion: v2 name: rustqueue description: Kubernetes-native share-nothing NSQ-compatible message queue type: application -version: 0.8.1 -appVersion: "0.8.1" +version: 0.8.2 +appVersion: "0.8.2" kubeVersion: ">=1.28.0-0" keywords: - messaging diff --git a/deploy/helm/rustqueue/values.yaml b/deploy/helm/rustqueue/values.yaml index a63a45a..5250874 100644 --- a/deploy/helm/rustqueue/values.yaml +++ b/deploy/helm/rustqueue/values.yaml @@ -5,7 +5,7 @@ operator: replicas: 2 image: repository: rustqueue-operator - tag: "0.8.1" + tag: "0.8.2" pullPolicy: IfNotPresent log: rustqueue_operator=info resources: @@ -20,7 +20,7 @@ operator: queue: create: true name: rustqueue - image: rustqueue:0.8.1 + image: rustqueue:0.8.2 imagePullPolicy: IfNotPresent minBrokers: 1 maxBrokers: 500 diff --git a/docs/architecture/share-nothing-v7.md b/docs/architecture/share-nothing-v7.md index 3096be6..0fc378e 100644 --- a/docs/architecture/share-nothing-v7.md +++ b/docs/architecture/share-nothing-v7.md @@ -1,7 +1,7 @@ # RustQueue format v7 share-nothing architecture Status: accepted implementation contract -Target release: 0.8.1 +Target release: 0.8.2 Data format: v7, clean directories only ## 1. Goal @@ -57,25 +57,31 @@ There is no broker-to-broker message path and no cluster consensus path. - A direct `SUB` sees only the addressed broker. Lookup-based consumers connect to all discovered owners of the topic. -### 2.3 Bootstrap retention +### 2.3 Unrouted and bootstrap retention -Every broker retains each topic message for at least 90 seconds even when no -channel exists. The Kodo compatibility profile requires 180 seconds so the -default Go consumer can survive one failed 60-second Lookupd poll, including -its 30% initial jitter. When a local channel is created, its initial cursor -starts at the oldest message still inside this bootstrap window. +Every broker persists the earliest Topic position accepted while no durable +Channel exists. Normal GC cannot cross that unrouted boundary, regardless of +the bootstrap window or a Broker restart. The first durable Channel starts +immediately before the boundary and can therefore consume every successfully +published message from that zero-Channel interval. Deleting the last durable +Channel starts a new boundary at the current Topic tail. Ephemeral Channels do +not satisfy or clear this durability boundary. -This deliberately prefers duplicates over misses during the normal discovery -window. The guarantee is bounded: +Once at least one durable Channel exists, a later Channel starts at the oldest +message still inside the configured bootstrap window. The default is 90 +seconds. The Kodo compatibility profile requires 180 seconds so the default Go +consumer can survive one failed 60-second Lookupd poll, including its 30% +initial jitter. -- If discovery and `SUB` complete within 90 seconds, messages accepted by a new - fallback owner remain consumable by the newly created local channel. -- In the Kodo profile, the equivalent bound is 180 seconds and includes a - second default lookup poll after one failed request. -- A newly created channel may receive a small amount of data published before - its `SUB`. -- If all consumers are absent for longer than the bootstrap window and a topic - appears on a previously unused broker, complete replay is not guaranteed. +This separates two guarantees: + +- A Topic with no durable Channel retains all acknowledged publishes until its + first durable Channel is created. +- Additional Channels prefer duplicates over misses during the normal + discovery window, but do not receive unbounded Topic history. +- Explicit Topic empty/delete and opt-in protective eviction remain + intentional destructive operations; protective eviction writes an audit + record before advancing the unrouted boundary. ### 2.4 Routing and scale @@ -280,7 +286,8 @@ discard is a separate, audited administrative action. A complete segment may be deleted only when it is older than: - every durable local channel retention cursor; -- the 90-second bootstrap floor; +- the persisted unrouted boundary when no durable Channel exists; +- the configured bootstrap time floor; - every active reader reference; - every pending DLQ outbox reference. @@ -397,9 +404,11 @@ The implementation is complete only when all of the following pass: GC delete boundaries reopen successfully without losing acknowledged live records. 3. Tail corruption truncates safely; middle corruption isolates the topic. -4. Discovery indexes a new fallback owner within 5 seconds. An official Go - consumer using the unchanged 60-second poll and 30% jitter then subscribes - before the 90-second bootstrap floor expires; its ledger has `missing=0`. +4. A Topic published before any durable Channel survives the configured + bootstrap interval, normal GC, and Broker restart. A durable Channel created + afterward receives the complete acknowledged ledger with `missing=0`. + Discovery still indexes a new fallback owner within 5 seconds, and an + official Go consumer uses the unchanged 60-second poll and 30% jitter. 5. Official Go and Python NSQ clients pass direct, lookup, compression, TLS, AUTH, MPUB, DPUB, REQ, TOUCH, fan-out, sampling, and ephemeral cases. 6. Broker restart reattaches its PVC and resumes its local backlog. diff --git a/docs/operations/kubernetes.md b/docs/operations/kubernetes.md index 0c0ed85..0666854 100644 --- a/docs/operations/kubernetes.md +++ b/docs/operations/kubernetes.md @@ -76,7 +76,7 @@ Canary approval is optional: ```sh helm upgrade rustqueue deploy/helm/rustqueue \ --namespace rustqueue \ - --set queue.image=registry.example/rustqueue:0.8.1 \ + --set queue.image=registry.example/rustqueue:0.8.2 \ --set queue.rollout.requireCanaryApproval=true rustqueuectl -n rustqueue rollout approve @@ -88,7 +88,7 @@ Useful controls: rustqueuectl -n rustqueue rollout pause rustqueuectl -n rustqueue rollout resume rustqueuectl -n rustqueue rollout retry -rustqueuectl -n rustqueue rollout rollback registry.example/rustqueue:0.8.1 +rustqueuectl -n rustqueue rollout rollback registry.example/rustqueue:0.8.2 rustqueuectl -n rustqueue rollout forward ``` diff --git a/docs/releases/v0.8.2.md b/docs/releases/v0.8.2.md new file mode 100644 index 0000000..8d97d2b --- /dev/null +++ b/docs/releases/v0.8.2.md @@ -0,0 +1,91 @@ +# RustQueue 0.8.2 + +RustQueue 0.8.2 aligns no-Channel retention with NSQ, hardens the direct Broker +data path, and corrects premature Channel group commits. It completed short +OrbStack correctness and regression preflights, but the optional 60-run +performance qualification was not run and this release makes no formal +throughput claim. + +## No-Channel durability + +- A Topic now persists the earliest position published while it has no durable + Channel. Normal GC cannot advance past this boundary, even after the 90-second + bootstrap window or a Broker restart. +- The first durable Channel consumes from that boundary. Deleting the last + durable Channel starts a new boundary at the current Topic tail, so later + acknowledged publishes remain available to its replacement. +- Existing v7 Topic manifests without the new optional field are reconciled + from the earliest retained position during open. The disk format remains v7 + and no migration command is required. +- Later Channels created while another durable Channel already exists retain + the existing 90-second bootstrap behavior. + +## Broker data path + +- A Channel commit worker now keeps collecting `FIN` and `REQ` requests until + the end of its existing 1 ms coalescing window. A transiently empty receiver + no longer forces an early durable commit. +- The group remains bounded at 64 requests and the queue remains bounded at + 1,024 requests. Every successful operation still crosses the affected + Channel WAL `fsync` before the client can observe completion. +- Publish acknowledgement semantics, at-least-once delivery and on-disk format + v7 are unchanged. + +## Benchmark correctness + +- Consumer shutdown preserves an NSQ frame that was only partly read when + `CLS` begins. The benchmark can finish that frame without restarting the + parser in the message body. +- A warmup with consumers is fully delivered and durably acknowledged before + the measured Topic starts. Warmup backlog can no longer inflate the measured + drain time or Broker RSS. +- A nonzero benchmark exit, missing delivery, unexpected duplicate, incomplete + drain, or nonzero final Channel depth, in-flight count or deferred count is a + hard failure before statistics are evaluated. + +## Benchmark protocol and release decision + +The optional full qualification uses OrbStack only, with both Broker and load +generator fixed at 2 vCPU and 2 GiB. It compares the exact `v0.8.1` tag with +one exact candidate commit using a fresh volume for every run. + +Each case runs as 10 alternating pairs (`A -> B`, then `B -> A`), with a +30-second warmup and 120-second measurement. Consumer cases have a 1,800-second +upper bound for the mandatory complete drain; a timeout remains a hard failure: + +- raw write ceiling: one Topic, no consumer, 16 producers, 1 KiB messages and + `MPUB` batch 64; +- sustainable throughput: one Topic and one Channel, 16 producers, 16 + consumers, 1 KiB messages and `MPUB` batch 64, followed by a complete drain; +- low-load guard: one Topic and one Channel, 16 producers, 16 consumers, 1 KiB + `PUB`, and a fixed total arrival rate of 100 messages/s. + +The evaluator uses a deterministic paired bootstrap of the geometric mean +candidate/baseline ratio. A throughput regression is reported only when the +one-sided 95% upper bound is below `0.95`. Low-load PUB ACK p99 or peak Broker +RSS regresses only when the one-sided 95% lower bound is above `1.10`. +Saturation RSS remains in every run record but is descriptive because a faster +candidate completes more work in the same measurement window. + +Raw JSON, stderr and RSS samples remain under the ignored +`benchmarks/results/` directory. By release decision, RustQueue 0.8.2 does not +include the optional compact 60-run artifact. Release readiness instead +requires the non-Kubernetes production gate, native x86_64 and aarch64 builds, +package checksums, and the existing compatibility tests. + +## Compatibility + +- The on-disk format remains v7; no data migration is required from 0.8.1. +- NSQ V2 commands, TLS/mTLS, AUTH, compression and the opt-in Kodo profile keep + their 0.8.1 compatibility contract. +- Proxy, Operator, Console and Kodo routing are outside the performance change. + +## Assets + +- `rustqueue-0.8.2-linux-x86_64.tar.gz`: Linux x86_64 binaries, Console UI and + example configuration +- `rustqueue-0.8.2-linux-aarch64.tar.gz`: Linux ARM64 binaries, Console UI and + example configuration +- `rustqueue-0.8.2-source.tar.gz`: source archive for the tagged commit +- `rustqueue-0.8.2.tgz`: Helm Chart +- `SHA256SUMS-0.8.2`: SHA-256 checksums for all assets diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index c02910f..9318bb9 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -2017,7 +2017,7 @@ dependencies = [ [[package]] name = "rustqueue-discovery" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "axum", @@ -2055,7 +2055,7 @@ dependencies = [ [[package]] name = "rustqueue-protocol" -version = "0.8.1" +version = "0.8.2" dependencies = [ "bytes", "serde", @@ -2065,7 +2065,7 @@ dependencies = [ [[package]] name = "rustqueue-proxy" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "axum", @@ -2085,7 +2085,7 @@ dependencies = [ [[package]] name = "rustqueue-queue" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "bytes", @@ -2105,7 +2105,7 @@ dependencies = [ [[package]] name = "rustqueue-storage" -version = "0.8.1" +version = "0.8.2" dependencies = [ "anyhow", "crc32c", @@ -2117,7 +2117,7 @@ dependencies = [ [[package]] name = "rustqueue-telemetry" -version = "0.8.1" +version = "0.8.2" dependencies = [ "serde", ] diff --git a/scripts/acceptance-k8s.sh b/scripts/acceptance-k8s.sh index 055b452..7840f91 100755 --- a/scripts/acceptance-k8s.sh +++ b/scripts/acceptance-k8s.sh @@ -544,6 +544,8 @@ done kubectl -n "$NAMESPACE" patch rustqueue "$QUEUE" --type=merge \ -p "{\"spec\":{\"image\":\"$BROKER_IMAGE_A\"}}" >/dev/null wait_queue_ready 300 +kubectl -n "$NAMESPACE" rollout status deployment/"$QUEUE-discovery" --timeout=180s +kubectl -n "$NAMESPACE" rollout status daemonset/"$QUEUE-proxy" --timeout=180s run_curl proxy-health-after-recovery -fsS "http://$QUEUE-proxy:4151/v1/health" echo "OrbStack Kubernetes share-nothing v7 acceptance passed" diff --git a/scripts/acceptance-multi-broker-k8s.sh b/scripts/acceptance-multi-broker-k8s.sh index 384ec56..a9f23d0 100755 --- a/scripts/acceptance-multi-broker-k8s.sh +++ b/scripts/acceptance-multi-broker-k8s.sh @@ -343,9 +343,9 @@ require jq } if [[ "$BUILD_IMAGES" == "1" ]]; then - BUILD_VERSION=0.8.1-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image + BUILD_VERSION=0.8.2-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image docker tag rustqueue:dev "$BROKER_IMAGE_A" - BUILD_VERSION=0.8.1-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist + BUILD_VERSION=0.8.2-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist docker tag rustqueue:dev "$BROKER_IMAGE_B" [[ "$(docker image inspect "$BROKER_IMAGE_A" -f '{{.Id}}')" != \ "$(docker image inspect "$BROKER_IMAGE_B" -f '{{.Id}}')" ]] || { diff --git a/scripts/benchmark-qualify.sh b/scripts/benchmark-qualify.sh new file mode 100755 index 0000000..fe8a21d --- /dev/null +++ b/scripts/benchmark-qualify.sh @@ -0,0 +1,689 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +RELEASE="${RELEASE:-0.8.2}" +BASELINE_REF="${BASELINE_REF:-v0.8.1}" +CANDIDATE_REF="${CANDIDATE_REF:-HEAD}" +PAIRS="${PAIRS:-10}" +WARMUP_SECONDS="${WARMUP_SECONDS:-30}" +MEASUREMENT_SECONDS="${MEASUREMENT_SECONDS:-120}" +BOOTSTRAP_ITERATIONS="${BOOTSTRAP_ITERATIONS:-100000}" +BOOTSTRAP_SEED="${BOOTSTRAP_SEED:-802}" +DRAIN_TIMEOUT_SECONDS="${DRAIN_TIMEOUT_SECONDS:-1800}" +CASES="${CASES:-raw_write sustainable low_load_latency}" +QUALIFICATION_DEV="${QUALIFICATION_DEV:-0}" +KEEP_IMAGES="${KEEP_IMAGES:-0}" +RESULT_ROOT="$ROOT/benchmarks/results" +EVIDENCE_OUTPUT="${EVIDENCE_OUTPUT:-$ROOT/benchmarks/qualifications/v0.8.2-orbstack.json}" +RUN_ID="$(date -u +%Y%m%dT%H%M%SZ)-$$" +RUN_DIR="$RESULT_ROOT/qualification-$RUN_ID" +RUNS_FILE="$RUN_DIR/runs.ndjson" +INPUT_FILE="$RUN_DIR/input.json" +EVIDENCE_FILE="$RUN_DIR/evidence.json" +BUILD_ROOT="$(mktemp -d "${TMPDIR:-/tmp}/rustqueue-qualify.XXXXXX")" +BASELINE_SOURCE="$BUILD_ROOT/baseline" +CANDIDATE_SOURCE="$BUILD_ROOT/candidate" +TARGET_ROOT="$RESULT_ROOT/.qualification-target" +BASELINE_RUNTIME="$BUILD_ROOT/baseline-runtime" +CANDIDATE_RUNTIME="$BUILD_ROOT/candidate-runtime" +TOOLS_RUNTIME="$BUILD_ROOT/tools-runtime" +PREFIX="rq-q-$$" +NETWORK="$PREFIX-net" +BASELINE_IMAGE="rustqueue:qualify-baseline-$$" +CANDIDATE_IMAGE="rustqueue:qualify-candidate-$$" +TOOLS_IMAGE="rustqueue:qualify-tools-$$" +ACTIVE_BROKER="" +ACTIVE_VOLUME="" +SAMPLER_PID="" +SEQUENCE=0 + +die() { + printf 'benchmark qualification: %s\n' "$*" >&2 + exit 1 +} + +require_positive_integer() { + local name=$1 value=$2 + [[ "$value" =~ ^[1-9][0-9]*$ ]] || die "$name must be a positive integer" +} + +cleanup_run() { + if [[ -n "$SAMPLER_PID" ]]; then + kill "$SAMPLER_PID" >/dev/null 2>&1 || true + wait "$SAMPLER_PID" >/dev/null 2>&1 || true + SAMPLER_PID="" + fi + if [[ -n "$ACTIVE_BROKER" ]]; then + docker rm -f "$ACTIVE_BROKER" >/dev/null 2>&1 || true + ACTIVE_BROKER="" + fi + if [[ -n "$ACTIVE_VOLUME" ]]; then + docker volume rm "$ACTIVE_VOLUME" >/dev/null 2>&1 || true + ACTIVE_VOLUME="" + fi +} + +cleanup() { + cleanup_run + docker network rm "$NETWORK" >/dev/null 2>&1 || true + if [[ "$KEEP_IMAGES" != 1 ]]; then + docker image rm "$BASELINE_IMAGE" "$CANDIDATE_IMAGE" "$TOOLS_IMAGE" \ + >/dev/null 2>&1 || true + fi + rm -rf "$BUILD_ROOT" +} +trap cleanup EXIT INT TERM + +for command in docker git jq tar awk; do + command -v "$command" >/dev/null 2>&1 || die "required command is missing: $command" +done +require_positive_integer PAIRS "$PAIRS" +require_positive_integer WARMUP_SECONDS "$WARMUP_SECONDS" +require_positive_integer MEASUREMENT_SECONDS "$MEASUREMENT_SECONDS" +require_positive_integer BOOTSTRAP_ITERATIONS "$BOOTSTRAP_ITERATIONS" +require_positive_integer DRAIN_TIMEOUT_SECONDS "$DRAIN_TIMEOUT_SECONDS" + +CASE_COUNT=0 +seen_cases=" " +for scenario in $CASES; do + case "$scenario" in + raw_write|sustainable|low_load_latency) ;; + *) die "unknown case $scenario" ;; + esac + [[ "$seen_cases" != *" $scenario "* ]] || die "case $scenario is duplicated" + seen_cases="$seen_cases$scenario " + CASE_COUNT=$((CASE_COUNT + 1)) +done +[[ "$CASE_COUNT" -gt 0 ]] || die "CASES must enable at least one case" + +docker_context="$(docker context show)" +docker_os="$(docker info --format '{{.OperatingSystem}}')" +if [[ "$docker_context" != "orbstack" && "$docker_os" != *OrbStack* ]]; then + die "Docker must use OrbStack (context=$docker_context, os=$docker_os)" +fi + +baseline_commit="$(git -C "$ROOT" rev-parse --verify "$BASELINE_REF^{commit}")" +candidate_commit="$(git -C "$ROOT" rev-parse --verify "$CANDIDATE_REF^{commit}")" +tag_commit="$(git -C "$ROOT" rev-parse --verify "v0.8.1^{commit}")" +BASELINE_TARGET="$TARGET_ROOT/$baseline_commit" +CANDIDATE_TARGET="$TARGET_ROOT/$candidate_commit" +[[ "$baseline_commit" == "$tag_commit" ]] || + die "baseline must resolve to the exact v0.8.1 tag commit" + +case "$EVIDENCE_OUTPUT" in + "$ROOT/benchmarks/qualifications/"*) + [[ "$QUALIFICATION_DEV" == 0 ]] || + die "development runs cannot publish committed qualification evidence" + [[ "$PAIRS" == 10 && "$WARMUP_SECONDS" == 30 && "$MEASUREMENT_SECONDS" == 120 ]] || + die "committed evidence requires 10 pairs, 30s warmup and 120s measurement" + [[ "$DRAIN_TIMEOUT_SECONDS" == 1800 ]] || + die "committed evidence requires an 1800s complete-drain timeout" + [[ "$CASES" == "raw_write sustainable low_load_latency" ]] || + die "committed evidence requires all three qualification cases" + ;; +esac + +mkdir -p \ + "$RUN_DIR" \ + "$BASELINE_SOURCE" \ + "$CANDIDATE_SOURCE" \ + "$BASELINE_TARGET" \ + "$CANDIDATE_TARGET" \ + "$BASELINE_RUNTIME" \ + "$CANDIDATE_RUNTIME" \ + "$TOOLS_RUNTIME" +: >"$RUNS_FILE" +git -C "$ROOT" archive "$baseline_commit" | tar -x -C "$BASELINE_SOURCE" +git -C "$ROOT" archive "$candidate_commit" | tar -x -C "$CANDIDATE_SOURCE" + +read_workspace_version() { + awk ' + /^\[workspace\.package\]$/ { workspace = 1; next } + /^\[/ { workspace = 0 } + workspace && $1 == "version" { + gsub(/"/, "", $3) + print $3 + exit + } + ' "$1/Cargo.toml" +} + +baseline_version="$(read_workspace_version "$BASELINE_SOURCE")" +candidate_version="$(read_workspace_version "$CANDIDATE_SOURCE")" +[[ "$baseline_version" == 0.8.1 ]] || + die "baseline workspace version is $baseline_version, expected 0.8.1" +if [[ "$QUALIFICATION_DEV" == 0 && "$candidate_version" != "$RELEASE" ]]; then + die "candidate workspace version is $candidate_version, expected $RELEASE" +fi + +candidate_context="$CANDIDATE_SOURCE" +tool_source="$candidate_commit" +if [[ "$QUALIFICATION_DEV" == 1 ]]; then + candidate_context="$ROOT" + candidate_version="$(read_workspace_version "$ROOT")" + CANDIDATE_TARGET="$BUILD_ROOT/candidate-target" + mkdir -p "$CANDIDATE_TARGET" + tool_source="worktree" +fi + +build_release() { + local source=$1 target=$2 version=$3 + shift 3 + local -a binary_args=() + for binary in "$@"; do + binary_args+=(--bin "$binary") + done + docker run --rm \ + -e RUSTUP_TOOLCHAIN=1.88.0 \ + -e CARGO_INCREMENTAL=0 \ + -e CARGO_TARGET_DIR=/target \ + -e "RUSTQUEUE_BUILD_VERSION=$version" \ + -v "$source:/work" \ + -v "$target:/target" \ + -v rustqueue-cargo-registry:/usr/local/cargo/registry \ + -v rustqueue-rustup:/usr/local/rustup \ + -w /work \ + rust:1.88-bookworm \ + cargo build --locked --release "${binary_args[@]}" +} + +printf 'Compiling exact baseline Broker %s (%s)\n' "$BASELINE_REF" "$baseline_commit" +build_release "$BASELINE_SOURCE" "$BASELINE_TARGET" "$baseline_version" rustqueued +cp "$BASELINE_TARGET/release/rustqueued" "$BASELINE_RUNTIME/rustqueued" +if command -v shasum >/dev/null 2>&1; then + baseline_binary_sha256="$( + shasum -a 256 "$BASELINE_RUNTIME/rustqueued" | awk '{print $1}' + )" +else + baseline_binary_sha256="$( + sha256sum "$BASELINE_RUNTIME/rustqueued" | awk '{print $1}' + )" +fi +docker build --target broker \ + -f "$ROOT/benchmarks/Dockerfile.qualify" \ + -t "$BASELINE_IMAGE" "$BASELINE_RUNTIME" + +printf 'Compiling candidate Broker %s (%s)\n' "$CANDIDATE_REF" "$candidate_commit" +build_release \ + "$candidate_context" \ + "$CANDIDATE_TARGET" \ + "$candidate_version" \ + rustqueued \ + rustqueue-bench \ + rustqueue-qualify +cp "$CANDIDATE_TARGET/release/rustqueued" "$CANDIDATE_RUNTIME/rustqueued" +cp "$CANDIDATE_TARGET/release/rustqueue-bench" "$TOOLS_RUNTIME/rustqueue-bench" +cp "$CANDIDATE_TARGET/release/rustqueue-qualify" "$TOOLS_RUNTIME/rustqueue-qualify" +if command -v shasum >/dev/null 2>&1; then + candidate_binary_sha256="$( + shasum -a 256 "$CANDIDATE_RUNTIME/rustqueued" | awk '{print $1}' + )" +else + candidate_binary_sha256="$( + sha256sum "$CANDIDATE_RUNTIME/rustqueued" | awk '{print $1}' + )" +fi +docker build --target broker \ + -f "$ROOT/benchmarks/Dockerfile.qualify" \ + -t "$CANDIDATE_IMAGE" "$CANDIDATE_RUNTIME" +printf 'Building common load generator and qualification evaluator\n' +docker build --target tools \ + -f "$ROOT/benchmarks/Dockerfile.qualify" \ + -t "$TOOLS_IMAGE" "$TOOLS_RUNTIME" + +docker network create "$NETWORK" >/dev/null + +sample_rss() { + local output=$1 broker=$2 + { + printf 'timestamp_utc\trss_bytes\n' + while docker inspect "$broker" >/dev/null 2>&1; do + rss="$( + docker exec "$broker" awk '/VmRSS:/ { print $2 * 1024 }' /proc/1/status \ + 2>/dev/null || true + )" + if [[ "$rss" =~ ^[0-9]+$ ]]; then + printf '%s\t%s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$rss" + fi + sleep 1 + done + } >"$output" +} + +wait_for_broker() { + local broker=$1 + for _ in $(seq 1 60); do + if docker exec "$broker" curl -fsS http://127.0.0.1:4151/v1/health \ + >/dev/null 2>&1; then + return + fi + sleep 1 + done + docker logs "$broker" >&2 || true + die "Broker $broker did not become healthy" +} + +run_variant() { + local scenario=$1 pair=$2 position=$3 variant=$4 + local image commit producers consumers batch rate + case "$variant" in + baseline) + image="$BASELINE_IMAGE" + commit="$baseline_commit" + ;; + candidate) + image="$CANDIDATE_IMAGE" + commit="$candidate_commit" + ;; + *) + die "unknown variant $variant" + ;; + esac + case "$scenario" in + raw_write) + producers=16 + consumers=0 + batch=64 + rate="" + ;; + sustainable) + producers=16 + consumers=16 + batch=64 + rate="" + ;; + low_load_latency) + producers=16 + consumers=16 + batch=1 + rate=100 + ;; + *) + die "unknown case $scenario" + ;; + esac + + SEQUENCE=$((SEQUENCE + 1)) + local label="${scenario}-p${pair}-${variant}" + local report="$RUN_DIR/$label.json" + local error_log="$RUN_DIR/$label.stderr.log" + local rss_file="$RUN_DIR/$label.rss.tsv" + ACTIVE_BROKER="$PREFIX-b$SEQUENCE" + ACTIVE_VOLUME="$PREFIX-v$SEQUENCE" + + docker volume create "$ACTIVE_VOLUME" >/dev/null + docker run -d --name "$ACTIVE_BROKER" --network "$NETWORK" \ + --cpus 2 --memory 2g \ + -e RUSTQUEUE_DATA_PATH=/data \ + -v "$ACTIVE_VOLUME:/data" \ + "$image" >/dev/null + wait_for_broker "$ACTIVE_BROKER" + sample_rss "$rss_file" "$ACTIVE_BROKER" & + SAMPLER_PID=$! + + printf '[%02d/%02d] %s pair=%d position=%d variant=%s\n' \ + "$SEQUENCE" "$((PAIRS * CASE_COUNT * 2))" "$scenario" "$pair" "$position" "$variant" + local -a bench_args=( + --address "$ACTIVE_BROKER:4150" + --topic "q-$scenario-$pair-$variant" + --messages 100000 + --message-bytes 1024 + --producers "$producers" + --consumers "$consumers" + --batch-size "$batch" + --warmup-seconds "$WARMUP_SECONDS" + --duration-seconds "$MEASUREMENT_SECONDS" + --drain-timeout-seconds "$DRAIN_TIMEOUT_SECONDS" + --json + ) + if [[ -n "$rate" ]]; then + bench_args+=(--rate "$rate") + fi + set +e + docker run --rm --network "$NETWORK" --cpus 2 --memory 2g \ + --entrypoint /usr/local/bin/rustqueue-bench "$TOOLS_IMAGE" \ + "${bench_args[@]}" >"$report" 2>"$error_log" + benchmark_status=$? + set -e + + kill "$SAMPLER_PID" >/dev/null 2>&1 || true + wait "$SAMPLER_PID" >/dev/null 2>&1 || true + SAMPLER_PID="" + if [[ "$benchmark_status" -ne 0 ]]; then + cat "$error_log" >&2 + cleanup_run + die "$label exited with status $benchmark_status" + fi + jq -e \ + --argjson producers "$producers" \ + --argjson consumers "$consumers" \ + --argjson batch "$batch" \ + '.message_bytes == 1024 + and .producers == $producers + and .consumers == $consumers + and .batch_size == $batch + and .messages > 0 + and .publish_messages_per_second > 0 + and .latency_us_p99 > 0' \ + "$report" >/dev/null || { + cleanup_run + die "$label produced an invalid benchmark report" + } + + final_depth=null + final_in_flight=null + final_deferred=null + if [[ "$consumers" -gt 0 ]]; then + topic="$(jq -r '.topic' "$report")" + channel="$(jq -r '.channel' "$report")" + stats="$( + docker exec "$ACTIVE_BROKER" curl -fsS \ + "http://127.0.0.1:4151/stats?format=json&include_clients=false&topic=$topic&channel=$channel" + )" + channel_stats="$( + jq -c --arg topic "$topic" --arg channel "$channel" ' + [.topics[] + | select(.topic_name == $topic) + | .channels[] + | select(.channel_name == $channel)][0] + ' <<<"$stats" + )" + [[ "$channel_stats" != null ]] || { + cleanup_run + die "$label final Channel stats are missing" + } + final_depth="$(jq -r '.depth' <<<"$channel_stats")" + final_in_flight="$(jq -r '.in_flight_count' <<<"$channel_stats")" + final_deferred="$(jq -r '.deferred_count' <<<"$channel_stats")" + fi + + prometheus="$( + docker exec "$ACTIVE_BROKER" curl -fsS http://127.0.0.1:4151/metrics + )" + metric_value() { + local name=$1 + awk -v name="$name" '$1 == name { print $2; found = 1; exit } END { if (!found) exit 1 }' \ + <<<"$prometheus" + } + publish_group_commits="$(metric_value rustqueue_publish_group_commits_total)" + publish_group_requests="$(metric_value rustqueue_publish_group_requests_total)" + publish_group_max_requests="$(metric_value rustqueue_publish_group_max_requests)" + channel_group_commits="$(metric_value rustqueue_channel_group_commits_total)" + channel_group_requests="$(metric_value rustqueue_channel_group_requests_total)" + channel_group_max_requests="$(metric_value rustqueue_channel_group_max_requests)" + channel_fsync_count="$(metric_value rustqueue_channel_fsync_duration_seconds_count)" + channel_fsync_sum="$(metric_value rustqueue_channel_fsync_duration_seconds_sum)" + channel_group_wait_count="$( + metric_value rustqueue_channel_group_commit_wait_duration_seconds_count + )" + channel_group_wait_sum="$( + metric_value rustqueue_channel_group_commit_wait_duration_seconds_sum + )" + consumer_fetch_batches="$(metric_value rustqueue_consumer_fetch_batches_total)" + consumer_fetch_messages="$(metric_value rustqueue_consumer_fetch_messages_total)" + aggregate_channel_depth="$(metric_value rustqueue_channel_depth_total)" + aggregate_channel_in_flight="$(metric_value rustqueue_channel_in_flight_total)" + aggregate_channel_deferred="$(metric_value rustqueue_channel_deferred_total)" + if [[ "$consumers" -gt 0 ]] && + [[ "$aggregate_channel_depth" -ne 0 || + "$aggregate_channel_in_flight" -ne 0 || + "$aggregate_channel_deferred" -ne 0 ]]; then + cleanup_run + die "$label left aggregate Channel backlog after drain" + fi + + rss_peak="$(awk 'NR > 1 && $2 > peak { peak=$2 } END { print peak + 0 }' "$rss_file")" + [[ "$rss_peak" -gt 0 ]] || { + cleanup_run + die "$label did not capture Broker RSS" + } + + jq -c \ + --arg case "$scenario" \ + --arg variant "$variant" \ + --arg commit "$commit" \ + --argjson pair "$pair" \ + --argjson sequence "$SEQUENCE" \ + --argjson position "$position" \ + --argjson status "$benchmark_status" \ + --argjson rss_peak "$rss_peak" \ + --argjson final_depth "$final_depth" \ + --argjson final_in_flight "$final_in_flight" \ + --argjson final_deferred "$final_deferred" \ + --argjson publish_group_commits "$publish_group_commits" \ + --argjson publish_group_requests "$publish_group_requests" \ + --argjson publish_group_max_requests "$publish_group_max_requests" \ + --argjson channel_group_commits "$channel_group_commits" \ + --argjson channel_group_requests "$channel_group_requests" \ + --argjson channel_group_max_requests "$channel_group_max_requests" \ + --argjson channel_fsync_count "$channel_fsync_count" \ + --argjson channel_fsync_sum "$channel_fsync_sum" \ + --argjson channel_group_wait_count "$channel_group_wait_count" \ + --argjson channel_group_wait_sum "$channel_group_wait_sum" \ + --argjson consumer_fetch_batches "$consumer_fetch_batches" \ + --argjson consumer_fetch_messages "$consumer_fetch_messages" \ + --argjson aggregate_channel_depth "$aggregate_channel_depth" \ + --argjson aggregate_channel_in_flight "$aggregate_channel_in_flight" \ + --argjson aggregate_channel_deferred "$aggregate_channel_deferred" \ + '{ + case: $case, + pair: $pair, + sequence: $sequence, + position_in_pair: $position, + variant: $variant, + commit: $commit, + benchmark_exit_code: $status, + metrics: { + messages, + received_unique_messages, + duplicate_messages, + missing_messages, + delivery_verified, + delivery_complete, + drain_timed_out, + final_channel_depth: $final_depth, + final_in_flight: $final_in_flight, + final_deferred: $final_deferred, + publish_messages_per_second, + receive_messages_per_second, + pub_ack_p99_us: .latency_us_p99, + rss_peak_bytes: $rss_peak, + broker_profile: { + publish_group_commits: $publish_group_commits, + publish_group_requests: $publish_group_requests, + publish_group_max_requests: $publish_group_max_requests, + channel_group_commits: $channel_group_commits, + channel_group_requests: $channel_group_requests, + channel_group_max_requests: $channel_group_max_requests, + channel_fsync_count: $channel_fsync_count, + channel_fsync_sum_seconds: $channel_fsync_sum, + channel_group_wait_count: $channel_group_wait_count, + channel_group_wait_sum_seconds: $channel_group_wait_sum, + consumer_fetch_batches: $consumer_fetch_batches, + consumer_fetch_messages: $consumer_fetch_messages, + aggregate_channel_depth: $aggregate_channel_depth, + aggregate_channel_in_flight: $aggregate_channel_in_flight, + aggregate_channel_deferred: $aggregate_channel_deferred + } + } + }' "$report" >>"$RUNS_FILE" + cleanup_run +} + +for scenario in $CASES; do + for pair in $(seq 1 "$PAIRS"); do + if (( pair % 2 == 1 )); then + run_variant "$scenario" "$pair" 1 baseline + run_variant "$scenario" "$pair" 2 candidate + else + run_variant "$scenario" "$pair" 1 candidate + run_variant "$scenario" "$pair" 2 baseline + fi + done +done + +baseline_image_id="$(docker image inspect "$BASELINE_IMAGE" --format '{{.Id}}')" +candidate_image_id="$(docker image inspect "$CANDIDATE_IMAGE" --format '{{.Id}}')" +docker_server_version="$(docker version --format '{{.Server.Version}}')" +docker_architecture="$(docker info --format '{{.Architecture}}')" +docker_cpus="$(docker info --format '{{.NCPU}}')" +docker_memory_bytes="$(docker info --format '{{.MemTotal}}')" +docker_storage_driver="$(docker info --format '{{.Driver}}')" +orbstack_version="$(orbctl version 2>/dev/null | head -n 1 || printf unknown)" +macos_version="$(sw_vers -productVersion 2>/dev/null || printf unknown)" +hardware_model="$(sysctl -n hw.model 2>/dev/null || printf unknown)" + +jq -n \ + --arg docker_context "$docker_context" \ + --arg docker_os "$docker_os" \ + --arg docker_server_version "$docker_server_version" \ + --arg docker_architecture "$docker_architecture" \ + --argjson docker_cpus "$docker_cpus" \ + --argjson docker_memory_bytes "$docker_memory_bytes" \ + --arg docker_storage_driver "$docker_storage_driver" \ + --arg orbstack_version "$orbstack_version" \ + --arg macos_version "$macos_version" \ + --arg hardware_model "$hardware_model" \ + --arg tool_source "$tool_source" \ + '{ + platform: "OrbStack on macOS", + comparison_scope: "same-host relative only", + macos_version: $macos_version, + hardware_model: $hardware_model, + orbstack_version: $orbstack_version, + docker: { + context: $docker_context, + operating_system: $docker_os, + server_version: $docker_server_version, + architecture: $docker_architecture, + cpus: $docker_cpus, + memory_bytes: $docker_memory_bytes, + storage_driver: $docker_storage_driver + }, + resource_limits: { + broker: {cpus: 2, memory_bytes: 2147483648}, + load_generator: {cpus: 2, memory_bytes: 2147483648} + }, + tool_source: $tool_source + }' >"$RUN_DIR/environment-core.json" + +if command -v shasum >/dev/null 2>&1; then + environment_fingerprint="$(shasum -a 256 "$RUN_DIR/environment-core.json" | awk '{print $1}')" +else + environment_fingerprint="$(sha256sum "$RUN_DIR/environment-core.json" | awk '{print $1}')" +fi +jq --arg fingerprint "$environment_fingerprint" \ + '. + {fingerprint_sha256: $fingerprint}' \ + "$RUN_DIR/environment-core.json" >"$RUN_DIR/environment.json" + +jq -n \ + --argjson pairs "$PAIRS" \ + --argjson warmup "$WARMUP_SECONDS" \ + --argjson measurement "$MEASUREMENT_SECONDS" \ + --argjson drain_timeout "$DRAIN_TIMEOUT_SECONDS" \ + --argjson iterations "$BOOTSTRAP_ITERATIONS" \ + --argjson seed "$BOOTSTRAP_SEED" \ + --arg cases "$CASES" \ + '{ + pairs: $pairs, + warmup_seconds: $warmup, + measurement_seconds: $measurement, + drain_timeout_seconds: $drain_timeout, + alternating_order: "AB_then_BA", + bootstrap_iterations: $iterations, + bootstrap_seed: $seed, + throughput_regression_ratio: 0.95, + latency_rss_regression_ratio: 1.10, + scenarios: [ + { + name: "raw_write", + topic_count: 1, + channel_count: 0, + producers: 16, + consumers: 0, + message_bytes: 1024, + batch_size: 64, + rate: "saturation", + headline_metric: "publish_messages_per_second" + }, + { + name: "sustainable", + topic_count: 1, + channel_count: 1, + producers: 16, + consumers: 16, + message_bytes: 1024, + batch_size: 64, + rate: "saturation", + headline_metric: "receive_messages_per_second", + completion_contract: "depth=0, in_flight=0, deferred=0, missing=0, duplicates=0" + }, + { + name: "low_load_latency", + topic_count: 1, + channel_count: 1, + producers: 16, + consumers: 16, + message_bytes: 1024, + batch_size: 1, + messages_per_second: 100, + headline_metric: "pub_ack_p99_us", + completion_contract: "depth=0, in_flight=0, deferred=0, missing=0, duplicates=0" + } + ] + } + | .scenarios = [ + .scenarios[] + | select(.name as $name | ($cases | split(" ") | index($name)) != null) + ]' >"$RUN_DIR/protocol.json" + +generated_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)" +jq -n \ + --arg release "$RELEASE" \ + --arg generated_at "$generated_at" \ + --arg baseline_ref "$BASELINE_REF" \ + --arg baseline_commit "$baseline_commit" \ + --arg baseline_image "$baseline_image_id" \ + --arg baseline_binary_sha256 "$baseline_binary_sha256" \ + --arg candidate_ref "$CANDIDATE_REF" \ + --arg candidate_commit "$candidate_commit" \ + --arg candidate_image "$candidate_image_id" \ + --arg candidate_binary_sha256 "$candidate_binary_sha256" \ + --slurpfile environment "$RUN_DIR/environment.json" \ + --slurpfile protocol "$RUN_DIR/protocol.json" \ + --slurpfile runs "$RUNS_FILE" \ + '{ + schema_version: 1, + release: $release, + generated_at_utc: $generated_at, + baseline: { + revision: $baseline_ref, + commit: $baseline_commit, + image_id: $baseline_image, + binary_sha256: $baseline_binary_sha256 + }, + candidate: { + revision: $candidate_ref, + commit: $candidate_commit, + image_id: $candidate_image, + binary_sha256: $candidate_binary_sha256 + }, + environment: $environment[0], + protocol: $protocol[0], + runs: $runs + }' >"$INPUT_FILE" + +set +e +docker run --rm \ + -v "$RUN_DIR:/results:ro" \ + --entrypoint /usr/local/bin/rustqueue-qualify \ + "$TOOLS_IMAGE" --input /results/input.json >"$EVIDENCE_FILE" +qualification_status=$? +set -e + +mkdir -p "$(dirname "$EVIDENCE_OUTPUT")" +cp "$EVIDENCE_FILE" "$EVIDENCE_OUTPUT" +printf 'Raw qualification artifacts: %s\n' "$RUN_DIR" +printf 'Compact qualification evidence: %s\n' "$EVIDENCE_OUTPUT" +exit "$qualification_status" diff --git a/scripts/verify-release-version.sh b/scripts/verify-release-version.sh index 111ebaa..62df953 100755 --- a/scripts/verify-release-version.sh +++ b/scripts/verify-release-version.sh @@ -94,4 +94,65 @@ grep -Fq "Current release: [v$EXPECTED]" "$ROOT/README.md" || { exit 1 } +qualification="$ROOT/benchmarks/qualifications/v$EXPECTED-orbstack.json" +if [[ "$EXPECTED" == "0.8.2" && -f "$qualification" ]]; then + command -v jq >/dev/null 2>&1 || { + echo "jq is required to verify Broker qualification evidence" >&2 + exit 1 + } + jq -e --arg version "$EXPECTED" ' + .schema_version == 1 + and .release == $version + and .baseline.revision == "v0.8.1" + and (.baseline.commit | test("^[0-9a-f]{40}$")) + and (.candidate.commit | test("^[0-9a-f]{40}$")) + and (.baseline.binary_sha256 | test("^[0-9a-f]{64}$")) + and (.candidate.binary_sha256 | test("^[0-9a-f]{64}$")) + and .environment.platform == "OrbStack on macOS" + and .environment.tool_source == .candidate.commit + and .environment.resource_limits.broker.cpus == 2 + and .environment.resource_limits.broker.memory_bytes == 2147483648 + and .environment.resource_limits.load_generator.cpus == 2 + and .environment.resource_limits.load_generator.memory_bytes == 2147483648 + and .protocol.pairs == 10 + and .protocol.warmup_seconds == 30 + and .protocol.measurement_seconds == 120 + and .protocol.alternating_order == "AB_then_BA" + and .protocol.throughput_regression_ratio == 0.95 + and .protocol.latency_rss_regression_ratio == 1.10 + and ([.protocol.scenarios[].name] | sort) + == ["low_load_latency", "raw_write", "sustainable"] + and (.runs | length) == 60 + and all(.runs[]; .benchmark_exit_code == 0) + and all( + .runs[] + | select(.case != "raw_write"); + .metrics.delivery_verified == true + and .metrics.delivery_complete == true + and .metrics.drain_timed_out == false + and .metrics.missing_messages == 0 + and .metrics.duplicate_messages == 0 + and .metrics.final_channel_depth == 0 + and .metrics.final_in_flight == 0 + and .metrics.final_deferred == 0 + and .metrics.broker_profile.aggregate_channel_depth == 0 + and .metrics.broker_profile.aggregate_channel_in_flight == 0 + and .metrics.broker_profile.aggregate_channel_deferred == 0 + ) + and ([.statistics[] | "\(.case):\(.metric)"] | sort) == [ + "low_load_latency:pub_ack_p99_us", + "low_load_latency:rss_peak_bytes", + "raw_write:publish_messages_per_second", + "sustainable:receive_messages_per_second" + ] + and all(.statistics[]; .regression == false) + and .verdict.status == "pass" + and (.verdict.hard_failures | length) == 0 + and (.verdict.regressions | length) == 0 + ' "$qualification" >/dev/null || { + echo "Broker qualification evidence is invalid or failed" >&2 + exit 1 + } +fi + echo "RustQueue release metadata is consistent at $EXPECTED"