test(cluster): replace fixed sleeps with readiness polling and fix SWIM probe test

Replace fixed tokio::time::sleep calls in tests with bounded polling
loops that break as soon as the expected condition is met. This makes
tests deterministic under heavy parallel load (500+ unit tests sharing
the same CPU pool) rather than depending on timing assumptions.

In the SWIM indirect_ack_saves_target test, remove start_paused — paused
time auto-advances timeouts before channel-woken tasks are polled, making
the indirect path race its own timeout. With real time the 40 ms probe
window is ample for in-memory delivery. Add a recv loop on the local
endpoint to resolve inflight probes when Acks arrive, and handle both
Ping and PingReq in the helper so the test is correct regardless of
which node the scheduler picks as the direct target.

In the cluster harness, replace the 200 ms sleep before peers join with
a topology-readiness poll, and serialise node 2 joining before node 3 to
avoid overwhelming the bootstrap leader's join handler under load.

Extend the descriptor lease planner wait budget from 3 s to 10 s to
match the extended election timeout range used in tests.

Author: farhan-syah

nodedb-cluster/src/raft_loop/loop_core.rs

@@ -580,35 +580,38 @@ mod tests {
         let sr1h = shutdown_tx.subscribe();
         tokio::spawn(async move { t1.serve(rl1_h, sr1h).await });
 
-        tokio::time::sleep(Duration::from_millis(200)).await;
-
-        assert!(
-            rl1.applier.count() >= 1,
-            "node 1 should have committed at least the no-op, got {}",
-            rl1.applier.count()
-        );
+        // Poll until node 1 commits at least the no-op (election done).
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+        loop {
+            if rl1.applier.count() >= 1 {
+                break;
+            }
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "node 1 should have committed at least the no-op, got {}",
+                rl1.applier.count()
+            );
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
 
         let (_gid, idx) = rl1.propose(0, b"distributed-cmd".to_vec()).unwrap();
         assert!(idx >= 2);
 
-        tokio::time::sleep(Duration::from_millis(200)).await;
-
-        assert!(
-            rl1.applier.count() >= 2,
-            "node 1: expected >= 2 applied, got {}",
-            rl1.applier.count()
-        );
-
-        assert!(
-            rl2.applier.count() >= 1,
-            "node 2: expected >= 1 applied, got {}",
-            rl2.applier.count()
-        );
-        assert!(
-            rl3.applier.count() >= 1,
-            "node 3: expected >= 1 applied, got {}",
-            rl3.applier.count()
-        );
+        // Poll until all nodes replicate the proposed command.
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+        loop {
+            if rl1.applier.count() >= 2 && rl2.applier.count() >= 1 && rl3.applier.count() >= 1 {
+                break;
+            }
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "replication timed out: n1={}, n2={}, n3={}",
+                rl1.applier.count(),
+                rl2.applier.count(),
+                rl3.applier.count()
+            );
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
 
         shutdown_tx.send(true).unwrap();
     }

nodedb-cluster/src/swim/detector/probe_round.rs

@@ -413,13 +413,16 @@ mod tests {
         );
     }
 
-    #[tokio::test(start_paused = true)]
+    #[tokio::test]
     async fn indirect_ack_saves_target() {
+        // No `start_paused` — paused time auto-advances timeouts
+        // before polling channel-woken tasks, making the indirect
+        // path race the timeout. With real time, the 40ms probe
+        // timeout is ample for the in-memory fabric (sub-µs delivery).
         let fab = TransportFabric::new();
-        let local = Arc::new(fab.bind(addr(7000)).await) as Arc<dyn Transport>;
-        // Target bound but silent on the direct channel.
+        let local = Arc::new(fab.bind(addr(7000)).await);
         let _silent = fab.bind(addr(7001)).await;
-        let helper = fab.bind(addr(7002)).await;
+        let helper = Arc::new(fab.bind(addr(7002)).await);
         let list = membership_with_peers(
             "local",
             7000,
@@ -432,38 +435,74 @@ mod tests {
         let mut sched = ProbeScheduler::with_seed(1);
         let inflight = Arc::new(InflightProbes::new());
 
-        // Helper task: forwards any PingReq it sees into an Ack via the
-        // inflight registry. Paused-runtime auto-advance drives the
-        // direct-ping timeout on the main task.
-        let inflight_helper = Arc::clone(&inflight);
+        // Helper task: respond to Ping (direct probe) with Ack and to
+        // PingReq (indirect probe) with a forwarded Ack — mirrors the
+        // production runner recv-loop + handle_ping_req path. The
+        // scheduler may pick n2 as direct target or as indirect helper
+        // depending on the shuffle seed, so both must be handled.
+        let helper_t: Arc<dyn Transport> = helper.clone();
         let responder = tokio::spawn(async move {
             loop {
-                let (_from, msg) = match helper.recv().await {
+                let (from, msg) = match helper_t.recv().await {
                     Ok(v) => v,
                     Err(_) => return,
                 };
-                if let SwimMessage::PingReq(req) = msg {
-                    inflight_helper
-                        .resolve(
-                            req.probe_id,
-                            SwimMessage::Ack(Ack {
-                                probe_id: req.probe_id,
-                                from: req.target.clone(),
-                                incarnation: Incarnation::new(9),
-                                piggyback: vec![],
-                            }),
-                        )
-                        .await;
-                    return;
+                match msg {
+                    SwimMessage::Ping(ping) => {
+                        let _ = helper_t
+                            .send(
+                                from,
+                                SwimMessage::Ack(Ack {
+                                    probe_id: ping.probe_id,
+                                    from: NodeId::new("n2"),
+                                    incarnation: Incarnation::new(9),
+                                    piggyback: vec![],
+                                }),
+                            )
+                            .await;
+                        return;
+                    }
+                    SwimMessage::PingReq(req) => {
+                        let _ = helper_t
+                            .send(
+                                from,
+                                SwimMessage::Ack(Ack {
+                                    probe_id: req.probe_id,
+                                    from: req.target.clone(),
+                                    incarnation: Incarnation::new(9),
+                                    piggyback: vec![],
+                                }),
+                            )
+                            .await;
+                        return;
+                    }
+                    _ => {}
                 }
             }
         });
 
+        // Recv-loop on the local endpoint: resolves inflight probes
+        // when Acks arrive — mirrors the production runner recv-loop.
+        let recv_t: Arc<dyn Transport> = local.clone();
+        let recv_inflight = Arc::clone(&inflight);
+        let recv_loop = tokio::spawn(async move {
+            loop {
+                let (_from, msg) = match recv_t.recv().await {
+                    Ok(v) => v,
+                    Err(_) => return,
+                };
+                if let SwimMessage::Ack(ref ack) = msg {
+                    recv_inflight.resolve(ack.probe_id, msg).await;
+                }
+            }
+        });
+
+        let local_dyn: Arc<dyn Transport> = local.clone();
         let dissemination = Arc::new(DisseminationQueue::new());
         let outcome = ProbeRound {
             scheduler: &mut sched,
             membership: &list,
-            transport: &local,
+            transport: &local_dyn,
             inflight: &inflight,
             dissemination: &dissemination,
             probe_timeout: cfg().probe_timeout,
@@ -476,9 +515,8 @@ mod tests {
         .execute()
         .await
         .expect("run");
-        let _ = responder.await;
-        // Either direct (unlikely — n1 is silent) or indirect ack via n2.
-        // Whichever path fires, the outcome must be Acked.
+        responder.abort();
+        recv_loop.abort();
         assert!(matches!(outcome, ProbeOutcome::Acked { .. }));
     }
 

nodedb-cluster/tests/common/mod.rs

@@ -187,6 +187,8 @@ impl TestNode {
                 max_backoff_secs: 2,
             },
             swim_udp_addr: None,
+            election_timeout_min: std::time::Duration::from_millis(150),
+            election_timeout_max: std::time::Duration::from_millis(300),
         };
 
         let lifecycle = ClusterLifecycleTracker::new();

nodedb/tests/cluster_execute_request.rs

@@ -162,8 +162,19 @@ async fn execute_request_cross_node_dispatch() {
         .await
         .expect("create collection");
 
-    // Give the metadata applier on all nodes a moment to replicate.
-    tokio::time::sleep(Duration::from_millis(400)).await;
+    // Wait for the collection to be visible on every node.
+    common::cluster_harness::wait_for(
+        "cross_node_kv visible on all nodes",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 1)
+        },
+    )
+    .await;
 
     // Node 2 sends the request; node 1 (bootstrap leader) receives it.
     let sender_transport = cluster.nodes[1]

nodedb/tests/common/cluster_harness/cluster.rs

@@ -17,7 +17,18 @@ impl TestCluster {
     /// via node 1's pre-bound address. Waits until every node sees
     /// topology_size == 3 (10s deadline).
     pub async fn spawn_three() -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
-        Self::spawn_three_with_tuning(ClusterTransportTuning::default()).await
+        Self::spawn_three_with_tuning(ClusterTransportTuning {
+            // Fast health pings so the HealthMonitor re-broadcasts
+            // topology within ~1s if the initial join broadcast was missed.
+            health_ping_interval_secs: 1,
+            // Fast election timeouts so the metadata Raft group elects a
+            // leader well within the 10s convergence deadline, even under
+            // heavy parallel test load.
+            election_timeout_min_secs: 1,
+            election_timeout_max_secs: 2,
+            ..ClusterTransportTuning::default()
+        })
+        .await
     }
 
     /// Spawn a 3-node cluster with a custom `ClusterTransportTuning`.
@@ -29,12 +40,34 @@ impl TestCluster {
     ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
         let node1 = TestClusterNode::spawn_with_tuning(1, vec![], tuning.clone()).await?;
 
-        // Give node 1's transport + raft loop a moment to start
-        // accepting before peers dial in.
-        tokio::time::sleep(Duration::from_millis(200)).await;
+        // Wait until node 1 has bootstrapped (topology shows itself)
+        // before peers try to join. The old fixed 200ms sleep was too
+        // short under heavy host load (e.g. 500+ parallel unit tests
+        // sharing the same CPU pool), causing peers to dial before
+        // node 1's transport was ready — failing topology convergence.
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while node1.topology_size() < 1 {
+            if std::time::Instant::now() >= deadline {
+                return Err("node 1 failed to bootstrap within 10s".into());
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
 
         let seeds = vec![node1.listen_addr];
         let node2 = TestClusterNode::spawn_with_tuning(2, seeds.clone(), tuning.clone()).await?;
+
+        // Wait for node 2's join to be reflected before spawning node 3.
+        // Under load, spawning both peers simultaneously can overwhelm the
+        // bootstrap leader's join handler, causing neither join to complete
+        // within the topology convergence deadline.
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while node1.topology_size() < 2 {
+            if std::time::Instant::now() >= deadline {
+                return Err("node 2 failed to join within 10s".into());
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+
         let node3 = TestClusterNode::spawn_with_tuning(3, seeds, tuning).await?;
 
         let cluster = Self {
@@ -44,7 +77,7 @@ impl TestCluster {
         wait_for(
             "all 3 nodes report topology_size == 3",
             Duration::from_secs(10),
-            Duration::from_millis(100),
+            Duration::from_millis(50),
             || cluster.nodes.iter().all(|n| n.topology_size() == 3),
         )
         .await;

nodedb/tests/common/cluster_harness/node.rs

@@ -129,6 +129,7 @@ impl TestClusterNode {
             &cluster_settings,
             transport.clone(),
             &data_dir_path,
+            &tuning,
         )
         .await?;
 

nodedb/tests/descriptor_lease_planner_integration.rs

@@ -16,7 +16,7 @@ use common::cluster_harness::{TestCluster, wait_for};
 use nodedb_cluster::{DescriptorId, DescriptorKind};
 
 const TENANT: u32 = 1;
-const WAIT_BUDGET: Duration = Duration::from_secs(3);
+const WAIT_BUDGET: Duration = Duration::from_secs(10);
 const POLL: Duration = Duration::from_millis(20);
 
 fn coll_id(name: &str) -> DescriptorId {