fix(cluster): single global sender counter and self-addressed replay guard

Replace per-target outbound seq counters in `PeerSeqSender` with a single
monotonic counter. The receiver's replay window is keyed by the sender's
`local_node_id`, so per-target counters from the same sender collide in
that window: seq=1 sent to A and seq=1 sent to B are indistinguishable
to any node that receives from this sender. A single counter makes every
outbound frame globally unique from the receiver's perspective.

Also skip the inbound replay-window check when `from_node_id` matches
the local node. In single-node tests and genuine self-dispatch, the
client and server share one `AuthContext`, so one `peer_seq_in` window
is touched by both the server-side request accept and the client-side
response accept. Without this guard, the second accept trips on the
first — the frame was never replayed; the window simply saw both
directions on the same entry.

Update `transport_security.rs` to use `tokio::sync::Mutex` for the
insecure-counter serialization lock so the guard can be held across
`.await` points without tripping `clippy::await_holding_lock`.

Author: farhan-syah

nodedb-cluster/src/rpc_codec/peer_seq.rs

@@ -32,42 +32,33 @@ use crate::error::{ClusterError, Result};
 /// Size of the inbound replay-detection window.
 pub const REPLAY_WINDOW: u64 = 64;
 
-/// Per-peer outbound monotonic counter. One counter per (local_node,
-/// remote_peer) pair. Thread-safe via atomics.
+/// Outbound monotonic counter for this `AuthContext`. One counter total
+/// — not one per target — because the receiver's replay window is keyed
+/// by the *sender's* `local_node_id`. If this sender used a per-target
+/// counter, two distinct targets' traffic would share the same window on
+/// any node that receives from both: seq=1 from target=A and seq=1 from
+/// target=B collide in the receiver's `window[sender_id]`. A single
+/// counter makes every outbound seq globally unique per sender.
 #[derive(Default, Debug)]
 pub struct PeerSeqSender {
-    counters: RwLock<HashMap<u64, AtomicU64>>,
+    counter: AtomicU64,
 }
 
 impl PeerSeqSender {
     pub fn new() -> Self {
         Self::default()
     }
 
-    /// Reserve and return the next sequence number for frames sent to
-    /// `peer_id`. Sequence starts at 1 and is strictly increasing.
-    pub fn next(&self, peer_id: u64) -> u64 {
-        // Fast path: counter already exists.
-        {
-            let guard = self.counters.read().unwrap_or_else(|p| p.into_inner());
-            if let Some(counter) = guard.get(&peer_id) {
-                return counter.fetch_add(1, Ordering::Relaxed) + 1;
-            }
-        }
-        // Slow path: create counter under the write lock.
-        let mut guard = self.counters.write().unwrap_or_else(|p| p.into_inner());
-        let counter = guard.entry(peer_id).or_insert_with(|| AtomicU64::new(0));
-        counter.fetch_add(1, Ordering::Relaxed) + 1
+    /// Reserve and return the next outbound sequence number. Starts at 1
+    /// and is strictly increasing across all targets for this sender.
+    pub fn next(&self) -> u64 {
+        self.counter.fetch_add(1, Ordering::Relaxed) + 1
     }
 
     /// Current counter value (0 if no frames have been sent). Test-only.
     #[cfg(test)]
-    pub fn peek(&self, peer_id: u64) -> u64 {
-        let guard = self.counters.read().unwrap_or_else(|p| p.into_inner());
-        guard
-            .get(&peer_id)
-            .map(|c| c.load(Ordering::Relaxed))
-            .unwrap_or(0)
+    pub fn peek(&self) -> u64 {
+        self.counter.load(Ordering::Relaxed)
     }
 }
 
@@ -157,18 +148,24 @@ mod tests {
     #[test]
     fn outbound_counter_starts_at_one() {
         let s = PeerSeqSender::new();
-        assert_eq!(s.next(1), 1);
-        assert_eq!(s.next(1), 2);
-        assert_eq!(s.next(1), 3);
+        assert_eq!(s.next(), 1);
+        assert_eq!(s.next(), 2);
+        assert_eq!(s.next(), 3);
     }
 
     #[test]
-    fn outbound_counters_are_independent_per_peer() {
+    fn outbound_counter_is_single_across_all_targets() {
+        // The outbound counter is intentionally shared across targets: the
+        // receiver's replay window is keyed by the sender's local_node_id,
+        // so per-target counters would collide in the same window. A
+        // single monotonic counter guarantees every emitted seq is unique
+        // from the receiver's point of view regardless of which target
+        // the sender was aiming at.
         let s = PeerSeqSender::new();
-        assert_eq!(s.next(1), 1);
-        assert_eq!(s.next(2), 1);
-        assert_eq!(s.next(1), 2);
-        assert_eq!(s.next(2), 2);
+        assert_eq!(s.next(), 1);
+        assert_eq!(s.next(), 2);
+        assert_eq!(s.next(), 3);
+        assert_eq!(s.next(), 4);
     }
 
     #[test]

nodedb-cluster/src/swim/wire/authenticated.rs

@@ -94,10 +94,13 @@ impl SwimAuth {
 
 /// Encode `msg` and wrap it in an authenticated envelope destined for
 /// `to`. Returns the bytes to hand to `UdpSocket::send_to`.
-pub fn wrap(auth: &SwimAuth, to: SocketAddr, msg: &SwimMessage) -> Result<Vec<u8>, SwimError> {
+///
+/// `to` is retained in the signature as documentation (callers pass the
+/// destination they are about to `send_to`), but the outbound seq is a
+/// single sender-global counter — see `PeerSeqSender`.
+pub fn wrap(auth: &SwimAuth, _to: SocketAddr, msg: &SwimMessage) -> Result<Vec<u8>, SwimError> {
     let inner = codec::encode(msg)?;
-    let to_hash = addr_hash(to);
-    let seq = auth.seq_out.next(to_hash);
+    let seq = auth.seq_out.next();
     let mut out = Vec::with_capacity(auth_envelope::ENVELOPE_OVERHEAD + inner.len());
     auth_envelope::write_envelope(auth.local_addr_hash, seq, &inner, &auth.mac_key, &mut out)
         .map_err(|e| SwimError::Encode {

nodedb-cluster/src/transport/client/send.rs

@@ -19,15 +19,6 @@ use crate::transport::server;
 
 use super::transport::NexarTransport;
 
-/// Sentinel peer id for the outbound seq counter when the real peer id is
-/// not yet known (bootstrap/join — see [`send_rpc_to_addr`]).
-///
-/// Using a fixed key means two concurrent bootstrap attempts on the same
-/// transport share a counter, which is fine — each attempt produces a
-/// strictly-higher seq than the last, and neither attempt cares about
-/// pairing its outbound seq with any specific remote id.
-const BOOTSTRAP_PEER_ID: u64 = 0;
-
 impl NexarTransport {
     /// Send an RPC to an address directly (for bootstrap/join before peer
     /// IDs are known).
@@ -45,7 +36,7 @@ impl NexarTransport {
     }
 
     async fn send_rpc_to_addr_inner(&self, addr: SocketAddr, rpc: RaftRpc) -> Result<RaftRpc> {
-        let envelope = self.wrap_outbound(BOOTSTRAP_PEER_ID, &rpc)?;
+        let envelope = self.wrap_outbound(&rpc)?;
 
         let conn = self
             .listener
@@ -94,7 +85,7 @@ impl NexarTransport {
                 tokio::time::sleep(delay).await;
             }
 
-            let envelope = self.wrap_inner(target, &inner)?;
+            let envelope = self.wrap_inner(&inner)?;
             match self.try_send_once(target, &envelope).await {
                 Ok(resp) => {
                     self.circuit_breaker.record_success(target);
@@ -156,15 +147,15 @@ impl NexarTransport {
         Ok(self.parse_inbound(&response_envelope))
     }
 
-    /// Encode and wrap an RPC for a known peer id.
-    fn wrap_outbound(&self, target: u64, rpc: &RaftRpc) -> Result<Vec<u8>> {
+    /// Encode and wrap an RPC in an authenticated envelope.
+    fn wrap_outbound(&self, rpc: &RaftRpc) -> Result<Vec<u8>> {
         let inner = rpc_codec::encode(rpc)?;
-        self.wrap_inner(target, &inner)
+        self.wrap_inner(&inner)
     }
 
     /// Wrap an already-encoded inner frame in an authenticated envelope.
-    fn wrap_inner(&self, target: u64, inner: &[u8]) -> Result<Vec<u8>> {
-        let seq = self.auth.peer_seq_out.next(target);
+    fn wrap_inner(&self, inner: &[u8]) -> Result<Vec<u8>> {
+        let seq = self.auth.peer_seq_out.next();
         let mut out = Vec::with_capacity(auth_envelope::ENVELOPE_OVERHEAD + inner.len());
         auth_envelope::write_envelope(
             self.auth.local_node_id,
@@ -178,11 +169,22 @@ impl NexarTransport {
 
     /// Parse an inbound envelope: verify MAC, check replay window, decode
     /// inner RPC.
+    ///
+    /// Self-addressed frames skip the replay-window check. In a single-node
+    /// test (or when a node genuinely dispatches an RPC to itself over the
+    /// transport) the client and server share one `AuthContext`, which
+    /// means one `peer_seq_in` window is updated by *both* the server-side
+    /// request-accept and the client-side response-accept. Without this
+    /// guard the second accept trips on its own first — the envelope
+    /// was never replayed, the same window simply saw traffic from both
+    /// directions for `peer_id == local_node_id`.
     fn parse_inbound(&self, envelope: &[u8]) -> Result<RaftRpc> {
         let (fields, inner_frame) = auth_envelope::parse_envelope(envelope, &self.auth.mac_key)?;
-        self.auth
-            .peer_seq_in
-            .accept(fields.from_node_id, fields.seq)?;
+        if fields.from_node_id != self.auth.local_node_id {
+            self.auth
+                .peer_seq_in
+                .accept(fields.from_node_id, fields.seq)?;
+        }
         rpc_codec::decode(inner_frame)
     }
 }

nodedb-cluster/src/transport/server.rs

@@ -107,7 +107,17 @@ async fn handle_stream<H: RaftRpcHandler>(
         let (fields, inner_frame) = auth_envelope::parse_envelope(&envelope, &auth.mac_key)?;
 
         // 2. Replay window — under the advertised from_node_id (MAC-verified).
-        auth.peer_seq_in.accept(fields.from_node_id, fields.seq)?;
+        //    Self-addressed frames skip the window: when a node dispatches
+        //    an RPC to itself over the transport, the shared `AuthContext`
+        //    means one window is updated by both the server-side request
+        //    accept (here) and the client-side response accept (in
+        //    `send.rs::parse_inbound`). Skipping when `from == local`
+        //    keeps the two flows from tripping on each other's entries —
+        //    a self-addressed frame can't have been replayed by an
+        //    external attacker by definition.
+        if fields.from_node_id != auth.local_node_id {
+            auth.peer_seq_in.accept(fields.from_node_id, fields.seq)?;
+        }
 
         // 3. Decode inner RPC and hand to handler.
         let request = rpc_codec::decode(inner_frame)?;
@@ -116,7 +126,7 @@ async fn handle_stream<H: RaftRpcHandler>(
         // 4. Wrap the response in its own envelope. `from = local_node_id`,
         //    `seq = next outbound seq scoped to the caller`.
         let response_inner = rpc_codec::encode(&response)?;
-        let response_seq = auth.peer_seq_out.next(fields.from_node_id);
+        let response_seq = auth.peer_seq_out.next();
         let mut response_envelope =
             Vec::with_capacity(auth_envelope::ENVELOPE_OVERHEAD + response_inner.len());
         auth_envelope::write_envelope(

nodedb-cluster/tests/transport_security.rs

@@ -20,13 +20,17 @@
 //! Tests build transports on ephemeral loopback ports and do not depend
 //! on any cluster infrastructure beyond the transport layer itself.
 
-use std::sync::{Arc, Mutex, OnceLock};
+use std::sync::{Arc, OnceLock};
 use std::time::Duration;
+use tokio::sync::Mutex;
 
 /// Serializes tests that construct `TransportCredentials::Insecure`, so the
 /// process-global `insecure_transport_count()` observed by
 /// `observability_insecure_counter_monotonic` is not racing with other
 /// concurrent tokio tests that also bump the counter.
+///
+/// Uses `tokio::sync::Mutex` so the guard can be held across `.await`
+/// points without tripping `clippy::await_holding_lock`.
 fn insecure_counter_guard() -> &'static Mutex<()> {
     static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
     LOCK.get_or_init(|| Mutex::new(()))
@@ -193,7 +197,7 @@ async fn l1_different_ca_mtls_rejects_handshake() {
 /// self-signed cert not signed by the client's CA).
 #[tokio::test]
 async fn l1_insecure_server_rejected_by_mtls_client() {
-    let _guard = insecure_counter_guard().lock().unwrap();
+    let _guard = insecure_counter_guard().lock().await;
     let (_ca, client_creds) = generate_node_credentials("nodedb").unwrap();
     let server = Arc::new(
         NexarTransport::new(
@@ -352,7 +356,7 @@ async fn l3_swim_rejects_mismatched_mac_key() {
 /// L.5 / observability: every `Insecure` construction bumps the counter.
 #[tokio::test]
 async fn observability_insecure_counter_monotonic() {
-    let _guard = insecure_counter_guard().lock().unwrap();
+    let _guard = insecure_counter_guard().lock().await;
     let before = insecure_transport_count();
     let _ = NexarTransport::new(
         99,