test(actor): deflake debounce/derived/liveness timing tests (#665)

Replace fixed-sleep-then-assert with condition-based waiting (wait_for
helper) in debounce.rs, derived.rs, and lib.rs.  No production code
changed — all fixes are test-only.

Root cause per test:

debounce_single_message / debounce_separate_bursts
  Test race: fixed sleeps (100ms / 80ms) smaller than the timer + actor
  message-delivery chain under parallel suite load.  Fixed by polling
  count.load() with a 2000ms timeout instead of sleeping.

debounce_rapid_messages / throttle_* tests
  Same class of race on the positive "eventually happens" assertions.
  Hardened proactively.  throttle_rate_limited's "still throttled"
  assertion keeps a short fixed sleep (20ms << 100ms cooldown) with a
  comment explaining why it is a genuine timing invariant.

debounce_timer_reset
  The intermediate assert_eq!(count, 0) at 30ms is a real timing
  invariant (timer must not have fired) — kept as-is with an explanatory
  comment.  The final assert_eq!(count, 1) replaced with wait_for.

derived_caches_unchanged
  Test race: DerivedActor::started() spawns a task to compute the
  initial UpdateCache.  If the test's Subscribe(counter) message is
  processed by DerivedActor *before* that warm-up UpdateCache arrives,
  the actor's cache is still None when UpdateCache(2) arrives later —
  triggering a dirty→idle→Notify cycle that increments count to 1
  (expected 0).
  Fixed by polling two consecutive d.get().await calls until they return
  Arc::ptr_eq (both served from self.cached, not fresh computes).  Once
  ptr_eq holds, self.cached is definitely Some(2), so any subsequent
  UpdateCache(2) — including any late warm-up — deduplicates to no-op.
  The counter is then subscribed and Set(4) tested.

derived_notifies_on_value_change
  Removed redundant 50ms warm-up sleep; uses wait_for for the positive
  "subscriber notified" assertion (count >= 1) instead.

ask_dead_actor_returns_closed / send_dead_actor_returns_send_error
/ recipient_dead_actor / system_shutdown_terminates_ctx_spawned_child
  Test race: system.shutdown().await returns when done_rx fires (at
  end of run_mailbox), but the tokio task hasn't yet dropped the
  Receiver<T>, so tx.is_closed() (== is_alive()) can still return
  false for a brief window.  Fixed by polling !addr.is_alive() with
  1ms retries (max 200ms) before asserting channel-closed semantics.

Verification:
- Each of the 3 originally-reported flaky tests: 30/30 green in isolation
- cargo test -p willow-actor --lib: 5/5 consecutive runs, 94/94 tests each
- just check: zero warnings, all tests green

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: intendednull

crates/actor/src/debounce.rs

@@ -159,6 +159,25 @@ mod tests {
     use std::sync::atomic::{AtomicU32, Ordering};
     use std::sync::Arc;
 
+    // Poll `condition` every 10 ms until it returns true or `timeout_ms`
+    // elapses. Panics with `msg` on timeout.
+    //
+    // Use this for POSITIVE "X eventually happens" assertions in tests that
+    // depend on async delivery — actor timers and message routing can be
+    // slower than the test's wall clock under parallel suite load.
+    async fn wait_for(condition: impl Fn() -> bool, timeout_ms: u64, msg: &str) {
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(timeout_ms);
+        loop {
+            if condition() {
+                return;
+            }
+            if std::time::Instant::now() >= deadline {
+                panic!("wait_for timeout after {timeout_ms}ms: {msg}");
+            }
+            runtime::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    }
+
     #[derive(Clone)]
     struct Ping(u32);
     impl Message for Ping {
@@ -207,7 +226,15 @@ mod tests {
         let debounce = system.spawn(Debounce::new(collector.into(), Duration::from_millis(50)));
 
         debounce.do_send(Enqueue(Ping(1))).unwrap();
-        runtime::sleep(Duration::from_millis(100)).await;
+
+        // Wait for the debounce flush to arrive — actor timer + message
+        // delivery can be slower than fixed sleeps under parallel suite load.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "debounce should have forwarded exactly one message",
+        )
+        .await;
 
         assert_eq!(count.load(Ordering::SeqCst), 1);
         assert_eq!(*last.lock().unwrap(), Some(1));
@@ -223,7 +250,14 @@ mod tests {
         for i in 0..5 {
             debounce.do_send(Enqueue(Ping(i))).unwrap();
         }
-        runtime::sleep(Duration::from_millis(150)).await;
+
+        // Wait for the single coalesced flush to arrive.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "debounce should have forwarded exactly one coalesced message",
+        )
+        .await;
 
         assert_eq!(count.load(Ordering::SeqCst), 1);
         assert_eq!(*last.lock().unwrap(), Some(4));
@@ -237,11 +271,27 @@ mod tests {
         let debounce = system.spawn(Debounce::new(collector.into(), Duration::from_millis(60)));
 
         debounce.do_send(Enqueue(Ping(1))).unwrap();
+        // Sleep for half the debounce window, then reset the timer with Ping(2).
         runtime::sleep(Duration::from_millis(30)).await;
         debounce.do_send(Enqueue(Ping(2))).unwrap();
         runtime::sleep(Duration::from_millis(30)).await;
-        assert_eq!(count.load(Ordering::SeqCst), 0);
-        runtime::sleep(Duration::from_millis(50)).await;
+        // Debounce window restarted from Ping(2); 30ms << 60ms — must not fire.
+        // This is an intentional timing invariant: fixed sleep is correct here
+        // because we're asserting a "not yet" condition within a known timed window.
+        assert_eq!(
+            count.load(Ordering::SeqCst),
+            0,
+            "timer should not have fired yet"
+        );
+
+        // Wait for the eventual flush via condition-based polling.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "debounce should have forwarded Ping(2) after quiet period",
+        )
+        .await;
+
         assert_eq!(count.load(Ordering::SeqCst), 1);
         assert_eq!(*last.lock().unwrap(), Some(2));
         system.shutdown().await;
@@ -253,13 +303,26 @@ mod tests {
         let (collector, count, _) = setup_collector(&system);
         let debounce = system.spawn(Debounce::new(collector.into(), Duration::from_millis(30)));
 
+        // First burst — wait for it to flush before sending the second burst,
+        // so the two bursts are truly separate and each yields exactly one flush.
         debounce.do_send(Enqueue(Ping(1))).unwrap();
         debounce.do_send(Enqueue(Ping(2))).unwrap();
-        runtime::sleep(Duration::from_millis(80)).await;
-
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "first burst should have flushed",
+        )
+        .await;
+
+        // Second burst — wait for its flush too.
         debounce.do_send(Enqueue(Ping(3))).unwrap();
         debounce.do_send(Enqueue(Ping(4))).unwrap();
-        runtime::sleep(Duration::from_millis(80)).await;
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 2,
+            2000,
+            "second burst should have flushed",
+        )
+        .await;
 
         assert_eq!(count.load(Ordering::SeqCst), 2);
         system.shutdown().await;
@@ -272,7 +335,15 @@ mod tests {
         let throttle = system.spawn(Throttle::new(collector.into(), Duration::from_millis(100)));
 
         throttle.do_send(Enqueue(Ping(1))).unwrap();
-        runtime::sleep(Duration::from_millis(20)).await;
+
+        // Throttle forwards the first message immediately (before any cooldown
+        // fires). Wait for it to arrive rather than guessing a fixed sleep.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "throttle should forward the first message immediately",
+        )
+        .await;
 
         assert_eq!(count.load(Ordering::SeqCst), 1);
         assert_eq!(*last.lock().unwrap(), Some(1));
@@ -286,12 +357,29 @@ mod tests {
         let throttle = system.spawn(Throttle::new(collector.into(), Duration::from_millis(100)));
 
         throttle.do_send(Enqueue(Ping(1))).unwrap();
-        runtime::sleep(Duration::from_millis(10)).await;
+        // First message is forwarded immediately. Wait for it so the cooldown
+        // window is known to have started before we queue more messages.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "throttle should forward first message",
+        )
+        .await;
+
         throttle.do_send(Enqueue(Ping(2))).unwrap();
         throttle.do_send(Enqueue(Ping(3))).unwrap();
-        runtime::sleep(Duration::from_millis(20)).await;
 
-        assert_eq!(count.load(Ordering::SeqCst), 1);
+        // Cooldown is 100ms. Sleep for 20ms — well inside the cooldown window —
+        // and assert count is still 1. This is an intentional timing invariant:
+        // the cooldown MUST suppress extra messages during this window. A fixed
+        // sleep is correct here because we're asserting a "not yet" condition
+        // within a known timed window (20ms << 100ms cooldown).
+        runtime::sleep(Duration::from_millis(20)).await;
+        assert_eq!(
+            count.load(Ordering::SeqCst),
+            1,
+            "throttle should suppress messages during cooldown"
+        );
         system.shutdown().await;
     }
 
@@ -302,9 +390,24 @@ mod tests {
         let throttle = system.spawn(Throttle::new(collector.into(), Duration::from_millis(50)));
 
         throttle.do_send(Enqueue(Ping(1))).unwrap();
-        runtime::sleep(Duration::from_millis(10)).await;
+        // Wait for first forward so we know the cooldown has started before
+        // queuing Ping(2).
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "throttle should forward first message",
+        )
+        .await;
+
         throttle.do_send(Enqueue(Ping(2))).unwrap();
-        runtime::sleep(Duration::from_millis(100)).await;
+
+        // After cooldown expires, pending Ping(2) should be forwarded.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 2,
+            2000,
+            "throttle should forward pending message after cooldown",
+        )
+        .await;
 
         assert_eq!(count.load(Ordering::SeqCst), 2);
         assert_eq!(*last.lock().unwrap(), Some(2));
@@ -318,10 +421,24 @@ mod tests {
         let throttle = system.spawn(Throttle::new(collector.into(), Duration::from_millis(50)));
 
         throttle.do_send(Enqueue(Ping(1))).unwrap();
-        runtime::sleep(Duration::from_millis(10)).await;
+        // Wait for first forward so cooldown is active before queuing.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 1,
+            2000,
+            "throttle should forward first message",
+        )
+        .await;
+
         throttle.do_send(Enqueue(Ping(2))).unwrap();
         throttle.do_send(Enqueue(Ping(3))).unwrap();
-        runtime::sleep(Duration::from_millis(100)).await;
+
+        // After cooldown, only the last pending (Ping(3)) should be forwarded.
+        wait_for(
+            || count.load(Ordering::SeqCst) >= 2,
+            2000,
+            "throttle should forward only latest pending message after cooldown",
+        )
+        .await;
 
         assert_eq!(count.load(Ordering::SeqCst), 2);
         assert_eq!(*last.lock().unwrap(), Some(3));