feat(cluster): start health monitor and harden topology convergence

Start the HealthMonitor task from start_raft so that periodic pings
run for the lifetime of the node. Without it, topology updates relied
solely on the fire-and-forget broadcast during the join flow; when that
broadcast was lost (peer QUIC server not yet accepting), the peer never
converged to the full topology.

On each successful pong, push the current topology to peers whose
reported version is behind ours. This closes the convergence gap for
peers that missed the initial broadcast.

Bound QUIC handshake attempts by rpc_timeout so a hung connection does
not block for the full 30 s idle timeout when a peer is slow to accept.

Expose the cluster catalog via ClusterHandle so the HealthMonitor can
persist topology changes detected during failure/recovery. Increase the
pgwire retry budget from 3 attempts (350 ms) to 5 attempts (750 ms) to
tolerate longer Raft leader drains under load.

Author: farhan-syah

nodedb-cluster/src/health.rs

@@ -151,14 +151,38 @@ impl HealthMonitor {
         }
     }
 
-    /// Handle a successful pong — reset failure count, mark node Active if needed.
-    fn handle_pong(&self, peer_id: u64, _pong: &PongResponse) -> bool {
+    /// Handle a successful pong — reset failure count, mark node Active
+    /// if needed, and push topology if the peer is behind.
+    fn handle_pong(&self, peer_id: u64, pong: &PongResponse) -> bool {
         // Reset failure count.
         {
             let mut failures = self.ping_failures.lock().unwrap_or_else(|p| p.into_inner());
             failures.remove(&peer_id);
         }
 
+        // Push topology to peers with a stale version. This closes
+        // the convergence gap when the fire-and-forget broadcast
+        // during the join flow is lost (e.g. peer QUIC server not
+        // yet accepting at that instant).
+        let our_version = {
+            let topo = self.topology.read().unwrap_or_else(|p| p.into_inner());
+            topo.version()
+        };
+        if pong.topology_version < our_version {
+            debug!(
+                peer_id,
+                peer_version = pong.topology_version,
+                our_version,
+                "peer has stale topology, pushing update"
+            );
+            let transport = self.transport.clone();
+            let topology = self.topology.clone();
+            let self_id = self.node_id;
+            tokio::spawn(async move {
+                broadcast_topology_to_peer(self_id, peer_id, &topology, &transport).await;
+            });
+        }
+
         // If node was not Active, mark it Active.
         let mut topo = self.topology.write().unwrap_or_else(|p| p.into_inner());
         if let Some(node) = topo.get_node(peer_id)
@@ -264,6 +288,34 @@ pub fn broadcast_topology(
     }
 }
 
+/// Send a topology update to a single peer that has a stale version.
+async fn broadcast_topology_to_peer(
+    _self_node_id: u64,
+    peer_id: u64,
+    topology: &RwLock<ClusterTopology>,
+    transport: &NexarTransport,
+) {
+    let update = {
+        let topo = topology.read().unwrap_or_else(|p| p.into_inner());
+        RaftRpc::TopologyUpdate(TopologyUpdate {
+            version: topo.version(),
+            nodes: topo
+                .all_nodes()
+                .map(|n| JoinNodeInfo {
+                    node_id: n.node_id,
+                    addr: n.addr.clone(),
+                    state: n.state.as_u8(),
+                    raft_groups: n.raft_groups.clone(),
+                    wire_version: n.wire_version,
+                })
+                .collect(),
+        })
+    };
+    if let Err(e) = transport.send_rpc(peer_id, update).await {
+        debug!(peer_id, error = %e, "targeted topology push failed");
+    }
+}
+
 /// Handle an incoming Ping RPC — return a Pong with our topology version.
 pub fn handle_ping(node_id: u64, topology_version: u64, _req: &PingRequest) -> RaftRpc {
     RaftRpc::Pong(PongResponse {

nodedb-cluster/src/transport/client.rs

@@ -245,15 +245,23 @@ impl NexarTransport {
                 .ok_or(ClusterError::NodeUnreachable { node_id: target })?
         };
 
-        // Connect.
-        let conn = self
+        // Connect — bounded by rpc_timeout so a hung QUIC handshake
+        // (peer not yet serving) doesn't block for the full 30s idle timeout.
+        let connecting = self
             .listener
             .endpoint()
             .connect_with(self.client_config.clone(), addr, SNI_HOSTNAME)
             .map_err(|e| ClusterError::Transport {
                 detail: format!("connect to node {target} at {addr}: {e}"),
-            })?
+            })?;
+        let conn = tokio::time::timeout(self.rpc_timeout, connecting)
             .await
+            .map_err(|_| ClusterError::Transport {
+                detail: format!(
+                    "handshake timeout ({}ms) with node {target} at {addr}",
+                    self.rpc_timeout.as_millis()
+                ),
+            })?
             .map_err(|e| ClusterError::Transport {
                 detail: format!("handshake with node {target} at {addr}: {e}"),
             })?;

nodedb/src/control/cluster/handle.rs

@@ -33,4 +33,8 @@ pub struct ClusterHandle {
     /// stays `Clone` while still guaranteeing single-transfer
     /// semantics at runtime.
     pub multi_raft: Mutex<Option<nodedb_cluster::MultiRaft>>,
+    /// Cluster catalog (redb-backed topology + routing persistence).
+    /// Shared with the `HealthMonitor` for persisting topology changes
+    /// on failure detection and recovery.
+    pub catalog: Arc<nodedb_cluster::ClusterCatalog>,
 }

nodedb/src/control/cluster/init.rs

@@ -38,7 +38,7 @@ pub async fn init_cluster(
         "cluster QUIC transport bound"
     );
 
-    init_cluster_with_transport(config, transport, data_dir).await
+    init_cluster_with_transport(config, transport, data_dir, transport_tuning).await
 }
 
 /// Initialize the cluster using a pre-bound QUIC transport.
@@ -56,13 +56,15 @@ pub async fn init_cluster_with_transport(
     config: &ClusterSettings,
     transport: Arc<nodedb_cluster::NexarTransport>,
     data_dir: &std::path::Path,
+    transport_tuning: &ClusterTransportTuning,
 ) -> crate::Result<ClusterHandle> {
     // 2. Open cluster catalog.
     let catalog_path = data_dir.join("cluster.redb");
-    let catalog =
+    let catalog = Arc::new(
         nodedb_cluster::ClusterCatalog::open(&catalog_path).map_err(|e| crate::Error::Config {
             detail: format!("cluster catalog: {e}"),
-        })?;
+        })?,
+    );
 
     // 3. Bootstrap, join, or restart.
     let cluster_config = nodedb_cluster::ClusterConfig {
@@ -75,6 +77,12 @@ pub async fn init_cluster_with_transport(
         force_bootstrap: config.force_bootstrap,
         join_retry: join_retry_policy_from_env(),
         swim_udp_addr: None,
+        election_timeout_min: std::time::Duration::from_secs(
+            transport_tuning.election_timeout_min_secs,
+        ),
+        election_timeout_max: std::time::Duration::from_secs(
+            transport_tuning.election_timeout_max_secs,
+        ),
     };
 
     let lifecycle = nodedb_cluster::ClusterLifecycleTracker::new();
@@ -105,6 +113,7 @@ pub async fn init_cluster_with_transport(
         applied_index_watcher,
         node_id: config.node_id,
         multi_raft: Mutex::new(Some(state.multi_raft)),
+        catalog,
     })
 }
 

nodedb/src/control/cluster/start_raft.rs

@@ -112,7 +112,7 @@ pub fn start_raft(
     // Start the RPC server (accepts inbound QUIC connections).
     let transport_serve = handle.transport.clone();
     let rl_handler = raft_loop.clone();
-    let sr_serve = shutdown_rx;
+    let sr_serve = shutdown_rx.clone();
     tokio::spawn(async move {
         if let Err(e) = transport_serve.serve(rl_handler, sr_serve).await {
             tracing::error!(error = %e, "raft RPC server failed");
@@ -138,6 +138,27 @@ pub fn start_raft(
         );
     }
 
+    // Start the health monitor (periodic pings, failure detection,
+    // topology re-broadcast). Without this, topology updates are
+    // only propagated via the fire-and-forget broadcast during the
+    // join flow — if that single broadcast is lost (peer QUIC server
+    // not yet accepting), the peer never converges.
+    let health_config = nodedb_cluster::HealthConfig {
+        ping_interval: Duration::from_secs(transport_tuning.health_ping_interval_secs),
+        failure_threshold: transport_tuning.health_failure_threshold,
+    };
+    let health_monitor = Arc::new(nodedb_cluster::HealthMonitor::new(
+        handle.node_id,
+        handle.transport.clone(),
+        handle.topology.clone(),
+        handle.catalog.clone(),
+        health_config,
+    ));
+    let sr_health = shutdown_rx;
+    tokio::spawn(async move {
+        health_monitor.run(sr_health).await;
+    });
+
     info!(node_id = handle.node_id, "raft loop and RPC server started");
 
     Ok(ready_rx)

nodedb/src/control/server/pgwire/handler/retry.rs

@@ -18,8 +18,8 @@
 //!
 //! ## Retry budget
 //!
-//! Three attempts total with 50ms, 100ms, 200ms backoff between
-//! them — roughly 350ms of tolerance for a drain to complete.
+//! Five attempts total with 50/100/200/400 ms backoff between
+//! them — roughly 750ms of tolerance for a drain to complete.
 //! The `DEFAULT_DRAIN_TIMEOUT` from `metadata_proposer` is 35s,
 //! so in practice either drain completes within our retry budget
 //! (the proposer is actively draining and is probably close to
@@ -31,13 +31,17 @@ use std::time::Duration;
 use crate::error::Error;
 
 /// Maximum number of attempts (including the initial call).
-const MAX_ATTEMPTS: usize = 3;
+const MAX_ATTEMPTS: usize = 5;
 
 /// Backoff durations BETWEEN attempts. `BACKOFFS[i]` is the sleep
 /// duration before attempt `i + 1`. Length must be
 /// `MAX_ATTEMPTS - 1`.
-const BACKOFFS: [Duration; MAX_ATTEMPTS - 1] =
-    [Duration::from_millis(50), Duration::from_millis(100)];
+const BACKOFFS: [Duration; MAX_ATTEMPTS - 1] = [
+    Duration::from_millis(50),
+    Duration::from_millis(100),
+    Duration::from_millis(200),
+    Duration::from_millis(400),
+];
 
 /// Run `op` up to `MAX_ATTEMPTS` times. Retries only on
 /// `Error::RetryableSchemaChanged`. Any other error (including