feat(message_bus): async fire-and-forget transport for VSR consensus

The previous cache/connection.rs was a single-stream blob that
held a RefCell<TcpStream> across .await, awaited kernel write
completion in the send path, and serialized fan-out through one
shared lock. Under VSR pipelining a slow peer stalled sends to
every other peer in the same dispatch round, killing parallel
quorum collection. Reentrant sends to the same peer would also
panic on BorrowMutError.

Rebuilt the crate around a per-connection writer task that
drains a bounded mpsc and submits batches via a single
write_vectored_all syscall, with an independent read half
handled by a dedicated reader task. send_to_* is now sync
try_send under the async fn shell: zero awaits, returns
SendError::Backpressure on full so VSR can recover via WAL
retransmission or view-change timeouts.

The lifecycle module owns a root Shutdown / ShutdownToken plus
a ConnectionRegistry<K> that tracks the per-peer Sender and
both task handles for graceful drain. The directional rule
(lower id dials, higher id accepts) eliminates the dialed-
both-ways race without a tiebreaker. Message::into_frozen()
removes the per-send memcpy.

Socket split and lifecycle plumbing (C1, C2, Cx2)

Swapped the dup(2)-based socket_split module for compio's
native TcpStream::into_split(), deleting ~50 lines of unsafe,
three dup_stream copies, and an entire module file. Reader and
writer tasks now take OwnedReadHalf / OwnedWriteHalf directly.
The writer task, on write_vectored_all error, removes its own
registry entry so a stale Sender cannot accept further sends;
on graceful shutdown path it defers removal to the root drain
so DrainOutcome.clean counts both halves. The reader self-remove
path went through a new ConnectionRegistry::close_peer helper
that orders "close sender, await writer, drop reader handle"
to prevent a mid-writev cancellation landing a truncated frame.
All listeners + the outbound connector apply TCP SO_KEEPALIVE
via socket2 (TCP_KEEPIDLE=10s, TCP_KEEPINTVL=5s, TCP_KEEPCNT=3)
so a silently dead peer is detected within ~25s, well inside
the VSR view-change window.

Shutdown loop-drain (C3, N5)

track_background is back to an unconditional push. shutdown
now loop-drains the background-task vec until empty, so a
task pushed mid-shutdown (e.g. a reader that observed the
token and registered its own cleanup) is still awaited.
DrainOutcome gains background_clean / background_force fields
to make the bg count observable.

Cluster config wedge fixes (C5, C6)

replica_id is now Option<u8> on both CurrentNodeConfig and
OtherNodeConfig. Startup-time ClusterConfig::validate rejects
missing ids, duplicate ids, and ids >= total replica count
so a misconfigured cluster cannot wedge into a permanent view
change on boot. CurrentNodeConfig gained the TransportPorts
field that was already present on OtherNodeConfig; the bus
now has a place to read its own tcp_replica bind port from.

Per-connection body-byte limiter (C7)

A new FrameRateLimiter (token bucket, 32 MiB/s sustained, 256
MiB burst by default) gates the body-read allocation in
framing::read_message. A peer claiming a valid 64 MiB body
goes through once then has to wait seconds before the next
burst; the hard ceiling on MAX_MESSAGE_SIZE stays as-is.

Zero-copy framing (W1, W3)

framing::read_message was allocating three Vecs and copying
the header + body twice to reassemble the final Owned. It is
now a single Owned<MESSAGE_ALIGN>::with_capacity(HEADER_SIZE)
handed straight to compio's read_exact, grown in-place via
reserve_exact after the size field is parsed, then a second
read_exact into owned.slice(HEADER_SIZE..total_size). The
backing AVec's buffer is reused across both reads: one alloc
(with at most one in-place realloc) and zero memcpys of the
data. iggy_binary_protocol's Owned<ALIGN> now implements
IoBuf / IoBufMut / SetLen so compio can drive the read
directly. A compile-time static assert guards the hardcoded
offset_of!(GenericHeader, size) == 48 the reader relies on.

Router blocking-send fix (C8)

core/shard/src/router.rs was calling the blocking
crossfire::MTx::send from the compio reactor, which could park
the io_uring thread and stall every other connection on the
shard when an inbox filled up. Both dispatch and dispatch_request
now use try_send and log-drop on Full / Disconnected; consensus
recovers via WAL retransmit or view change.

Misc (Cx1, W4, W6, W7, W8, W9, W10, N3, N4)

- connector.rs's dead is_err branch on insert() is replaced
  with an expect() documenting the directional-rule invariant.
- SendError::Io is removed (no constructor anywhere).
- replica_listener doc adds an explicit trusted-network-only
  warning for the handshake (no mTLS / shared secret).
- writer_task MAX_BATCH doc drops the "tunable per-deployment"
  claim.
- Duplicate client id insert is now unreachable!() with a
  helpful diagnostic.
- chain-replication failure paths use structured
  tracing::warn!(error = ?e, ...) so oncall can grep the
  SendError variant.
- MessageBus trait doc advertises the no-yield property of
  send_to_* in the production impl.
- writer_task's license-header "czpressed" typo fixed.
- transports/mod.rs 40-line sketch compressed to a tracking
  reference for IGGY-112.

Deferred with TODO(hubcio)

- Chain-replicate-before-journal-append ordering (C4) is
  flagged at both metadata and partitions sites; an ordering
  decision issue is needed, not a patch in this PR.
- Fan-out deep_copy (W2) in plane_helpers and the three shard
  send sites is flagged; requires a trait-level change to
  MessageBus::send_to_* taking Frozen<MESSAGE_ALIGN> so the
  primary freezes once and fan-out is refcount bumps.

Scope

Consensus, metadata, partitions, shard, and simulator are
updated to the new MessageBus shape; core/server is left
untouched as legacy code. MESSAGE_ALIGN promoted to pub in
iggy_binary_protocol::consensus is consistent with the
existing Message<GenericHeader> leak through the MessageBus
trait and is accepted for this PR (N8).

Follow-up issues to file:
- W2 MessageBus::send_to_*(.., Frozen<MESSAGE_ALIGN>) trait-
  level fan-out fix (benchmark-driven)
- W5 SendError::Backpressure(Message) carrying payload back
  to the caller (benchmark-driven)
- W11 SimOutbox bounded mode (simulator coverage)
- N6 ConnectionRegistry<u8> -> Vec<Option<Entry>> micro-opt
- N7 drain parallel via FuturesUnordered
- C4 VSR chain-replicate-vs-append ordering decision
- IGGY-112 Transport trait family + mTLS handshake

Author: hubcio

Cargo.lock

@@ -147,7 +147,7 @@ capacity_builder: 0.5.0, "MIT",
 capacity_builder_macros: 0.3.0, "MIT",
 cargo-platform: 0.3.2, "Apache-2.0 OR MIT",
 cargo_metadata: 0.23.1, "MIT",
-cc: 1.2.60, "Apache-2.0 OR MIT",
+cc: 1.2.59, "Apache-2.0 OR MIT",
 cesu8: 1.1.0, "Apache-2.0 OR MIT",
 cexpr: 0.6.0, "Apache-2.0 OR MIT",
 cfg-if: 1.0.4, "Apache-2.0 OR MIT",
@@ -451,7 +451,7 @@ hwlocality-sys: 0.7.0, "MIT",
 hybrid-array: 0.4.10, "Apache-2.0 OR MIT",
 hyper: 1.9.0, "MIT",
 hyper-named-pipe: 0.1.0, "Apache-2.0",
-hyper-rustls: 0.27.8, "Apache-2.0 OR ISC OR MIT",
+hyper-rustls: 0.27.7, "Apache-2.0 OR ISC OR MIT",
 hyper-timeout: 0.5.2, "Apache-2.0 OR MIT",
 hyper-util: 0.1.20, "MIT",
 hyperlocal: 0.9.1, "MIT",
@@ -532,7 +532,7 @@ jni-sys: 0.4.1, "Apache-2.0 OR MIT",
 jni-sys-macros: 0.4.1, "Apache-2.0 OR MIT",
 jobserver: 0.1.34, "Apache-2.0 OR MIT",
 journal: 0.1.0, "Apache-2.0",
-js-sys: 0.3.95, "Apache-2.0 OR MIT",
+js-sys: 0.3.94, "Apache-2.0 OR MIT",
 jsonwebtoken: 10.3.0, "MIT",
 jwalk: 0.8.1, "MIT",
 keccak: 0.2.0, "Apache-2.0 OR MIT",
@@ -654,11 +654,11 @@ once_cell: 1.21.4, "Apache-2.0 OR MIT",
 once_cell_polyfill: 1.70.2, "Apache-2.0 OR MIT",
 opaque-debug: 0.3.1, "Apache-2.0 OR MIT",
 opendal: 0.55.0, "Apache-2.0",
-openssl: 0.10.77, "Apache-2.0",
+openssl: 0.10.76, "Apache-2.0",
 openssl-macros: 0.1.1, "Apache-2.0 OR MIT",
 openssl-probe: 0.2.1, "Apache-2.0 OR MIT",
 openssl-src: 300.6.0+3.6.2, "Apache-2.0 OR MIT",
-openssl-sys: 0.9.113, "MIT",
+openssl-sys: 0.9.112, "MIT",
 opentelemetry: 0.31.0, "Apache-2.0",
 opentelemetry-appender-tracing: 0.31.1, "Apache-2.0",
 opentelemetry-http: 0.31.0, "Apache-2.0",
@@ -708,7 +708,7 @@ pin-utils: 0.1.0, "Apache-2.0 OR MIT",
 pinned: 0.1.0, "Apache-2.0 OR MIT",
 pkcs1: 0.7.5, "Apache-2.0 OR MIT",
 pkcs8: 0.10.2, "Apache-2.0 OR MIT",
-pkg-config: 0.3.33, "Apache-2.0 OR MIT",
+pkg-config: 0.3.32, "Apache-2.0 OR MIT",
 plain: 0.2.3, "Apache-2.0 OR MIT",
 png: 0.17.16, "Apache-2.0 OR MIT",
 png: 0.18.1, "Apache-2.0 OR MIT",
@@ -761,13 +761,13 @@ r-efi: 5.3.0, "Apache-2.0 OR LGPL-2.1-or-later OR MIT",
 r-efi: 6.0.0, "Apache-2.0 OR LGPL-2.1-or-later OR MIT",
 radium: 0.7.0, "MIT",
 rand: 0.8.5, "Apache-2.0 OR MIT",
-rand: 0.9.4, "Apache-2.0 OR MIT",
+rand: 0.9.2, "Apache-2.0 OR MIT",
 rand: 0.10.1, "Apache-2.0 OR MIT",
 rand_chacha: 0.3.1, "Apache-2.0 OR MIT",
 rand_chacha: 0.9.0, "Apache-2.0 OR MIT",
 rand_core: 0.6.4, "Apache-2.0 OR MIT",
 rand_core: 0.9.5, "Apache-2.0 OR MIT",
-rand_core: 0.10.1, "Apache-2.0 OR MIT",
+rand_core: 0.10.0, "Apache-2.0 OR MIT",
 rand_xoshiro: 0.8.0, "Apache-2.0 OR MIT",
 rav1e: 0.8.1, "BSD-2-Clause",
 ravif: 0.13.0, "BSD-3-Clause",
@@ -826,7 +826,7 @@ rustls-pemfile: 2.2.0, "Apache-2.0 OR ISC OR MIT",
 rustls-pki-types: 1.14.0, "Apache-2.0 OR MIT",
 rustls-platform-verifier: 0.6.2, "Apache-2.0 OR MIT",
 rustls-platform-verifier-android: 0.1.1, "Apache-2.0 OR MIT",
-rustls-webpki: 0.103.11, "ISC",
+rustls-webpki: 0.103.10, "ISC",
 rustversion: 1.0.22, "Apache-2.0 OR MIT",
 rustybuzz: 0.20.1, "MIT",
 ryu: 1.0.23, "Apache-2.0 OR BSL-1.0",
@@ -907,7 +907,7 @@ sqlx-macros-core: 0.8.6, "Apache-2.0 OR MIT",
 sqlx-mysql: 0.8.6, "Apache-2.0 OR MIT",
 sqlx-postgres: 0.8.6, "Apache-2.0 OR MIT",
 sqlx-sqlite: 0.8.6, "Apache-2.0 OR MIT",
-sse-stream: 0.2.2, "Apache-2.0 OR MIT",
+sse-stream: 0.2.1, "Apache-2.0 OR MIT",
 stable_deref_trait: 1.2.1, "Apache-2.0 OR MIT",
 static-toml: 1.3.0, "MIT",
 static_assertions: 1.1.0, "Apache-2.0 OR MIT",
@@ -1063,19 +1063,19 @@ wasi: 0.11.1+wasi-snapshot-preview1, "Apache-2.0 OR Apache-2.0 WITH LLVM-excepti
 wasip2: 1.0.2+wasi-0.2.9, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
 wasip3: 0.4.0+wasi-0.3.0-rc-2026-01-06, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
 wasite: 0.1.0, "Apache-2.0 OR BSL-1.0 OR MIT",
-wasm-bindgen: 0.2.118, "Apache-2.0 OR MIT",
-wasm-bindgen-futures: 0.4.68, "Apache-2.0 OR MIT",
-wasm-bindgen-macro: 0.2.118, "Apache-2.0 OR MIT",
-wasm-bindgen-macro-support: 0.2.118, "Apache-2.0 OR MIT",
-wasm-bindgen-shared: 0.2.118, "Apache-2.0 OR MIT",
+wasm-bindgen: 0.2.117, "Apache-2.0 OR MIT",
+wasm-bindgen-futures: 0.4.67, "Apache-2.0 OR MIT",
+wasm-bindgen-macro: 0.2.117, "Apache-2.0 OR MIT",
+wasm-bindgen-macro-support: 0.2.117, "Apache-2.0 OR MIT",
+wasm-bindgen-shared: 0.2.117, "Apache-2.0 OR MIT",
 wasm-encoder: 0.244.0, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
 wasm-metadata: 0.244.0, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
 wasm-streams: 0.4.2, "Apache-2.0 OR MIT",
 wasm-streams: 0.5.0, "Apache-2.0 OR MIT",
 wasm_dep_analyzer: 0.3.0, "MIT",
 wasmparser: 0.244.0, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
 wasmtimer: 0.4.3, "MIT",
-web-sys: 0.3.95, "Apache-2.0 OR MIT",
+web-sys: 0.3.94, "Apache-2.0 OR MIT",
 web-time: 1.1.0, "Apache-2.0 OR MIT",
 webpki-root-certs: 1.0.6, "CDLA-Permissive-2.0",
 webpki-roots: 0.26.11, "CDLA-Permissive-2.0",

DEPENDENCIES.md

@@ -15,13 +15,14 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::mem::MaybeUninit;
 use std::ops::{Deref, DerefMut, RangeBounds};
 use std::ptr::NonNull;
 use std::slice;
 use std::sync::atomic::{AtomicUsize, Ordering, fence};
 
 use aligned_vec::{AVec, ConstAlign};
-use compio_buf::IoBuf;
+use compio_buf::{IoBuf, IoBufMut, ReserveError, ReserveExactError, SetLen};
 
 #[derive(Debug, Clone)]
 pub struct Owned<const ALIGN: usize = 4096> {
@@ -91,6 +92,16 @@ impl<const ALIGN: usize> Owned<ALIGN> {
         Self { inner }
     }
 
+    /// Allocate a buffer with `capacity` reserved bytes and `len == 0`.
+    /// Intended for use with compio's [`IoBufMut`] read path: the caller
+    /// hands the `Owned` to `read_exact`, which fills it in-place and
+    /// advances the length via [`SetLen::set_len`].
+    pub fn with_capacity(capacity: usize) -> Self {
+        Self {
+            inner: AVec::with_capacity(ALIGN, capacity),
+        }
+    }
+
     pub fn copy_from_slice(data: &[u8]) -> Self {
         let mut inner: AVec<u8, ConstAlign<ALIGN>> = AVec::new(ALIGN);
         inner.extend_from_slice(data);
@@ -133,6 +144,54 @@ impl<const ALIGN: usize> Owned<ALIGN> {
     }
 }
 
+// SAFETY contract for the compio IO buf impls below:
+// * `as_init` returns the initialized prefix (len bytes).
+// * `as_uninit` returns the full allocation up to capacity; compio writes
+//   bytes into it during reads and THEN calls `SetLen::set_len` to advance
+//   the initialized region, which is exactly what `AVec::set_len` expects.
+// * `buf_capacity` is the AVec capacity, not the initialized length.
+impl<const ALIGN: usize> IoBuf for Owned<ALIGN> {
+    fn as_init(&self) -> &[u8] {
+        self.inner.as_slice()
+    }
+}
+
+impl<const ALIGN: usize> IoBufMut for Owned<ALIGN> {
+    fn as_uninit(&mut self) -> &mut [MaybeUninit<u8>] {
+        let cap = self.inner.capacity();
+        let ptr = self.inner.as_mut_ptr().cast::<MaybeUninit<u8>>();
+        unsafe { slice::from_raw_parts_mut(ptr, cap) }
+    }
+
+    fn buf_capacity(&mut self) -> usize {
+        self.inner.capacity()
+    }
+
+    fn reserve(&mut self, len: usize) -> Result<(), ReserveError> {
+        // `reserve(additional)` contract: capacity >= current_len + additional
+        // after the call. `AVec::reserve` matches this contract exactly.
+        if self.inner.capacity() - self.inner.len() >= len {
+            return Ok(());
+        }
+        self.inner.reserve(len);
+        Ok(())
+    }
+
+    fn reserve_exact(&mut self, len: usize) -> Result<(), ReserveExactError> {
+        if self.inner.capacity() - self.inner.len() >= len {
+            return Ok(());
+        }
+        self.inner.reserve_exact(len);
+        Ok(())
+    }
+}
+
+impl<const ALIGN: usize> SetLen for Owned<ALIGN> {
+    unsafe fn set_len(&mut self, len: usize) {
+        unsafe { self.inner.set_len(len) };
+    }
+}
+
 impl<const ALIGN: usize> Prefix<ALIGN> {
     pub fn copy_from_slice(src: &[u8]) -> Self {
         Self {

core/binary_protocol/src/consensus/iobuf.rs

@@ -24,7 +24,7 @@ use crate::consensus::{
 use smallvec::SmallVec;
 use std::{marker::PhantomData, mem::size_of};
 
-const MESSAGE_ALIGN: usize = 4096;
+pub const MESSAGE_ALIGN: usize = 4096;
 
 pub trait MessageBacking<H>
 where

core/binary_protocol/src/consensus/message.rs

@@ -66,7 +66,7 @@ pub use header::{
     PrepareOkHeader, ReplyHeader, RequestHeader, StartViewChangeHeader, StartViewHeader,
 };
 pub use message::{
-    ConsensusMessage, FragmentedBacking, Message, MessageBacking, MessageBag, MutableBacking,
-    RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind,
+    ConsensusMessage, FragmentedBacking, MESSAGE_ALIGN, Message, MessageBacking, MessageBag,
+    MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind,
 };
 pub use operation::Operation;

core/binary_protocol/src/consensus/mod.rs

@@ -36,12 +36,28 @@ pub struct NodeConfig {
 pub struct CurrentNodeConfig {
     pub name: String,
     pub ip: String,
+    /// Explicit numeric replica ID for VSR consensus (0-based).
+    ///
+    /// Required when `cluster.enabled` is true. `None` is a configuration
+    /// error caught by [`ClusterConfig::validate`]; a silent default of 0
+    /// would wedge the cluster into a permanent view change as soon as a
+    /// second replica joined.
+    pub replica_id: Option<u8>,
+    /// Bind ports for the current node. Symmetric with [`OtherNodeConfig`]
+    /// so the consensus listener can pick a dedicated `tcp_replica` port
+    /// without borrowing it from the client-facing `tcp.address`.
+    #[serde(default)]
+    pub ports: TransportPorts,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
 pub struct OtherNodeConfig {
     pub name: String,
     pub ip: String,
+    /// Explicit numeric replica ID for VSR consensus (0-based).
+    ///
+    /// Required when `cluster.enabled` is true. See [`CurrentNodeConfig::replica_id`].
+    pub replica_id: Option<u8>,
     pub ports: TransportPorts,
 }
 
@@ -51,4 +67,6 @@ pub struct TransportPorts {
     pub quic: Option<u16>,
     pub http: Option<u16>,
     pub websocket: Option<u16>,
+    /// Dedicated port for replica-to-replica consensus traffic.
+    pub tcp_replica: Option<u16>,
 }

core/configs/src/server_config/cluster.rs

@@ -583,6 +583,8 @@ impl Default for NodeConfig {
             current: CurrentNodeConfig {
                 name: SERVER_CONFIG.cluster.node.current.name.parse().unwrap(),
                 ip: SERVER_CONFIG.cluster.node.current.ip.parse().unwrap(),
+                replica_id: Some(SERVER_CONFIG.cluster.node.current.replica_id as u8),
+                ports: TransportPorts::default(),
             },
             others: SERVER_CONFIG
                 .cluster
@@ -592,11 +594,13 @@ impl Default for NodeConfig {
                 .map(|other| OtherNodeConfig {
                     name: other.name.parse().unwrap(),
                     ip: other.ip.parse().unwrap(),
+                    replica_id: Some(other.replica_id as u8),
                     ports: TransportPorts {
                         tcp: Some(other.ports.tcp as u16),
                         quic: Some(other.ports.quic as u16),
                         http: Some(other.ports.http as u16),
                         websocket: Some(other.ports.websocket as u16),
+                        tcp_replica: Some(other.ports.tcp_replica as u16),
                     },
                 })
                 .collect(),