diff --git a/Cargo.lock b/Cargo.lock index d1f83af..0713b21 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2283,7 +2283,7 @@ dependencies = [ [[package]] name = "rustqueue-bench" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "clap", @@ -2295,7 +2295,7 @@ dependencies = [ [[package]] name = "rustqueue-console" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "axum", @@ -2323,7 +2323,7 @@ dependencies = [ [[package]] name = "rustqueue-discovery" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "axum", @@ -2346,7 +2346,7 @@ dependencies = [ [[package]] name = "rustqueue-operator" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "axum", @@ -2373,7 +2373,7 @@ dependencies = [ [[package]] name = "rustqueue-protocol" -version = "0.8.2" +version = "0.8.3" dependencies = [ "bytes", "serde", @@ -2383,7 +2383,7 @@ dependencies = [ [[package]] name = "rustqueue-proxy" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "axum", @@ -2404,7 +2404,7 @@ dependencies = [ [[package]] name = "rustqueue-queue" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "bytes", @@ -2426,7 +2426,7 @@ dependencies = [ [[package]] name = "rustqueue-server" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "async-compression", @@ -2462,7 +2462,7 @@ dependencies = [ [[package]] name = "rustqueue-storage" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "crc32c", @@ -2475,14 +2475,14 @@ dependencies = [ [[package]] name = "rustqueue-telemetry" -version = "0.8.2" +version = "0.8.3" dependencies = [ "serde", ] [[package]] name = "rustqueuectl" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "clap", diff --git a/Cargo.toml b/Cargo.toml index 9e96ce3..1bc891d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,7 +16,7 @@ members = [ exclude = ["fuzz"] [workspace.package] -version = "0.8.2" +version = "0.8.3" edition = "2021" license = "Apache-2.0" rust-version = "1.88" diff --git a/Makefile b/Makefile index 4c63bd4..a88054e 100644 --- a/Makefile +++ b/Makefile @@ -80,6 +80,10 @@ helm-lint: --set queue.imagePullPolicy=Never ! helm template rustqueue deploy/helm/rustqueue \ --set queue.kodoCompatibility.cleanupEnabled=true + ! helm template rustqueue deploy/helm/rustqueue \ + --set queue.kodoCompatibility.enabled=true \ + --set queue.imagePullPolicy=Never \ + --set queue.publishAckMode=nsq_relaxed ! rg -n 'x-kubernetes-preserve-unknown-fields:[[:space:]]*false' \ deploy/helm/rustqueue/crds diff --git a/README.md b/README.md index 251a328..b8b9c4c 100644 --- a/README.md +++ b/README.md @@ -8,16 +8,17 @@ [![Kubernetes](https://img.shields.io/badge/kubernetes-1.28%2B-326CE5.svg)](https://kubernetes.io/) [Architecture](docs/architecture/share-nothing-v7.md) · +[NSQ performance boundaries](docs/architecture/nsq-performance.md) · [Kubernetes operations](docs/operations/kubernetes.md) · [Console operations](docs/operations/console.md) · -[v0.8.2 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2) +[v0.8.3 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.3) -RustQueue 0.8.2 is a Kubernetes-native, NSQ V2-compatible message queue for +RustQueue 0.8.3 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.2](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2). +> Current release: [v0.8.3](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.3). > 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. @@ -29,7 +30,7 @@ The complete architecture and reliability contract is documented in | Area | Contract | | --- | --- | -| Durability | `PUB`/`MPUB`/`DPUB` return only after local segment `fsync`; `FIN`/`REQ` use a durable channel WAL | +| Durability | Default `PUB`/`MPUB`/`DPUB` return after local segment `fsync`; opt-in `write_ack` and `nsq_relaxed` return after append with explicit crash-loss windows; `FIN`/`REQ` use a durable channel WAL | | Delivery | At least once; a restart may redeliver a message without a durable `FIN` | | Compatibility | NSQ V2 core commands, lookup, standard Stats fields, TLS/mTLS, AUTH, Snappy, Deflate, fan-out and ephemeral channels | | Kodo | Default-off compatibility profile: stable publish Gateways from `/nodes`, real Broker owners from `/lookup`, and no upstream Kodo change | @@ -43,56 +44,45 @@ 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.2 - -- **NSQ-aligned no-Channel durability.** A Topic with no durable Channel now - persists its unrouted start position and normal GC cannot cross it. The first - durable Channel receives every acknowledged publish from that interval, even - when creation happens after the bootstrap window or a Broker restart. -- **Direct-Broker preflight.** Reproducible OrbStack tooling compares the exact - `v0.8.1` tag with one candidate commit using fresh volumes, fixed - 2 vCPU / 2 GiB limits and alternating paired runs. RustQueue 0.8.2 completed - short correctness and regression preflights but does not claim completion of - the optional 60-run performance qualification. -- **Bounded Channel coalescing.** The durable Channel worker now keeps - collecting `FIN` and `REQ` requests throughout its existing bounded 1 ms - window instead of committing at the first transient queue gap. The - 64-request ceiling, channel WAL `fsync` boundary and at-least-once contract - are unchanged. The mechanism and short preflight are not a formal throughput - guarantee. -- **Reliable benchmark shutdown and warmup.** The benchmark preserves a - partially read NSQ frame while closing consumers, and a consumer warmup is - fully drained before measurement. Missing, duplicate or non-drained delivery - remains a hard failure. -- **Explicit regression policy.** Raw write and end-to-end sustainable - throughput fail only when a one-sided paired 95% bootstrap interval is - wholly below `0.95`. Fixed-rate PUB ACK p99 and comparable fixed-rate peak - RSS fail only when the interval is wholly above `1.10`. - -The patch keeps disk format v7 and remains wire-compatible with the NSQ/Kodo -contract from 0.8.1. See the -[v0.8.2 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.2) -for the validation boundaries. - -## Download 0.8.2 +## What's new in 0.8.3 + +- **Deadline-indexed delivery.** Channel and TCP-session leases now use an + ordered deadline index instead of rescanning every in-flight delivery on + each fetch or session event. `TOUCH`, `FIN`, `REQ`, completion, and disconnect + update the same index, so high-RDY consumers avoid stale timer buildup. +- **NSQ scheduler parity.** NSQ uses an in-flight priority queue; RustQueue now + matches that scheduler shape while keeping token-checked at-least-once + delivery and the durable Channel WAL acknowledgement boundary. +- **Durability-aware comparison.** Benchmark documentation now distinguishes + RustQueue's acknowledgement-after-fsync semantics from NSQ diskqueue writes + and its optional memory queue. `--sync-every=1` is reported as an NSQ write + profile, not as an equal durability claim. +- **No format migration.** The disk format remains v7, and the NSQ/Kodo wire + contract is unchanged from 0.8.2. + +See the [v0.8.3 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.3) +and [NSQ performance boundaries](docs/architecture/nsq-performance.md) for +the contract and benchmark interpretation. + +## Download 0.8.3 Every release contains native Linux binaries, the Console UI, source, the Helm Chart and a checksum manifest: | Asset | Contents | | --- | --- | -| `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 | +| `rustqueue-0.8.3-linux-x86_64.tar.gz` | Linux x86_64 binaries, Console UI and example configuration | +| `rustqueue-0.8.3-linux-aarch64.tar.gz` | Linux ARM64 binaries, Console UI and example configuration | +| `rustqueue-0.8.3-source.tar.gz` | Source archive for the tagged commit | +| `rustqueue-0.8.3.tgz` | Helm Chart | +| `SHA256SUMS-0.8.3` | SHA-256 checksums for every downloadable artifact | ```sh arch="$(uname -m)" -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" +curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.3/rustqueue-0.8.3-linux-${arch}.tar.gz" +curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.3/SHA256SUMS-0.8.3" +sha256sum --check --ignore-missing SHA256SUMS-0.8.3 +tar -xzf "rustqueue-0.8.3-linux-${arch}.tar.gz" ``` ## Architecture @@ -111,7 +101,10 @@ consumer -> discovery /lookup -> every broker that owns the topic operator -> eligible nodes -> StatefulSet ordinal + retained RWO PVC ``` -- A successful `PUB`, `MPUB`, or `DPUB` has passed local segment `fsync`. +- In the default `durable` mode, a successful `PUB`, `MPUB`, or `DPUB` has + passed local segment `fsync`. `write_ack` returns after append but delays + consumption until background fsync; `nsq_relaxed` returns and exposes the + append immediately. Both relaxed modes can lose their unsynced tail. - Concurrent publishes to one topic use a bounded group commit (up to 64 requests or 8 MiB, with at most 1 ms coalescing delay) and all wait for the same durable boundary before receiving `OK`. @@ -230,7 +223,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.2 \ + --set queue.image=registry.example/rustqueue:0.8.3 \ --set queue.storageClassName=ssd-rwo ``` @@ -467,7 +460,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.2 CI/CD workflow publishes a Release only after the non-Kubernetes +The v0.8.3 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 @@ -546,14 +539,25 @@ transfers only small revision/readiness heads; nevertheless a consumer still needs one connection per actual Topic owner. This is a share-nothing cost, not an unbounded or zero-cost scaling claim. -Latency histograms cover publish and channel-WAL fsync, publish and FIN/REQ -group-commit queueing, publish and channel ACK, payload reads, scrub/GC, proxy -backend calls, and discovery registry polling. Queue aggregates have fixed -cardinality by default; `[metrics].detailed_queue_metrics` enables bounded +Latency histograms cover publish and channel-WAL fsync, Topic-lock wait/hold +for publish and delivery reservation, publish and FIN/REQ group-commit +queueing, publish and channel ACK, payload reads, scrub/GC, proxy backend +calls, and discovery registry polling. Queue aggregates have fixed cardinality +by default; `[metrics].detailed_queue_metrics` enables bounded per-topic/channel series up to `max_detailed_series`. Delivery-budget bytes, waiters and cumulative waits are exported as bounded aggregate gauges/counters. +`[queue].publish_ack_mode` defaults to `"durable"`: successful publish commands +follow local segment `fsync`. `"write_ack"` returns after append while consumers +remain bounded by `last_durable_position`; `"nsq_relaxed"` also exposes the +appended tail immediately. Background sync runs at the first of +`relaxed_sync_messages`, `relaxed_sync_bytes`, or `relaxed_sync_interval_ms`. +`rustqueue_publish_unsynced_messages`, `rustqueue_publish_unsynced_bytes`, and +`rustqueue_publish_sync_lag_seconds` expose the crash-loss window, with bounded +per-Topic variants when detailed metrics are enabled. A sync failure stops new +writes. Keep both relaxed profiles separate from durable-PUB results. + ## Storage and upgrades RustQueue 0.8 keeps disk format v7. Format v7 is a clean break: a v6 or older diff --git a/benchmarks/qualifications/README.md b/benchmarks/qualifications/README.md index cbbb3dd..3901fd3 100644 --- a/benchmarks/qualifications/README.md +++ b/benchmarks/qualifications/README.md @@ -7,14 +7,14 @@ qualification. Run it on OrbStack from a committed candidate: make benchmark-qualify ``` -The default protocol compares the exact `v0.8.1` tag with `HEAD`, uses fresh +The default protocol compares the exact `v0.8.2` tag with `HEAD`, uses fresh Docker volumes, fixes Broker and load-generator containers at 2 vCPU / 2 GiB, and runs all three cases as 10 alternating pairs. A full run writes the -reviewable artifact to `v0.8.2-orbstack.json`. Consumer cases must drain -completely and have a fixed 1,800-second timeout so the v0.8.1 durable `FIN` +reviewable artifact to `v0.8.3-orbstack.json`. Consumer cases must drain +completely and have a fixed 1,800-second timeout so the v0.8.2 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 +The RustQueue 0.8.3 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. diff --git a/console-ui/package.json b/console-ui/package.json index a5b4f4b..a5acb86 100644 --- a/console-ui/package.json +++ b/console-ui/package.json @@ -1,7 +1,7 @@ { "name": "rustqueue-console-ui", "private": true, - "version": "0.8.2", + "version": "0.8.3", "packageManager": "pnpm@11.9.0", "type": "module", "scripts": { diff --git a/crates/console/src/model.rs b/crates/console/src/model.rs index 7ca4de9..6e030c3 100644 --- a/crates/console/src/model.rs +++ b/crates/console/src/model.rs @@ -322,6 +322,10 @@ mod tests { message_count: 1, segment_count: 1, segment_bytes: 128, + last_durable_position: 1, + unsynced_messages: 0, + unsynced_bytes: 0, + sync_lag_ms: 0, channels: Vec::new(), }); let mut head = BrokerObservationHead { diff --git a/crates/operator/src/controller/mod.rs b/crates/operator/src/controller/mod.rs index f880993..29deb73 100644 --- a/crates/operator/src/controller/mod.rs +++ b/crates/operator/src/controller/mod.rs @@ -1366,10 +1366,20 @@ fn validate(cluster: &RustQueue, active_feature_floor: u32) -> anyhow::Result<() || cluster.spec.max_topics == 0 || cluster.spec.max_publish_workers == 0 || cluster.spec.publish_worker_idle_seconds == 0 + || (cluster.spec.publish_ack_mode != "durable" + && (cluster.spec.relaxed_sync_messages == 0 + || cluster.spec.relaxed_sync_bytes < 4096 + || cluster.spec.relaxed_sync_interval_ms == 0)) || cluster.spec.max_detailed_metric_series == 0 { bail!("queue limits are outside the stable v7 contract"); } + if !matches!( + cluster.spec.publish_ack_mode.as_str(), + "durable" | "write_ack" | "nsq_relaxed" + ) { + bail!("publishAckMode must be durable, write_ack, or nsq_relaxed"); + } validate_message_storage_contract( cluster.spec.max_message_bytes, effective_storage_feature_level(cluster.spec.storage_feature_level, active_feature_floor), @@ -1381,13 +1391,14 @@ fn validate(cluster: &RustQueue, active_feature_floor: u32) -> anyhow::Result<() || cluster.spec.bootstrap_retention_seconds < KODO_BOOTSTRAP_RETENTION_SECONDS || cluster.spec.max_message_bytes != 100 * 1024 * 1024 || cluster.spec.connection_delivery_inflight_bytes < 128 * 1024 * 1024 - || cluster.spec.node_delivery_inflight_bytes < 512 * 1024 * 1024) + || cluster.spec.node_delivery_inflight_bytes < 512 * 1024 * 1024 + || cluster.spec.publish_ack_mode != "durable") { bail!( "Kodo compatibility requires exactly 3 brokers, storageFeatureLevel 2, \ bootstrapRetentionSeconds >= 180, \ maxMessageBytes 104857600, connectionDeliveryInflightBytes >= 134217728, \ - and nodeDeliveryInflightBytes >= 536870912" + nodeDeliveryInflightBytes >= 536870912, and publishAckMode durable" ); } if !(630..=86_400).contains(&cluster.spec.kodo_compatibility.cutover_grace_seconds) { @@ -2037,5 +2048,11 @@ mod tests { .unwrap_err() .to_string() .contains("bootstrapRetentionSeconds >= 180")); + cluster.spec.bootstrap_retention_seconds = KODO_BOOTSTRAP_RETENTION_SECONDS; + cluster.spec.publish_ack_mode = "nsq_relaxed".into(); + assert!(validate(&cluster, 2) + .unwrap_err() + .to_string() + .contains("publishAckMode durable")); } } diff --git a/crates/operator/src/controller/status.rs b/crates/operator/src/controller/status.rs index 0f8222c..3060f7f 100644 --- a/crates/operator/src/controller/status.rs +++ b/crates/operator/src/controller/status.rs @@ -213,6 +213,10 @@ mod tests { max_topics: 100, max_publish_workers: 32, publish_worker_idle_seconds: 60, + publish_ack_mode: "durable".into(), + relaxed_sync_messages: 2_500, + relaxed_sync_bytes: 8 * 1024 * 1024, + relaxed_sync_interval_ms: 10, detailed_queue_metrics: false, max_detailed_metric_series: 1_000, registry_secret_name: None, diff --git a/crates/operator/src/crd.rs b/crates/operator/src/crd.rs index 6993d80..74dec16 100644 --- a/crates/operator/src/crd.rs +++ b/crates/operator/src/crd.rs @@ -57,6 +57,14 @@ pub struct RustQueueSpec { pub max_publish_workers: usize, #[serde(default = "default_publish_worker_idle_seconds")] pub publish_worker_idle_seconds: u64, + #[serde(default = "default_publish_ack_mode")] + pub publish_ack_mode: String, + #[serde(default = "default_relaxed_sync_messages")] + pub relaxed_sync_messages: usize, + #[serde(default = "default_relaxed_sync_bytes")] + pub relaxed_sync_bytes: usize, + #[serde(default = "default_relaxed_sync_interval_ms")] + pub relaxed_sync_interval_ms: u64, #[serde(default)] pub detailed_queue_metrics: bool, #[serde(default = "default_max_detailed_metric_series")] @@ -349,6 +357,18 @@ fn default_max_publish_workers() -> usize { fn default_publish_worker_idle_seconds() -> u64 { 60 } +fn default_publish_ack_mode() -> String { + "durable".into() +} +fn default_relaxed_sync_messages() -> usize { + 2_500 +} +fn default_relaxed_sync_bytes() -> usize { + 8 * 1024 * 1024 +} +fn default_relaxed_sync_interval_ms() -> u64 { + 10 +} fn default_max_detailed_metric_series() -> usize { 1_000 } @@ -381,6 +401,7 @@ mod tests { assert!(schema .to_string() .contains("proxyTcpMaxConnectionAgeSeconds")); + assert!(schema.to_string().contains("publishAckMode")); assert!(schema.to_string().contains("kodoCompatibility")); assert!(schema.to_string().contains("decommissionConfirmed")); assert!(schema.to_string().contains("producerRestartNonce")); diff --git a/crates/operator/src/resources.rs b/crates/operator/src/resources.rs index 83928c4..29229d8 100644 --- a/crates/operator/src/resources.rs +++ b/crates/operator/src/resources.rs @@ -534,7 +534,7 @@ fn broker_config(cluster: &RustQueue, secret_name: &str) -> String { "" }; let mut output = format!( - "{kodo_network}[storage]\ndata_path = \"/data\"\nfeature_level = {}\nmax_segment_bytes = {max_segment_bytes}\nmin_free_bytes = {}\ndisk_high_watermark_percent = {}\ndisk_low_watermark_percent = {}\nprotective_eviction_enabled = {}\ndisk_pressure_grace_seconds = {}\nmaintenance_startup_delay_seconds = {}\n\n[queue]\nbootstrap_retention_seconds = {}\nmax_message_bytes = {}\nmax_topics = {}\nmax_publish_workers = {}\npublish_worker_idle_seconds = {}\n\n[limits]\nmax_body_bytes = {max_body_bytes}\nnode_publish_inflight_bytes = {node_publish_inflight_bytes}\nconnection_publish_inflight_bytes = {connection_publish_inflight_bytes}\nnode_delivery_inflight_bytes = {}\nconnection_delivery_inflight_bytes = {}\ndisconnect_on_retriable_publish_error = false\n\n[metrics]\ndetailed_queue_metrics = {}\nmax_detailed_series = {}\n\n[security]\nadmin_token_file = \"/run/secrets/rustqueue/admin-token\"\n{publish_token}registry_token_file = \"/run/secrets/rustqueue/registry-token\"\nconsole_token_file = \"/run/secrets/rustqueue/console-token\"\n{kodo_cleanup_token}console_management_enabled = {}\nkodo_cleanup_enabled = {}\n# secret: {secret_name}\n", + "{kodo_network}[storage]\ndata_path = \"/data\"\nfeature_level = {}\nmax_segment_bytes = {max_segment_bytes}\nmin_free_bytes = {}\ndisk_high_watermark_percent = {}\ndisk_low_watermark_percent = {}\nprotective_eviction_enabled = {}\ndisk_pressure_grace_seconds = {}\nmaintenance_startup_delay_seconds = {}\n\n[queue]\nbootstrap_retention_seconds = {}\nmax_message_bytes = {}\nmax_topics = {}\nmax_publish_workers = {}\npublish_worker_idle_seconds = {}\npublish_ack_mode = \"{}\"\nrelaxed_sync_messages = {}\nrelaxed_sync_bytes = {}\nrelaxed_sync_interval_ms = {}\n\n[limits]\nmax_body_bytes = {max_body_bytes}\nnode_publish_inflight_bytes = {node_publish_inflight_bytes}\nconnection_publish_inflight_bytes = {connection_publish_inflight_bytes}\nnode_delivery_inflight_bytes = {}\nconnection_delivery_inflight_bytes = {}\ndisconnect_on_retriable_publish_error = false\n\n[metrics]\ndetailed_queue_metrics = {}\nmax_detailed_series = {}\n\n[security]\nadmin_token_file = \"/run/secrets/rustqueue/admin-token\"\n{publish_token}registry_token_file = \"/run/secrets/rustqueue/registry-token\"\nconsole_token_file = \"/run/secrets/rustqueue/console-token\"\n{kodo_cleanup_token}console_management_enabled = {}\nkodo_cleanup_enabled = {}\n# secret: {secret_name}\n", cluster.spec.storage_feature_level, cluster.spec.min_free_bytes, cluster.spec.disk_high_watermark_percent, @@ -547,6 +547,10 @@ fn broker_config(cluster: &RustQueue, secret_name: &str) -> String { cluster.spec.max_topics, cluster.spec.max_publish_workers, cluster.spec.publish_worker_idle_seconds, + cluster.spec.publish_ack_mode, + cluster.spec.relaxed_sync_messages, + cluster.spec.relaxed_sync_bytes, + cluster.spec.relaxed_sync_interval_ms, cluster.spec.node_delivery_inflight_bytes, cluster.spec.connection_delivery_inflight_bytes, cluster.spec.detailed_queue_metrics, @@ -650,6 +654,10 @@ mod tests { max_topics: 10_000, max_publish_workers: 1_024, publish_worker_idle_seconds: 60, + publish_ack_mode: "durable".into(), + relaxed_sync_messages: 2_500, + relaxed_sync_bytes: 8 * 1024 * 1024, + relaxed_sync_interval_ms: 10, detailed_queue_metrics: false, max_detailed_metric_series: 1_000, registry_secret_name: None, @@ -710,6 +718,10 @@ mod tests { config.contains("console_token_file") && config.contains("detailed_queue_metrics = false") && config.contains("max_detailed_series = 1000") + && config.contains("publish_ack_mode = \"durable\"") + && config.contains("relaxed_sync_messages = 2500") + && config.contains("relaxed_sync_bytes = 8388608") + && config.contains("relaxed_sync_interval_ms = 10") && config.contains("maintenance_startup_delay_seconds = 30") && config.contains("node_delivery_inflight_bytes = 536870912") && config.contains("connection_delivery_inflight_bytes = 33554432") diff --git a/crates/queue/src/broker.rs b/crates/queue/src/broker.rs index 951f970..b8b05a8 100644 --- a/crates/queue/src/broker.rs +++ b/crates/queue/src/broker.rs @@ -31,14 +31,64 @@ use rustqueue_storage::{ binary_capabilities, ensure_data_format, prepare_compatibility, BinaryCapabilities, CompatibilityState, StorageError, BASE_STORAGE_FEATURE_LEVEL, }; +use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::PathBuf; +use std::str::FromStr; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use thiserror::Error; const FEATURE_LEVEL_2_MAX_MESSAGE_BYTES: usize = 100 * 1024 * 1024; +pub const RELAXED_SYNC_MIN_BYTES: usize = 4096; + +#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum PublishAckMode { + #[default] + Durable, + /// Acknowledge after append, but do not expose the message to consumers + /// until a background fsync makes it durable. + WriteAck, + /// Acknowledge and expose the message after append, before background fsync. + NsqRelaxed, +} + +impl PublishAckMode { + pub const fn is_relaxed(self) -> bool { + matches!(self, Self::WriteAck | Self::NsqRelaxed) + } + + pub const fn exposes_unsynced(self) -> bool { + matches!(self, Self::NsqRelaxed) + } + + pub const fn as_str(self) -> &'static str { + match self { + Self::Durable => "durable", + Self::WriteAck => "write_ack", + Self::NsqRelaxed => "nsq_relaxed", + } + } +} + +#[derive(Clone, Copy, Debug, Error)] +#[error("publish acknowledgement mode must be durable, write_ack, or nsq_relaxed")] +pub struct InvalidPublishAckMode; + +impl FromStr for PublishAckMode { + type Err = InvalidPublishAckMode; + + fn from_str(value: &str) -> Result { + match value { + "durable" => Ok(Self::Durable), + "write_ack" => Ok(Self::WriteAck), + "nsq_relaxed" => Ok(Self::NsqRelaxed), + _ => Err(InvalidPublishAckMode), + } + } +} #[derive(Clone, Debug)] pub struct BrokerConfig { @@ -52,6 +102,10 @@ pub struct BrokerConfig { pub max_topics: usize, pub max_publish_workers: usize, pub publish_worker_idle: Duration, + pub publish_ack_mode: PublishAckMode, + pub relaxed_sync_messages: usize, + pub relaxed_sync_bytes: usize, + pub relaxed_sync_interval: Duration, pub entry_cache_bytes: usize, pub message_index_cache_bytes: usize, pub payload_read_workers: usize, @@ -75,6 +129,10 @@ impl Default for BrokerConfig { max_topics: 10_000, max_publish_workers: 1_024, publish_worker_idle: Duration::from_secs(60), + publish_ack_mode: PublishAckMode::Durable, + relaxed_sync_messages: 2_500, + relaxed_sync_bytes: 8 * 1024 * 1024, + relaxed_sync_interval: Duration::from_millis(10), entry_cache_bytes: 64 * 1024 * 1024, message_index_cache_bytes: 64 * 1024 * 1024, payload_read_workers: 0, @@ -193,14 +251,20 @@ impl Broker { if config.max_topics == 0 || config.max_publish_workers == 0 || config.publish_worker_idle.is_zero() + || (config.publish_ack_mode.is_relaxed() + && (config.relaxed_sync_messages == 0 + || config.relaxed_sync_bytes < RELAXED_SYNC_MIN_BYTES + || config.relaxed_sync_interval.is_zero())) { return Err(BrokerError::InvalidRecord( - "topic and publish worker limits must be greater than zero".into(), + "topic and publish worker limits must be greater than zero; relaxed sync requires at least one message, 4096 bytes, and a non-zero interval" + .into(), )); } let now = Instant::now(); if now.checked_add(config.message_timeout).is_none() || now.checked_add(config.publish_worker_idle).is_none() + || now.checked_add(config.relaxed_sync_interval).is_none() { return Err(BrokerError::InvalidRecord( "broker timeouts exceed the platform timer range".into(), @@ -399,6 +463,7 @@ impl Broker { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, Some(&channel))?; let handle = broker.get_or_create_topic_locked(&topic)?; + let _commit_gate = handle.commit_gate.lock(); if handle .state .lock() @@ -420,11 +485,9 @@ impl Broker { self.storage_task(move || { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, Some(&channel))?; - broker - .topic(&topic)? - .state - .lock() - .delete_channel(&channel)?; + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + handle.state.lock().delete_channel(&channel)?; broker.bump_registry()?; Ok(()) }) @@ -438,7 +501,10 @@ impl Broker { self.storage_task(move || { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, None)?; - broker.topic(&topic)?.state.lock().set_paused(paused) + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + let result = handle.state.lock().set_paused(paused); + result }) .await } @@ -456,11 +522,10 @@ impl Broker { self.storage_task(move || { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, Some(&channel))?; - broker - .topic(&topic)? - .state - .lock() - .set_channel_paused(&channel, paused) + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + let result = handle.state.lock().set_channel_paused(&channel, paused); + result }) .await } @@ -472,7 +537,10 @@ impl Broker { self.storage_task(move || { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, None)?; - broker.topic(&topic)?.state.lock().empty_topic() + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + let result = handle.state.lock().empty_topic(); + result }) .await } @@ -485,7 +553,10 @@ impl Broker { self.storage_task(move || { let _lifecycle = broker.inner.topic_lifecycle.lock(); broker.ensure_management_access(&topic, Some(&channel))?; - broker.topic(&topic)?.state.lock().empty_channel(&channel) + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + let result = handle.state.lock().empty_channel(&channel); + result }) .await } diff --git a/crates/queue/src/broker/delivery.rs b/crates/queue/src/broker/delivery.rs index 1491d75..cd7d540 100644 --- a/crates/queue/src/broker/delivery.rs +++ b/crates/queue/src/broker/delivery.rs @@ -114,7 +114,13 @@ impl Broker { } let handle = Arc::clone(&batch.handle); let action = { + let topic_lock_started = Instant::now(); let mut topic_state = handle.state.lock(); + self.inner + .metrics + .delivery_topic_lock_wait + .observe(topic_lock_started.elapsed()); + let _topic_lock_hold = self.inner.metrics.delivery_topic_lock_hold.timer(); self.ensure_management_access(topic, Some(channel))?; let action = topic_state.reserve_batch( channel, diff --git a/crates/queue/src/broker/group_commit.rs b/crates/queue/src/broker/group_commit.rs index 981941e..b4741e3 100644 --- a/crates/queue/src/broker/group_commit.rs +++ b/crates/queue/src/broker/group_commit.rs @@ -1,5 +1,6 @@ -use super::{Broker, BrokerError, BrokerInner}; +use super::{Broker, BrokerError, BrokerInner, PublishAckMode}; use crate::model::PublishGroupCommitStats; +use crate::topic::PendingSync; use bytes::Bytes; use parking_lot::Mutex; use std::collections::HashMap; @@ -8,7 +9,7 @@ use std::sync::{Arc, Weak}; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, oneshot}; -const QUEUE_CAPACITY: usize = 1024; +const QUEUE_CAPACITY: usize = 1_024; const MAX_GROUP_REQUESTS: usize = 64; const MAX_GROUP_BYTES: usize = 8 * 1024 * 1024; const COALESCE_DELAY: Duration = Duration::from_millis(1); @@ -145,9 +146,8 @@ impl Broker { self.publish_inner(topic, bodies, delay, None).await } - /// Keeps `guard` alive until the queued body has either been rejected or - /// crossed the durable publish boundary. This makes caller-side admission - /// accounting cancellation-safe after a request enters group commit. + /// Keeps `guard` alive until the request crosses the configured publish + /// acknowledgement boundary, including cancellation after it is queued. pub async fn publish_guarded( &self, topic: &str, @@ -199,10 +199,14 @@ impl Broker { } } - fn commit_publish_group(&self, topic: &str, requests: Vec) { + fn commit_publish_group( + &self, + topic: &str, + requests: Vec, + ) -> Result<(), BrokerError> { if let Err(error) = self.ensure_storage_healthy() { fail_requests(requests, &error); - return; + return Err(error); } let message_count = requests .iter() @@ -211,26 +215,34 @@ impl Broker { let mut metadata = match self.reserve_message_metadata(message_count) { Ok(reservation) => reservation, Err(error) => { - self.observe_storage_result::<()>(Err(copy_error(&error))) - .ok(); fail_requests(requests, &error); - return; + return self.worker_result(error); } }; let handle = match self.get_or_create_topic(topic) { Ok(handle) => handle, Err(error) => { - self.observe_storage_result::<()>(Err(copy_error(&error))) - .ok(); fail_requests(requests, &error); - return; + return self.worker_result(error); } }; - let mut topic_state = handle.state.lock(); + let commit_gate = handle.commit_gate.lock(); + if let Err(error) = self.ensure_storage_healthy() { + fail_requests(requests, &error); + return Err(error); + } if let Err(error) = self.ensure_management_access(topic, None) { fail_requests(requests, &error); - return; + return Ok(()); } + let topic_lock_started = Instant::now(); + let mut topic_state = handle.state.lock(); + self.inner + .metrics + .publish_topic_lock_wait + .observe(topic_lock_started.elapsed()); + let topic_lock_hold = self.inner.metrics.publish_topic_lock_hold.timer(); + let deliverable_before = topic_state.deliverable_position(); let mut pending = Vec::with_capacity(requests.len()); let mut requests = requests.into_iter(); @@ -238,10 +250,10 @@ impl Broker { let PublishRequest { bodies, delay, + encoded_bytes, enqueued_at, reply, guard, - .. } = request; self.inner .metrics @@ -254,18 +266,19 @@ impl Broker { false, &mut metadata, ) { - Ok(ids) => pending.push(PendingPublish { - ids, - reply, - _guard: guard, - }), + Ok(ids) => { + topic_state.record_unsynced(bodies.len(), encoded_bytes); + pending.push(PendingPublish { + ids, + reply, + _guard: guard, + }); + } Err(error) if is_storage_error(&error) => { - self.observe_storage_result::<()>(Err(copy_error(&error))) - .ok(); let _ = reply.send(Err(copy_error(&error))); fail_pending(pending); fail_requests(requests, &BrokerError::StorageUnavailable); - return; + return self.worker_result(error); } Err(error) => { let _ = reply.send(Err(error)); @@ -274,34 +287,133 @@ impl Broker { } if pending.is_empty() { - return; + return Ok(()); } drop(metadata); if self.inner.message_index_cache.over_budget() { if let Err(error) = topic_state.spill_message_metadata() { - self.observe_storage_result::<()>(Err(copy_error(&error))) - .ok(); fail_pending(pending); - return; + return self.worker_result(error); + } + } + let ack_mode = self.inner.config.publish_ack_mode; + let visibility_advanced; + if ack_mode == PublishAckMode::Durable { + let durable_through = topic_state.last_position(); + let sync_file = match topic_state.clone_log_for_sync() { + Ok(file) => file, + Err(error) => { + fail_pending(pending); + return self.worker_result(error); + } + }; + drop(topic_state); + drop(topic_lock_hold); + rustqueue_storage::crash_failpoint("publish_after_append_before_fsync"); + let sync_result = { + let _timer = self.inner.metrics.fsync.timer(); + sync_file.sync_data().map_err(BrokerError::from) + }; + if let Err(error) = sync_result { + let topic_lock_started = Instant::now(); + let topic_state = handle.state.lock(); + self.inner + .metrics + .publish_topic_lock_wait + .observe(topic_lock_started.elapsed()); + let _topic_lock_hold = self.inner.metrics.publish_topic_lock_hold.timer(); + topic_state.mark_log_sync_failed(); + fail_pending(pending); + return self.worker_result(error); } + let topic_lock_started = Instant::now(); + let mut topic_state = handle.state.lock(); + self.inner + .metrics + .publish_topic_lock_wait + .observe(topic_lock_started.elapsed()); + let topic_lock_hold = self.inner.metrics.publish_topic_lock_hold.timer(); + topic_state.mark_durable_through(durable_through); + visibility_advanced = topic_state.deliverable_position() > deliverable_before; + drop(topic_state); + drop(topic_lock_hold); + rustqueue_storage::crash_failpoint("publish_after_fsync_before_reply"); + } else { + if ack_mode.exposes_unsynced() { + let deliverable_through = topic_state.last_position(); + topic_state.mark_deliverable_through(deliverable_through); + } + visibility_advanced = topic_state.deliverable_position() > deliverable_before; + drop(topic_state); + drop(topic_lock_hold); + } + + drop(commit_gate); + self.inner.publish_groups.record(pending.len()); + if visibility_advanced { + handle.signal(); + } + for pending in pending { + let _ = pending.reply.send(Ok(pending.ids)); } + Ok(()) + } + + fn sync_pending_topic(&self, topic: &str) -> Result<(), BrokerError> { + self.ensure_storage_healthy()?; + let handle = match self.topic(topic) { + Ok(handle) => handle, + Err(BrokerError::TopicNotFound) => return Ok(()), + Err(error) => return Err(error), + }; + let commit_gate = handle.commit_gate.lock(); + let topic_lock_started = Instant::now(); + let topic_state = handle.state.lock(); + self.inner + .metrics + .publish_topic_lock_wait + .observe(topic_lock_started.elapsed()); + let topic_lock_hold = self.inner.metrics.publish_topic_lock_hold.timer(); + if topic_state.pending_sync().is_none() { + return Ok(()); + } + let durable_through = topic_state.last_position(); + let sync_file = topic_state.clone_log_for_sync()?; + drop(topic_state); + drop(topic_lock_hold); rustqueue_storage::crash_failpoint("publish_after_append_before_fsync"); let sync_result = { let _timer = self.inner.metrics.fsync.timer(); - topic_state.sync_log() + sync_file.sync_data().map_err(BrokerError::from) }; if let Err(error) = sync_result { - self.observe_storage_result::<()>(Err(copy_error(&error))) - .ok(); - fail_pending(pending); - return; + let topic_state = handle.state.lock(); + topic_state.mark_log_sync_failed(); + drop(topic_state); + return self.observe_storage_result(Err(error)); } - rustqueue_storage::crash_failpoint("publish_after_fsync_before_reply"); + let mut topic_state = handle.state.lock(); + topic_state.mark_durable_through(durable_through); drop(topic_state); - self.inner.publish_groups.record(pending.len()); + drop(commit_gate); handle.signal(); - for pending in pending { - let _ = pending.reply.send(Ok(pending.ids)); + Ok(()) + } + + fn pending_topic_sync(&self, topic: &str) -> Option { + self.inner + .topics + .read() + .get(topic) + .and_then(|handle| handle.state.lock().pending_sync()) + } + + fn worker_result(&self, error: BrokerError) -> Result<(), BrokerError> { + if is_storage_error(&error) { + let _ = self.observe_storage_result::<()>(Err(copy_error(&error))); + Err(BrokerError::StorageUnavailable) + } else { + Ok(()) } } } @@ -314,21 +426,68 @@ async fn run_worker( mut receiver: mpsc::Receiver, ) { let mut carry = None; + let mut pending_sync = None; loop { + let Some(inner) = broker.upgrade() else { + return; + }; + let relaxed_ack_mode = inner.config.publish_ack_mode.is_relaxed(); + let sync_interval = inner.config.relaxed_sync_interval; + let sync_messages = inner.config.relaxed_sync_messages as u64; + let sync_bytes = inner.config.relaxed_sync_bytes as u64; + drop(inner); + + if pending_sync + .is_some_and(|pending| sync_is_due(pending, sync_messages, sync_bytes, sync_interval)) + { + if !run_pending_sync(&broker, &topic).await { + return; + } + pending_sync = pending_topic_sync(&broker, &topic, relaxed_ack_mode); + continue; + } + let first = match carry.take() { Some(request) => request, - None => match tokio::time::timeout(idle_timeout, receiver.recv()).await { - Ok(Some(request)) => request, - Ok(None) => return, - Err(_) => { - let Some(inner) = broker.upgrade() else { - return; - }; - if inner.publish_groups.retire_idle(&topic, worker_id) { - return; + None => match pending_sync { + Some(pending) => { + let deadline = tokio::time::Instant::from_std( + pending + .since + .checked_add(sync_interval) + .expect("validated relaxed sync interval"), + ); + tokio::select! { + request = receiver.recv() => match request { + Some(request) => request, + None => { + let _ = run_pending_sync(&broker, &topic).await; + return; + } + }, + _ = tokio::time::sleep_until(deadline) => { + if !run_pending_sync(&broker, &topic).await { + return; + } + pending_sync = + pending_topic_sync(&broker, &topic, relaxed_ack_mode); + continue; + } } - continue; } + None => match tokio::time::timeout(idle_timeout, receiver.recv()).await { + Ok(Some(request)) => request, + Ok(None) => return, + Err(_) => { + let Some(inner) = broker.upgrade() else { + return; + }; + if inner.publish_groups.retire_idle(&topic, worker_id) { + return; + } + continue; + } + }, }, }; let (requests, next) = collect_group(first, &mut receiver).await; @@ -338,20 +497,65 @@ async fn run_worker( }; let worker_broker = Broker { inner }; let worker_topic = topic.clone(); - if tokio::task::spawn_blocking(move || { + match tokio::task::spawn_blocking(move || { worker_broker.commit_publish_group(&worker_topic, requests) }) .await - .is_err() { + Ok(Ok(())) => {} + Ok(Err(_)) | Err(_) => { + if let Some(inner) = broker.upgrade() { + inner.storage_healthy.store(false, Ordering::Release); + } + return; + } + } + pending_sync = pending_topic_sync(&broker, &topic, relaxed_ack_mode); + } +} + +fn pending_topic_sync( + broker: &Weak, + topic: &str, + relaxed_ack_mode: bool, +) -> Option { + if !relaxed_ack_mode { + return None; + } + broker + .upgrade() + .and_then(|inner| Broker { inner }.pending_topic_sync(topic)) +} + +async fn run_pending_sync(broker: &Weak, topic: &str) -> bool { + let Some(inner) = broker.upgrade() else { + return false; + }; + let worker_broker = Broker { inner }; + let worker_topic = topic.to_owned(); + match tokio::task::spawn_blocking(move || worker_broker.sync_pending_topic(&worker_topic)).await + { + Ok(Ok(())) => true, + Ok(Err(_)) | Err(_) => { if let Some(inner) = broker.upgrade() { inner.storage_healthy.store(false, Ordering::Release); } - return; + false } } } +fn sync_is_due( + pending: PendingSync, + message_limit: u64, + byte_limit: u64, + interval: Duration, +) -> bool { + pending.messages >= message_limit + || pending.bytes >= byte_limit + || pending.since.elapsed() >= interval +} + async fn collect_group( first: PublishRequest, receiver: &mut mpsc::Receiver, @@ -484,4 +688,40 @@ mod tests { assert_eq!(group.len(), 1); assert_eq!(carry.unwrap().encoded_bytes, 3 * 1024 * 1024); } + + #[test] + fn relaxed_sync_uses_the_first_reached_threshold() { + let interval = Duration::from_millis(10); + let now = Instant::now(); + assert!(sync_is_due( + PendingSync { + messages: 2_500, + bytes: 1, + since: now, + }, + 2_500, + 8 * 1024 * 1024, + interval, + )); + assert!(sync_is_due( + PendingSync { + messages: 1, + bytes: 8 * 1024 * 1024, + since: now, + }, + 2_500, + 8 * 1024 * 1024, + interval, + )); + assert!(sync_is_due( + PendingSync { + messages: 1, + bytes: 1, + since: now - interval, + }, + 2_500, + 8 * 1024 * 1024, + interval, + )); + } } diff --git a/crates/queue/src/broker/io.rs b/crates/queue/src/broker/io.rs index 0ec08b2..3e2dfb4 100644 --- a/crates/queue/src/broker/io.rs +++ b/crates/queue/src/broker/io.rs @@ -299,6 +299,8 @@ impl Broker { self.validate_publish_request_with_limit(topic, bodies, self.durable_message_read_limit())?; let mut metadata = self.reserve_message_metadata(bodies.len())?; let handle = self.get_or_create_topic(topic)?; + let _commit_gate = handle.commit_gate.lock(); + self.ensure_storage_healthy()?; let mut state = handle.state.lock(); self.ensure_management_access(topic, None)?; let ids = self.append_publish_to_topic(&mut state, bodies, delay, true, &mut metadata)?; @@ -360,8 +362,20 @@ impl Broker { durable: bool, metadata: &mut crate::topic::index::MetadataReservation, ) -> Result, BrokerError> { - let first_position = state.next_position(); let first_id = self.reserve_ids(bodies.len())?; + self.append_reserved_publish_to_topic(state, first_id, bodies, delay, durable, metadata) + } + + pub(super) fn append_reserved_publish_to_topic( + &self, + state: &mut Topic, + first_id: u64, + bodies: &[Bytes], + delay: Duration, + durable: bool, + metadata: &mut crate::topic::index::MetadataReservation, + ) -> Result, BrokerError> { + let first_position = state.next_position(); let batch = batch::encode(first_position, first_id, bodies)?; let timestamp = now_ns(); let available = now_ms().saturating_add(delay.as_millis().min(i64::MAX as u128) as i64); @@ -499,4 +513,35 @@ mod tests { )); assert_eq!(broker.stats().topics[0].message_count, 0); } + + #[tokio::test] + async fn internal_publish_remains_durable_in_nsq_relaxed_mode() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::NsqRelaxed, + relaxed_sync_messages: usize::MAX, + relaxed_sync_bytes: usize::MAX, + relaxed_sync_interval: Duration::from_secs(60), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + + broker + .publish( + "events", + vec![Bytes::from_static(b"relaxed")], + Duration::ZERO, + ) + .await + .unwrap(); + assert_eq!(broker.stats().topics[0].unsynced_messages, 1); + broker + .publish_durable_body_sync("events", &[Bytes::from_static(b"durable")], Duration::ZERO) + .unwrap(); + let stats = broker.stats(); + assert_eq!(stats.topics[0].last_durable_position, 2); + assert_eq!(stats.topics[0].unsynced_messages, 0); + } } diff --git a/crates/queue/src/broker/maintenance.rs b/crates/queue/src/broker/maintenance.rs index d911e8e..116ed3f 100644 --- a/crates/queue/src/broker/maintenance.rs +++ b/crates/queue/src/broker/maintenance.rs @@ -48,13 +48,21 @@ impl Broker { }; let mut removed = 0; for offset in 0..selected { - let (_, topic) = &topics[(start + offset) % topics.len()]; - let mut topic = topic.state.lock(); + let (_, handle) = &topics[(start + offset) % topics.len()]; + let _commit_gate = handle.commit_gate.lock(); + let mut topic = handle.state.lock(); let retained = broker.inner.payload_reader.retained_paths(); let name = topic.name.clone(); let ids = outbox_ids.get(&name).cloned().unwrap_or_default(); - removed += + let deliverable_before = topic.deliverable_position(); + let compacted = topic.compact(broker.inner.config.bootstrap_retention, &retained, &ids)?; + let visibility_advanced = topic.deliverable_position() > deliverable_before; + drop(topic); + if visibility_advanced { + handle.signal(); + } + removed += compacted; } broker.inner.payload_reader.prune_deleted_files(); Ok(removed) @@ -77,10 +85,17 @@ impl Broker { let Some((_, topic)) = candidate else { return Ok(None); }; - let mut topic = topic.state.lock(); + let _commit_gate = topic.commit_gate.lock(); + let mut state = topic.state.lock(); let retained = broker.inner.payload_reader.retained_paths(); - let result = topic + let deliverable_before = state.deliverable_position(); + let result = state .protective_evict_oldest(&broker.inner.config.data_path.join("audit"), &retained)?; + let visibility_advanced = state.deliverable_position() > deliverable_before; + drop(state); + if visibility_advanced { + topic.signal(); + } broker.inner.payload_reader.prune_deleted_files(); Ok(result) }) @@ -180,7 +195,9 @@ impl Broker { let broker = self.clone(); self.storage_task(move || { for topic in broker.inner.topics.read().values() { + let _commit_gate = topic.commit_gate.lock(); topic.state.lock().sync()?; + topic.signal(); } Ok(()) }) diff --git a/crates/queue/src/broker/management.rs b/crates/queue/src/broker/management.rs index f890ebe..f7c0c85 100644 --- a/crates/queue/src/broker/management.rs +++ b/crates/queue/src/broker/management.rs @@ -100,15 +100,18 @@ impl Broker { } } TopicManagementAction::Pause | TopicManagementAction::Unpause => { - broker - .topic(&topic)? + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + handle .state .lock() .set_paused(action == TopicManagementAction::Pause)?; changed = true; } TopicManagementAction::Empty => { - broker.topic(&topic)?.state.lock().empty_topic()?; + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + handle.state.lock().empty_topic()?; changed = true; } TopicManagementAction::Delete | TopicManagementAction::Tombstone => { @@ -209,6 +212,7 @@ impl Broker { } else { None }; + let _idle_commit_gate = idle_handle.as_ref().map(|handle| handle.commit_gate.lock()); let mut idle_state = idle_handle.as_ref().map(|handle| handle.state.lock()); if let Some(topic_state) = idle_state.as_mut() { let channel_exists = topic_state @@ -246,6 +250,7 @@ impl Broker { match action { ChannelManagementAction::Create => { let handle = broker.get_or_create_topic_locked(&topic)?; + let _commit_gate = handle.commit_gate.lock(); changed = handle .state .lock() @@ -258,15 +263,18 @@ impl Broker { handle.signal(); } ChannelManagementAction::Pause | ChannelManagementAction::Unpause => { - broker - .topic(&topic)? + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + handle .state .lock() .set_channel_paused(&channel, action == ChannelManagementAction::Pause)?; changed = true; } ChannelManagementAction::Empty => { - broker.topic(&topic)?.state.lock().empty_channel(&channel)?; + let handle = broker.topic(&topic)?; + let _commit_gate = handle.commit_gate.lock(); + handle.state.lock().empty_channel(&channel)?; changed = true; } ChannelManagementAction::Delete | ChannelManagementAction::Tombstone => { @@ -288,6 +296,7 @@ impl Broker { changed = true; } } else if let Ok(handle) = broker.topic(&topic) { + let _commit_gate = handle.commit_gate.lock(); let mut topic_state = handle.state.lock(); let mut fences = broker.inner.fences.lock(); if set_channel_fence(&mut fences, &topic, &channel, until, require_idle) { diff --git a/crates/queue/src/broker/metadata_budget.rs b/crates/queue/src/broker/metadata_budget.rs index bc01ee7..c58ff46 100644 --- a/crates/queue/src/broker/metadata_budget.rs +++ b/crates/queue/src/broker/metadata_budget.rs @@ -20,7 +20,17 @@ impl Broker { .sort_by_key(|topic| std::cmp::Reverse(topic.state.lock().active_metadata_count())); let mut progressed = false; for topic in topics { - if topic.state.lock().spill_message_metadata()? > 0 { + let _commit_gate = topic.commit_gate.lock(); + let (spilled, visibility_advanced) = { + let mut state = topic.state.lock(); + let deliverable_before = state.deliverable_position(); + let spilled = state.spill_message_metadata()?; + (spilled, state.deliverable_position() > deliverable_before) + }; + if visibility_advanced { + topic.signal(); + } + if spilled > 0 { progressed = true; } if let Some(reservation) = self.inner.message_index_cache.try_reserve(messages) { diff --git a/crates/queue/src/broker/stats.rs b/crates/queue/src/broker/stats.rs index 5b91e45..c8ee19e 100644 --- a/crates/queue/src/broker/stats.rs +++ b/crates/queue/src/broker/stats.rs @@ -23,11 +23,12 @@ impl Broker { let mut topics = Vec::new(); let mut remaining = max_series; for handle in handles { + let _commit_gate = handle.commit_gate.lock(); let mut topic = handle.state.lock(); topic.add_aggregate_stats(&mut aggregate); - if detailed && remaining > 0 { + if detailed && remaining >= 5 { let mut stats = topic.stats(); - remaining = remaining.saturating_sub(1); + remaining = remaining.saturating_sub(5); let channel_limit = remaining / 4; stats.channels.truncate(channel_limit); remaining = remaining.saturating_sub(stats.channels.len().saturating_mul(4)); @@ -55,7 +56,10 @@ impl Broker { }; handles .into_iter() - .map(|topic| topic.state.lock().stats()) + .map(|topic| { + let _commit_gate = topic.commit_gate.lock(); + topic.state.lock().stats() + }) .collect() } diff --git a/crates/queue/src/broker/topics.rs b/crates/queue/src/broker/topics.rs index 8f3b82e..2da62c4 100644 --- a/crates/queue/src/broker/topics.rs +++ b/crates/queue/src/broker/topics.rs @@ -84,7 +84,9 @@ impl Broker { let Some(handle) = self.inner.topics.read().get(name).cloned() else { return Ok(false); }; + let commit_gate = handle.commit_gate.lock(); handle.state.lock().mark_deleted()?; + drop(commit_gate); self.inner.topics.write().remove(name); let directory = topic_directory(&self.inner.config.data_path, name); if Arc::strong_count(&handle) == 1 diff --git a/crates/queue/src/broker_tests.rs b/crates/queue/src/broker_tests.rs index d43118a..e394261 100644 --- a/crates/queue/src/broker_tests.rs +++ b/crates/queue/src/broker_tests.rs @@ -77,6 +77,372 @@ async fn cancelled_publish_keeps_admission_guard_until_commit_finishes() { .unwrap(); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn relaxed_publish_waits_for_append_before_acknowledging() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::WriteAck, + relaxed_sync_interval: Duration::from_secs(60), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_topic("events").await.unwrap(); + + let handle = broker.topic("events").unwrap(); + let (locked_tx, locked_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let lock_thread = std::thread::spawn(move || { + let _lock = handle.commit_gate.lock(); + locked_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + locked_rx.recv().unwrap(); + + let mut publish = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"write-ack")], + Duration::ZERO, + ) + .await + } + }); + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut publish) + .await + .is_err(), + "write_ack must not acknowledge queue admission before append" + ); + + release_tx.send(()).unwrap(); + lock_thread.join().unwrap(); + assert_eq!(publish.await.unwrap().unwrap().len(), 1); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn queued_publish_rechecks_storage_health_before_append() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::WriteAck, + relaxed_sync_interval: Duration::from_secs(60), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_topic("events").await.unwrap(); + + let handle = broker.topic("events").unwrap(); + let (locked_tx, locked_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let lock_thread = std::thread::spawn(move || { + let _lock = handle.commit_gate.lock(); + locked_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + locked_rx.recv().unwrap(); + + let mut publish = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"must-not-append")], + Duration::ZERO, + ) + .await + } + }); + tokio::time::timeout(Duration::from_secs(1), async { + while broker.inner.publish_groups.stats().active_workers == 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut publish) + .await + .is_err(), + "the publish must be waiting at the Topic commit boundary" + ); + broker + .observe_storage_result::<()>(Err(BrokerError::StorageUnavailable)) + .unwrap_err(); + + release_tx.send(()).unwrap(); + lock_thread.join().unwrap(); + assert!(matches!( + publish.await.unwrap(), + Err(BrokerError::StorageUnavailable) + )); + assert_eq!( + broker.filtered_stats(Some("events"), None).topics[0].message_count, + 0 + ); +} + +#[tokio::test] +async fn write_ack_hides_the_unsynced_tail_until_flush() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::WriteAck, + relaxed_sync_messages: usize::MAX, + relaxed_sync_bytes: usize::MAX, + relaxed_sync_interval: Duration::from_secs(60), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + + let ids = broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"pending")], + Duration::ZERO, + ) + .await + .unwrap(); + let stats = broker.filtered_stats(Some("events"), None); + assert_eq!(stats.topics[0].last_durable_position, 0); + assert_eq!(stats.topics[0].unsynced_messages, 1); + assert!(stats.topics[0].unsynced_bytes > 0); + assert!(broker + .next_message("events", "workers", Some(Duration::ZERO)) + .await + .unwrap() + .is_none()); + + broker.flush().await.unwrap(); + let delivery = broker + .next_message("events", "workers", Some(Duration::ZERO)) + .await + .unwrap() + .unwrap(); + assert_eq!(delivery.id, ids[0]); + assert_eq!(&*delivery.body, b"pending"); + let stats = broker.filtered_stats(Some("events"), None); + assert_eq!(stats.topics[0].last_durable_position, 1); + assert_eq!(stats.topics[0].unsynced_messages, 0); + assert_eq!(stats.topics[0].unsynced_bytes, 0); +} + +#[tokio::test] +async fn nsq_relaxed_exposes_the_unsynced_tail_immediately() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::NsqRelaxed, + relaxed_sync_messages: usize::MAX, + relaxed_sync_bytes: usize::MAX, + relaxed_sync_interval: Duration::from_secs(60), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + + let ids = broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"visible")], + Duration::ZERO, + ) + .await + .unwrap(); + let stats = broker.filtered_stats(Some("events"), None); + assert_eq!(stats.topics[0].last_durable_position, 0); + assert_eq!(stats.topics[0].unsynced_messages, 1); + let delivery = broker + .next_message("events", "workers", Some(Duration::ZERO)) + .await + .unwrap() + .unwrap(); + assert_eq!(delivery.id, ids[0]); + assert_eq!(&*delivery.body, b"visible"); +} + +#[tokio::test] +async fn write_ack_timer_makes_the_tail_durable_and_visible() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::WriteAck, + relaxed_sync_messages: usize::MAX, + relaxed_sync_bytes: usize::MAX, + relaxed_sync_interval: Duration::from_millis(10), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"timer")], + Duration::ZERO, + ) + .await + .unwrap(); + + let delivery = broker + .fetch_batch( + "events", + "workers", + 1, + usize::MAX, + Duration::from_secs(1), + None, + ) + .await + .expect("the relaxed sync interval must flush an idle Topic") + .pop() + .unwrap(); + assert_eq!(&*delivery.body, b"timer"); + assert_eq!( + broker.filtered_stats(Some("events"), None).topics[0].unsynced_messages, + 0 + ); +} + +#[tokio::test] +async fn relaxed_sync_failure_stops_new_writes() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + publish_ack_mode: PublishAckMode::WriteAck, + relaxed_sync_messages: usize::MAX, + relaxed_sync_bytes: usize::MAX, + relaxed_sync_interval: Duration::from_millis(100), + ..BrokerConfig::default() + }) + .unwrap(); + broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"pending")], + Duration::ZERO, + ) + .await + .unwrap(); + broker + .topic("events") + .unwrap() + .state + .lock() + .mark_log_sync_failed(); + + tokio::time::timeout(Duration::from_secs(1), async { + while broker.storage_healthy() { + tokio::task::yield_now().await; + } + }) + .await + .expect("a failed background sync must isolate the broker"); + assert!(matches!( + broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"rejected")], + Duration::ZERO, + ) + .await, + Err(BrokerError::StorageUnavailable) + )); +} + +#[tokio::test] +async fn pending_publish_is_not_reservable_before_its_fsync_completes() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + + let handle = broker.topic("events").unwrap(); + let mut metadata = broker.reserve_message_metadata(1).unwrap(); + let (sync_file, durable_through) = { + let _commit_gate = handle.commit_gate.lock(); + let mut state = handle.state.lock(); + broker + .append_publish_to_topic( + &mut state, + &[bytes::Bytes::from_static(b"pending")], + Duration::ZERO, + false, + &mut metadata, + ) + .unwrap(); + let durable_through = state.last_position(); + let sync_file = state.clone_log_for_sync().unwrap(); + (sync_file, durable_through) + }; + + assert!(broker + .next_message("events", "workers", None) + .await + .unwrap() + .is_none()); + + sync_file.sync_data().unwrap(); + handle.state.lock().mark_durable_through(durable_through); + handle.signal(); + + let message = broker + .next_message("events", "workers", None) + .await + .unwrap() + .unwrap(); + assert_eq!(&*message.body, b"pending"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn delivery_reservation_does_not_wait_for_the_publish_commit_gate() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + broker + .publish( + "events", + vec![bytes::Bytes::from_static(b"already-durable")], + Duration::ZERO, + ) + .await + .unwrap(); + + let handle = broker.topic("events").unwrap(); + let (locked_tx, locked_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let lock_thread = std::thread::spawn(move || { + // This is the interval after append where a publisher is syncing its + // file outside the Topic state lock. + let _commit_gate = handle.commit_gate.lock(); + locked_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + }); + locked_rx.recv().unwrap(); + + let delivery = tokio::time::timeout( + Duration::from_secs(1), + broker.next_message("events", "workers", None), + ) + .await; + release_tx.send(()).unwrap(); + lock_thread.join().unwrap(); + + let message = delivery.unwrap().unwrap().unwrap(); + assert_eq!(&*message.body, b"already-durable"); +} + #[tokio::test] async fn startup_replays_dlq_outbox_before_finishing_the_source() { let root = tempdir().unwrap(); @@ -1359,6 +1725,63 @@ async fn broker_metadata_budget_spills_active_tails_across_topics() { assert!(broker.inner.message_index_cache.resident_bytes() <= 64 * 1024); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn metadata_spill_wakes_a_write_ack_consumer() { + let root = tempdir().unwrap(); + let broker = Broker::open(BrokerConfig { + data_path: root.path().into(), + message_index_cache_bytes: 128 * 1024, + publish_ack_mode: PublishAckMode::WriteAck, + relaxed_sync_messages: usize::MAX, + relaxed_sync_bytes: usize::MAX, + relaxed_sync_interval: Duration::from_secs(60), + ..BrokerConfig::default() + }) + .unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + broker + .publish("events", vec![vec![1u8]; 64], Duration::ZERO) + .await + .unwrap(); + assert_eq!( + broker.filtered_stats(Some("events"), None).topics[0].unsynced_messages, + 64 + ); + + let waiting = tokio::spawn({ + let broker = broker.clone(); + async move { + broker + .fetch_batch( + "events", + "workers", + 1, + usize::MAX, + Duration::from_secs(1), + None, + ) + .await + } + }); + tokio::time::sleep(Duration::from_millis(20)).await; + broker + .publish("other", vec![vec![2u8]; 1_024], Duration::ZERO) + .await + .unwrap(); + let stats = broker.filtered_stats(Some("events"), Some("workers")); + assert_eq!(stats.topics[0].unsynced_messages, 0); + assert_eq!(stats.topics[0].last_durable_position, 64); + assert_eq!(stats.topics[0].channels[0].depth, 64); + + let delivery = waiting + .await + .unwrap() + .unwrap() + .pop() + .expect("metadata spill must wake consumers after making the tail durable"); + assert_eq!(&*delivery.body, &[1]); +} + #[tokio::test] async fn concurrent_publishes_wait_for_metadata_spill_instead_of_rejecting() { let root = tempdir().unwrap(); diff --git a/crates/queue/src/channel.rs b/crates/queue/src/channel.rs index b9ba2e0..6f95be4 100644 --- a/crates/queue/src/channel.rs +++ b/crates/queue/src/channel.rs @@ -3,6 +3,7 @@ use crate::model::ChannelStats; use crate::BrokerError; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::sync::Arc; use std::time::{Duration, Instant}; #[derive(Clone, Debug, Deserialize, Serialize)] @@ -77,12 +78,16 @@ pub(crate) struct ChannelState { next_position: u64, in_flight: HashMap, in_flight_ids: HashMap, + in_flight_deadlines: BTreeSet<(Instant, u64, u64)>, redelivery: BTreeSet, attempts: HashMap, next_token: u64, max_ack_gap: usize, requeue_count: u64, timeout_count: u64, + // Rebuilt from Topic segment ranges on open. Lost relaxed positions are + // never reused, so this shared index needs no additional v7 persistence. + absent_ranges: Arc<[(u64, u64)]>, } pub(crate) enum MessageAvailability { @@ -111,12 +116,14 @@ impl ChannelState { next_position: barrier_position.saturating_add(1), in_flight: HashMap::new(), in_flight_ids: HashMap::new(), + in_flight_deadlines: BTreeSet::new(), redelivery: BTreeSet::new(), attempts: HashMap::new(), next_token: 1, max_ack_gap: max_ack_gap.max(1), requeue_count: 0, timeout_count: 0, + absent_ranges: Arc::from(Vec::new()), } } @@ -149,12 +156,14 @@ impl ChannelState { next_position, in_flight: HashMap::new(), in_flight_ids: HashMap::new(), + in_flight_deadlines: BTreeSet::new(), redelivery, attempts: checkpoint.attempts.into_iter().collect(), next_token: 1, max_ack_gap: max_ack_gap.max(1), requeue_count: checkpoint.requeue_count, timeout_count: checkpoint.timeout_count, + absent_ranges: Arc::from(Vec::new()), }) } @@ -242,6 +251,13 @@ impl ChannelState { let mut absent = Vec::new(); let redelivery: Vec<_> = self.redelivery.iter().copied().collect(); for position in redelivery { + if self.is_absent(position) { + self.redelivery.remove(&position); + self.requeued_until.remove(&position); + self.attempts.remove(&position); + self.advance_ack_floor(); + continue; + } if self .requeued_until .get(&position) @@ -258,7 +274,10 @@ impl ChannelState { return NextCandidate::Ready(position); } MessageAvailability::Missing => return NextCandidate::Load(position), - MessageAvailability::Absent => absent.push(position), + MessageAvailability::Absent => { + absent.push(position); + self.acknowledge(position); + } MessageAvailability::Ready(_) => {} } } @@ -267,11 +286,12 @@ impl ChannelState { self.requeued_until.remove(&position); } while self.next_position <= last_position { - if self.next_position - > self - .ack_floor_position - .saturating_add(self.max_ack_gap as u64) - { + if let Some(end) = self.absent_range_end(self.next_position) { + self.next_position = end.saturating_add(1); + self.advance_ack_floor(); + continue; + } + if self.present_distance(self.next_position) > self.max_ack_gap as u64 { return NextCandidate::None; } let position = self.next_position; @@ -288,6 +308,7 @@ impl ChannelState { MessageAvailability::Missing => return NextCandidate::Load(position), MessageAvailability::Absent => { self.next_position = self.next_position.saturating_add(1); + self.acknowledge(position); continue; } }; @@ -302,19 +323,50 @@ impl ChannelState { NextCandidate::None } + pub fn set_absent_ranges(&mut self, ranges: Arc<[(u64, u64)]>) { + debug_assert!(ranges + .windows(2) + .all(|pair| pair[0].0 <= pair[0].1 && pair[0].1 < pair[1].0)); + debug_assert!(ranges.last().is_none_or(|range| range.0 <= range.1)); + debug_assert!(self.in_flight.is_empty()); + self.acknowledged + .retain(|position| !position_in_ranges(&ranges, *position)); + self.redelivery + .retain(|position| !position_in_ranges(&ranges, *position)); + self.requeued_until + .retain(|position, _| !position_in_ranges(&ranges, *position)); + self.attempts + .retain(|position, _| !position_in_ranges(&ranges, *position)); + self.absent_ranges = ranges; + self.advance_ack_floor(); + } + + pub fn recovered_position_high_watermark(&self) -> u64 { + self.ack_floor_position + .max(self.acknowledged.last().copied().unwrap_or(0)) + .max( + self.requeued_until + .last_key_value() + .map_or(0, |(position, _)| *position), + ) + .max(self.attempts.keys().copied().max().unwrap_or(0)) + } + pub fn reserve(&mut self, position: u64, id: u64, timeout: Duration) -> (u64, u16) { let token = self.next_token; self.next_token = self.next_token.wrapping_add(1).max(1); let attempts = self.attempts.entry(position).or_insert(0); *attempts = attempts.saturating_add(1); + let deadline = Instant::now() + timeout; self.in_flight.insert( position, InFlight { id, - deadline: Instant::now() + timeout, + deadline, token, }, ); + self.in_flight_deadlines.insert((deadline, position, token)); self.in_flight_ids.insert(id, position); (token, *attempts) } @@ -369,10 +421,17 @@ impl ChannelState { } pub fn touch_until(&mut self, position: u64, deadline: Instant) -> bool { - let Some(flight) = self.in_flight.get_mut(&position) else { + let Some(flight) = self.in_flight.get(&position) else { return false; }; - flight.deadline = deadline; + self.in_flight_deadlines + .remove(&(flight.deadline, position, flight.token)); + let token = flight.token; + self.in_flight + .get_mut(&position) + .expect("in-flight delivery remains present") + .deadline = deadline; + self.in_flight_deadlines.insert((deadline, position, token)); true } @@ -393,12 +452,18 @@ impl ChannelState { } fn expired_positions(&self, now: Instant) -> Vec { - self.in_flight - .iter() - .filter_map(|(position, flight)| (flight.deadline <= now).then_some(*position)) + self.in_flight_deadlines + .range(..=(now, u64::MAX, u64::MAX)) + .map(|(_, position, _)| *position) .collect() } + fn has_expired_in_flight(&self, now: Instant) -> bool { + self.in_flight_deadlines + .first() + .is_some_and(|(deadline, _, _)| *deadline <= now) + } + fn expire_positions(&mut self, positions: &[u64]) { for position in positions { self.remove_in_flight(*position); @@ -439,7 +504,11 @@ impl ChannelState { scheduled: &BTreeSet, now_ms: i64, ) -> (u64, u64, u64, u64) { - let total = last_position.saturating_sub(self.ack_floor_position); + let total = last_position + .saturating_sub(self.ack_floor_position) + .saturating_sub( + self.absent_count(self.ack_floor_position.saturating_add(1), last_position), + ); let scheduled_count = scheduled .iter() .filter(|position| self.is_outstanding(**position, last_position)) @@ -464,6 +533,7 @@ impl ChannelState { fn is_outstanding(&self, position: u64, last_position: u64) -> bool { position > self.ack_floor_position && position <= last_position + && !self.is_absent(position) && !self.acknowledged.contains(&position) && !self.in_flight.contains_key(&position) } @@ -477,26 +547,89 @@ impl ChannelState { return; } self.acknowledged.insert(position); - while self - .acknowledged - .remove(&self.ack_floor_position.saturating_add(1)) - { - self.ack_floor_position = self.ack_floor_position.saturating_add(1); + self.advance_ack_floor(); + } + + fn advance_ack_floor(&mut self) { + loop { + let next = self.ack_floor_position.saturating_add(1); + if next == self.ack_floor_position { + break; + } + if let Some(end) = self.absent_range_end(next) { + self.ack_floor_position = end; + continue; + } + if self.acknowledged.remove(&next) { + self.ack_floor_position = next; + continue; + } + break; + } + self.next_position = self + .next_position + .max(self.ack_floor_position.saturating_add(1)); + } + + fn is_absent(&self, position: u64) -> bool { + self.absent_range_end(position).is_some() + } + + fn absent_range_end(&self, position: u64) -> Option { + let index = self + .absent_ranges + .partition_point(|(_, end)| *end < position); + self.absent_ranges + .get(index) + .filter(|(start, _)| position >= *start) + .map(|(_, end)| *end) + } + + fn absent_count(&self, first: u64, last: u64) -> u64 { + if first > last { + return 0; } + self.absent_ranges + .iter() + .map(|(range_first, range_last)| { + let overlap_first = first.max(*range_first); + let overlap_last = last.min(*range_last); + if overlap_first <= overlap_last { + overlap_last.saturating_sub(overlap_first).saturating_add(1) + } else { + 0 + } + }) + .fold(0u64, u64::saturating_add) + } + + fn present_distance(&self, position: u64) -> u64 { + position + .saturating_sub(self.ack_floor_position) + .saturating_sub(self.absent_count(self.ack_floor_position.saturating_add(1), position)) } fn remove_in_flight(&mut self, position: u64) -> bool { let Some(delivery) = self.in_flight.remove(&position) else { return false; }; + self.in_flight_deadlines + .remove(&(delivery.deadline, position, delivery.token)); self.in_flight_ids.remove(&delivery.id); true } } +fn position_in_ranges(ranges: &[(u64, u64)], position: u64) -> bool { + let index = ranges.partition_point(|(_, end)| *end < position); + ranges + .get(index) + .is_some_and(|(start, _)| position >= *start) +} + impl ChannelRuntime { pub fn has_expired_in_flight(&self) -> bool { - !self.state.expired_positions(Instant::now()).is_empty() + self.state.has_expired_in_flight(Instant::now()) } pub fn expire_in_flight(&mut self) -> Result { @@ -530,6 +663,45 @@ impl ChannelRuntime { mod tests { use super::*; + #[test] + fn recovered_position_gaps_do_not_consume_the_ack_window() { + let mut channel = ChannelState::new("workers".into(), 0, false, 2); + channel.set_absent_ranges(Arc::from(vec![(2, 10)])); + let stats = channel.stats(11, &BTreeSet::new(), i64::MAX); + assert_eq!(stats.depth, 2); + assert_eq!(stats.message_count, 11); + + assert!(matches!( + channel.next_candidate(0, 11, |position| match position { + 1 | 11 => MessageAvailability::Ready(0), + _ => panic!("known gaps must not be looked up"), + }), + NextCandidate::Ready(1) + )); + channel.reserve(1, 100, Duration::from_secs(30)); + assert!(matches!( + channel.next_candidate(0, 11, |position| match position { + 1 | 11 => MessageAvailability::Ready(0), + _ => panic!("known gaps must not be looked up"), + }), + NextCandidate::Ready(11) + )); + channel.reserve(11, 110, Duration::from_secs(30)); + channel.apply(&ChannelCommand::Finish { + position: 11, + message_id: 110, + }); + channel.apply(&ChannelCommand::Finish { + position: 1, + message_id: 100, + }); + + let stats = channel.stats(11, &BTreeSet::new(), i64::MAX); + assert_eq!(stats.ack_cursor, 11); + assert_eq!(stats.ack_gap, 0); + assert_eq!(stats.depth, 0); + } + #[test] fn cancelling_an_unhanded_delivery_restores_attempt_count() { let mut channel = ChannelState::new("workers".into(), 0, false, 16); @@ -575,6 +747,23 @@ mod tests { ); } + #[test] + fn in_flight_deadline_index_tracks_touch_and_finish() { + let mut channel = ChannelState::new("workers".into(), 0, false, 16); + channel.reserve(1, 10, Duration::from_secs(60)); + channel.reserve(2, 20, Duration::from_secs(60)); + + let now = Instant::now(); + assert!(channel.touch_until(1, now - Duration::from_secs(1))); + assert_eq!(channel.expired_positions(now), vec![1]); + + channel.apply(&ChannelCommand::Finish { + position: 1, + message_id: 10, + }); + assert!(channel.expired_positions(now).is_empty()); + } + #[test] fn cumulative_counters_and_message_count_survive_checkpoint_and_empty() { let mut channel = ChannelState::new("workers".into(), 0, false, 16); diff --git a/crates/queue/src/lib.rs b/crates/queue/src/lib.rs index 1bc31e4..9a1498e 100644 --- a/crates/queue/src/lib.rs +++ b/crates/queue/src/lib.rs @@ -14,7 +14,10 @@ mod payload_reader; mod telemetry; mod topic; -pub use broker::{Broker, BrokerConfig, BrokerError}; +pub use broker::{ + Broker, BrokerConfig, BrokerError, InvalidPublishAckMode, PublishAckMode, + RELAXED_SYNC_MIN_BYTES, +}; pub use delivery_budget::DeliveryHold; pub use delivery_guard::DeliveryGuard; pub use eviction::ProtectiveEviction; diff --git a/crates/queue/src/model.rs b/crates/queue/src/model.rs index 7fb5730..707c107 100644 --- a/crates/queue/src/model.rs +++ b/crates/queue/src/model.rs @@ -74,6 +74,9 @@ pub struct QueueAggregateStats { pub message_count: u64, pub segment_count: u64, pub segment_bytes: u64, + pub unsynced_messages: u64, + pub unsynced_bytes: u64, + pub sync_lag_ms: u64, pub channel_count: u64, pub channel_depth: u64, pub channel_in_flight: u64, @@ -87,6 +90,11 @@ impl QueueAggregateStats { self.message_count = self.message_count.saturating_add(topic.message_count); self.segment_count = self.segment_count.saturating_add(topic.segment_count); self.segment_bytes = self.segment_bytes.saturating_add(topic.segment_bytes); + self.unsynced_messages = self + .unsynced_messages + .saturating_add(topic.unsynced_messages); + self.unsynced_bytes = self.unsynced_bytes.saturating_add(topic.unsynced_bytes); + self.sync_lag_ms = self.sync_lag_ms.max(topic.sync_lag_ms); for channel in &topic.channels { self.channel_count = self.channel_count.saturating_add(1); self.channel_depth = self.channel_depth.saturating_add(channel.depth); @@ -103,7 +111,11 @@ impl QueueAggregateStats { pub struct BrokerLatencyStats { pub fsync: HistogramSnapshot, pub group_commit_wait: HistogramSnapshot, + pub publish_topic_lock_wait: HistogramSnapshot, + pub publish_topic_lock_hold: HistogramSnapshot, pub publish_ack: HistogramSnapshot, + pub delivery_topic_lock_wait: HistogramSnapshot, + pub delivery_topic_lock_hold: HistogramSnapshot, pub channel_fsync: HistogramSnapshot, pub channel_group_commit_wait: HistogramSnapshot, pub channel_ack: HistogramSnapshot, @@ -146,6 +158,14 @@ pub struct TopicStats { pub segment_count: u64, #[serde(default)] pub segment_bytes: u64, + #[serde(default)] + pub last_durable_position: u64, + #[serde(default)] + pub unsynced_messages: u64, + #[serde(default)] + pub unsynced_bytes: u64, + #[serde(default)] + pub sync_lag_ms: u64, pub channels: Vec, } diff --git a/crates/queue/src/telemetry.rs b/crates/queue/src/telemetry.rs index c483a18..0344919 100644 --- a/crates/queue/src/telemetry.rs +++ b/crates/queue/src/telemetry.rs @@ -6,7 +6,11 @@ use std::sync::Arc; pub(crate) struct QueueMetrics { pub fsync: Arc, pub group_commit_wait: Arc, + pub publish_topic_lock_wait: Arc, + pub publish_topic_lock_hold: Arc, pub publish_ack: Arc, + pub delivery_topic_lock_wait: Arc, + pub delivery_topic_lock_hold: Arc, pub channel_fsync: Arc, pub channel_group_commit_wait: Arc, pub channel_ack: Arc, @@ -20,7 +24,11 @@ impl QueueMetrics { BrokerLatencyStats { fsync: self.fsync.snapshot(), group_commit_wait: self.group_commit_wait.snapshot(), + publish_topic_lock_wait: self.publish_topic_lock_wait.snapshot(), + publish_topic_lock_hold: self.publish_topic_lock_hold.snapshot(), publish_ack: self.publish_ack.snapshot(), + delivery_topic_lock_wait: self.delivery_topic_lock_wait.snapshot(), + delivery_topic_lock_hold: self.delivery_topic_lock_hold.snapshot(), channel_fsync: self.channel_fsync.snapshot(), channel_group_commit_wait: self.channel_group_commit_wait.snapshot(), channel_ack: self.channel_ack.snapshot(), diff --git a/crates/queue/src/topic.rs b/crates/queue/src/topic.rs index 130ee7c..b9e0f13 100644 --- a/crates/queue/src/topic.rs +++ b/crates/queue/src/topic.rs @@ -19,13 +19,22 @@ use parking_lot::Mutex; use rustqueue_protocol::validate_name; use rustqueue_storage::{RecordHeader, RecordKind, SegmentLog}; use std::collections::HashMap; +use std::fs::File; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; pub(crate) const MAX_CHANNELS_PER_TOPIC: usize = 1_024; +#[derive(Clone, Copy, Debug)] +pub(crate) struct PendingSync { + pub messages: u64, + pub bytes: u64, + pub since: Instant, +} + pub(crate) struct TopicHandle { + pub commit_gate: Mutex<()>, pub state: Mutex, pub wake: tokio::sync::watch::Sender, } @@ -37,6 +46,12 @@ pub(crate) struct Topic { manifest: TopicManifest, log: SegmentLog, messages: MessageIndex, + position_gaps: Arc<[(u64, u64)]>, + deliverable_position: u64, + durable_position: u64, + unsynced_messages: u64, + unsynced_bytes: u64, + unsynced_since: Option, channels: HashMap, max_ack_gap: usize, durable_channel_counters: bool, @@ -69,6 +84,7 @@ impl TopicHandle { } topic.reconcile_unrouted_boundary()?; Ok(Arc::new(Self { + commit_gate: Mutex::new(()), state: Mutex::new(topic), wake, })) @@ -101,6 +117,7 @@ impl TopicHandle { )?; let (wake, _) = tokio::sync::watch::channel(0); Ok(Arc::new(Self { + commit_gate: Mutex::new(()), state: Mutex::new(Topic { name: name.into(), directory: directory.into(), @@ -108,6 +125,12 @@ impl TopicHandle { manifest, log, messages: MessageIndex::new(index_cache), + position_gaps: Arc::from(Vec::new()), + deliverable_position: 0, + durable_position: 0, + unsynced_messages: 0, + unsynced_bytes: 0, + unsynced_since: None, channels: HashMap::new(), max_ack_gap, durable_channel_counters: storage_feature_level >= 2, @@ -198,6 +221,7 @@ impl Topic { let recovered_next = messages .last_position() .map_or(1, |position| position.saturating_add(1)); + let deliverable_position = messages.last_position().unwrap_or(0); if manifest.next_position < recovered_next { manifest.next_position = recovered_next; store_atomic(&manifest_path, &manifest)?; @@ -210,6 +234,12 @@ impl Topic { manifest, log, messages, + position_gaps: Arc::from(Vec::new()), + deliverable_position, + durable_position: deliverable_position, + unsynced_messages: 0, + unsynced_bytes: 0, + unsynced_since: None, channels: HashMap::new(), max_ack_gap, durable_channel_counters: storage_feature_level >= 2, @@ -239,6 +269,40 @@ impl Topic { }, ); } + let recovered_next = self + .messages + .last_position() + .map_or(1, |position| position.saturating_add(1)); + let channel_next = self + .channels + .values() + .map(|channel| channel.state.recovered_position_high_watermark()) + .max() + .unwrap_or(0) + .checked_add(1) + .ok_or_else(|| { + BrokerError::InvalidRecord("channel position range is exhausted".into()) + })?; + if self.manifest.next_position < channel_next { + self.manifest.next_position = channel_next; + store_atomic(&self.manifest_path, &self.manifest)?; + self.published_count = self.manifest.next_position.saturating_sub(1); + } + self.position_gaps = Arc::from( + self.messages + .position_gaps(self.manifest.next_position.saturating_sub(1)), + ); + for channel in self.channels.values_mut() { + channel + .state + .set_absent_ranges(Arc::clone(&self.position_gaps)); + } + if self.manifest.next_position > recovered_next && self.active_metadata_count() > 0 { + // A relaxed tail can disappear after its position was recorded by + // Topic metadata or a durable Channel command. Put the surviving + // prefix in its own segment so later appends preserve that gap. + self.spill_message_metadata()?; + } Ok(()) } @@ -271,6 +335,7 @@ impl Topic { message_id: first_id, available_at_ms, }; + let previous_last_position = self.last_position(); let previous_segment = self.log.current_segment_path().to_path_buf(); let parts = batch.parts(); let location = self @@ -284,22 +349,67 @@ impl Topic { .saturating_add(batch.entries.len() as u64); if previous_segment != self.log.current_segment_path() { self.persist_segment_index(&previous_segment)?; + self.durable_position = self.durable_position.max(previous_last_position); + self.deliverable_position = self.deliverable_position.max(previous_last_position); + self.unsynced_messages = 0; + self.unsynced_bytes = 0; + self.unsynced_since = None; } self.manifest.next_position = self .manifest .next_position .saturating_add(batch.entries.len() as u64); + if durable { + self.mark_durable_through(self.last_position()); + } Ok(ids) } - pub fn sync_log(&self) -> Result<(), BrokerError> { - self.log.sync()?; - Ok(()) + pub fn clone_log_for_sync(&self) -> Result { + Ok(self.log.clone_current_for_sync()?) + } + + pub fn mark_log_sync_failed(&self) { + self.log.mark_sync_failed(); + } + + pub fn mark_deliverable_through(&mut self, position: u64) { + debug_assert!(position <= self.last_position()); + self.deliverable_position = self.deliverable_position.max(position); + } + + pub fn record_unsynced(&mut self, messages: usize, bytes: usize) { + if messages == 0 || self.durable_position >= self.written_position() { + return; + } + self.unsynced_messages = self.unsynced_messages.saturating_add(messages as u64); + self.unsynced_bytes = self.unsynced_bytes.saturating_add(bytes as u64); + self.unsynced_since.get_or_insert_with(Instant::now); + } + + pub fn pending_sync(&self) -> Option { + (self.unsynced_messages > 0).then(|| PendingSync { + messages: self.unsynced_messages, + bytes: self.unsynced_bytes, + since: self.unsynced_since.unwrap_or_else(Instant::now), + }) + } + + pub fn mark_durable_through(&mut self, position: u64) { + debug_assert!(position <= self.last_position()); + self.durable_position = self.durable_position.max(position); + self.deliverable_position = self.deliverable_position.max(position); + if self.durable_position >= self.written_position() { + self.unsynced_messages = 0; + self.unsynced_bytes = 0; + self.unsynced_since = None; + } } fn seal_log(&mut self) -> Result<(), BrokerError> { let previous_segment = self.log.current_segment_path().to_path_buf(); self.log.seal()?; + self.mark_durable_through(self.last_position()); if previous_segment != self.log.current_segment_path() { self.persist_segment_index(&previous_segment)?; } @@ -338,6 +448,12 @@ impl Topic { pub fn last_position(&self) -> u64 { self.manifest.next_position.saturating_sub(1) } + fn written_position(&self) -> u64 { + self.messages.last_position().unwrap_or(0) + } + pub fn deliverable_position(&self) -> u64 { + self.deliverable_position + } fn has_durable_channels(&self) -> bool { self.channels @@ -422,7 +538,8 @@ impl Topic { .retain_from_timestamp(cutoff, self.manifest.next_position) .saturating_sub(1) }; - let state = ChannelState::new(name.into(), barrier, ephemeral, self.max_ack_gap); + let mut state = ChannelState::new(name.into(), barrier, ephemeral, self.max_ack_gap); + state.set_absent_ranges(Arc::clone(&self.position_gaps)); let store = if ephemeral { None } else { @@ -484,7 +601,7 @@ impl Topic { pub fn channel_counts(&mut self, channel: &str) -> Result<(u64, u64, u64), BrokerError> { let now_ms = now_ms(); let scheduled = self.messages.deferred_positions(now_ms); - let last_position = self.last_position(); + let last_position = self.deliverable_position; let channel = self .channels .get(channel) diff --git a/crates/queue/src/topic/delivery.rs b/crates/queue/src/topic/delivery.rs index 0b097fb..11782a6 100644 --- a/crates/queue/src/topic/delivery.rs +++ b/crates/queue/src/topic/delivery.rs @@ -26,7 +26,7 @@ impl Topic { if self.manifest.paused { return Ok(ReserveBatch::Ready(Vec::new())); } - let last = self.last_position(); + let last = self.deliverable_position; let messages = &self.messages; let channel = self .channels diff --git a/crates/queue/src/topic/index.rs b/crates/queue/src/topic/index.rs index c822655..a43c0e4 100644 --- a/crates/queue/src/topic/index.rs +++ b/crates/queue/src/topic/index.rs @@ -234,6 +234,37 @@ impl MessageIndex { .or_else(|| self.active.front().map(|message| message.position)) } + pub(crate) fn position_gaps(&self, last_position: u64) -> Vec<(u64, u64)> { + let mut gaps = Vec::new(); + let mut next = 1u64; + let active = self + .active + .front() + .zip(self.active.back()) + .map(|(first, last)| (first.position, last.position)); + for (first, last) in self + .sealed + .iter() + .map(|segment| (segment.first_position, segment.last_position)) + .chain(active) + { + if first > last_position { + break; + } + if next < first { + gaps.push((next, first.saturating_sub(1).min(last_position))); + } + let Some(after) = last.checked_add(1) else { + return gaps; + }; + next = after; + } + if next <= last_position { + gaps.push((next, last_position)); + } + gaps + } + pub(crate) fn last_timestamp_ns(&self) -> Option { self.active .back() @@ -360,12 +391,18 @@ impl MessageIndex { let mut expected = self .active .back() - .map(|message| message.position.saturating_add(1)) - .or_else(|| { - self.sealed - .back() - .map(|segment| segment.last_position.saturating_add(1)) - }); + .map(|message| message.position.saturating_add(1)); + if expected.is_none() + && self.sealed.back().is_some_and(|segment| { + messages + .first() + .is_some_and(|message| message.position <= segment.last_position) + }) + { + return Err(BrokerError::InvalidRecord( + "topic message position ranges overlap".into(), + )); + } for message in &messages { if expected.is_some_and(|expected| message.position != expected) { return Err(BrokerError::InvalidRecord( @@ -424,11 +461,11 @@ impl MessageIndex { || self .sealed .back() - .is_some_and(|previous| previous.last_position + 1 != segment.first_position) + .is_some_and(|previous| previous.last_position >= segment.first_position) || self .active .front() - .is_some_and(|message| segment.last_position + 1 != message.position) + .is_some_and(|message| segment.last_position >= message.position) { return Err(BrokerError::InvalidRecord( "sealed topic message range is invalid".into(), diff --git a/crates/queue/src/topic/maintenance.rs b/crates/queue/src/topic/maintenance.rs index b42f2d8..0dfcc67 100644 --- a/crates/queue/src/topic/maintenance.rs +++ b/crates/queue/src/topic/maintenance.rs @@ -37,10 +37,11 @@ impl Topic { } pub fn stats(&mut self) -> TopicStats { - let last = self.last_position(); + let last = self.deliverable_position; let now_ms = now_ms(); let scheduled = self.messages.deferred_positions(now_ms); let (segment_count, segment_bytes) = self.log.storage_usage(); + let pending_sync = self.pending_sync(); let mut channels: Vec<_> = self .channels .values_mut() @@ -54,12 +55,18 @@ impl Topic { message_count: self.messages.total_count(), segment_count, segment_bytes, + last_durable_position: self.durable_position, + unsynced_messages: pending_sync.map_or(0, |pending| pending.messages), + unsynced_bytes: pending_sync.map_or(0, |pending| pending.bytes), + sync_lag_ms: pending_sync.map_or(0, |pending| { + pending.since.elapsed().as_millis().min(u64::MAX as u128) as u64 + }), channels, } } pub fn add_aggregate_stats(&mut self, aggregate: &mut QueueAggregateStats) { - let last = self.last_position(); + let last = self.deliverable_position; let now_ms = now_ms(); let scheduled = self.messages.deferred_positions(now_ms); let (segment_count, segment_bytes) = self.log.storage_usage(); @@ -69,6 +76,14 @@ impl Topic { .saturating_add(self.messages.total_count()); aggregate.segment_count = aggregate.segment_count.saturating_add(segment_count); aggregate.segment_bytes = aggregate.segment_bytes.saturating_add(segment_bytes); + if let Some(pending) = self.pending_sync() { + aggregate.unsynced_messages = + aggregate.unsynced_messages.saturating_add(pending.messages); + aggregate.unsynced_bytes = aggregate.unsynced_bytes.saturating_add(pending.bytes); + aggregate.sync_lag_ms = aggregate + .sync_lag_ms + .max(pending.since.elapsed().as_millis().min(u64::MAX as u128) as u64); + } for channel in self.channels.values_mut() { let (depth, in_flight, deferred, ack_gap) = channel.state.metric_counts(last, &scheduled, now_ms); @@ -208,6 +223,7 @@ impl Topic { pub fn sync(&mut self) -> Result<(), BrokerError> { self.log.sync()?; + self.mark_durable_through(self.last_position()); self.checkpoint_channels()?; store_atomic(&self.manifest_path, &self.manifest)?; Ok(()) diff --git a/crates/queue/tests/recovery.rs b/crates/queue/tests/recovery.rs index 6738d3d..7ca0e9b 100644 --- a/crates/queue/tests/recovery.rs +++ b/crates/queue/tests/recovery.rs @@ -1,4 +1,4 @@ -use rustqueue_queue::{Broker, BrokerConfig}; +use rustqueue_queue::{Broker, BrokerConfig, PublishAckMode}; use std::collections::BTreeSet; use std::time::Duration; @@ -93,6 +93,110 @@ async fn durable_fin_req_and_unfinished_delivery_recover_at_least_once() { assert_eq!(only_new.id, next); } +#[tokio::test] +async fn nsq_relaxed_preserves_a_gap_after_an_acknowledged_tail_is_lost() { + let root = tempfile::tempdir().unwrap(); + let mut cfg = config(root.path()); + cfg.publish_ack_mode = PublishAckMode::NsqRelaxed; + cfg.relaxed_sync_messages = usize::MAX; + cfg.relaxed_sync_bytes = usize::MAX; + cfg.relaxed_sync_interval = Duration::from_secs(60); + cfg.max_ack_gap = 1; + let broker = Broker::open(cfg.clone()).unwrap(); + broker.create_channel("events", "workers").await.unwrap(); + broker.create_channel("events", "lagging").await.unwrap(); + let durable_id = broker + .publish("events", vec![b"durable-prefix".to_vec()], Duration::ZERO) + .await + .unwrap()[0]; + broker.flush().await.unwrap(); + let segment = root + .path() + .join("topics") + .join(hex::encode("events")) + .join("segments") + .join("segment-00000000000000000001.rqlog"); + let durable_len = std::fs::metadata(&segment).unwrap().len(); + let durable = broker + .next_message("events", "workers", None) + .await + .unwrap() + .unwrap(); + assert_eq!(durable.id, durable_id); + broker + .finish("events", "workers", durable_id) + .await + .unwrap(); + let durable = broker + .next_message("events", "lagging", None) + .await + .unwrap() + .unwrap(); + assert_eq!(durable.id, durable_id); + broker + .finish("events", "lagging", durable_id) + .await + .unwrap(); + + let lost_id = broker + .publish("events", vec![b"lost-tail".to_vec()], Duration::ZERO) + .await + .unwrap()[0]; + let delivery = broker + .next_message("events", "workers", None) + .await + .unwrap() + .unwrap(); + assert_eq!(delivery.id, lost_id); + broker.finish("events", "workers", lost_id).await.unwrap(); + drop(broker); + + std::fs::OpenOptions::new() + .write(true) + .open(segment) + .unwrap() + .set_len(durable_len) + .unwrap(); + + let mut durable_cfg = config(root.path()); + durable_cfg.max_ack_gap = 1; + let broker = Broker::open(durable_cfg.clone()).unwrap(); + let replacement_id = broker + .publish("events", vec![b"replacement".to_vec()], Duration::ZERO) + .await + .unwrap()[0]; + assert_ne!(replacement_id, lost_id); + let lagging = broker.filtered_stats(Some("events"), Some("lagging")); + assert_eq!(lagging.topics[0].channels[0].depth, 1); + let replacement = broker + .next_message("events", "workers", None) + .await + .unwrap() + .expect("the durable FIN for the lost tail must not skip the next position"); + assert_eq!(replacement.id, replacement_id); + assert_eq!(&*replacement.body, b"replacement"); + let lagging = broker + .next_message("events", "lagging", None) + .await + .unwrap() + .expect("a lost position must not consume the Channel ACK window"); + assert_eq!(lagging.id, replacement_id); + broker + .finish("events", "lagging", replacement_id) + .await + .unwrap(); + broker.flush().await.unwrap(); + drop(broker); + + let broker = Broker::open(durable_cfg).unwrap(); + let replacement = broker + .next_message("events", "workers", None) + .await + .unwrap() + .expect("message-index recovery must preserve the internal position gap"); + assert_eq!(replacement.id, replacement_id); +} + #[tokio::test] async fn channels_fan_out_and_ephemeral_channels_do_not_survive_restart() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index c4c46aa..6ee62c1 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -3,6 +3,7 @@ mod environment; mod validation; use environment::read_optional_secret; +use rustqueue_queue::{PublishAckMode, RELAXED_SYNC_MIN_BYTES}; use serde::{Deserialize, Serialize}; use std::fs; use std::net::SocketAddr; @@ -93,6 +94,10 @@ pub struct QueueConfig { pub max_topics: usize, pub max_publish_workers: usize, pub publish_worker_idle_seconds: u64, + pub publish_ack_mode: PublishAckMode, + pub relaxed_sync_messages: usize, + pub relaxed_sync_bytes: usize, + pub relaxed_sync_interval_ms: u64, pub bootstrap_retention_seconds: u64, pub message_retention_seconds: u64, pub max_delivery_attempts: u16, @@ -237,6 +242,10 @@ impl Default for QueueConfig { max_topics: 10_000, max_publish_workers: 1_024, publish_worker_idle_seconds: 60, + publish_ack_mode: PublishAckMode::Durable, + relaxed_sync_messages: 2_500, + relaxed_sync_bytes: 8 * 1024 * 1024, + relaxed_sync_interval_ms: 10, bootstrap_retention_seconds: 90, message_retention_seconds: 0, max_delivery_attempts: 16, @@ -344,8 +353,12 @@ impl Config { || self.queue.max_topics == 0 || self.queue.max_publish_workers == 0 || self.queue.publish_worker_idle_seconds == 0 + || (self.queue.publish_ack_mode.is_relaxed() + && (self.queue.relaxed_sync_messages == 0 + || self.queue.relaxed_sync_bytes < RELAXED_SYNC_MIN_BYTES + || self.queue.relaxed_sync_interval_ms == 0)) { - bail!("queue limits must be greater than zero"); + bail!("queue limits must be greater than zero; relaxed sync requires at least one message, 4096 bytes, and a non-zero interval"); } if self.storage.entry_cache_bytes == 0 || self.storage.message_index_cache_bytes == 0 { bail!("storage caches must be greater than zero"); diff --git a/crates/server/src/config/environment.rs b/crates/server/src/config/environment.rs index a274180..deeff37 100644 --- a/crates/server/src/config/environment.rs +++ b/crates/server/src/config/environment.rs @@ -70,6 +70,22 @@ impl Config { "RUSTQUEUE_PUBLISH_WORKER_IDLE_SECONDS", &mut self.queue.publish_worker_idle_seconds, )?; + set_from_env( + "RUSTQUEUE_PUBLISH_ACK_MODE", + &mut self.queue.publish_ack_mode, + )?; + set_from_env( + "RUSTQUEUE_RELAXED_SYNC_MESSAGES", + &mut self.queue.relaxed_sync_messages, + )?; + set_from_env( + "RUSTQUEUE_RELAXED_SYNC_BYTES", + &mut self.queue.relaxed_sync_bytes, + )?; + set_from_env( + "RUSTQUEUE_RELAXED_SYNC_INTERVAL_MS", + &mut self.queue.relaxed_sync_interval_ms, + )?; set_from_env("RUSTQUEUE_MAX_BODY_BYTES", &mut self.limits.max_body_bytes)?; set_from_env( "RUSTQUEUE_HTTP_BODY_TIMEOUT_MS", diff --git a/crates/server/src/config/tests.rs b/crates/server/src/config/tests.rs index 06f4af9..4132b77 100644 --- a/crates/server/src/config/tests.rs +++ b/crates/server/src/config/tests.rs @@ -13,6 +13,24 @@ fn default_bootstrap_window_covers_official_lookup_poll_jitter() { assert!(Config::default().queue.bootstrap_retention_seconds >= 78); } +#[test] +fn relaxed_acknowledgement_requires_bounded_sync_thresholds() { + let mut config = Config::default(); + assert_eq!(config.queue.publish_ack_mode, PublishAckMode::Durable); + + config.queue.publish_ack_mode = PublishAckMode::WriteAck; + config.queue.relaxed_sync_bytes = RELAXED_SYNC_MIN_BYTES - 1; + assert!(config.validate().is_err()); + + config.queue.relaxed_sync_bytes = RELAXED_SYNC_MIN_BYTES; + config.queue.relaxed_sync_messages = 0; + assert!(config.validate().is_err()); + + config.queue.relaxed_sync_messages = 1; + config.queue.relaxed_sync_interval_ms = 0; + assert!(config.validate().is_err()); +} + #[test] fn maintenance_has_a_startup_quiet_period() { assert_eq!( diff --git a/crates/server/src/http.rs b/crates/server/src/http.rs index c0f972b..73f8d0c 100644 --- a/crates/server/src/http.rs +++ b/crates/server/src/http.rs @@ -380,6 +380,10 @@ mod tests { message_count: 3, segment_count: 1, segment_bytes: 256, + last_durable_position: 3, + unsynced_messages: 0, + unsynced_bytes: 0, + sync_lag_ms: 0, channels: vec![ChannelStats { name: "workers".into(), depth: 2, diff --git a/crates/server/src/http/native.rs b/crates/server/src/http/native.rs index 2774573..2f1ac79 100644 --- a/crates/server/src/http/native.rs +++ b/crates/server/src/http/native.rs @@ -332,6 +332,10 @@ fn observation_head(state: &AppState) -> Value { "delivery_budget": state.broker.delivery_budget_stats(), "limits": { "max_message_bytes": state.config.queue.max_message_bytes, + "publish_ack_mode": state.config.queue.publish_ack_mode.as_str(), + "relaxed_sync_messages": state.config.queue.relaxed_sync_messages, + "relaxed_sync_bytes": state.config.queue.relaxed_sync_bytes, + "relaxed_sync_interval_ms": state.config.queue.relaxed_sync_interval_ms, "message_index_cache_bytes": state.config.storage.message_index_cache_bytes, "max_connections": state.config.limits.max_connections, }, diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 7693a1f..043d563 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -71,6 +71,12 @@ async fn main() -> anyhow::Result<()> { publish_worker_idle: std::time::Duration::from_secs( config.queue.publish_worker_idle_seconds, ), + publish_ack_mode: config.queue.publish_ack_mode, + relaxed_sync_messages: config.queue.relaxed_sync_messages, + relaxed_sync_bytes: config.queue.relaxed_sync_bytes, + relaxed_sync_interval: std::time::Duration::from_millis( + config.queue.relaxed_sync_interval_ms, + ), entry_cache_bytes: config.storage.entry_cache_bytes, message_index_cache_bytes: config.storage.message_index_cache_bytes, payload_read_workers: config.storage.payload_read_workers, diff --git a/crates/server/src/metrics.rs b/crates/server/src/metrics.rs index cced357..a7cf51b 100644 --- a/crates/server/src/metrics.rs +++ b/crates/server/src/metrics.rs @@ -70,6 +70,7 @@ pub fn render_broker(stats: &BrokerStats, config: &MetricsConfig) -> String { let channel_in_flight = stats.aggregate.channel_in_flight; let channel_deferred = stats.aggregate.channel_deferred; let channel_ack_gap = stats.aggregate.channel_ack_gap; + let sync_lag_seconds = stats.aggregate.sync_lag_ms as f64 / 1_000.0; let mut output = format!( "# TYPE rustqueue_publish_group_commits_total counter\n\ rustqueue_publish_group_commits_total {}\n\ @@ -99,6 +100,12 @@ pub fn render_broker(stats: &BrokerStats, config: &MetricsConfig) -> String { rustqueue_topics {topic_count}\n\ # TYPE rustqueue_topic_messages_total gauge\n\ rustqueue_topic_messages_total {message_count}\n\ + # TYPE rustqueue_publish_unsynced_messages gauge\n\ + rustqueue_publish_unsynced_messages {}\n\ + # TYPE rustqueue_publish_unsynced_bytes gauge\n\ + rustqueue_publish_unsynced_bytes {}\n\ + # TYPE rustqueue_publish_sync_lag_seconds gauge\n\ + rustqueue_publish_sync_lag_seconds {sync_lag_seconds}\n\ # TYPE rustqueue_channels gauge\n\ rustqueue_channels {channel_count}\n\ # TYPE rustqueue_channel_depth_total gauge\n\ @@ -127,6 +134,8 @@ pub fn render_broker(stats: &BrokerStats, config: &MetricsConfig) -> String { stats.channel_group_commit.active_workers, stats.channel_group_commit.retired_workers, stats.channel_group_commit.rejected_workers, + stats.aggregate.unsynced_messages, + stats.aggregate.unsynced_bytes, stats.delivery_budget.in_flight_bytes, stats.delivery_budget.waiters, stats.delivery_budget.waits_total, @@ -143,11 +152,31 @@ pub fn render_broker(stats: &BrokerStats, config: &MetricsConfig) -> String { "Time a publish request waits before group commit processing.", &stats.latency.group_commit_wait, ), + ( + "rustqueue_publish_topic_lock_wait_duration_seconds", + "Time a publish group waits to acquire its Topic state lock.", + &stats.latency.publish_topic_lock_wait, + ), + ( + "rustqueue_publish_topic_lock_hold_duration_seconds", + "Time a publish group holds its Topic state lock.", + &stats.latency.publish_topic_lock_hold, + ), ( "rustqueue_publish_ack_duration_seconds", "End-to-end broker publish acknowledgement latency.", &stats.latency.publish_ack, ), + ( + "rustqueue_delivery_topic_lock_wait_duration_seconds", + "Time a delivery reservation waits to acquire its Topic state lock.", + &stats.latency.delivery_topic_lock_wait, + ), + ( + "rustqueue_delivery_topic_lock_hold_duration_seconds", + "Time a delivery reservation holds its Topic state lock.", + &stats.latency.delivery_topic_lock_hold, + ), ( "rustqueue_channel_fsync_duration_seconds", "Time spent making a FIN or REQ group durable.", @@ -187,6 +216,7 @@ pub fn render_broker(stats: &BrokerStats, config: &MetricsConfig) -> String { fn render_detailed_queue_metrics(output: &mut String, stats: &BrokerStats, config: &MetricsConfig) { let desired = usize::try_from(stats.aggregate.topic_count) .unwrap_or(usize::MAX) + .saturating_mul(5) .saturating_add( usize::try_from(stats.aggregate.channel_count) .unwrap_or(usize::MAX) @@ -196,19 +226,31 @@ fn render_detailed_queue_metrics(output: &mut String, stats: &BrokerStats, confi if config.detailed_queue_metrics { output.push_str( "# TYPE rustqueue_topic_messages gauge\n\ + # TYPE rustqueue_topic_last_durable_position gauge\n\ + # TYPE rustqueue_topic_publish_unsynced_messages gauge\n\ + # TYPE rustqueue_topic_publish_unsynced_bytes gauge\n\ + # TYPE rustqueue_topic_publish_sync_lag_seconds gauge\n\ # TYPE rustqueue_channel_depth gauge\n\ # TYPE rustqueue_channel_in_flight gauge\n\ # TYPE rustqueue_channel_deferred gauge\n\ # TYPE rustqueue_channel_ack_gap gauge\n", ); for topic in &stats.topics { - if emitted < config.max_detailed_series { + if emitted.saturating_add(5) <= config.max_detailed_series { let topic_label = format!("topic=\"{}\"", escape_label(&topic.name)); output.push_str(&format!( - "rustqueue_topic_messages{{{topic_label}}} {}\n", - topic.message_count + "rustqueue_topic_messages{{{topic_label}}} {}\n\ + rustqueue_topic_last_durable_position{{{topic_label}}} {}\n\ + rustqueue_topic_publish_unsynced_messages{{{topic_label}}} {}\n\ + rustqueue_topic_publish_unsynced_bytes{{{topic_label}}} {}\n\ + rustqueue_topic_publish_sync_lag_seconds{{{topic_label}}} {}\n", + topic.message_count, + topic.last_durable_position, + topic.unsynced_messages, + topic.unsynced_bytes, + topic.sync_lag_ms as f64 / 1_000.0, )); - emitted += 1; + emitted += 5; } for channel in &topic.channels { if emitted.saturating_add(4) > config.max_detailed_series { @@ -372,6 +414,9 @@ mod tests { message_count: 7, segment_count: 1, segment_bytes: 100, + unsynced_messages: 2, + unsynced_bytes: 32, + sync_lag_ms: 7, channel_count: 1, channel_depth: 3, channel_in_flight: 2, @@ -385,6 +430,10 @@ mod tests { message_count: 7, segment_count: 1, segment_bytes: 100, + last_durable_position: 5, + unsynced_messages: 2, + unsynced_bytes: 32, + sync_lag_ms: 7, channels: vec![ChannelStats { name: "workers".into(), depth: 3, @@ -408,9 +457,14 @@ mod tests { let output = render_broker(&broker_stats(), &MetricsConfig::default()); assert!(output.contains("rustqueue_topic_messages_total 7\n")); assert!(output.contains("rustqueue_channel_depth_total 3\n")); + assert!(output.contains("rustqueue_publish_topic_lock_wait_duration_seconds_count 0\n")); + assert!(output.contains("rustqueue_delivery_topic_lock_hold_duration_seconds_count 0\n")); + assert!(output.contains("rustqueue_publish_unsynced_messages 2\n")); + assert!(output.contains("rustqueue_publish_unsynced_bytes 32\n")); + assert!(output.contains("rustqueue_publish_sync_lag_seconds 0.007\n")); assert!(!output.contains("topic=\"events\"")); assert!(output.contains("rustqueue_detailed_queue_metric_series 0\n")); - assert!(output.contains("rustqueue_detailed_queue_metric_series_omitted 5\n")); + assert!(output.contains("rustqueue_detailed_queue_metric_series_omitted 9\n")); } #[test] @@ -419,12 +473,13 @@ mod tests { &broker_stats(), &MetricsConfig { detailed_queue_metrics: true, - max_detailed_series: 4, + max_detailed_series: 5, }, ); assert!(output.contains("rustqueue_topic_messages{topic=\"events\"} 7\n")); + assert!(output.contains("rustqueue_topic_last_durable_position{topic=\"events\"} 5\n")); assert!(!output.contains("rustqueue_channel_depth{topic=")); - assert!(output.contains("rustqueue_detailed_queue_metric_series 1\n")); + assert!(output.contains("rustqueue_detailed_queue_metric_series 5\n")); assert!(output.contains("rustqueue_detailed_queue_metric_series_omitted 4\n")); } diff --git a/crates/server/src/tcp.rs b/crates/server/src/tcp.rs index f7cb83d..f181808 100644 --- a/crates/server/src/tcp.rs +++ b/crates/server/src/tcp.rs @@ -33,7 +33,7 @@ use rustqueue_protocol::{ }; use rustqueue_queue::{Broker, BrokerError, DeliveryGuard}; use serde_json::json; -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeSet, HashMap, HashSet}; use std::future::Future; use std::net::SocketAddr; use std::pin::Pin; @@ -144,6 +144,7 @@ struct SessionState { subscription: Option, rdy: u64, in_flight: HashMap, + in_flight_deadlines: BTreeSet<(Instant, u64, u64)>, pending_channel_ops: HashSet, closing: bool, client_identity: ClientIdentity, @@ -173,6 +174,44 @@ impl SessionState { } self.in_flight.get(&id).copied() } + + fn record_delivery(&mut self, id: u64, delivery: InFlightDelivery) { + if let Some(previous) = self.in_flight.insert(id, delivery) { + self.in_flight_deadlines + .remove(&(previous.deadline, id, previous.token)); + } + if !self.pending_channel_ops.contains(&id) { + self.in_flight_deadlines + .insert((delivery.deadline, id, delivery.token)); + } + } + + fn mark_channel_operation_pending(&mut self, id: u64) { + if !self.pending_channel_ops.insert(id) { + return; + } + if let Some(delivery) = self.in_flight.get(&id) { + self.in_flight_deadlines + .remove(&(delivery.deadline, id, delivery.token)); + } + } + + fn restore_delivery_deadline(&mut self, id: u64) { + if self.pending_channel_ops.contains(&id) { + return; + } + if let Some(delivery) = self.in_flight.get(&id) { + self.in_flight_deadlines + .insert((delivery.deadline, id, delivery.token)); + } + } + + fn remove_delivery(&mut self, id: u64) -> Option { + let delivery = self.in_flight.remove(&id)?; + self.in_flight_deadlines + .remove(&(delivery.deadline, id, delivery.token)); + Some(delivery) + } } #[allow(clippy::too_many_arguments)] @@ -340,6 +379,7 @@ async fn handle_connection( subscription: None, rdy: 0, in_flight: HashMap::new(), + in_flight_deadlines: BTreeSet::new(), pending_channel_ops: HashSet::new(), closing: false, client_identity: ClientIdentity { diff --git a/crates/server/src/tcp/commands.rs b/crates/server/src/tcp/commands.rs index 0298568..cf7e81a 100644 --- a/crates/server/src/tcp/commands.rs +++ b/crates/server/src/tcp/commands.rs @@ -240,14 +240,14 @@ pub(super) async fn process_command( if let Err(error) = result { write_broker_error(writer, "E_FIN_FAILED", error).await?; } else { - state.in_flight.insert( + state.record_delivery( id, InFlightDelivery { deadline, ..delivery }, ); - state.pending_channel_ops.insert(id); + state.mark_channel_operation_pending(id); } } Command::Requeue { id, delay_ms } => { @@ -284,14 +284,14 @@ pub(super) async fn process_command( if let Err(error) = result { write_broker_error(writer, "E_REQ_FAILED", error).await?; } else { - state.in_flight.insert( + state.record_delivery( id, InFlightDelivery { deadline, ..delivery }, ); - state.pending_channel_ops.insert(id); + state.mark_channel_operation_pending(id); } } Command::Touch(id) => { @@ -313,7 +313,7 @@ pub(super) async fn process_command( ); match touch_result { Ok(deadline) => { - state.in_flight.insert( + state.record_delivery( id, InFlightDelivery { deadline, diff --git a/crates/server/src/tcp/session.rs b/crates/server/src/tcp/session.rs index fd685c8..6dfcc94 100644 --- a/crates/server/src/tcp/session.rs +++ b/crates/server/src/tcp/session.rs @@ -82,7 +82,7 @@ pub(super) async fn run_session( loop { let expired = expire_client_deadlines( &mut state.in_flight, - &state.pending_channel_ops, + &mut state.in_flight_deadlines, Instant::now(), ); if expired { @@ -196,8 +196,7 @@ pub(super) async fn run_session( pending_fetch = Some(PendingFetch { request, future }); } } - let in_flight_deadline = - next_client_deadline(&state.in_flight, &state.pending_channel_ops); + let in_flight_deadline = next_client_deadline(&state.in_flight_deadlines); tokio::select! { command = command_rx.recv() => { @@ -380,14 +379,14 @@ pub(super) async fn run_session( token, ) .map_err(|error| anyhow::anyhow!(error.to_string()))?; - state.in_flight.insert( + state.record_delivery( delivery.id, InFlightDelivery { deadline, token, }, ); - state.pending_channel_ops.insert(delivery.id); + state.mark_channel_operation_pending(delivery.id); delivery_guard.accept(delivery.id); continue; } @@ -410,7 +409,7 @@ pub(super) async fn run_session( .accept_with_token(delivery.id) .ok_or_else(|| anyhow::anyhow!("delivery token is missing"))?; debug_assert_eq!(accepted_token, token); - state.in_flight.insert( + state.record_delivery( delivery.id, InFlightDelivery { deadline: handoff_deadline, @@ -434,9 +433,12 @@ pub(super) async fn run_session( ) .map_err(|error| anyhow::anyhow!(error.to_string()))?; for (id, token) in handed_off { - if let Some(delivery) = state.in_flight.get_mut(&id) { + if let Some(delivery) = state.in_flight.get(&id).copied() { debug_assert_eq!(delivery.token, token); - delivery.deadline = deadline; + state.record_delivery( + id, + InFlightDelivery { deadline, ..delivery }, + ); } } } @@ -573,24 +575,28 @@ fn delivery_is_outstanding(in_flight: &HashMap, id: u64) fn expire_client_deadlines( in_flight: &mut HashMap, - pending_channel_ops: &HashSet, + deadlines: &mut BTreeSet<(Instant, u64, u64)>, now: Instant, ) -> bool { - let before = in_flight.len(); - in_flight.retain(|id, delivery| pending_channel_ops.contains(id) || delivery.deadline > now); - in_flight.len() != before + let mut expired = false; + while let Some((deadline, id, token)) = deadlines.first().copied() { + if deadline > now { + break; + } + deadlines.remove(&(deadline, id, token)); + if in_flight + .get(&id) + .is_some_and(|delivery| delivery.deadline == deadline && delivery.token == token) + { + in_flight.remove(&id); + expired = true; + } + } + expired } -fn next_client_deadline( - in_flight: &HashMap, - pending_channel_ops: &HashSet, -) -> Option { - in_flight - .iter() - .filter_map(|(id, delivery)| { - (!pending_channel_ops.contains(id)).then_some(delivery.deadline) - }) - .min() +fn next_client_deadline(deadlines: &BTreeSet<(Instant, u64, u64)>) -> Option { + deadlines.first().map(|(deadline, _, _)| *deadline) } fn apply_channel_op_completion( @@ -600,13 +606,14 @@ fn apply_channel_op_completion( ) -> Result<(), (&'static str, BrokerError)> { state.pending_channel_ops.remove(&completion.id); if let Err(error) = completion.result { + state.restore_delivery_deadline(completion.id); if broker_storage_error(&error) { metrics.storage_errors.fetch_add(1, Ordering::Relaxed); } return Err((completion.kind.error_code(), error)); } - state.in_flight.remove(&completion.id); + state.remove_delivery(completion.id); if let Some(subscription) = &state.subscription { match completion.kind { ChannelOpKind::Finish => subscription.lease.observe_finish(), @@ -670,6 +677,30 @@ mod tests { InFlightDelivery { deadline, token } } + fn session_state() -> SessionState { + SessionState { + identified: false, + encrypted: false, + tls_common_name: String::new(), + heartbeat: None, + message_timeout: Duration::from_secs(60), + output_buffer_size: 1, + output_buffer_timeout: None, + sample_rate: 0, + sample_cursor: 0, + auth: None, + auth_secret: None, + _auth_reservation: None, + subscription: None, + rdy: 0, + in_flight: HashMap::new(), + in_flight_deadlines: BTreeSet::new(), + pending_channel_ops: HashSet::new(), + closing: false, + client_identity: ClientIdentity::default(), + } + } + #[tokio::test] async fn pending_fetch_survives_an_unrelated_ready_branch() { let request = FetchRequest { @@ -718,28 +749,39 @@ mod tests { (7, delivery(now - Duration::from_millis(1), 70)), (8, delivery(now + Duration::from_secs(1), 80)), ]); - assert!(expire_client_deadlines( - &mut in_flight, - &HashSet::new(), - now - )); + let mut deadlines = BTreeSet::from([ + (now - Duration::from_millis(1), 7, 70), + (now + Duration::from_secs(1), 8, 80), + ]); + assert!(expire_client_deadlines(&mut in_flight, &mut deadlines, now)); assert_eq!(in_flight.keys().copied().collect::>(), vec![8]); assert!(!expire_client_deadlines( &mut in_flight, - &HashSet::new(), + &mut deadlines, now )); } #[test] - fn pending_channel_operations_do_not_expire() { + fn pending_channel_operations_unschedule_and_restore_the_delivery_deadline() { let now = Instant::now(); - let mut in_flight = HashMap::from([(7, delivery(now - Duration::from_millis(1), 70))]); - let pending = HashSet::from([7]); + let mut state = session_state(); + let deadline = now + Duration::from_secs(1); + state.record_delivery(7, delivery(deadline, 70)); + assert_eq!( + next_client_deadline(&state.in_flight_deadlines), + Some(deadline) + ); + + state.mark_channel_operation_pending(7); + assert_eq!(next_client_deadline(&state.in_flight_deadlines), None); - assert!(!expire_client_deadlines(&mut in_flight, &pending, now)); - assert_eq!(next_client_deadline(&in_flight, &pending), None); - assert!(in_flight.contains_key(&7)); + state.pending_channel_ops.remove(&7); + state.restore_delivery_deadline(7); + assert_eq!( + next_client_deadline(&state.in_flight_deadlines), + Some(deadline) + ); } #[test] diff --git a/crates/storage/src/segment.rs b/crates/storage/src/segment.rs index 28468f9..3efc5c4 100644 --- a/crates/storage/src/segment.rs +++ b/crates/storage/src/segment.rs @@ -377,6 +377,19 @@ impl SegmentLog { result } + pub fn clone_current_for_sync(&self) -> Result { + self.ensure_available()?; + let result = self.current.try_clone().map_err(StorageError::from); + if result.is_err() { + self.isolate(); + } + result + } + + pub fn mark_sync_failed(&self) { + self.isolate(); + } + pub fn read(&self, index: u64) -> Result, StorageError> { let Some(location) = self.location(index)? else { return Ok(None); diff --git a/deploy/helm/rustqueue/Chart.yaml b/deploy/helm/rustqueue/Chart.yaml index 5750021..9d8ca3a 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.2 -appVersion: "0.8.2" +version: 0.8.3 +appVersion: "0.8.3" kubeVersion: ">=1.28.0-0" keywords: - messaging diff --git a/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml b/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml index c34dfde..9a4fa76 100644 --- a/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml +++ b/deploy/helm/rustqueue/crds/rustqueue.io_rustqueues.yaml @@ -53,6 +53,10 @@ spec: maxTopics: {type: integer, format: int64, minimum: 1, default: 10000} maxPublishWorkers: {type: integer, format: int64, minimum: 1, default: 1024} publishWorkerIdleSeconds: {type: integer, format: int64, minimum: 1, default: 60} + publishAckMode: {type: string, enum: [durable, write_ack, nsq_relaxed], default: durable} + relaxedSyncMessages: {type: integer, format: int64, minimum: 1, default: 2500} + relaxedSyncBytes: {type: integer, format: int64, minimum: 4096, default: 8388608} + relaxedSyncIntervalMs: {type: integer, format: int64, minimum: 1, default: 10} detailedQueueMetrics: {type: boolean, default: false} maxDetailedMetricSeries: {type: integer, format: int64, minimum: 1, default: 1000} registrySecretName: {type: string, nullable: true} diff --git a/deploy/helm/rustqueue/templates/queue.yaml b/deploy/helm/rustqueue/templates/queue.yaml index 59558e2..cf2010f 100644 --- a/deploy/helm/rustqueue/templates/queue.yaml +++ b/deploy/helm/rustqueue/templates/queue.yaml @@ -5,6 +5,9 @@ {{- if and .Values.queue.kodoCompatibility.enabled (ne .Values.queue.imagePullPolicy "Never") (not (regexMatch "@sha256:[a-f0-9]{64}$" .Values.queue.image)) }} {{- fail "queue.kodoCompatibility.enabled requires queue.image pinned by @sha256, or queue.imagePullPolicy=Never for a preloaded local image" }} {{- end }} +{{- if and .Values.queue.kodoCompatibility.enabled (ne .Values.queue.publishAckMode "durable") }} +{{- fail "queue.kodoCompatibility.enabled requires queue.publishAckMode=durable" }} +{{- end }} apiVersion: rustqueue.io/v1alpha1 kind: RustQueue metadata: @@ -35,6 +38,10 @@ spec: maxTopics: {{ .Values.queue.maxTopics | int64 }} maxPublishWorkers: {{ .Values.queue.maxPublishWorkers | int64 }} publishWorkerIdleSeconds: {{ .Values.queue.publishWorkerIdleSeconds | int64 }} + publishAckMode: {{ .Values.queue.publishAckMode | quote }} + relaxedSyncMessages: {{ .Values.queue.relaxedSyncMessages | int64 }} + relaxedSyncBytes: {{ .Values.queue.relaxedSyncBytes | int64 }} + relaxedSyncIntervalMs: {{ .Values.queue.relaxedSyncIntervalMs | int64 }} detailedQueueMetrics: {{ .Values.queue.detailedQueueMetrics }} maxDetailedMetricSeries: {{ .Values.queue.maxDetailedMetricSeries | int64 }} consoleManagementEnabled: {{ and .Values.console.enabled .Values.console.management.enabled }} diff --git a/deploy/helm/rustqueue/values.yaml b/deploy/helm/rustqueue/values.yaml index 5250874..eaeb4a6 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.2" + tag: "0.8.3" pullPolicy: IfNotPresent log: rustqueue_operator=info resources: @@ -20,7 +20,7 @@ operator: queue: create: true name: rustqueue - image: rustqueue:0.8.2 + image: rustqueue:0.8.3 imagePullPolicy: IfNotPresent minBrokers: 1 maxBrokers: 500 @@ -46,6 +46,12 @@ queue: maxTopics: 10000 maxPublishWorkers: 1024 publishWorkerIdleSeconds: 60 + # Default is strict local-fsync acknowledgement. write_ack and nsq_relaxed + # are explicit crash-loss trade-offs and are not supported by Kodo compatibility. + publishAckMode: durable + relaxedSyncMessages: 2500 + relaxedSyncBytes: 8388608 + relaxedSyncIntervalMs: 10 detailedQueueMetrics: false maxDetailedMetricSeries: 1000 registrySecretName: "" diff --git a/docs/architecture/nsq-performance.md b/docs/architecture/nsq-performance.md new file mode 100644 index 0000000..11dba9d --- /dev/null +++ b/docs/architecture/nsq-performance.md @@ -0,0 +1,87 @@ +# RustQueue and NSQ performance boundaries + +This comparison uses NSQ `v1.3.0`, the version used by +`scripts/benchmark-compare.sh`. It distinguishes protocol compatibility from +durability and scheduling behavior: equal TCP commands do not imply equal work +before an acknowledgement is returned. + +## What RustQueue 0.8.3 changes + +RustQueue now keeps each Channel's in-flight deadlines in an ordered index. +The Broker can find expired deliveries without scanning every in-flight message, +and a TCP session can wait for its next client deadline without scanning its +entire RDY window on every event. `TOUCH`, `FIN`, `REQ`, disconnect, and +delivery handoff remove or update the same index, so stale deadlines do not +accumulate. + +The on-disk format remains v7. A successful `FIN` or `REQ` still crosses the +Channel WAL group `fsync` before the broker replies; this release changes the +runtime scheduler, not the acknowledged-delivery contract. + +For publish groups, RustQueue now releases the Topic state lock while the +active segment file is syncing. A separate commit gate keeps tail mutations +serialized, and reservation stops at a durable position that advances only +after the sync succeeds. Existing durable messages can therefore continue to +flow while the next group waits for its acknowledged-durability boundary. + +## Material differences from NSQ + +| Area | NSQ v1.3.0 | RustQueue | Performance consequence | +| --- | --- | --- | --- | +| In-flight timeout lookup | `inFlightPQ` priority queue plus ID map | Ordered deadline index plus ID map | Both avoid a full in-flight scan. RustQueue 0.8.3 closes the prior CPU and lock-contention gap at high RDY. | +| Publish acknowledgement | `Topic.put` queues to memory or `go-diskqueue`; the diskqueue writer replies after `writeOne`, and performs its scheduled `sync` on a later I/O-loop iteration | Default `durable` mode appends, calls segment `fsync`, then replies | RustQueue has higher durable-PUB latency by design. NSQ `--sync-every=1` is not an acknowledgement-after-fsync equivalent. | +| FIN / REQ acknowledgement | In-flight state is memory-resident; shutdown flushes outstanding messages to the backend | Channel state is appended to a WAL and group-fsynced before success | RustQueue trades throughput for a stronger confirmed-ack crash boundary. Do not compare raw consume rate without stating this difference. | +| Relaxed publish path | Configurable `mem-queue-size`; diskqueue replies after write and syncs later by count or timer | `write_ack` replies after append but delays delivery until fsync; `nsq_relaxed` replies and delivers after append | Compare each mode under its own durability label. The script keeps NSQ `mem-queue-size=0`, so the relaxed comparison isolates diskqueue sync cadence rather than adding NSQ's memory queue. | +| Topic fan-out and locking | A Topic message pump allocates a message object per additional Channel and feeds each Channel queue | One Topic segment is shared by cursor-based Channels; publish append/rotation remains serialized, but group `fsync` runs outside the reservation lock and delivery stops at the durable tail | RustQueue avoids one durable payload log per Channel without making existing durable reservations wait for the next PUB fsync. | + +## How to read benchmark results + +`scripts/benchmark-compare.sh` always produces three distinct durability +profiles: + +- `rustqueue-local-fsync`: a PUB acknowledgement follows the local segment + group `fsync`. +- `nsq-sync-every-1`: NSQ requests a diskqueue sync every write, but its PUB + response follows the write before the next I/O-loop sync. +- `nsq-sync-every-2500`: NSQ's more relaxed sync cadence. + +Treat the first two as a storage-write comparison, not an equal +acknowledged-durability comparison. Record publish acknowledgement latency, +complete unique delivery, duplicates, drain state, and RSS alongside throughput. +The benchmark aborts when a consumer run has missing deliveries, unexpected +duplicates, or an incomplete drain. + +Set `RUN_RELAXED=1` to add two separately named RustQueue profiles: + +- `rustqueue-write-ack`: ACK after append, consume after background fsync. +- `rustqueue-nsq-relaxed`: ACK and consume after append. + +Both use the first reached `RELAXED_SYNC_MESSAGES` (default 2500), +`RELAXED_SYNC_BYTES` (default 8 MiB), or `RELAXED_SYNC_INTERVAL_MS` (default +10 ms) boundary. Never combine either result with `rustqueue-local-fsync`. +Track `rustqueue_publish_unsynced_messages`, +`rustqueue_publish_unsynced_bytes`, and +`rustqueue_publish_sync_lag_seconds` alongside throughput and ACK latency. + +## Next candidates to measure before changing semantics + +1. Profile Channel WAL `FIN`/`REQ` commit time against reservation latency; + those durable state transitions still use the Topic state lock. +2. Profile `write_ack` and `nsq_relaxed` under mixed producer/consumer load, + including ACK latency, durable-position lag, RSS, and the time from append + acknowledgement to fsync. Keep both separate from the durable-PUB result. +3. Compare many-Channel fan-out with equal memory budgets. NSQ allocates a + message object per Channel and can persist each Channel backlog separately; + RustQueue's shared segment reduces durable payload duplication but must also + keep cursor and metadata lock costs bounded. +4. Keep NSQ's memory queue as a separate profile. Neither RustQueue relaxed + mode bypasses the segment append, so comparing either one with a non-zero + NSQ `mem-queue-size` measures an additional architectural difference. +5. Inspect publish latency outliers at segment rotation and durable message-ID + block reservation. These paths sync metadata infrequently and should not be + inferred from median steady-state throughput. + +Primary source paths: NSQ's +[`channel.go`](https://github.com/nsqio/nsq/blob/v1.3.0/nsqd/channel.go), +[`topic.go`](https://github.com/nsqio/nsq/blob/v1.3.0/nsqd/topic.go), and +[`go-diskqueue`](https://github.com/nsqio/go-diskqueue/blob/v1.1.0/diskqueue.go). diff --git a/docs/architecture/share-nothing-v7.md b/docs/architecture/share-nothing-v7.md index 0fc378e..94920e4 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.2 +Target release: 0.8.3 Data format: v7, clean directories only ## 1. Goal @@ -33,8 +33,22 @@ There is no broker-to-broker message path and no cluster consensus path. message. - The broker PVC is the only copy. Permanent PVC loss means permanent message loss. -- `PUB`, `MPUB`, and `DPUB` succeed only after the local message segment has - passed `fsync`. +- By default, `PUB`, `MPUB`, and `DPUB` succeed only after the local message + segment has passed `fsync`. +- `queue.publish_ack_mode = "write_ack"` is an explicit weaker alternative: + success follows append/write, while consumers remain bounded by the last + background-fsynced position. A crash or power loss can lose the acknowledged + tail. +- `queue.publish_ack_mode = "nsq_relaxed"` also exposes the appended tail + immediately. A crash can therefore lose messages that were both acknowledged + and delivered. Neither relaxed mode is a durable-PUB benchmark result. +- Recovery derives lost-tail gaps from surviving Topic metadata and durable + Channel commands. This prevents an acknowledged vanished position from + aliasing a later publish; derived gap ranges do not consume Channel depth or + the out-of-order ACK window. +- Relaxed background fsync runs when the first configured message, byte, or + interval threshold is reached. Any sync failure isolates local storage and + stops subsequent writes. - `FIN` and `REQ` succeed only after the local channel WAL has passed `fsync`. - Local group commit may combine requests, but never weakens the fsync boundary. - Message ID ranges are reserved durably in large blocks. A restart may leave diff --git a/docs/operations/kubernetes.md b/docs/operations/kubernetes.md index 0666854..c903795 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.2 \ + --set queue.image=registry.example/rustqueue:0.8.3 \ --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.2 +rustqueuectl -n rustqueue rollout rollback registry.example/rustqueue:0.8.3 rustqueuectl -n rustqueue rollout forward ``` diff --git a/docs/releases/v0.8.3.md b/docs/releases/v0.8.3.md new file mode 100644 index 0000000..0dcb2be --- /dev/null +++ b/docs/releases/v0.8.3.md @@ -0,0 +1,99 @@ +# RustQueue 0.8.3 + +RustQueue 0.8.3 removes full in-flight deadline scans from the hot delivery +path while preserving the v7 storage format and durable acknowledgement +contract. + +## Deadline-indexed delivery + +- A durable Channel now keeps its active delivery deadlines in an ordered + index alongside its ID lookup. Expiry checks visit only due deliveries rather + than scanning the entire in-flight set on every fetch. +- TCP sessions use the same pattern for their local delivery leases. The next + timer comes directly from the earliest active lease; `TOUCH`, `FIN`, `REQ`, + delivery handoff, completion, and disconnect keep the index consistent. +- `FIN` and `REQ` operations in the asynchronous Channel pipeline are removed + from the session deadline index while they are pending, then restored only + if the operation fails. A successful operation removes both the ID lookup + and deadline entry. + +This is most relevant with high `RDY` values, long message timeouts, slow +consumers, and many active consumer connections. It lowers avoidable CPU work +and Topic-lock pressure without changing duplicate, timeout, or at-least-once +semantics. + +## Topic fsync and delivery concurrency + +- A publish group still serializes appends for one Topic and acknowledges only + after the segment group has been fsynced. +- The active segment file is now synced after releasing the Topic state lock. + A separate commit gate keeps append, rotation, retention, and management + changes serialized until that durable boundary completes. +- Delivery reservation reads only through a durable position advanced after a + successful sync. Consumers can therefore reserve earlier durable messages + during the next group's fsync, but never observe the uncommitted tail. +- Prometheus now exposes publish and delivery Topic-lock wait/hold histograms + so mixed PUB/consume contention can be measured directly. + +## Explicit publish durability modes + +- `queue.publish_ack_mode = "durable"` remains the default and preserves + acknowledgement-after-fsync semantics. +- `queue.publish_ack_mode = "write_ack"` returns PUB, MPUB, and DPUB success + after append/write, but consumers see only through `last_durable_position`. +- `queue.publish_ack_mode = "nsq_relaxed"` returns after append/write and makes + that tail immediately consumable, matching NSQ's relaxed diskqueue boundary + more closely. +- Both relaxed modes sync at the first of 2,500 messages, 8 MiB, or 10 ms by + default. A sync failure isolates the broker from further writes. +- Unsynced messages, bytes, last durable position, and sync lag are exposed in + broker statistics and aggregate/bounded per-Topic Prometheus metrics. +- Recovery preserves non-overlapping position gaps when an acknowledged + relaxed tail disappears, preventing an older durable Channel ACK from + skipping a later message. Recovered gaps are derived from segment ranges, + Topic metadata, and durable Channel high-water marks; they do not consume + Channel depth or its bounded ACK window. + +## NSQ comparison + +NSQ v1.3.0 tracks in-flight timeouts with a priority queue, which exposed the +previous RustQueue scan as a direct scheduler mismatch. RustQueue now has the +same asymptotic deadline lookup shape, while keeping its own token-checked +delivery model. + +Important throughput results still require a durability label: + +- RustQueue's default successful `PUB`, `MPUB`, and `DPUB` response follows a + local segment group fsync. `write_ack` and `nsq_relaxed` are separately + labelled weaker acknowledgement profiles. Successful `FIN` and `REQ` always + cross a Channel WAL group fsync. +- NSQ's diskqueue returns a write response before its next scheduled sync. + Therefore NSQ `--sync-every=1` is not equal to RustQueue's + acknowledgement-after-fsync contract, and NSQ's default memory queue can + further widen a raw-throughput result. +- `scripts/benchmark-compare.sh` continues to run NSQ with + `--mem-queue-size=0` and emits strict and relaxed sync profiles separately. + Consumer runs require complete unique delivery, no unexpected duplicates, + and a drained Channel before a result is accepted. + +[`docs/architecture/nsq-performance.md`](../architecture/nsq-performance.md) +records the concrete source-level differences and the next measurements worth +making before any durability trade-off. + +## Compatibility + +- On-disk format remains v7; no migration is required from 0.8.2. +- NSQ V2 commands, TLS/mTLS, AUTH, compression, lookup, and the opt-in Kodo + profile retain their existing compatibility contract. +- The Broker remains share-nothing: one Broker owns one durable RWO PVC and + the release does not add replica-based message durability. + +## Assets + +- `rustqueue-0.8.3-linux-x86_64.tar.gz`: Linux x86_64 binaries, Console UI and + example configuration +- `rustqueue-0.8.3-linux-aarch64.tar.gz`: Linux ARM64 binaries, Console UI and + example configuration +- `rustqueue-0.8.3-source.tar.gz`: source archive for the tagged commit +- `rustqueue-0.8.3.tgz`: Helm Chart +- `SHA256SUMS-0.8.3`: SHA-256 checksums for all assets diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index 9318bb9..0bfb529 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -2017,7 +2017,7 @@ dependencies = [ [[package]] name = "rustqueue-discovery" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "axum", @@ -2055,7 +2055,7 @@ dependencies = [ [[package]] name = "rustqueue-protocol" -version = "0.8.2" +version = "0.8.3" dependencies = [ "bytes", "serde", @@ -2065,7 +2065,7 @@ dependencies = [ [[package]] name = "rustqueue-proxy" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "axum", @@ -2085,7 +2085,7 @@ dependencies = [ [[package]] name = "rustqueue-queue" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "bytes", @@ -2105,7 +2105,7 @@ dependencies = [ [[package]] name = "rustqueue-storage" -version = "0.8.2" +version = "0.8.3" dependencies = [ "anyhow", "crc32c", @@ -2117,7 +2117,7 @@ dependencies = [ [[package]] name = "rustqueue-telemetry" -version = "0.8.2" +version = "0.8.3" dependencies = [ "serde", ] diff --git a/rustqueue.example.toml b/rustqueue.example.toml index 0048cdd..ef81feb 100644 --- a/rustqueue.example.toml +++ b/rustqueue.example.toml @@ -48,6 +48,14 @@ max_topics = 10000 # Each topic accepts at most 1024 channels to bound channel WAL/index growth. max_publish_workers = 1024 publish_worker_idle_seconds = 60 +# "durable" returns PUB success and exposes messages only after local fsync. +# "write_ack" returns after append but exposes messages only after background fsync. +# "nsq_relaxed" returns and exposes messages after append, before background fsync. +publish_ack_mode = "durable" +# Relaxed modes sync when the first message, byte, or time threshold is reached. +relaxed_sync_messages = 2500 +relaxed_sync_bytes = 8388608 +relaxed_sync_interval_ms = 10 # Lets consumers discover a newly owning broker without losing the first messages. bootstrap_retention_seconds = 90 message_retention_seconds = 0 diff --git a/scripts/acceptance-multi-broker-k8s.sh b/scripts/acceptance-multi-broker-k8s.sh index a9f23d0..ef27de9 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.2-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image + BUILD_VERSION=0.8.3-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image docker tag rustqueue:dev "$BROKER_IMAGE_A" - BUILD_VERSION=0.8.2-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist + BUILD_VERSION=0.8.3-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-compare.sh b/scripts/benchmark-compare.sh index 38b498d..e246019 100755 --- a/scripts/benchmark-compare.sh +++ b/scripts/benchmark-compare.sh @@ -12,16 +12,27 @@ payloads=${PAYLOADS:-"100 1024 10240"} warmup_seconds=${WARMUP_SECONDS:-60} duration_seconds=${DURATION_SECONDS:-600} fixed_rate=${FIXED_RATE:-100} +run_relaxed=${RUN_RELAXED:-0} +relaxed_sync_messages=${RELAXED_SYNC_MESSAGES:-2500} +relaxed_sync_bytes=${RELAXED_SYNC_BYTES:-8388608} +relaxed_sync_interval_ms=${RELAXED_SYNC_INTERVAL_MS:-10} sampler_pid= mkdir -p "$result_dir" +case "$run_relaxed" in + 0|1) ;; + *) echo "RUN_RELAXED must be 0 or 1" >&2; exit 2 ;; +esac + cleanup() { if [ -n "$sampler_pid" ]; then kill "$sampler_pid" >/dev/null 2>&1 || true wait "$sampler_pid" >/dev/null 2>&1 || true fi - docker rm -f rq-bench-rustqueue rq-bench-nsq-strict rq-bench-nsq-default >/dev/null 2>&1 || true + docker rm -f rq-bench-rustqueue rq-bench-rustqueue-write-ack \ + rq-bench-rustqueue-nsq-relaxed rq-bench-nsq-strict rq-bench-nsq-default >/dev/null 2>&1 || true docker volume rm rq-bench-rustqueue-data rq-bench-nsq-strict-data \ + rq-bench-rustqueue-write-ack-data rq-bench-rustqueue-nsq-relaxed-data \ rq-bench-nsq-default-data >/dev/null 2>&1 || true } trap cleanup EXIT INT TERM @@ -33,6 +44,24 @@ cleanup docker run -d --name rq-bench-rustqueue --cpus 2 --memory 2g \ -e RUSTQUEUE_DATA_PATH=/data -v rq-bench-rustqueue-data:/data \ -p 14150:4150 -p 14151:4151 rustqueue:bench >/dev/null +if [ "$run_relaxed" = 1 ]; then + docker run -d --name rq-bench-rustqueue-write-ack --cpus 2 --memory 2g \ + -e RUSTQUEUE_DATA_PATH=/data \ + -e RUSTQUEUE_PUBLISH_ACK_MODE=write_ack \ + -e RUSTQUEUE_RELAXED_SYNC_MESSAGES="$relaxed_sync_messages" \ + -e RUSTQUEUE_RELAXED_SYNC_BYTES="$relaxed_sync_bytes" \ + -e RUSTQUEUE_RELAXED_SYNC_INTERVAL_MS="$relaxed_sync_interval_ms" \ + -v rq-bench-rustqueue-write-ack-data:/data -p 17150:4150 -p 17151:4151 \ + rustqueue:bench >/dev/null + docker run -d --name rq-bench-rustqueue-nsq-relaxed --cpus 2 --memory 2g \ + -e RUSTQUEUE_DATA_PATH=/data \ + -e RUSTQUEUE_PUBLISH_ACK_MODE=nsq_relaxed \ + -e RUSTQUEUE_RELAXED_SYNC_MESSAGES="$relaxed_sync_messages" \ + -e RUSTQUEUE_RELAXED_SYNC_BYTES="$relaxed_sync_bytes" \ + -e RUSTQUEUE_RELAXED_SYNC_INTERVAL_MS="$relaxed_sync_interval_ms" \ + -v rq-bench-rustqueue-nsq-relaxed-data:/data -p 18150:4150 -p 18151:4151 \ + rustqueue:bench >/dev/null +fi docker run -d --name rq-bench-nsq-strict --cpus 2 --memory 2g \ -v rq-bench-nsq-strict-data:/data -p 15150:4150 -p 15151:4151 \ nsqio/nsq:v1.3.0 /nsqd --mem-queue-size=0 --sync-every=1 --data-path=/data >/dev/null @@ -40,7 +69,9 @@ docker run -d --name rq-bench-nsq-default --cpus 2 --memory 2g \ -v rq-bench-nsq-default-data:/data -p 16150:4150 -p 16151:4151 \ nsqio/nsq:v1.3.0 /nsqd --mem-queue-size=0 --sync-every=2500 --data-path=/data >/dev/null -for port in 14151 15151 16151; do +ports="14151 15151 16151" +if [ "$run_relaxed" = 1 ]; then ports="$ports 17151 18151"; fi +for port in $ports; do until curl -fsS "http://127.0.0.1:$port/ping" >/dev/null; do sleep 1; done done @@ -75,17 +106,28 @@ for payload in $payloads; do run=1 while [ "$run" -le "$runs" ]; do run_bench rustqueue-local-fsync 14150 rq-bench-rustqueue "$payload" "$run" + if [ "$run_relaxed" = 1 ]; then + run_bench rustqueue-write-ack 17150 rq-bench-rustqueue-write-ack "$payload" "$run" + run_bench rustqueue-nsq-relaxed 18150 rq-bench-rustqueue-nsq-relaxed "$payload" "$run" + fi run_bench nsq-sync-every-1 15150 rq-bench-nsq-strict "$payload" "$run" run_bench nsq-sync-every-2500 16150 rq-bench-nsq-default "$payload" "$run" run_bench rustqueue-local-fsync-fixed 14150 rq-bench-rustqueue "$payload" "$run" "$fixed_rate" + if [ "$run_relaxed" = 1 ]; then + run_bench rustqueue-write-ack-fixed 17150 rq-bench-rustqueue-write-ack "$payload" "$run" "$fixed_rate" + run_bench rustqueue-nsq-relaxed-fixed 18150 rq-bench-rustqueue-nsq-relaxed "$payload" "$run" "$fixed_rate" + fi run_bench nsq-sync-every-1-fixed 15150 rq-bench-nsq-strict "$payload" "$run" "$fixed_rate" run=$((run + 1)) done done +profiles="rustqueue-local-fsync nsq-sync-every-1 nsq-sync-every-2500 rustqueue-local-fsync-fixed nsq-sync-every-1-fixed" +if [ "$run_relaxed" = 1 ]; then + profiles="$profiles rustqueue-write-ack rustqueue-nsq-relaxed rustqueue-write-ack-fixed rustqueue-nsq-relaxed-fixed" +fi for payload in $payloads; do - for name in rustqueue-local-fsync nsq-sync-every-1 nsq-sync-every-2500 \ - rustqueue-local-fsync-fixed nsq-sync-every-1-fixed; do + for name in $profiles; do jq -s ' def median_by($field): map(select(.[$field] != null)) diff --git a/scripts/benchmark-qualify.sh b/scripts/benchmark-qualify.sh index fe8a21d..bd3f14b 100755 --- a/scripts/benchmark-qualify.sh +++ b/scripts/benchmark-qualify.sh @@ -2,8 +2,8 @@ set -euo pipefail ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" -RELEASE="${RELEASE:-0.8.2}" -BASELINE_REF="${BASELINE_REF:-v0.8.1}" +RELEASE="${RELEASE:-0.8.3}" +BASELINE_REF="${BASELINE_REF:-v0.8.2}" CANDIDATE_REF="${CANDIDATE_REF:-HEAD}" PAIRS="${PAIRS:-10}" WARMUP_SECONDS="${WARMUP_SECONDS:-30}" @@ -15,7 +15,7 @@ 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}" +EVIDENCE_OUTPUT="${EVIDENCE_OUTPUT:-$ROOT/benchmarks/qualifications/v0.8.3-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" @@ -103,13 +103,15 @@ if [[ "$docker_context" != "orbstack" && "$docker_os" != *OrbStack* ]]; then die "Docker must use OrbStack (context=$docker_context, os=$docker_os)" fi +[[ "$BASELINE_REF" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]] || + die "baseline reference must be a version tag" 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}")" +tag_commit="$(git -C "$ROOT" rev-parse --verify "refs/tags/$BASELINE_REF^{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" + die "baseline must resolve to the exact $BASELINE_REF tag commit" case "$EVIDENCE_OUTPUT" in "$ROOT/benchmarks/qualifications/"*) @@ -151,8 +153,9 @@ read_workspace_version() { 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" +expected_baseline_version="${BASELINE_REF#v}" +[[ "$baseline_version" == "$expected_baseline_version" ]] || + die "baseline workspace version is $baseline_version, expected $expected_baseline_version" if [[ "$QUALIFICATION_DEV" == 0 && "$candidate_version" != "$RELEASE" ]]; then die "candidate workspace version is $candidate_version, expected $RELEASE" fi