From fd6756522640d0aaecd439b4caef44f53d52765e Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 10:39:34 +0800 Subject: [PATCH 1/7] feat: qualify RustQueue 0.8.2 broker performance --- Cargo.lock | 22 +- Cargo.toml | 2 +- Makefile | 5 +- README.md | 79 ++- benchmarks/Dockerfile.qualify | 16 + benchmarks/qualifications/README.md | 18 + console-ui/package.json | 2 +- crates/bench/src/bin/rustqueue-qualify.rs | 604 +++++++++++++++++++ crates/bench/src/consumer.rs | 225 +++++-- crates/bench/src/main.rs | 68 ++- crates/queue/src/broker/channel_commit.rs | 21 +- deploy/helm/rustqueue/Chart.yaml | 4 +- deploy/helm/rustqueue/values.yaml | 4 +- docs/architecture/share-nothing-v7.md | 2 +- docs/operations/kubernetes.md | 4 +- docs/releases/v0.8.2.md | 77 +++ fuzz/Cargo.lock | 12 +- scripts/acceptance-k8s.sh | 2 + scripts/acceptance-multi-broker-k8s.sh | 4 +- scripts/benchmark-qualify.sh | 685 ++++++++++++++++++++++ scripts/verify-release-version.sh | 65 ++ 21 files changed, 1776 insertions(+), 145 deletions(-) create mode 100644 benchmarks/Dockerfile.qualify create mode 100644 benchmarks/qualifications/README.md create mode 100644 crates/bench/src/bin/rustqueue-qualify.rs create mode 100644 docs/releases/v0.8.2.md create mode 100755 scripts/benchmark-qualify.sh 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..9d9a227 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,52 @@ 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. +## What's new in 0.8.2 + +- **Direct-Broker qualification.** A reproducible OrbStack gate compares the + exact `v0.8.1` tag with one candidate commit using fresh volumes, fixed + 2 vCPU / 2 GiB limits and alternating paired runs. Raw reports remain local; + the compact environment, run metrics, deterministic bootstrap result and + verdict are committed with the release. +- **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 qualification record, rather than the mechanism alone, + determines whether a performance claim is justified. +- **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 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. +0.8.1. See the +[v0.8.2 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2) +for the qualification protocol and validation record. -## Download 0.8.1 +## 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 @@ -223,7 +222,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 +459,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..855b2f6 --- /dev/null +++ b/benchmarks/qualifications/README.md @@ -0,0 +1,18 @@ +# Broker qualification evidence + +`scripts/benchmark-qualify.sh` is the pre-release RustQueue Broker performance +gate. 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 release cases as 10 alternating pairs. A full run writes the +reviewable release artifact to `v0.8.2-orbstack.json`. + +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..45dfd35 --- /dev/null +++ b/crates/bench/src/bin/rustqueue-qualify.rs @@ -0,0 +1,604 @@ +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, + 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.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..94ed228 100644 --- a/crates/queue/src/broker/channel_commit.rs +++ b/crates/queue/src/broker/channel_commit.rs @@ -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; @@ -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/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..6338242 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 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..aa1e4ba --- /dev/null +++ b/docs/releases/v0.8.2.md @@ -0,0 +1,77 @@ +# RustQueue 0.8.2 + +RustQueue 0.8.2 qualifies the direct Broker data path and corrects premature +Channel group commits. The comparison bypasses Proxy and uses the production +defaults; only the committed qualification record may substantiate a +performance claim, and it does not generalize beyond the same OrbStack host +used for the paired runs. + +## 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. + +## Qualification protocol + +The release 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: + +- 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. The compact, machine-readable release record +is +[`benchmarks/qualifications/v0.8.2-orbstack.json`](../../benchmarks/qualifications/v0.8.2-orbstack.json). +The release metadata gate requires that artifact to contain a passing verdict +before `v0.8.2` can be tagged. + +## 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..88b1092 --- /dev/null +++ b/scripts/benchmark-qualify.sh @@ -0,0 +1,685 @@ +#!/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:-300}" +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" + [[ "$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 iterations "$BOOTSTRAP_ITERATIONS" \ + --argjson seed "$BOOTSTRAP_SEED" \ + --arg cases "$CASES" \ + '{ + pairs: $pairs, + warmup_seconds: $warmup, + measurement_seconds: $measurement, + 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..f75edc2 100755 --- a/scripts/verify-release-version.sh +++ b/scripts/verify-release-version.sh @@ -94,4 +94,69 @@ grep -Fq "Current release: [v$EXPECTED]" "$ROOT/README.md" || { exit 1 } +qualification="$ROOT/benchmarks/qualifications/v$EXPECTED-orbstack.json" +if [[ "$EXPECTED" == "0.8.2" ]]; then + [[ -f "$qualification" ]] || { + echo "Broker qualification evidence is missing: $qualification" >&2 + exit 1 + } + 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" From 04da3a57312b3ba103bae0dafac095bd992f1a59 Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 12:35:40 +0800 Subject: [PATCH 2/7] test: allow complete broker qualification drain --- benchmarks/qualifications/README.md | 4 +++- crates/bench/src/bin/rustqueue-qualify.rs | 7 +++++++ docs/releases/v0.8.2.md | 3 ++- scripts/benchmark-qualify.sh | 6 +++++- 4 files changed, 17 insertions(+), 3 deletions(-) diff --git a/benchmarks/qualifications/README.md b/benchmarks/qualifications/README.md index 855b2f6..5558fae 100644 --- a/benchmarks/qualifications/README.md +++ b/benchmarks/qualifications/README.md @@ -10,7 +10,9 @@ 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 release cases as 10 alternating pairs. A full run writes the -reviewable release artifact to `v0.8.2-orbstack.json`. +reviewable release 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. Per-run benchmark JSON, stderr, RSS samples and the evaluator input stay under the ignored `benchmarks/results/` directory. Development runs may shorten the diff --git a/crates/bench/src/bin/rustqueue-qualify.rs b/crates/bench/src/bin/rustqueue-qualify.rs index 45dfd35..33c9700 100644 --- a/crates/bench/src/bin/rustqueue-qualify.rs +++ b/crates/bench/src/bin/rustqueue-qualify.rs @@ -43,6 +43,7 @@ struct Protocol { pairs: usize, warmup_seconds: u64, measurement_seconds: u64, + drain_timeout_seconds: u64, alternating_order: String, bootstrap_iterations: usize, bootstrap_seed: u64, @@ -259,6 +260,12 @@ fn validate_input(input: &QualificationInput) -> anyhow::Result<()> { 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"); } diff --git a/docs/releases/v0.8.2.md b/docs/releases/v0.8.2.md index aa1e4ba..ca8e8a2 100644 --- a/docs/releases/v0.8.2.md +++ b/docs/releases/v0.8.2.md @@ -36,7 +36,8 @@ 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: +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; diff --git a/scripts/benchmark-qualify.sh b/scripts/benchmark-qualify.sh index 88b1092..fe8a21d 100755 --- a/scripts/benchmark-qualify.sh +++ b/scripts/benchmark-qualify.sh @@ -10,7 +10,7 @@ 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:-300}" +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}" @@ -117,6 +117,8 @@ case "$EVIDENCE_OUTPUT" in 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" ;; @@ -579,6 +581,7 @@ 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" \ @@ -586,6 +589,7 @@ jq -n \ pairs: $pairs, warmup_seconds: $warmup, measurement_seconds: $measurement, + drain_timeout_seconds: $drain_timeout, alternating_order: "AB_then_BA", bootstrap_iterations: $iterations, bootstrap_seed: $seed, From 0f858c5197133bd0eefeea42ad2e2febc880c8af Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 14:09:47 +0800 Subject: [PATCH 3/7] perf: remove channel in-flight entries in constant time --- crates/queue/src/channel.rs | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) 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); From adc96c265075d21ae05e9c8d3287b630411b9516 Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 14:24:47 +0800 Subject: [PATCH 4/7] perf: enlarge durable channel commit groups --- crates/queue/src/broker/channel_commit.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/queue/src/broker/channel_commit.rs b/crates/queue/src/broker/channel_commit.rs index 94ed228..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 { @@ -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(); } From b4cb4d5aba808fd21a93f29c5bb85bd0602fccdf Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 14:54:11 +0800 Subject: [PATCH 5/7] perf: reduce durable channel checkpoint frequency --- crates/queue/src/channel_store.rs | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) 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(); From aa4669eb18eebf95f5970a31a920ce6e60944e27 Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 15:51:42 +0800 Subject: [PATCH 6/7] fix: retain unrouted topic messages --- README.md | 22 ++++-- crates/queue/src/broker_tests.rs | 15 +++- crates/queue/src/metadata.rs | 2 + crates/queue/src/topic.rs | 88 ++++++++++++++++++++++ crates/queue/src/topic/index.rs | 7 ++ crates/queue/src/topic/maintenance.rs | 18 ++++- crates/queue/tests/recovery.rs | 101 ++++++++++++++++++++++++++ docs/architecture/share-nothing-v7.md | 55 ++++++++------ docs/releases/v0.8.2.md | 24 ++++-- 9 files changed, 293 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index 9d9a227..e3246cc 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,10 @@ your workload before deploying to production. ## 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 qualification.** A reproducible OrbStack gate compares the exact `v0.8.1` tag with one candidate commit using fresh volumes, fixed 2 vCPU / 2 GiB limits and alternating paired runs. Raw reports remain local; @@ -65,8 +69,8 @@ your workload before deploying to production. 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 the NSQ/Kodo compatibility contract from -0.8.1. See the +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 qualification protocol and validation record. @@ -123,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. 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/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/docs/architecture/share-nothing-v7.md b/docs/architecture/share-nothing-v7.md index 6338242..0fc378e 100644 --- a/docs/architecture/share-nothing-v7.md +++ b/docs/architecture/share-nothing-v7.md @@ -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 - -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. - -This deliberately prefers duplicates over misses during the normal discovery -window. The guarantee is bounded: - -- 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. +### 2.3 Unrouted and bootstrap retention + +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. + +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. + +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/releases/v0.8.2.md b/docs/releases/v0.8.2.md index ca8e8a2..0a81fa4 100644 --- a/docs/releases/v0.8.2.md +++ b/docs/releases/v0.8.2.md @@ -1,10 +1,24 @@ # RustQueue 0.8.2 -RustQueue 0.8.2 qualifies the direct Broker data path and corrects premature -Channel group commits. The comparison bypasses Proxy and uses the production -defaults; only the committed qualification record may substantiate a -performance claim, and it does not generalize beyond the same OrbStack host -used for the paired runs. +RustQueue 0.8.2 aligns no-Channel retention with NSQ, qualifies the direct +Broker data path, and corrects premature Channel group commits. The comparison +bypasses Proxy and uses the production defaults; only the committed +qualification record may substantiate a performance claim, and it does not +generalize beyond the same OrbStack host used for the paired runs. + +## 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 From b5cd697ecaf73235a3e787c8fadd10e791f6ba94 Mon Sep 17 00:00:00 2001 From: livesite Date: Wed, 29 Jul 2026 16:42:24 +0800 Subject: [PATCH 7/7] release: prepare RustQueue 0.8.2 --- README.md | 16 ++++++++-------- benchmarks/qualifications/README.md | 13 +++++++++---- docs/releases/v0.8.2.md | 27 +++++++++++++-------------- scripts/verify-release-version.sh | 6 +----- 4 files changed, 31 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index e3246cc..251a328 100644 --- a/README.md +++ b/README.md @@ -49,17 +49,17 @@ your workload before deploying to production. 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 qualification.** A reproducible OrbStack gate compares the - exact `v0.8.1` tag with one candidate commit using fresh volumes, fixed - 2 vCPU / 2 GiB limits and alternating paired runs. Raw reports remain local; - the compact environment, run metrics, deterministic bootstrap result and - verdict are committed with the release. +- **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 qualification record, rather than the mechanism alone, - determines whether a performance claim is justified. + 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 @@ -72,7 +72,7 @@ your workload before deploying to production. 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 qualification protocol and validation record. +for the validation boundaries. ## Download 0.8.2 diff --git a/benchmarks/qualifications/README.md b/benchmarks/qualifications/README.md index 5558fae..cbbb3dd 100644 --- a/benchmarks/qualifications/README.md +++ b/benchmarks/qualifications/README.md @@ -1,7 +1,7 @@ # Broker qualification evidence -`scripts/benchmark-qualify.sh` is the pre-release RustQueue Broker performance -gate. Run it on OrbStack from a committed candidate: +`scripts/benchmark-qualify.sh` is the optional RustQueue Broker performance +qualification. Run it on OrbStack from a committed candidate: ```sh make benchmark-qualify @@ -9,11 +9,16 @@ 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 release cases as 10 alternating pairs. A full run writes the -reviewable release artifact to `v0.8.2-orbstack.json`. Consumer cases must drain +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 diff --git a/docs/releases/v0.8.2.md b/docs/releases/v0.8.2.md index 0a81fa4..8d97d2b 100644 --- a/docs/releases/v0.8.2.md +++ b/docs/releases/v0.8.2.md @@ -1,10 +1,10 @@ # RustQueue 0.8.2 -RustQueue 0.8.2 aligns no-Channel retention with NSQ, qualifies the direct -Broker data path, and corrects premature Channel group commits. The comparison -bypasses Proxy and uses the production defaults; only the committed -qualification record may substantiate a performance claim, and it does not -generalize beyond the same OrbStack host used for the paired runs. +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 @@ -43,11 +43,11 @@ generalize beyond the same OrbStack host used for the paired runs. drain, or nonzero final Channel depth, in-flight count or deferred count is a hard failure before statistics are evaluated. -## Qualification protocol +## Benchmark protocol and release decision -The release 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. +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 @@ -68,11 +68,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. The compact, machine-readable release record -is -[`benchmarks/qualifications/v0.8.2-orbstack.json`](../../benchmarks/qualifications/v0.8.2-orbstack.json). -The release metadata gate requires that artifact to contain a passing verdict -before `v0.8.2` can be tagged. +`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 diff --git a/scripts/verify-release-version.sh b/scripts/verify-release-version.sh index f75edc2..62df953 100755 --- a/scripts/verify-release-version.sh +++ b/scripts/verify-release-version.sh @@ -95,11 +95,7 @@ grep -Fq "Current release: [v$EXPECTED]" "$ROOT/README.md" || { } qualification="$ROOT/benchmarks/qualifications/v$EXPECTED-orbstack.json" -if [[ "$EXPECTED" == "0.8.2" ]]; then - [[ -f "$qualification" ]] || { - echo "Broker qualification evidence is missing: $qualification" >&2 - exit 1 - } +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