fix(nodedb-cluster): eliminate 30s-per-peer stalls during cluster join

Two independent but compounding issues caused cluster join to hang for
tens of seconds on every startup when seeds were not yet bootstrapped:

1. The QUIC RPC timeout only covered the response-read phase. A
   handshake attempt against an unreachable or not-yet-listening peer
   blocked for the transport's internal idle timeout (~30 s), not the
   configured RPC timeout. In a 5-node race where every non-bootstrapper
   seed redirects to another non-bootstrapper, this multiplied to
   (N-1) × 30 s of wasted wall time per join attempt.

   Fixed by wrapping the entire send_rpc_to_addr operation — handshake,
   stream open, write, and read — in a single tokio::time::timeout
   bounded by self.rpc_timeout, and extracting the inner work into
   send_rpc_to_addr_inner so the public interface stays clean.

2. The seed work-list was a Vec used as a stack (pop), so seed order
   was unspecified. Under the single-elected-bootstrapper rule the
   lexicographically smallest address is the one peer that can actually
   answer during the initial race; hitting it last meant exhausting
   timeouts against every other seed first.

   Fixed by sorting seeds at the start of the join loop so the
   designated bootstrapper surfaces first, and switching to VecDeque
   so leader redirects are pushed to the front (push_front / pop_front)
   and consumed before unvisited seeds.

Author: farhan-syah

nodedb-cluster/src/bootstrap/join.rs

@@ -143,15 +143,30 @@ async fn try_join_once(
     transport: &NexarTransport,
     req_template: &JoinRequest,
 ) -> Result<ClusterState> {
-    // Work list: start with the configured seeds, prepend leader hints
-    // as they arrive. `HashSet` deduplicates so a redirect loop can't
-    // consume all attempts against the same address.
-    let mut work: Vec<SocketAddr> = config.seed_nodes.clone();
+    // Work list: try seeds in sorted order so the lexicographically
+    // smallest address — the designated bootstrapper under the
+    // single-elected-bootstrapper rule — is contacted first. This is
+    // critical during the initial 5-node race: every other seed points
+    // at a node that is itself still joining, so asking them first
+    // eats the full RPC timeout per non-bootstrapper before we reach
+    // the one peer that can actually answer. `HashSet` deduplicates
+    // so a redirect loop can't consume all attempts against the same
+    // address.
+    let mut work: std::collections::VecDeque<SocketAddr> =
+        config.seed_nodes.iter().copied().collect();
+    {
+        // Sort so the designated bootstrapper surfaces first. Leader
+        // redirects get prepended with push_front below, keeping the
+        // "most likely to answer" candidate at the head.
+        let mut sorted: Vec<SocketAddr> = work.drain(..).collect();
+        sorted.sort();
+        work.extend(sorted);
+    }
     let mut visited: HashSet<SocketAddr> = HashSet::new();
     let mut redirects: u32 = 0;
     let mut last_err: Option<ClusterError> = None;
 
-    while let Some(addr) = work.pop() {
+    while let Some(addr) = work.pop_front() {
         if !visited.insert(addr) {
             continue;
         }
@@ -172,7 +187,7 @@ async fn try_join_once(
                             "following leader redirect"
                         );
                         redirects += 1;
-                        work.push(leader);
+                        work.push_front(leader);
                         continue;
                     }
                     debug!(

nodedb-cluster/src/transport/client.rs

@@ -267,7 +267,23 @@ impl NexarTransport {
     }
 
     /// Send an RPC to an address directly (for bootstrap/join before peer IDs are known).
+    ///
+    /// The **entire** operation — handshake, stream open, write, read — is
+    /// bounded by `self.rpc_timeout`. Previously only the response read was
+    /// timed out, which meant a QUIC handshake against an unreachable peer
+    /// could block for the transport's internal idle timeout (~30 s per
+    /// address). That was fatal to cluster join races where every non-
+    /// bootstrapper seed points at another non-bootstrapper: each attempt
+    /// would stall 30 s × (N-1) peers, compounding across retry passes.
     pub async fn send_rpc_to_addr(&self, addr: SocketAddr, rpc: RaftRpc) -> Result<RaftRpc> {
+        tokio::time::timeout(self.rpc_timeout, self.send_rpc_to_addr_inner(addr, rpc))
+            .await
+            .map_err(|_| ClusterError::Transport {
+                detail: format!("RPC timeout ({}ms) to {addr}", self.rpc_timeout.as_millis()),
+            })?
+    }
+
+    async fn send_rpc_to_addr_inner(&self, addr: SocketAddr, rpc: RaftRpc) -> Result<RaftRpc> {
         let frame = rpc_codec::encode(&rpc)?;
 
         let conn = self
@@ -295,12 +311,7 @@ impl NexarTransport {
             detail: format!("finish send to {addr}: {e}"),
         })?;
 
-        let response_frame = tokio::time::timeout(self.rpc_timeout, server::read_frame(&mut recv))
-            .await
-            .map_err(|_| ClusterError::Transport {
-                detail: format!("RPC timeout ({}ms) to {addr}", self.rpc_timeout.as_millis()),
-            })??;
-
+        let response_frame = server::read_frame(&mut recv).await?;
         rpc_codec::decode(&response_frame)
     }