diff --git a/crates/tracedecay/src/daemon/invocation_tests/types_tests.rs b/crates/tracedecay/src/daemon/invocation_tests/types_tests.rs index a812192f7a..b9fe8aa12f 100644 --- a/crates/tracedecay/src/daemon/invocation_tests/types_tests.rs +++ b/crates/tracedecay/src/daemon/invocation_tests/types_tests.rs @@ -557,33 +557,34 @@ async fn replayed_admission_after_retryable_failure_completes_a_fresh_cycle() { "the failed cycle must not acknowledge the durable admission" ); - // The spool consumer replays the identical envelope until the producer - // work genuinely terminates. A replay that races the settling failure - // joins it and is dropped with it, so the consumer's next pass retries. - let replay_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); - while acknowledged.load(std::sync::atomic::Ordering::Relaxed) == 0 { - assert!( - std::time::Instant::now() < replay_deadline, - "a replayed admission must complete a fresh cycle after a retryable failure" - ); - let replay_acknowledged = Arc::clone(&acknowledged); - assert_eq!( - admit_registered_hook_orchestration( - envelope.clone(), - binding.clone(), - Some(hook_lifecycle()), - 1, - false, - Some(Arc::new(move || { - replay_acknowledged.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - })), - ), - HookOrchestrationAdmissionV1::Enqueued - ); - tokio::time::sleep(std::time::Duration::from_millis(5)).await; - } - assert!( - attempts.load(std::sync::atomic::Ordering::Relaxed) >= 2, + // One replay of the settled failure. Replaying again while that cycle is + // still running would only join it, and past the coalescing bound a join + // is backpressured by design. + let replay_acknowledged = Arc::clone(&acknowledged); + let replay_settled = Arc::new(tokio::sync::Notify::new()); + let replay_terminal = Arc::clone(&replay_settled); + assert_eq!( + admit_registered_hook_orchestration( + envelope.clone(), + binding.clone(), + Some(hook_lifecycle()), + 1, + false, + Some(Arc::new(move || { + replay_acknowledged.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + replay_terminal.notify_one(); + })), + ), + HookOrchestrationAdmissionV1::Enqueued, + "the settled failure must free its slot for the replay" + ); + tokio::time::timeout(std::time::Duration::from_secs(5), replay_settled.notified()) + .await + .expect("a replayed admission must complete a fresh cycle after a retryable failure"); + assert_eq!(acknowledged.load(std::sync::atomic::Ordering::Relaxed), 1); + assert_eq!( + attempts.load(std::sync::atomic::Ordering::Relaxed), + 2, "acknowledgement requires a fresh successful cycle, not the settled failure" ); unregister_hook_orchestration_runtime([23; 16], [25; 16], &runtime);