Remove sleeps from tests

Replace sleeps with the existing polling function instead. Exposes that the stress test isn't much of a stress test, it probably needs more data or something, but if it's heavy it should maybe not be part of `just test`

crates/tranquil-pds/tests/common/mod.rs

@@ -747,7 +747,40 @@ pub async fn spawn_cluster(pool: Option<sqlx::PgPool>, node_count: usize) -> Vec
     let first = &instances[0];
     APP_PORT.set(first.port).ok();
 
-    tokio::time::sleep(Duration::from_millis(2000)).await;
+    let caches: Vec<Arc<dyn Cache>> = instances
+        .iter()
+        .map(|instance| instance.cache.clone().expect("cluster node has no cache"))
+        .collect();
+    futures::future::join_all(caches.iter().enumerate().map(|(i, cache)| async move {
+        cache
+            .set(
+                &format!("__cluster_ready_{i}"),
+                "1",
+                Duration::from_secs(60),
+            )
+            .await
+            .expect("cluster readiness probe failed");
+    }))
+    .await;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        let ready = futures::future::join_all(caches.iter().flat_map(|cache| {
+            (0..caches.len()).map(move |i| async move {
+                cache.get(&format!("__cluster_ready_{i}")).await.is_some()
+            })
+        }))
+        .await
+        .into_iter()
+        .all(|seen| seen);
+        if ready {
+            break;
+        }
+        assert!(
+            tokio::time::Instant::now() < deadline,
+            "cluster nodes did not converge within 10s"
+        );
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
 
     instances
 }

crates/tranquil-ripple/tests/two_node_convergence.rs

@@ -36,8 +36,22 @@ async fn spawn_pair(
         .await
         .expect("node B failed to start");
 
-    tokio::time::sleep(Duration::from_millis(2000)).await;
-
+    cache_a
+        .set("__ready_a", "1", Duration::from_secs(60))
+        .await
+        .expect("readiness probe on A failed");
+    cache_b
+        .set("__ready_b", "1", Duration::from_secs(60))
+        .await
+        .expect("readiness probe on B failed");
+    poll_until(10_000, 50, || {
+        let cache_a = cache_a.clone();
+        let cache_b = cache_b.clone();
+        async move {
+            cache_b.get("__ready_a").await.is_some() && cache_a.get("__ready_b").await.is_some()
+        }
+    })
+    .await;
     ((cache_a, rl_a), (cache_b, rl_b))
 }
 
@@ -780,7 +794,5 @@ async fn two_node_stress_concurrent_load() {
         r.unwrap_or_else(|e| panic!("task {i} panicked: {e}"));
     });
 
-    tokio::time::sleep(Duration::from_secs(12)).await;
-
     shutdown.cancel();
 }