diff --git a/Cargo.lock b/Cargo.lock
index 0713b21..6e4eeff 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2283,7 +2283,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-bench"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "clap",
@@ -2295,7 +2295,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-console"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "axum",
@@ -2323,7 +2323,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-discovery"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "axum",
@@ -2346,7 +2346,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-operator"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "axum",
@@ -2373,7 +2373,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-protocol"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "bytes",
  "serde",
@@ -2383,7 +2383,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-proxy"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "axum",
@@ -2404,7 +2404,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-queue"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "bytes",
@@ -2426,7 +2426,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-server"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "async-compression",
@@ -2462,7 +2462,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-storage"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "crc32c",
@@ -2475,14 +2475,14 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-telemetry"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "serde",
 ]
 
 [[package]]
 name = "rustqueuectl"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "clap",
diff --git a/Cargo.toml b/Cargo.toml
index 1bc891d..363b4d1 100644
--- a/Cargo.toml
+++ b/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"
diff --git a/README.md b/README.md
index b8b9c4c..c3c4027 100644
--- a/README.md
+++ b/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
diff --git a/benchmarks/qualifications/README.md b/benchmarks/qualifications/README.md
index 3901fd3..4fbc170 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.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.
diff --git a/console-ui/package.json b/console-ui/package.json
index a5acb86..d80ae0a 100644
--- a/console-ui/package.json
+++ b/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": {
diff --git a/crates/bench/src/consumer.rs b/crates/bench/src/consumer.rs
index adf3695..b65566c 100644
--- a/crates/bench/src/consumer.rs
+++ b/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();
diff --git a/crates/queue/src/broker.rs b/crates/queue/src/broker.rs
index b8b05a8..6669880 100644
--- a/crates/queue/src/broker.rs
+++ b/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
         })
diff --git a/crates/queue/src/broker/channel_commit.rs b/crates/queue/src/broker/channel_commit.rs
index 2b2ec8e..75edbec 100644
--- a/crates/queue/src/broker/channel_commit.rs
+++ b/crates/queue/src/broker/channel_commit.rs
@@ -9,7 +9,7 @@ use std::time::{Duration, Instant};
 use tokio::sync::{mpsc, oneshot};
 
 const QUEUE_CAPACITY: usize = 1024;
-const MAX_GROUP_REQUESTS: usize = 256;
+const MAX_GROUP_REQUESTS: usize = QUEUE_CAPACITY;
 const COALESCE_DELAY: Duration = Duration::from_millis(1);
 
 pub(super) enum ChannelOperation {
@@ -43,7 +43,7 @@ struct WorkerEntry {
 }
 
 struct ChannelRequest {
-    channel: String,
+    channel: Arc<str>,
     operation: ChannelOperation,
     enqueued_at: Instant,
     reply: oneshot::Sender<Result<(), BrokerError>>,
@@ -72,7 +72,7 @@ impl ChannelGroups {
         &self,
         broker: &Broker,
         topic: &str,
-        channel: String,
+        channel: Arc<str>,
         operation: ChannelOperation,
     ) -> Result<(), BrokerError> {
         let sender = self.sender(broker, topic)?;
@@ -171,6 +171,7 @@ impl Broker {
                 return;
             }
         };
+        let _channel_commit_gate = handle.channel_commit_gate.lock();
         let mut topic_state = handle.state.lock();
         let mut pending = Vec::with_capacity(requests.len());
         let mut touched = BTreeSet::new();
@@ -187,7 +188,7 @@ impl Broker {
                 .metrics
                 .channel_group_commit_wait
                 .observe(enqueued_at.elapsed());
-            if let Err(error) = self.ensure_management_access(topic, Some(&channel)) {
+            if let Err(error) = self.ensure_management_access(topic, Some(channel.as_ref())) {
                 let _ = reply.send(Err(error));
                 continue;
             }
@@ -196,12 +197,12 @@ impl Broker {
                     id,
                     require_in_flight,
                     token,
-                } => topic_state.finish_buffered(&channel, id, require_in_flight, token),
+                } => topic_state.finish_buffered(channel.as_ref(), id, require_in_flight, token),
                 ChannelOperation::Requeue {
                     id,
                     available_at_ms,
                     token,
-                } => topic_state.requeue_buffered(&channel, id, available_at_ms, token),
+                } => topic_state.requeue_buffered(channel.as_ref(), id, available_at_ms, token),
             };
             match result {
                 Ok(()) => {
@@ -226,16 +227,35 @@ impl Broker {
         }
 
         rustqueue_storage::crash_failpoint("channel_group_after_append_before_fsync");
+        let syncs = match topic_state.prepare_channel_wal_syncs(touched.iter()) {
+            Ok(syncs) => syncs,
+            Err(error) => {
+                self.observe_storage_result::<()>(Err(copy_error(&error)))
+                    .ok();
+                fail_pending(pending);
+                return;
+            }
+        };
+        // Delivery reservations only need the Topic state lock. The Channel
+        // gate above keeps WAL/checkpoint writers serialized during this fsync.
+        drop(topic_state);
         let sync_result = {
             let _timer = self.inner.metrics.channel_fsync.timer();
-            topic_state.sync_channel_wals(touched.iter())
+            syncs
+                .into_iter()
+                .try_for_each(|wal| wal.sync_data().map_err(BrokerError::from))
         };
         if let Err(error) = sync_result {
+            handle
+                .state
+                .lock()
+                .mark_channel_wal_sync_failed(touched.iter());
             self.observe_storage_result::<()>(Err(copy_error(&error)))
                 .ok();
             fail_pending(pending);
             return;
         }
+        let mut topic_state = handle.state.lock();
         if let Err(error) = topic_state.checkpoint_channels_if_needed(touched.iter()) {
             self.observe_storage_result::<()>(Err(copy_error(&error)))
                 .ok();
@@ -273,7 +293,7 @@ async fn run_worker(
                 continue;
             }
         };
-        let requests = collect_group(first, &mut receiver).await;
+        let requests = collect_group(first, &mut receiver, COALESCE_DELAY).await;
         let Some(inner) = broker.upgrade() else {
             return;
         };
@@ -296,8 +316,9 @@ async fn run_worker(
 async fn collect_group(
     first: ChannelRequest,
     receiver: &mut mpsc::Receiver<ChannelRequest>,
+    coalesce_delay: Duration,
 ) -> Vec<ChannelRequest> {
-    let deadline = tokio::time::Instant::now() + COALESCE_DELAY;
+    let deadline = tokio::time::Instant::now() + coalesce_delay;
     let mut requests = vec![first];
     loop {
         if requests.len() >= MAX_GROUP_REQUESTS {
@@ -358,7 +379,9 @@ mod tests {
         }
         let first = receiver.recv().await.unwrap();
         assert_eq!(
-            collect_group(first, &mut receiver).await.len(),
+            collect_group(first, &mut receiver, COALESCE_DELAY)
+                .await
+                .len(),
             MAX_GROUP_REQUESTS
         );
         assert!(receiver.try_recv().is_ok());
@@ -369,13 +392,12 @@ mod tests {
         let (sender, mut receiver) = mpsc::channel(8);
         sender.send(request()).await.unwrap();
         sender.send(request()).await.unwrap();
-        let late_sender = tokio::spawn(async move {
-            tokio::task::yield_now().await;
-            sender.send(request()).await.unwrap();
-        });
-
         let first = receiver.recv().await.unwrap();
-        assert_eq!(collect_group(first, &mut receiver).await.len(), 3);
-        late_sender.await.unwrap();
+        let collector = tokio::spawn(async move {
+            collect_group(first, &mut receiver, Duration::from_secs(1)).await
+        });
+        tokio::task::yield_now().await;
+        sender.send(request()).await.unwrap();
+        assert_eq!(collector.await.unwrap().len(), 3);
     }
 }
diff --git a/crates/queue/src/broker/io.rs b/crates/queue/src/broker/io.rs
index 3e2dfb4..0413f83 100644
--- a/crates/queue/src/broker/io.rs
+++ b/crates/queue/src/broker/io.rs
@@ -11,7 +11,8 @@ pub(super) const SEQUENCE_RESERVATION: u64 = 1 << 20;
 
 impl Broker {
     pub async fn finish(&self, topic: &str, channel: &str, id: u64) -> Result<(), BrokerError> {
-        self.finish_with_token(topic, channel, id, None).await
+        self.finish_with_token(topic, Arc::from(channel), id, None)
+            .await
     }
 
     pub async fn finish_delivery(
@@ -20,6 +21,18 @@ impl Broker {
         channel: &str,
         id: u64,
         token: u64,
+    ) -> Result<(), BrokerError> {
+        self.finish_with_token(topic, Arc::from(channel), id, Some(token))
+            .await
+    }
+
+    /// Finishes a delivery while reusing a caller-owned shared Channel name.
+    pub async fn finish_delivery_shared(
+        &self,
+        topic: &str,
+        channel: Arc<str>,
+        id: u64,
+        token: u64,
     ) -> Result<(), BrokerError> {
         self.finish_with_token(topic, channel, id, Some(token))
             .await
@@ -28,19 +41,19 @@ impl Broker {
     async fn finish_with_token(
         &self,
         topic: &str,
-        channel: &str,
+        channel: Arc<str>,
         id: u64,
         token: Option<u64>,
     ) -> Result<(), BrokerError> {
         let _timer = self.inner.metrics.channel_ack.timer();
         self.ensure_storage_healthy()?;
-        self.ensure_management_access(topic, Some(channel))?;
+        self.ensure_management_access(topic, Some(channel.as_ref()))?;
         self.inner
             .channel_groups
             .submit(
                 self,
                 topic,
-                channel.to_owned(),
+                channel,
                 super::channel_commit::ChannelOperation::Finish {
                     id,
                     require_in_flight: true,
@@ -61,11 +74,10 @@ impl Broker {
         let channel = channel.to_owned();
         let result = self
             .storage_task(move || {
-                broker
-                    .topic(&topic)?
-                    .state
-                    .lock()
-                    .finish(&channel, id, false)
+                let handle = broker.topic(&topic)?;
+                let _channel_commit_gate = handle.channel_commit_gate.lock();
+                let result = handle.state.lock().finish(&channel, id, false);
+                result
             })
             .await;
         match result {
@@ -99,7 +111,7 @@ impl Broker {
         id: u64,
         delay: Duration,
     ) -> Result<(), BrokerError> {
-        self.requeue_with_token(topic, channel, id, None, delay)
+        self.requeue_with_token(topic, Arc::from(channel), id, None, delay)
             .await
     }
 
@@ -110,6 +122,19 @@ impl Broker {
         id: u64,
         token: u64,
         delay: Duration,
+    ) -> Result<(), BrokerError> {
+        self.requeue_with_token(topic, Arc::from(channel), id, Some(token), delay)
+            .await
+    }
+
+    /// Requeues a delivery while reusing a caller-owned shared Channel name.
+    pub async fn requeue_delivery_shared(
+        &self,
+        topic: &str,
+        channel: Arc<str>,
+        id: u64,
+        token: u64,
+        delay: Duration,
     ) -> Result<(), BrokerError> {
         self.requeue_with_token(topic, channel, id, Some(token), delay)
             .await
@@ -118,14 +143,14 @@ impl Broker {
     async fn requeue_with_token(
         &self,
         topic: &str,
-        channel: &str,
+        channel: Arc<str>,
         id: u64,
         token: Option<u64>,
         delay: Duration,
     ) -> Result<(), BrokerError> {
         let _timer = self.inner.metrics.channel_ack.timer();
         self.ensure_storage_healthy()?;
-        self.ensure_management_access(topic, Some(channel))?;
+        self.ensure_management_access(topic, Some(channel.as_ref()))?;
         let available = now_ms().saturating_add(delay.as_millis().min(i64::MAX as u128) as i64);
         let result = self
             .inner
@@ -133,7 +158,7 @@ impl Broker {
             .submit(
                 self,
                 topic,
-                channel.to_owned(),
+                channel,
                 super::channel_commit::ChannelOperation::Requeue {
                     id,
                     available_at_ms: available,
@@ -434,6 +459,7 @@ impl Broker {
                 return Err(error);
             }
             let finish = self.topic(&entry.source_topic).and_then(|topic| {
+                let _channel_commit_gate = topic.channel_commit_gate.lock();
                 topic
                     .state
                     .lock()
diff --git a/crates/queue/src/broker/maintenance.rs b/crates/queue/src/broker/maintenance.rs
index 116ed3f..3dcdbdb 100644
--- a/crates/queue/src/broker/maintenance.rs
+++ b/crates/queue/src/broker/maintenance.rs
@@ -50,6 +50,7 @@ impl Broker {
             for offset in 0..selected {
                 let (_, handle) = &topics[(start + offset) % topics.len()];
                 let _commit_gate = handle.commit_gate.lock();
+                let _channel_commit_gate = handle.channel_commit_gate.lock();
                 let mut topic = handle.state.lock();
                 let retained = broker.inner.payload_reader.retained_paths();
                 let name = topic.name.clone();
@@ -86,6 +87,7 @@ impl Broker {
                 return Ok(None);
             };
             let _commit_gate = topic.commit_gate.lock();
+            let _channel_commit_gate = topic.channel_commit_gate.lock();
             let mut state = topic.state.lock();
             let retained = broker.inner.payload_reader.retained_paths();
             let deliverable_before = state.deliverable_position();
@@ -158,6 +160,7 @@ impl Broker {
                 .read()
                 .values()
                 .try_fold(0usize, |total, topic| {
+                    let _channel_commit_gate = topic.channel_commit_gate.lock();
                     Ok(total.saturating_add(topic.state.lock().expire_in_flight()?))
                 })
         })
@@ -182,11 +185,10 @@ impl Broker {
         let topic = topic.to_owned();
         let channel = channel.to_owned();
         self.storage_task(move || {
-            broker
-                .topic(&topic)?
-                .state
-                .lock()
-                .expire_channel_in_flight(&channel)
+            let handle = broker.topic(&topic)?;
+            let _channel_commit_gate = handle.channel_commit_gate.lock();
+            let result = handle.state.lock().expire_channel_in_flight(&channel);
+            result
         })
         .await
     }
@@ -196,6 +198,7 @@ impl Broker {
         self.storage_task(move || {
             for topic in broker.inner.topics.read().values() {
                 let _commit_gate = topic.commit_gate.lock();
+                let _channel_commit_gate = topic.channel_commit_gate.lock();
                 topic.state.lock().sync()?;
                 topic.signal();
             }
@@ -209,6 +212,7 @@ impl Broker {
         let broker = self.clone();
         self.storage_task(move || {
             for topic in broker.inner.topics.read().values() {
+                let _channel_commit_gate = topic.channel_commit_gate.lock();
                 topic.state.lock().checkpoint_channels()?;
             }
             Ok(())
diff --git a/crates/queue/src/broker/management.rs b/crates/queue/src/broker/management.rs
index f7c0c85..178e589 100644
--- a/crates/queue/src/broker/management.rs
+++ b/crates/queue/src/broker/management.rs
@@ -111,6 +111,7 @@ impl Broker {
                 TopicManagementAction::Empty => {
                     let handle = broker.topic(&topic)?;
                     let _commit_gate = handle.commit_gate.lock();
+                    let _channel_commit_gate = handle.channel_commit_gate.lock();
                     handle.state.lock().empty_topic()?;
                     changed = true;
                 }
@@ -213,6 +214,9 @@ impl Broker {
                 None
             };
             let _idle_commit_gate = idle_handle.as_ref().map(|handle| handle.commit_gate.lock());
+            let _idle_channel_commit_gate = idle_handle
+                .as_ref()
+                .map(|handle| handle.channel_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
@@ -251,6 +255,7 @@ impl Broker {
                 ChannelManagementAction::Create => {
                     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();
                     changed = handle
                         .state
                         .lock()
@@ -265,6 +270,7 @@ impl Broker {
                 ChannelManagementAction::Pause | ChannelManagementAction::Unpause => {
                     let handle = broker.topic(&topic)?;
                     let _commit_gate = handle.commit_gate.lock();
+                    let _channel_commit_gate = handle.channel_commit_gate.lock();
                     handle
                         .state
                         .lock()
@@ -274,6 +280,7 @@ impl Broker {
                 ChannelManagementAction::Empty => {
                     let handle = broker.topic(&topic)?;
                     let _commit_gate = handle.commit_gate.lock();
+                    let _channel_commit_gate = handle.channel_commit_gate.lock();
                     handle.state.lock().empty_channel(&channel)?;
                     changed = true;
                 }
@@ -297,6 +304,7 @@ impl Broker {
                         }
                     } else if let Ok(handle) = broker.topic(&topic) {
                         let _commit_gate = handle.commit_gate.lock();
+                        let _channel_commit_gate = handle.channel_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/topics.rs b/crates/queue/src/broker/topics.rs
index 2da62c4..6e802ab 100644
--- a/crates/queue/src/broker/topics.rs
+++ b/crates/queue/src/broker/topics.rs
@@ -85,7 +85,9 @@ impl Broker {
             return Ok(false);
         };
         let commit_gate = handle.commit_gate.lock();
+        let channel_commit_gate = handle.channel_commit_gate.lock();
         handle.state.lock().mark_deleted()?;
+        drop(channel_commit_gate);
         drop(commit_gate);
         self.inner.topics.write().remove(name);
         let directory = topic_directory(&self.inner.config.data_path, name);
diff --git a/crates/queue/src/broker_tests.rs b/crates/queue/src/broker_tests.rs
index e394261..d7c991e 100644
--- a/crates/queue/src/broker_tests.rs
+++ b/crates/queue/src/broker_tests.rs
@@ -443,6 +443,48 @@ async fn delivery_reservation_does_not_wait_for_the_publish_commit_gate() {
     assert_eq!(&*message.body, b"already-durable");
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn delivery_reservation_does_not_wait_for_channel_wal_sync() {
+    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 || {
+        // Channel commits retain this gate while their WAL is syncing, but
+        // must release the Topic state lock so delivery can continue.
+        let _channel_commit_gate = handle.channel_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();
diff --git a/crates/queue/src/channel_store.rs b/crates/queue/src/channel_store.rs
index 2d2c993..866816a 100644
--- a/crates/queue/src/channel_store.rs
+++ b/crates/queue/src/channel_store.rs
@@ -10,6 +10,7 @@ const WAL_MAGIC: &[u8; 4] = b"RCW7";
 const CHECKPOINT_MAGIC: &[u8; 4] = b"RCC7";
 const HEADER_LEN: usize = 12;
 const MAX_COMMAND_BYTES: usize = 1024 * 1024;
+const ENCODED_COMMAND_BYTES: usize = 35;
 const MAX_CHECKPOINT_BYTES: u64 = 64 * 1024 * 1024;
 const CHECKPOINT_INTERVAL: usize = 8 * 1024;
 const MAX_RECOVERY_COMMANDS: usize = CHECKPOINT_INTERVAL * 2;
@@ -19,6 +20,7 @@ pub(crate) struct ChannelStore {
     checkpoint_path: PathBuf,
     wal_path: PathBuf,
     wal: File,
+    pending_wal: Vec<u8>,
     commands_since_checkpoint: usize,
     isolated: bool,
 }
@@ -47,6 +49,7 @@ impl ChannelStore {
             checkpoint_path,
             wal_path,
             wal,
+            pending_wal: Vec::new(),
             commands_since_checkpoint: 0,
             isolated: false,
         })
@@ -87,6 +90,7 @@ impl ChannelStore {
                 checkpoint_path: checkpoint_path.into(),
                 wal_path,
                 wal,
+                pending_wal: Vec::new(),
                 commands_since_checkpoint: commands.len(),
                 isolated: false,
             },
@@ -103,26 +107,23 @@ impl ChannelStore {
 
     pub fn append_buffered(&mut self, command: &ChannelCommand) -> Result<(), BrokerError> {
         self.ensure_available()?;
-        let body = encode_command(command);
-        if body.len() > MAX_COMMAND_BYTES {
-            return Err(BrokerError::InvalidRecord(
-                "channel command is too large".into(),
-            ));
-        }
+        let mut body = [0u8; ENCODED_COMMAND_BYTES];
+        let body_len = encode_command(command, &mut body);
+        let body = &body[..body_len];
         let mut header = [0u8; HEADER_LEN];
         header[0..4].copy_from_slice(WAL_MAGIC);
         header[4..8].copy_from_slice(&(body.len() as u32).to_be_bytes());
-        header[8..12].copy_from_slice(&crc32c::crc32c(&body).to_be_bytes());
-        if let Err(error) = self.append_bytes(&header, &body) {
-            self.isolated = true;
-            return Err(error);
-        }
+        header[8..12].copy_from_slice(&crc32c::crc32c(body).to_be_bytes());
+        self.pending_wal.reserve(HEADER_LEN + body.len());
+        self.pending_wal.extend_from_slice(&header);
+        self.pending_wal.extend_from_slice(body);
         self.commands_since_checkpoint += 1;
         Ok(())
     }
 
     pub fn sync(&mut self) -> Result<(), BrokerError> {
         self.ensure_available()?;
+        self.flush_pending()?;
         if let Err(error) = self.wal.sync_data() {
             self.isolated = true;
             return Err(error.into());
@@ -130,6 +131,19 @@ impl ChannelStore {
         Ok(())
     }
 
+    pub fn prepare_sync(&mut self) -> Result<File, BrokerError> {
+        self.ensure_available()?;
+        self.flush_pending()?;
+        self.wal.try_clone().map_err(|error| {
+            self.isolated = true;
+            error.into()
+        })
+    }
+
+    pub fn mark_sync_failed(&mut self) {
+        self.isolated = true;
+    }
+
     pub fn checkpoint_if_needed(&mut self, state: &ChannelState) -> Result<(), BrokerError> {
         if self.commands_since_checkpoint < CHECKPOINT_INTERVAL {
             return Ok(());
@@ -147,13 +161,20 @@ impl ChannelStore {
         Ok(())
     }
 
-    fn append_bytes(&mut self, header: &[u8], body: &[u8]) -> Result<(), BrokerError> {
-        self.wal.write_all(header)?;
-        self.wal.write_all(body)?;
+    fn flush_pending(&mut self) -> Result<(), BrokerError> {
+        if self.pending_wal.is_empty() {
+            return Ok(());
+        }
+        if let Err(error) = self.wal.write_all(&self.pending_wal) {
+            self.isolated = true;
+            return Err(error.into());
+        }
+        self.pending_wal.clear();
         Ok(())
     }
 
     fn write_checkpoint_and_reset(&mut self, state: &ChannelState) -> Result<(), BrokerError> {
+        debug_assert!(self.pending_wal.is_empty());
         write_checkpoint(&self.checkpoint_path, &state.checkpoint())?;
         self.wal = OpenOptions::new()
             .write(true)
@@ -357,16 +378,16 @@ fn recover_wal(path: &Path) -> Result<Vec<ChannelCommand>, BrokerError> {
     Ok(commands)
 }
 
-fn encode_command(command: &ChannelCommand) -> Vec<u8> {
-    let mut body = Vec::with_capacity(27);
+fn encode_command(command: &ChannelCommand, body: &mut [u8; ENCODED_COMMAND_BYTES]) -> usize {
     match *command {
         ChannelCommand::Finish {
             position,
             message_id,
         } => {
-            body.push(1);
-            body.extend_from_slice(&position.to_be_bytes());
-            body.extend_from_slice(&message_id.to_be_bytes());
+            body[0] = 1;
+            body[1..9].copy_from_slice(&position.to_be_bytes());
+            body[9..17].copy_from_slice(&message_id.to_be_bytes());
+            17
         }
         ChannelCommand::Requeue {
             position,
@@ -375,32 +396,38 @@ fn encode_command(command: &ChannelCommand) -> Vec<u8> {
             attempts,
             cumulative_count,
         } => {
-            body.push(if cumulative_count.is_some() { 7 } else { 2 });
-            body.extend_from_slice(&position.to_be_bytes());
-            body.extend_from_slice(&message_id.to_be_bytes());
-            body.extend_from_slice(&available_at_ms.to_be_bytes());
-            body.extend_from_slice(&attempts.to_be_bytes());
+            body[0] = if cumulative_count.is_some() { 7 } else { 2 };
+            body[1..9].copy_from_slice(&position.to_be_bytes());
+            body[9..17].copy_from_slice(&message_id.to_be_bytes());
+            body[17..25].copy_from_slice(&available_at_ms.to_be_bytes());
+            body[25..27].copy_from_slice(&attempts.to_be_bytes());
             if let Some(count) = cumulative_count {
-                body.extend_from_slice(&count.to_be_bytes());
+                body[27..35].copy_from_slice(&count.to_be_bytes());
+                35
+            } else {
+                27
             }
         }
         ChannelCommand::Pause { paused } => {
-            body.extend_from_slice(&[3, paused as u8]);
+            body[..2].copy_from_slice(&[3, paused as u8]);
+            2
         }
         ChannelCommand::Empty { through_position } => {
-            body.push(4);
-            body.extend_from_slice(&through_position.to_be_bytes());
+            body[0] = 4;
+            body[1..9].copy_from_slice(&through_position.to_be_bytes());
+            9
         }
         ChannelCommand::Evict { through_position } => {
-            body.push(5);
-            body.extend_from_slice(&through_position.to_be_bytes());
+            body[0] = 5;
+            body[1..9].copy_from_slice(&through_position.to_be_bytes());
+            9
         }
         ChannelCommand::Timeout { cumulative_count } => {
-            body.push(6);
-            body.extend_from_slice(&cumulative_count.to_be_bytes());
+            body[0] = 6;
+            body[1..9].copy_from_slice(&cumulative_count.to_be_bytes());
+            9
         }
     }
-    body
 }
 
 fn decode_command(body: &[u8]) -> Result<ChannelCommand, BrokerError> {
@@ -606,7 +633,9 @@ mod tests {
             position: 1,
             message_id: 7,
         };
-        let mut encoded = encode_command(&command);
+        let mut buffer = [0u8; ENCODED_COMMAND_BYTES];
+        let len = encode_command(&command, &mut buffer);
+        let mut encoded = buffer[..len].to_vec();
         encoded.push(0);
         assert!(decode_command(&encoded).is_err());
     }
@@ -620,11 +649,13 @@ mod tests {
             attempts: 1,
             cumulative_count: None,
         };
-        let encoded = encode_command(&legacy);
+        let mut encoded = [0u8; ENCODED_COMMAND_BYTES];
+        let len = encode_command(&legacy, &mut encoded);
+        let encoded = &encoded[..len];
         assert_eq!(encoded[0], 2);
         assert_eq!(encoded.len(), 27);
         assert!(matches!(
-            decode_command(&encoded).unwrap(),
+            decode_command(encoded).unwrap(),
             ChannelCommand::Requeue {
                 cumulative_count: None,
                 ..
@@ -638,7 +669,9 @@ mod tests {
             attempts: 1,
             cumulative_count: Some(1),
         };
-        let encoded = encode_command(&durable);
+        let mut encoded = [0u8; ENCODED_COMMAND_BYTES];
+        let len = encode_command(&durable, &mut encoded);
+        let encoded = &encoded[..len];
         assert_eq!(encoded[0], 7);
         assert_eq!(encoded.len(), 35);
     }
diff --git a/crates/queue/src/topic.rs b/crates/queue/src/topic.rs
index b9e0f13..03b3c69 100644
--- a/crates/queue/src/topic.rs
+++ b/crates/queue/src/topic.rs
@@ -35,6 +35,9 @@ pub(crate) struct PendingSync {
 
 pub(crate) struct TopicHandle {
     pub commit_gate: Mutex<()>,
+    // Durable Channel mutations must hold this while the Topic state lock is
+    // released for WAL fsync, so checkpoints cannot include an uncommitted ACK.
+    pub channel_commit_gate: Mutex<()>,
     pub state: Mutex<Topic>,
     pub wake: tokio::sync::watch::Sender<u64>,
 }
@@ -85,6 +88,7 @@ impl TopicHandle {
         topic.reconcile_unrouted_boundary()?;
         Ok(Arc::new(Self {
             commit_gate: Mutex::new(()),
+            channel_commit_gate: Mutex::new(()),
             state: Mutex::new(topic),
             wake,
         }))
@@ -118,6 +122,7 @@ impl TopicHandle {
         let (wake, _) = tokio::sync::watch::channel(0);
         Ok(Arc::new(Self {
             commit_gate: Mutex::new(()),
+            channel_commit_gate: Mutex::new(()),
             state: Mutex::new(Topic {
                 name: name.into(),
                 directory: directory.into(),
@@ -653,30 +658,46 @@ impl Topic {
         Ok(())
     }
 
-    pub fn sync_channel_wals<'a>(
+    pub fn prepare_channel_wal_syncs<'a>(
         &mut self,
-        channels: impl Iterator<Item = &'a String>,
-    ) -> Result<(), BrokerError> {
+        channels: impl Iterator<Item = &'a Arc<str>>,
+    ) -> Result<Vec<File>, BrokerError> {
+        let mut syncs = Vec::new();
         for name in channels {
             let runtime = self
                 .channels
-                .get_mut(name)
+                .get_mut(name.as_ref())
                 .ok_or(BrokerError::ChannelNotFound)?;
             if let Some(store) = runtime.store.as_mut() {
-                store.sync()?;
+                syncs.push(store.prepare_sync()?);
+            }
+        }
+        Ok(syncs)
+    }
+
+    pub fn mark_channel_wal_sync_failed<'a>(
+        &mut self,
+        channels: impl Iterator<Item = &'a Arc<str>>,
+    ) {
+        for name in channels {
+            if let Some(store) = self
+                .channels
+                .get_mut(name.as_ref())
+                .and_then(|runtime| runtime.store.as_mut())
+            {
+                store.mark_sync_failed();
             }
         }
-        Ok(())
     }
 
     pub fn checkpoint_channels_if_needed<'a>(
         &mut self,
-        channels: impl Iterator<Item = &'a String>,
+        channels: impl Iterator<Item = &'a Arc<str>>,
     ) -> Result<(), BrokerError> {
         for name in channels {
             let runtime = self
                 .channels
-                .get_mut(name)
+                .get_mut(name.as_ref())
                 .ok_or(BrokerError::ChannelNotFound)?;
             if let Some(store) = runtime.store.as_mut() {
                 store.checkpoint_if_needed(&runtime.state)?;
diff --git a/crates/server/src/tcp.rs b/crates/server/src/tcp.rs
index f181808..517d671 100644
--- a/crates/server/src/tcp.rs
+++ b/crates/server/src/tcp.rs
@@ -91,8 +91,8 @@ enum Compression {
 }
 
 struct Subscription {
-    topic: String,
-    channel: String,
+    topic: Arc<str>,
+    channel: Arc<str>,
     lease: SubscriptionLease,
 }
 
@@ -101,8 +101,8 @@ const DEFAULT_FETCH_WAIT_MS: u32 = 100;
 
 #[derive(Clone, Debug)]
 struct FetchRequest {
-    topic: String,
-    channel: String,
+    topic: Arc<str>,
+    channel: Arc<str>,
     timeout_ms: u64,
     max_messages: u16,
     max_bytes: u32,
diff --git a/crates/server/src/tcp/channel_ops.rs b/crates/server/src/tcp/channel_ops.rs
index 516b99b..8ed2a83 100644
--- a/crates/server/src/tcp/channel_ops.rs
+++ b/crates/server/src/tcp/channel_ops.rs
@@ -3,6 +3,10 @@ use futures::stream::{FuturesUnordered, StreamExt};
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 
+// Channel operations stay queued per session, but only this many hold active
+// broker futures while waiting for a durable group commit.
+const MAX_PENDING_CHANNEL_OPS: usize = 256;
+
 #[derive(Clone, Copy)]
 pub(super) enum ChannelOpKind {
     Finish,
@@ -21,15 +25,15 @@ impl ChannelOpKind {
 
 enum ChannelOp {
     Finish {
-        topic: String,
-        channel: String,
+        topic: Arc<str>,
+        channel: Arc<str>,
         id: u64,
         token: u64,
         sampled: bool,
     },
     Requeue {
-        topic: String,
-        channel: String,
+        topic: Arc<str>,
+        channel: Arc<str>,
         id: u64,
         token: u64,
         delay: Duration,
@@ -66,8 +70,8 @@ pub(super) struct ChannelOpSender {
 impl ChannelOpSender {
     pub(super) fn finish(
         &self,
-        topic: String,
-        channel: String,
+        topic: Arc<str>,
+        channel: Arc<str>,
         id: u64,
         token: u64,
     ) -> Result<(), BrokerError> {
@@ -82,8 +86,8 @@ impl ChannelOpSender {
 
     pub(super) fn finish_sampled(
         &self,
-        topic: String,
-        channel: String,
+        topic: Arc<str>,
+        channel: Arc<str>,
         id: u64,
         token: u64,
     ) -> Result<(), BrokerError> {
@@ -98,8 +102,8 @@ impl ChannelOpSender {
 
     pub(super) fn requeue(
         &self,
-        topic: String,
-        channel: String,
+        topic: Arc<str>,
+        channel: Arc<str>,
         id: u64,
         token: u64,
         delay: Duration,
@@ -146,19 +150,15 @@ async fn run_channel_ops(
     mut operations: mpsc::UnboundedReceiver<ChannelOp>,
     completions: mpsc::UnboundedSender<ChannelOpCompletion>,
 ) {
-    type Pending = Pin<Box<dyn Future<Output = ChannelOpCompletion> + Send>>;
-
-    let mut pending = FuturesUnordered::<Pending>::new();
+    let mut pending = FuturesUnordered::new();
     let mut receiving = true;
     while receiving || !pending.is_empty() {
         tokio::select! {
-            operation = operations.recv(), if receiving => {
+            operation = operations.recv(), if receiving && pending.len() < MAX_PENDING_CHANNEL_OPS => {
                 match operation {
                     Some(operation) => {
                         let broker = broker.clone();
-                        pending.push(Box::pin(async move {
-                            execute_channel_op(broker, operation).await
-                        }));
+                        pending.push(execute_channel_op(broker, operation));
                     }
                     None => receiving = false,
                 }
@@ -182,7 +182,11 @@ async fn execute_channel_op(broker: Broker, operation: ChannelOp) -> ChannelOpCo
             id,
             token,
             ..
-        } => broker.finish_delivery(&topic, &channel, id, token).await,
+        } => {
+            broker
+                .finish_delivery_shared(&topic, channel, id, token)
+                .await
+        }
         ChannelOp::Requeue {
             topic,
             channel,
@@ -191,7 +195,7 @@ async fn execute_channel_op(broker: Broker, operation: ChannelOp) -> ChannelOpCo
             delay,
         } => {
             broker
-                .requeue_delivery(&topic, &channel, id, token, delay)
+                .requeue_delivery_shared(&topic, channel, id, token, delay)
                 .await
         }
     };
diff --git a/crates/server/src/tcp/commands.rs b/crates/server/src/tcp/commands.rs
index cf7e81a..78e39ec 100644
--- a/crates/server/src/tcp/commands.rs
+++ b/crates/server/src/tcp/commands.rs
@@ -112,8 +112,8 @@ pub(super) async fn process_command(
             match create_result {
                 Ok(()) => {
                     state.subscription = Some(Subscription {
-                        topic,
-                        channel,
+                        topic: topic.into(),
+                        channel: channel.into(),
                         lease,
                     });
                     write_frame(writer, FrameType::Response, OK).await?;
diff --git a/deploy/helm/rustqueue/Chart.yaml b/deploy/helm/rustqueue/Chart.yaml
index 9d8ca3a..1328f92 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.3
-appVersion: "0.8.3"
+version: 0.8.4
+appVersion: "0.8.4"
 kubeVersion: ">=1.28.0-0"
 keywords:
   - messaging
diff --git a/deploy/helm/rustqueue/values.yaml b/deploy/helm/rustqueue/values.yaml
index eaeb4a6..c90514f 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.3"
+    tag: "0.8.4"
     pullPolicy: IfNotPresent
   log: rustqueue_operator=info
   resources:
@@ -20,7 +20,7 @@ operator:
 queue:
   create: true
   name: rustqueue
-  image: rustqueue:0.8.3
+  image: rustqueue:0.8.4
   imagePullPolicy: IfNotPresent
   minBrokers: 1
   maxBrokers: 500
diff --git a/docs/architecture/share-nothing-v7.md b/docs/architecture/share-nothing-v7.md
index 94920e4..8ede6da 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.3
+Target release: 0.8.4
 Data format: v7, clean directories only
 
 ## 1. Goal
diff --git a/docs/operations/kubernetes.md b/docs/operations/kubernetes.md
index c903795..b7ad420 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.3 \
+  --set queue.image=registry.example/rustqueue:0.8.4 \
   --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.3
+rustqueuectl -n rustqueue rollout rollback registry.example/rustqueue:0.8.4
 rustqueuectl -n rustqueue rollout forward
 ```
 
diff --git a/docs/releases/v0.8.4.md b/docs/releases/v0.8.4.md
new file mode 100644
index 0000000..116f002
--- /dev/null
+++ b/docs/releases/v0.8.4.md
@@ -0,0 +1,55 @@
+# RustQueue 0.8.4
+
+RustQueue 0.8.4 is a performance release for the durable Channel acknowledgement
+path. It reduces the work held under the Topic state lock and removes several
+per-message allocation points without changing the v7 storage or wire
+contracts.
+
+## Durable Channel commit path
+
+- `FIN` and `REQ` for one Topic are collected for up to 1 ms and committed in
+  one bounded Channel WAL group of at most 1,024 requests.
+- The Topic state lock is released while the Channel WAL files are synced. An
+  independent Channel commit gate still serializes append, checkpoint,
+  rotation, deletion and other durable Channel mutations around that boundary.
+- A successful `FIN` or `REQ` response still follows every affected Channel WAL
+  `fsync`. Delivery may reserve already durable messages while a later group is
+  syncing, but never observes an uncommitted tail.
+
+## Hot-path allocation reductions
+
+- Channel WAL commands use fixed-size stack encoding instead of allocating a
+  temporary command buffer for each frame.
+- TCP `SUB`, `FIN` and `REQ` paths carry shared `Arc<str>` Topic/Channel
+  identities through the broker, and touched Channel sets retain those shared
+  identities.
+- Channel operation futures no longer add an extra `Box` allocation. Each TCP
+  session has a bounded pool of 256 active Channel operations.
+- The benchmark consumer waits for the server EOF and verifies final Channel
+  depth, in-flight and deferred counts before accepting a drained result.
+
+## Compatibility and durability
+
+- The on-disk format remains v7; no migration is required from 0.8.3.
+- NSQ V2 commands, TLS/mTLS, AUTH, compression, lookup and the opt-in Kodo
+  profile retain their existing compatibility contracts.
+- Delivery remains at least once. Durable `FIN`/`REQ` acknowledgement semantics
+  and the single-copy share-nothing PVC model are unchanged.
+
+## Qualification protocol
+
+The qualification script defaults to comparing the exact `v0.8.3` tag with the
+candidate on OrbStack. The formal optional run uses fresh volumes, 2 vCPU and
+2 GiB for both Broker and load generator, 10 alternating pairs for each of the
+three cases, a 30-second warmup and a 120-second measurement. Short development
+A/B runs are useful for iteration but are not formal release evidence.
+
+## Release assets
+
+- `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 all downloadable assets
diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock
index 0bfb529..472abb4 100644
--- a/fuzz/Cargo.lock
+++ b/fuzz/Cargo.lock
@@ -2017,7 +2017,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-discovery"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "axum",
@@ -2055,7 +2055,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-protocol"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "bytes",
  "serde",
@@ -2065,7 +2065,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-proxy"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "axum",
@@ -2085,7 +2085,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-queue"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "bytes",
@@ -2105,7 +2105,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-storage"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "anyhow",
  "crc32c",
@@ -2117,7 +2117,7 @@ dependencies = [
 
 [[package]]
 name = "rustqueue-telemetry"
-version = "0.8.3"
+version = "0.8.4"
 dependencies = [
  "serde",
 ]
diff --git a/scripts/acceptance-multi-broker-k8s.sh b/scripts/acceptance-multi-broker-k8s.sh
index ef27de9..3ec9b60 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.3-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image
+  BUILD_VERSION=0.8.4-e2e-a MAX_STORAGE_FEATURE_LEVEL=1 make image
   docker tag rustqueue:dev "$BROKER_IMAGE_A"
-  BUILD_VERSION=0.8.3-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist
+  BUILD_VERSION=0.8.4-e2e-b MAX_STORAGE_FEATURE_LEVEL=2 make image-from-dist
   docker tag rustqueue:dev "$BROKER_IMAGE_B"
   [[ "$(docker image inspect "$BROKER_IMAGE_A" -f '{{.Id}}')" != \
      "$(docker image inspect "$BROKER_IMAGE_B" -f '{{.Id}}')" ]] || {
diff --git a/scripts/benchmark-qualify.sh b/scripts/benchmark-qualify.sh
index bd3f14b..29c168c 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.3}"
-BASELINE_REF="${BASELINE_REF:-v0.8.2}"
+RELEASE="${RELEASE:-0.8.4}"
+BASELINE_REF="${BASELINE_REF:-v0.8.3}"
 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.3-orbstack.json}"
+EVIDENCE_OUTPUT="${EVIDENCE_OUTPUT:-$ROOT/benchmarks/qualifications/v0.8.4-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"
@@ -37,6 +37,7 @@ ACTIVE_BROKER=""
 ACTIVE_VOLUME=""
 SAMPLER_PID=""
 SEQUENCE=0
+FINAL_DRAIN_ATTEMPTS=30
 
 die() {
   printf 'benchmark qualification: %s\n' "$*" >&2
@@ -103,15 +104,17 @@ 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 "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 $BASELINE_REF tag commit"
+if [[ "$QUALIFICATION_DEV" == 0 ]]; then
+  [[ "$BASELINE_REF" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]] ||
+    die "baseline reference must be a version tag"
+  tag_commit="$(git -C "$ROOT" rev-parse --verify "refs/tags/$BASELINE_REF^{commit}")"
+  [[ "$baseline_commit" == "$tag_commit" ]] ||
+    die "baseline must resolve to the exact $BASELINE_REF tag commit"
+fi
 
 case "$EVIDENCE_OUTPUT" in
   "$ROOT/benchmarks/qualifications/"*)
@@ -153,11 +156,12 @@ read_workspace_version() {
 
 baseline_version="$(read_workspace_version "$BASELINE_SOURCE")"
 candidate_version="$(read_workspace_version "$CANDIDATE_SOURCE")"
-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"
+if [[ "$QUALIFICATION_DEV" == 0 ]]; then
+  expected_baseline_version="${BASELINE_REF#v}"
+  [[ "$baseline_version" == "$expected_baseline_version" ]] ||
+    die "baseline workspace version is $baseline_version, expected $expected_baseline_version"
+  [[ "$candidate_version" == "$RELEASE" ]] ||
+    die "candidate workspace version is $candidate_version, expected $RELEASE"
 fi
 
 candidate_context="$CANDIDATE_SOURCE"
@@ -269,7 +273,7 @@ wait_for_broker() {
 
 run_variant() {
   local scenario=$1 pair=$2 position=$3 variant=$4
-  local image commit producers consumers batch rate
+  local image commit producers consumers batch rate attempt
   case "$variant" in
     baseline)
       image="$BASELINE_IMAGE"
@@ -380,25 +384,35 @@ run_variant() {
   if [[ "$consumers" -gt 0 ]]; then
     topic="$(jq -r '.topic' "$report")"
     channel="$(jq -r '.channel' "$report")"
-    stats="$(
-      docker exec "$ACTIVE_BROKER" curl -fsS \
-        "http://127.0.0.1:4151/stats?format=json&include_clients=false&topic=$topic&channel=$channel"
-    )"
-    channel_stats="$(
-      jq -c --arg topic "$topic" --arg channel "$channel" '
-        [.topics[]
-          | select(.topic_name == $topic)
-          | .channels[]
-          | select(.channel_name == $channel)][0]
-      ' <<<"$stats"
-    )"
-    [[ "$channel_stats" != null ]] || {
-      cleanup_run
-      die "$label final Channel stats are missing"
-    }
-    final_depth="$(jq -r '.depth' <<<"$channel_stats")"
-    final_in_flight="$(jq -r '.in_flight_count' <<<"$channel_stats")"
-    final_deferred="$(jq -r '.deferred_count' <<<"$channel_stats")"
+    for ((attempt = 1; attempt <= FINAL_DRAIN_ATTEMPTS; attempt++)); do
+      stats="$(
+        docker exec "$ACTIVE_BROKER" curl -fsS \
+          "http://127.0.0.1:4151/stats?format=json&include_clients=false&topic=$topic&channel=$channel"
+      )"
+      channel_stats="$(
+        jq -c --arg topic "$topic" --arg channel "$channel" '
+          [.topics[]
+            | select(.topic_name == $topic)
+            | .channels[]
+            | select(.channel_name == $channel)][0]
+        ' <<<"$stats"
+      )"
+      [[ "$channel_stats" != null ]] || {
+        cleanup_run
+        die "$label final Channel stats are missing"
+      }
+      final_depth="$(jq -r '.depth' <<<"$channel_stats")"
+      final_in_flight="$(jq -r '.in_flight_count' <<<"$channel_stats")"
+      final_deferred="$(jq -r '.deferred_count' <<<"$channel_stats")"
+      if [[ "$final_depth" -eq 0 && "$final_in_flight" -eq 0 && "$final_deferred" -eq 0 ]]; then
+        break
+      fi
+      if [[ "$attempt" -eq "$FINAL_DRAIN_ATTEMPTS" ]]; then
+        cleanup_run
+        die "$label left Channel backlog after drain"
+      fi
+      sleep 1
+    done
   fi
 
   prometheus="$(
diff --git a/site/index.html b/site/index.html
index e6940a5..0c37ef0 100644
--- a/site/index.html
+++ b/site/index.html
@@ -407,7 +407,7 @@ <h2>One Helm command.</h2>
             </div>
             <pre><code id="helm-command">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</code></pre>
           </div>
         </div>