feat(nodedb-cluster): add QUIC peer pre-warming on startup

Add `NexarTransport::warm_peer(node_id)` which dials a registered
peer and inserts the `quinn::Connection` into the connection cache.
First replicated request after boot reuses the cached connection
rather than paying a cold dial.

Add `control::cluster::warm_peers` module with `warm_known_peers()`
which fans out warm attempts in parallel across all topology peers
(excluding self) with a configurable per-peer timeout, and returns
a `PeerWarmReport` summarising successes and failures.

Author: farhan-syah

nodedb-cluster/src/transport/client.rs

@@ -152,6 +152,23 @@ impl NexarTransport {
         debug!(node_id, %addr, "peer registered");
     }
 
+    /// Pre-warm the QUIC connection cache for a peer by
+    /// performing the full dial + handshake and inserting
+    /// the connection into the peer cache. On success, the
+    /// next `send_rpc(target, ...)` skips the dial entirely
+    /// and reuses the cached `quinn::Connection`.
+    ///
+    /// Caller MUST have called [`register_peer`] first — this
+    /// function resolves the peer address from the
+    /// `peer_addrs` map. Used by the startup `warm_peers`
+    /// phase so the first replicated request after boot
+    /// doesn't pay a cold-connect penalty.
+    ///
+    /// [`register_peer`]: Self::register_peer
+    pub async fn warm_peer(&self, node_id: u64) -> Result<()> {
+        self.get_or_connect(node_id).await.map(|_| ())
+    }
+
     /// Run the inbound RPC accept loop until shutdown.
     ///
     /// For each incoming connection, spawns a task that accepts

nodedb/src/control/cluster/mod.rs

@@ -18,10 +18,12 @@ pub mod init;
 pub mod metadata_applier;
 pub mod spsc_applier;
 pub mod start_raft;
+pub mod warm_peers;
 
 pub use applied_index_watcher::AppliedIndexWatcher;
 pub use handle::ClusterHandle;
 pub use init::{init_cluster, init_cluster_with_transport};
 pub use metadata_applier::MetadataCommitApplier;
 pub use spsc_applier::SpscCommitApplier;
 pub use start_raft::start_raft;
+pub use warm_peers::{PeerWarmReport, warm_known_peers};

nodedb/src/control/cluster/warm_peers/mod.rs

@@ -0,0 +1,10 @@
+//! Pre-warm the QUIC peer cache after `TransportBind` so the
+//! first replicated request after boot doesn't pay a cold
+//! connect. Slots into the startup sequencer between
+//! `TransportBind` and `WarmPeers` phases.
+
+pub mod report;
+pub mod warm;
+
+pub use report::PeerWarmReport;
+pub use warm::warm_known_peers;

nodedb/src/control/cluster/warm_peers/report.rs

@@ -0,0 +1,118 @@
+//! Report returned by [`super::warm_known_peers`].
+//!
+//! Logged at INFO regardless of outcome so operators see
+//! `peer_warm: 3 succeeded, 1 failed in 412ms` in the
+//! startup log without grepping for individual lines.
+
+use std::fmt;
+use std::time::Duration;
+
+#[derive(Debug, Clone)]
+pub struct PeerWarmReport {
+    /// Number of peers attempted (everything in the topology
+    /// except this node).
+    pub attempted: usize,
+    /// Node ids that successfully warmed.
+    pub succeeded: Vec<u64>,
+    /// Per-peer failure with the underlying transport error
+    /// stringified — kept as String because the cluster
+    /// transport returns its own typed error which we don't
+    /// want to leak into this struct's API surface.
+    pub failed: Vec<(u64, String)>,
+    /// Total wall-clock for the whole warm phase. Includes
+    /// the parallel-dial time, NOT the sum of individual
+    /// per-peer dials.
+    pub elapsed: Duration,
+}
+
+impl PeerWarmReport {
+    /// Whether at least one peer was warmed. Useful for
+    /// tests; the production path logs the full report
+    /// regardless.
+    pub fn is_success(&self) -> bool {
+        !self.succeeded.is_empty()
+    }
+
+    /// Whether every attempted peer was reached.
+    pub fn is_complete(&self) -> bool {
+        self.failed.is_empty() && self.succeeded.len() == self.attempted
+    }
+
+    /// Empty report — used when the topology has no peers
+    /// (single-node mode).
+    pub fn empty() -> Self {
+        Self {
+            attempted: 0,
+            succeeded: Vec::new(),
+            failed: Vec::new(),
+            elapsed: Duration::ZERO,
+        }
+    }
+}
+
+impl fmt::Display for PeerWarmReport {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "peer_warm: {}/{} succeeded ({} failed) in {:?}",
+            self.succeeded.len(),
+            self.attempted,
+            self.failed.len(),
+            self.elapsed
+        )?;
+        for (id, err) in &self.failed {
+            write!(f, "\n  node {id}: {err}")?;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_report_is_complete_and_unsuccessful() {
+        let r = PeerWarmReport::empty();
+        assert!(r.is_complete());
+        assert!(!r.is_success());
+    }
+
+    #[test]
+    fn complete_report() {
+        let r = PeerWarmReport {
+            attempted: 2,
+            succeeded: vec![1, 2],
+            failed: vec![],
+            elapsed: Duration::from_millis(50),
+        };
+        assert!(r.is_complete());
+        assert!(r.is_success());
+    }
+
+    #[test]
+    fn partial_failure_not_complete() {
+        let r = PeerWarmReport {
+            attempted: 3,
+            succeeded: vec![1],
+            failed: vec![(2, "timeout".into()), (3, "refused".into())],
+            elapsed: Duration::from_millis(2000),
+        };
+        assert!(!r.is_complete());
+        assert!(r.is_success());
+    }
+
+    #[test]
+    fn display_includes_failures() {
+        let r = PeerWarmReport {
+            attempted: 2,
+            succeeded: vec![1],
+            failed: vec![(2, "connection refused".into())],
+            elapsed: Duration::from_millis(100),
+        };
+        let s = r.to_string();
+        assert!(s.contains("1/2 succeeded"));
+        assert!(s.contains("node 2"));
+        assert!(s.contains("connection refused"));
+    }
+}

nodedb/src/control/cluster/warm_peers/warm.rs

@@ -0,0 +1,176 @@
+//! Parallel peer-cache warm-up.
+//!
+//! Dials every node in the live topology except `self_id`,
+//! caps each dial at half the total deadline, and aggregates
+//! the outcome into a [`PeerWarmReport`]. Failures are
+//! non-fatal — operators see the per-peer failure in the
+//! returned report and the startup phase advances regardless.
+
+use std::collections::HashSet;
+use std::time::{Duration, Instant};
+
+use futures::future::join_all;
+use nodedb_cluster::{ClusterTopology, NexarTransport};
+
+use super::report::PeerWarmReport;
+
+/// Pre-warm every peer in `topology` except `self_id`.
+///
+/// `total_deadline` bounds the entire operation. Each
+/// individual peer dial is capped at `total_deadline / 2`
+/// so a single slow peer cannot eat the whole budget. All
+/// dials run in parallel, so the total wall-clock is
+/// roughly `max(per-peer-latency, total_deadline)`.
+///
+/// Empty topology → returns [`PeerWarmReport::empty`].
+/// Caller logs the returned report once; this function
+/// only logs per-peer failures if verbose tracing is
+/// enabled on the transport layer.
+pub async fn warm_known_peers(
+    transport: &NexarTransport,
+    topology: &ClusterTopology,
+    self_id: u64,
+    total_deadline: Duration,
+) -> PeerWarmReport {
+    let start = Instant::now();
+    let per_peer_deadline = total_deadline.checked_div(2).unwrap_or(total_deadline);
+
+    let peers: Vec<(u64, std::net::SocketAddr)> = topology
+        .all_nodes()
+        .filter(|n| n.node_id != self_id)
+        .filter_map(|n| n.socket_addr().map(|a| (n.node_id, a)))
+        .collect();
+
+    if peers.is_empty() {
+        return PeerWarmReport::empty();
+    }
+
+    let attempted = peers.len();
+
+    // Ensure the transport's address map has every peer.
+    // `register_peer` is idempotent — safe to call again.
+    for (id, addr) in &peers {
+        transport.register_peer(*id, *addr);
+    }
+
+    let futs = peers.iter().map(|(id, _addr)| {
+        let id = *id;
+        async move {
+            match tokio::time::timeout(per_peer_deadline, transport.warm_peer(id)).await {
+                Ok(Ok(())) => (id, Ok(())),
+                Ok(Err(e)) => (id, Err(format!("{e}"))),
+                Err(_) => (id, Err(format!("dial timeout after {per_peer_deadline:?}"))),
+            }
+        }
+    });
+
+    // Outer deadline bounds the entire parallel batch. On
+    // outer timeout, every still-pending future is dropped
+    // (cancelling the in-flight dials) and the missing peers
+    // are reported as deadline exceedances.
+    let outcome = tokio::time::timeout(total_deadline, join_all(futs)).await;
+
+    let mut succeeded: Vec<u64> = Vec::with_capacity(attempted);
+    let mut failed: Vec<(u64, String)> = Vec::new();
+
+    match outcome {
+        Ok(results) => {
+            for (id, outcome) in results {
+                match outcome {
+                    Ok(()) => succeeded.push(id),
+                    Err(msg) => failed.push((id, msg)),
+                }
+            }
+        }
+        Err(_) => {
+            // Outer deadline expired — we don't know which
+            // futures completed vs which were in flight, so
+            // mark every peer as a deadline exceedance. The
+            // report is accurate about the worst case.
+            for (id, _) in &peers {
+                failed.push((
+                    *id,
+                    format!("outer warm deadline exceeded after {total_deadline:?}"),
+                ));
+            }
+        }
+    }
+
+    // Sanity: account every attempted peer. If a future was
+    // dropped between matching on the outer result and
+    // collecting (shouldn't happen with join_all but cheap
+    // to double-check), any missing ids become failures.
+    let reported: HashSet<u64> = succeeded
+        .iter()
+        .copied()
+        .chain(failed.iter().map(|(id, _)| *id))
+        .collect();
+    for (id, _) in &peers {
+        if !reported.contains(id) {
+            failed.push((*id, "unaccounted peer — internal bookkeeping bug".into()));
+        }
+    }
+
+    PeerWarmReport {
+        attempted,
+        succeeded,
+        failed,
+        elapsed: start.elapsed(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nodedb_cluster::ClusterTopology;
+
+    #[tokio::test]
+    async fn empty_topology_returns_empty_report() {
+        let transport = test_transport();
+        let topo = ClusterTopology::new();
+        let report = warm_known_peers(&transport, &topo, 1, Duration::from_secs(1)).await;
+        assert_eq!(report.attempted, 0);
+        assert!(report.is_complete());
+        assert_eq!(report.elapsed, Duration::ZERO);
+    }
+
+    #[tokio::test]
+    async fn self_only_topology_skips_self() {
+        let transport = test_transport();
+        let mut topo = ClusterTopology::new();
+        topo.add_node(nodedb_cluster::NodeInfo::new(
+            1,
+            "127.0.0.1:65530".parse().unwrap(),
+            nodedb_cluster::NodeState::Active,
+        ));
+        let report = warm_known_peers(&transport, &topo, 1, Duration::from_secs(1)).await;
+        assert_eq!(report.attempted, 0);
+    }
+
+    #[tokio::test]
+    async fn dead_peer_is_reported_as_failure() {
+        let transport = test_transport();
+        let mut topo = ClusterTopology::new();
+        topo.add_node(nodedb_cluster::NodeInfo::new(
+            1,
+            "127.0.0.1:65530".parse().unwrap(),
+            nodedb_cluster::NodeState::Active,
+        ));
+        // 127.0.0.1:1 — reserved port, guaranteed to fail.
+        topo.add_node(nodedb_cluster::NodeInfo::new(
+            2,
+            "127.0.0.1:1".parse().unwrap(),
+            nodedb_cluster::NodeState::Active,
+        ));
+        let report = warm_known_peers(&transport, &topo, 1, Duration::from_millis(500)).await;
+        assert_eq!(report.attempted, 1);
+        assert_eq!(report.succeeded.len(), 0);
+        assert_eq!(report.failed.len(), 1);
+        assert_eq!(report.failed[0].0, 2);
+        assert!(report.elapsed <= Duration::from_millis(700));
+    }
+
+    fn test_transport() -> NexarTransport {
+        NexarTransport::new(999, "127.0.0.1:0".parse().unwrap()).expect("test transport bind")
+    }
+}