Release RustQueue 0.8.4

Cargo.toml

@@ -16,7 +16,7 @@ members = [
 exclude = ["fuzz"]
 
 [workspace.package]
-version = "0.8.3"
+version = "0.8.4"
 edition = "2021"
 license = "Apache-2.0"
 rust-version = "1.88"

README.md

@@ -11,14 +11,14 @@
 [NSQ performance boundaries](docs/architecture/nsq-performance.md) ·
 [Kubernetes operations](docs/operations/kubernetes.md) ·
 [Console operations](docs/operations/console.md) ·
-[v0.8.3 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.3)
+[v0.8.4 release](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.4)
 
-RustQueue 0.8.3 is a Kubernetes-native, NSQ V2-compatible message queue for
+RustQueue 0.8.4 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.3](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.3).
+> Current release: [v0.8.4](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.4).
 > 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.
@@ -44,45 +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.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)
+## What's new in 0.8.4
+
+- **Larger durable Channel groups.** `FIN` and `REQ` requests now share a
+  bounded 1 ms Channel WAL group commit of up to 1,024 requests. Every success
+  still crosses the affected WAL `fsync` before the client sees completion.
+- **More delivery concurrency.** Channel WAL `fsync` runs after releasing the
+  Topic state lock and uses an independent commit gate, so delivery reservations
+  can continue while the durable boundary is being written.
+- **Lower hot-path allocation cost.** WAL commands use stack encoding, TCP and
+  Channel operations share `Arc<str>` channel identities, Channel futures avoid
+  an extra `Box`, and each session bounds active Channel operations at 256.
+- **Stricter benchmark completion.** The benchmark consumer waits for EOF and
+  verifies a complete final drain before accepting a result.
+- **No compatibility change.** The v7 on-disk format and the durable `FIN`/`REQ`
+  fsync contract remain unchanged.
+
+See the [v0.8.4 release notes](https://github.com/SamuelSupe/rustqueue/releases/tag/v0.8.4)
 and [NSQ performance boundaries](docs/architecture/nsq-performance.md) for
 the contract and benchmark interpretation.
 
-## Download 0.8.3
+## Download 0.8.4
 
 Every release contains native Linux binaries, the Console UI, source, the Helm
 Chart and a checksum manifest:
 
 | Asset | Contents |
 | --- | --- |
-| `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 |
+| `rustqueue-0.8.4-linux-x86_64.tar.gz` | Linux x86_64 binaries, Console UI and example configuration |
+| `rustqueue-0.8.4-linux-aarch64.tar.gz` | Linux ARM64 binaries, Console UI and example configuration |
+| `rustqueue-0.8.4-source.tar.gz` | Source archive for the tagged commit |
+| `rustqueue-0.8.4.tgz` | Helm Chart |
+| `SHA256SUMS-0.8.4` | SHA-256 checksums for every downloadable artifact |
 
 ```sh
 arch="$(uname -m)"
-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"
+curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.4/rustqueue-0.8.4-linux-${arch}.tar.gz"
+curl -LO "https://github.com/SamuelSupe/rustqueue/releases/download/v0.8.4/SHA256SUMS-0.8.4"
+sha256sum --check --ignore-missing SHA256SUMS-0.8.4
+tar -xzf "rustqueue-0.8.4-linux-${arch}.tar.gz"
 ```
 
 ## Architecture
@@ -115,8 +115,10 @@ operator -> eligible nodes -> StatefulSet ordinal + retained RWO PVC
 - IDs are durably reserved in blocks, so restarts may create harmless gaps but
   never reuse a broker-scoped message ID.
 - Concurrent `FIN` and `REQ` for one topic share a bounded group commit (up to
-  64 requests with at most 1 ms coalescing delay). A successful response has
-  passed every affected local channel WAL `fsync`.
+  1,024 requests with at most 1 ms coalescing delay). The Topic state lock is
+  released during the WAL `fsync`, while an independent gate serializes durable
+  Channel mutations. A successful response has passed every affected local
+  channel WAL `fsync`.
 - Delivery is at least once. A restart redelivers messages without durable FIN.
 - The broker PVC is the only copy; permanent PVC loss loses its messages.
 - Topics and channels are broker-local. Lookup consumers union all owners.
@@ -223,7 +225,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.3 \
+  --set queue.image=registry.example/rustqueue:0.8.4 \
   --set queue.storageClassName=ssd-rwo
 ```
 
@@ -460,7 +462,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.3 CI/CD workflow publishes a Release only after the non-Kubernetes
+The v0.8.4 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

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.2` tag with `HEAD`, uses fresh
+The default protocol compares the exact `v0.8.3` 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.3-orbstack.json`. Consumer cases must drain
-completely and have a fixed 1,800-second timeout so the v0.8.2 durable `FIN`
+reviewable artifact to `v0.8.4-orbstack.json`. Consumer cases must drain
+completely and have a fixed 1,800-second timeout so the v0.8.3 durable `FIN`
 baseline is not rejected merely for exceeding a short operational timeout.
 
-The RustQueue 0.8.3 release does not make this optional 60-run artifact a
+The RustQueue 0.8.4 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.

console-ui/package.json

@@ -1,7 +1,7 @@
 {
   "name": "rustqueue-console-ui",
   "private": true,
-  "version": "0.8.3",
+  "version": "0.8.4",
   "packageManager": "pnpm@11.9.0",
   "type": "module",
   "scripts": {

crates/bench/src/consumer.rs

@@ -317,13 +317,28 @@ where
     R: AsyncRead + Unpin,
     W: AsyncWrite + Unpin,
 {
+    // CLOSE_WAIT acknowledges CLS, while the server may still be making
+    // already-received FINs durable. EOF is the completed session boundary.
+    let mut close_wait = false;
     loop {
-        let (frame_type, response) = match pending.take() {
-            Some(frame) => frame,
-            None => read_frame(reader).await?,
+        let frame = match pending.take() {
+            Some(frame) => Ok(frame),
+            None => read_frame(reader).await,
+        };
+        let (frame_type, response) = match frame {
+            Ok(frame) => frame,
+            Err(error)
+                if close_wait
+                    && error
+                        .downcast_ref::<std::io::Error>()
+                        .is_some_and(|error| error.kind() == std::io::ErrorKind::UnexpectedEof) =>
+            {
+                return Ok(());
+            }
+            Err(error) => return Err(error),
         };
         match frame_type {
-            0 if response == b"CLOSE_WAIT" => return Ok(()),
+            0 if response == b"CLOSE_WAIT" => close_wait = true,
             0 if response == b"_heartbeat_" => writer.write_all(b"NOP\n").await?,
             2 if response.len() >= 26 => {
                 writer.write_all(b"FIN ").await?;
@@ -420,6 +435,33 @@ mod tests {
         assert_eq!(result.snapshot.unique, 0);
     }
 
+    #[tokio::test]
+    async fn close_wait_does_not_complete_before_the_server_closes() {
+        let (client, mut server) = tokio::io::duplex(1024);
+        let (mut reader, mut writer) = tokio::io::split(client);
+        let (close_tx, close_rx) = oneshot::channel();
+        let server_task = tokio::spawn(async move {
+            server
+                .write_all(&test_frame(0, b"CLOSE_WAIT"))
+                .await
+                .unwrap();
+            server.flush().await.unwrap();
+            close_rx.await.unwrap();
+        });
+        let mut client_task =
+            tokio::spawn(async move { close_consumer(&mut reader, &mut writer, None).await });
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(50), &mut client_task)
+                .await
+                .is_err(),
+            "CLOSE_WAIT must not be treated as the durable session boundary"
+        );
+        close_tx.send(()).unwrap();
+        client_task.await.unwrap().unwrap();
+        server_task.await.unwrap();
+    }
+
     #[tokio::test]
     async fn stopping_during_a_partial_frame_preserves_the_frame_boundary() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

crates/queue/src/broker.rs

@@ -464,6 +464,7 @@ impl Broker {
             broker.ensure_management_access(&topic, Some(&channel))?;
             let handle = broker.get_or_create_topic_locked(&topic)?;
             let _commit_gate = handle.commit_gate.lock();
+            let _channel_commit_gate = handle.channel_commit_gate.lock();
             if handle
                 .state
                 .lock()
@@ -487,6 +488,7 @@ impl Broker {
             broker.ensure_management_access(&topic, Some(&channel))?;
             let handle = broker.topic(&topic)?;
             let _commit_gate = handle.commit_gate.lock();
+            let _channel_commit_gate = handle.channel_commit_gate.lock();
             handle.state.lock().delete_channel(&channel)?;
             broker.bump_registry()?;
             Ok(())
@@ -524,6 +526,7 @@ impl Broker {
             broker.ensure_management_access(&topic, Some(&channel))?;
             let handle = broker.topic(&topic)?;
             let _commit_gate = handle.commit_gate.lock();
+            let _channel_commit_gate = handle.channel_commit_gate.lock();
             let result = handle.state.lock().set_channel_paused(&channel, paused);
             result
         })
@@ -539,6 +542,7 @@ impl Broker {
             broker.ensure_management_access(&topic, None)?;
             let handle = broker.topic(&topic)?;
             let _commit_gate = handle.commit_gate.lock();
+            let _channel_commit_gate = handle.channel_commit_gate.lock();
             let result = handle.state.lock().empty_topic();
             result
         })
@@ -555,6 +559,7 @@ impl Broker {
             broker.ensure_management_access(&topic, Some(&channel))?;
             let handle = broker.topic(&topic)?;
             let _commit_gate = handle.commit_gate.lock();
+            let _channel_commit_gate = handle.channel_commit_gate.lock();
             let result = handle.state.lock().empty_channel(&channel);
             result
         })