Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 28 additions & 27 deletions crates/tracedecay/src/daemon/invocation_tests/types_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Loading