From 2e17ef99b850a1c09afdeca2e79e58b73530ab48 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:50:06 +0000 Subject: [PATCH] fix: correlate fired condition timeout replay time --- docs/replay-clock-source.md | 31 ++- regression-corpus-policy.json | 3 + scripts/ci/test-validate-regression-corpus.py | 17 ++ src/replay_clock.rs | 46 ++-- src/replay_clock/condition_timeout.rs | 146 ++++++++++ src/replay_clock/tests.rs | 2 + src/replay_clock/tests/condition_timeout.rs | 254 ++++++++++++++++++ ...ess-replay-clock-condition-fired-only.json | 86 ++++++ tests/replay_regression_corpus.rs | 4 + .../business_clock.rs | 20 ++ 10 files changed, 588 insertions(+), 21 deletions(-) create mode 100644 src/replay_clock/condition_timeout.rs create mode 100644 src/replay_clock/tests/condition_timeout.rs create mode 100644 tests/fixtures/replay-regressions/business-replay-clock-condition-fired-only.json diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md index 058f6b7..ec97d86 100644 --- a/docs/replay-clock-source.md +++ b/docs/replay-clock-source.md @@ -25,6 +25,7 @@ counter identity to order emitted events. | Scalar persisted signal delivery | One `SignalWaitOpened` followed by a unique `SignalApplied`, matching nonempty wait ID/name and explicit command sequence, with the persisted value consumed. A configured timeout requires valid numeric metadata and either no scheduled timer or the exact ordered schedule/cancel pair | Applied event time, preserving valid `reused_recorded_at` exactly as Core signal resume does | | Scalar persisted condition outcome | One ordered canonical wait open/resolution, matching physical wait, logical occurrence, explicit command counter, key/fingerprint and timeout definition | Consumed satisfied/timed-out resolution event time, preserving its precision; timer and scalar redrive helpers are not substituted | | Persisted remote Activity all group | Exact authored group/path and each consumed member lifecycle/type/sequence/execution identity | Core millisecond aggregate on success, full chosen precision; exact agreed failure at Core millisecond precision | +| Persisted scalar condition timeout without explicit resolution | One canonical wait and its exact ordered schedule/fire pair, with complete identity/definition/counters and unique physical timer | Executor uses the fired event’s own `recorded_at`; Core query has no corresponding resolution authority | | Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child/signal/condition correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable | | Replayed-instance query | Clock of the actual reconstructed workflow prefix, rather than a later visible row | Same scalar boundary rules; grouped executor aggregation is not query authority | | Direct update handler | Unique committed `UpdateAccepted` correlated to the handler's update ID | That event's validated time; missing/ambiguous acceptance is unavailable | @@ -122,8 +123,7 @@ numeric counters and complete matching definitions; legacy fallback parsing may still return a result when that time authority is missing. Each actual physical resolution in a logical reopen chain advances only when consumed. -A timeout fire without its persisted condition resolution supplies no new -condition clock. Core's live timeout branch has its own timer/fired-row time +The explicit resolution branch does not borrow a timeout fire's time. Core's live timeout branch has its own timer/fired-row time input; it is not inferred for Rust from an unrelated timer. Predicate callbacks still have explicit unavailability inside the invocation. Synchronous local true/zero-timeout paths preserve the already accepted previous-prefix behavior, @@ -166,3 +166,30 @@ milliseconds. The clock accepts persisted microseconds within the exact float microsecond range used by that comparator; finer or out-of-range evidence stays unavailable. Reference vectors cover positive/negative epochs, 499/500/501 microsecond edges, equal-millisecond candidates and timezone-equivalent instants. + +The persisted fired-only condition-timeout cohort follows Core +[`resumeTimedOutCondition`](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L895). +It consumes the selected `TimerFired.recorded_at`, including its precision. +The native producer already carries the matching physical/logical wait, +key/fingerprint, timer ID/kind, command sequence and delay through +[`wait_fields`](https://github.com/durable-workflow/server/blob/fadf8729265e077f280c8f71a87c06d8f73a682e/rust/src/runtime/wait_timeouts.rs#L38) +and the scheduled/fired history. It intentionally need not append a +`ConditionWaitTimedOut` event. This source change adds no field or producer. + +One ordered open/schedule/fire and complete matching metadata are required. +A cancelled, aliased, duplicate, reordered, superseded or otherwise ambiguous +physical wait cannot provide time. A logical reopen uses the final actually +consumed physical wait; an old orphan timeout cannot supply a pending reopen's +clock. Indexing reads identities once, and accessors only read the consumed +boundary. Payload `fired_at`, deadline, reuse, insertion aliases and an +unrepresented timer-row fallback do not replace the selected `recorded_at`. +Missing or invalid time is unavailable while legacy results remain unchanged. + +This fired-only authority is executor-specific. Current Core query replay stops +without an explicit condition resolution, so snapshot and reconstructed-query +clocks remain unavailable on this shape, including inside the reconstructed +workflow. Existing explicit persisted condition-resolution query behavior is +preserved. Grouped conditions, predicate-trigger retention and the separate PHP +occurrence-carrier gap remain open, as do published/native conformance and +release adoption. The corpus fixture is synthetic source evidence shaped from +the immutable writer, not a claim that a native experiment ran. diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index 66fc1d8..b46ba91 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -77,6 +77,9 @@ }, { "glob": "src/replay_clock/parallel_activity.rs" + }, + { + "glob": "src/replay_clock/condition_timeout.rs" } ] } diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py index 9739021..c77d21f 100644 --- a/scripts/ci/test-validate-regression-corpus.py +++ b/scripts/ci/test-validate-regression-corpus.py @@ -663,6 +663,23 @@ def matches(guard: dict[str, object]) -> bool: for guard in guards )) + def test_condition_timeout_clock_module_requires_replay_evidence(self) -> None: + source = "src/replay_clock/condition_timeout.rs" + path = self.root / source + path.parent.mkdir(parents=True) + path.write_text("fn boundary() -> Option { Some(0) }\n", encoding="utf-8") + guards = REPOSITORY_POLICY["categories"]["replay"]["guards"] + + def matches(guard: dict[str, object]) -> bool: + return VALIDATOR._guard_matches(self.root, "HEAD", {source}, guard) + + self.assertFalse(any(matches(guard) for guard in guards if guard["glob"] != source)) + self.assertTrue(any(matches(guard) for guard in guards)) + self.assertFalse(any( + VALIDATOR._guard_matches(self.root, "HEAD", {"src/replay_clock/tests/condition_timeout.rs"}, guard) + for guard in guards + )) + def test_parallel_activity_clock_module_requires_replay_evidence(self) -> None: source = "src/replay_clock/parallel_activity.rs" path = self.root / source diff --git a/src/replay_clock.rs b/src/replay_clock.rs index 4df81d2..1910428 100644 --- a/src/replay_clock.rs +++ b/src/replay_clock.rs @@ -13,6 +13,7 @@ use std::{ }; mod condition; +mod condition_timeout; mod parallel_activity; thread_local! { @@ -73,6 +74,26 @@ fn recorded_time(event: &HistoryEvent, reused: bool) -> Option> { ) } +fn boundary_time(event: &HistoryEvent, query: bool) -> Option> { + let internal = event.event_type == "TimerFired" && is_internal_timer_event(event); + if internal { + // Core query replay requires the explicit condition resolution. Its + // executor-only fired-time branch does not grant callback authority. + return if query { + None + } else { + parse(event.raw.get("recorded_at")) + }; + } + recorded_time( + event, + matches!( + event.event_type.as_str(), + "ActivityCompleted" | "TimerFired" | "SignalApplied" + ), + ) +} + fn start_time(history: &[HistoryEvent]) -> Option> { let mut starts = history .iter() @@ -372,8 +393,11 @@ impl Clock { boundaries.insert(sequence, index); } } + let condition_timeouts = condition_timeout::Index::new(history); boundaries.extend(conditions.into_iter().filter_map(|(sequence, indices)| { - condition::boundary(history, &indices, sequence).map(|index| (sequence, index)) + condition::boundary(history, &indices, sequence) + .or_else(|| condition_timeouts.boundary(history, &indices, sequence)) + .map(|index| (sequence, index)) })); // Index identities once. Timestamps are read only when that boundary // is actually consumed, never by scanning ahead from the accessor. @@ -453,15 +477,7 @@ impl WorkflowState { .boundaries .get(&sequence) .and_then(|index| self.history_events.get(*index)) - .and_then(|event| { - recorded_time( - event, - matches!( - event.event_type.as_str(), - "ActivityCompleted" | "TimerFired" | "SignalApplied" - ), - ) - }); + .and_then(|event| boundary_time(event, QUERY_REPLAY.with(Cell::get))); } } @@ -536,15 +552,7 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option Option> { diff --git a/src/replay_clock/condition_timeout.rs b/src/replay_clock/condition_timeout.rs new file mode 100644 index 0000000..fc97e35 --- /dev/null +++ b/src/replay_clock/condition_timeout.rs @@ -0,0 +1,146 @@ +use super::{HashMap, HistoryEvent, Value}; +use crate::recorded_parallel_group_path; + +fn counter(event: &HistoryEvent, sequence: u64) -> bool { + event.payload.get("sequence").and_then(Value::as_u64) == Some(sequence) + && event + .payload + .get("workflow_sequence") + .is_none_or(|value| value.as_u64() == Some(sequence)) +} + +pub(super) struct Index<'a> { + timers_by_wait: HashMap<&'a str, Vec>, + timers_by_id: HashMap<&'a str, Vec>, + conditions_by_wait: HashMap<&'a str, Vec>, + opens_by_occurrence: HashMap<&'a str, Vec>, +} + +fn link<'a>( + map: &mut HashMap<&'a str, Vec>, + event: &'a HistoryEvent, + key: &str, + position: usize, +) { + if let Some(value) = event.payload.get(key).and_then(Value::as_str) { + map.entry(value).or_default().push(position); + } +} + +impl<'a> Index<'a> { + pub(super) fn new(history: &'a [HistoryEvent]) -> Self { + let mut index = Self { + timers_by_wait: HashMap::new(), + timers_by_id: HashMap::new(), + conditions_by_wait: HashMap::new(), + opens_by_occurrence: HashMap::new(), + }; + for (position, event) in history.iter().enumerate() { + if event.event_type.starts_with("Timer") { + link( + &mut index.timers_by_wait, + event, + "condition_wait_id", + position, + ); + link(&mut index.timers_by_id, event, "timer_id", position); + } + if event.event_type.starts_with("ConditionWait") { + link( + &mut index.conditions_by_wait, + event, + "condition_wait_id", + position, + ); + } + if event.event_type == "ConditionWaitOpened" { + link( + &mut index.opens_by_occurrence, + event, + "condition_wait_occurrence_id", + position, + ); + } + } + index + } + + pub(super) fn boundary( + &self, + history: &[HistoryEvent], + indices: &[usize], + sequence: u64, + ) -> Option { + // A persisted explicit resolution has its own, different Core branch. + let [opened] = indices else { return None }; + let open = &history[*opened]; + if open.event_type != "ConditionWaitOpened" || !counter(open, sequence) { + return None; + } + let identity = [ + "condition_wait_id", + "condition_wait_occurrence_id", + "condition_key", + "condition_definition_fingerprint", + ]; + for key in identity { + open.payload + .get(key)? + .as_str() + .filter(|value| !value.is_empty())?; + } + let seconds = open.payload.get("timeout_seconds")?.as_u64()?; + let wait = open.payload.get("condition_wait_id")?.as_str()?; + if self.conditions_by_wait.get(wait)?.as_slice() != indices { + return None; + } + let timers = self.timers_by_wait.get(wait)?; + let [scheduled, fired] = timers.as_slice() else { + return None; + }; + if !(*opened < *scheduled && *scheduled < *fired) + || history[*scheduled].event_type != "TimerScheduled" + || history[*fired].event_type != "TimerFired" + { + return None; + } + let timer = history[*scheduled] + .payload + .get("timer_id")? + .as_str() + .filter(|value| !value.is_empty())?; + for &index in timers { + let event = &history[index]; + if !counter(event, sequence) + || event.payload.get("timer_id").and_then(Value::as_str) != Some(timer) + || event.payload.get("timer_kind").and_then(Value::as_str) + != Some("condition_timeout") + || event.payload.get("delay_seconds").and_then(Value::as_u64) != Some(seconds) + || recorded_parallel_group_path(&[event], sequence) + .ok()? + .is_some() + { + return None; + } + for key in identity { + if event.payload.get(key) != open.payload.get(key) { + return None; + } + } + } + // A mislabeled duplicate of this timer must not disappear into another + // wait's bucket. This is an identity refusal, never a future-time maximum. + if self.timers_by_id.get(timer)? != timers { + return None; + } + let occurrence = open.payload.get("condition_wait_occurrence_id")?.as_str()?; + let opens = self.opens_by_occurrence.get(occurrence)?; + let next = opens.partition_point(|&index| index <= *opened); + if opens.get(next).is_some_and(|&index| index < *fired) { + // A later physical open superseded this wait before its old timer + // fired. The legacy reopen resolver decides the final call outcome. + return None; + } + Some(*fired) + } +} diff --git a/src/replay_clock/tests.rs b/src/replay_clock/tests.rs index 45cecf2..fe88fc2 100644 --- a/src/replay_clock/tests.rs +++ b/src/replay_clock/tests.rs @@ -520,3 +520,5 @@ mod signal; mod condition; mod parallel_activity; + +mod condition_timeout; diff --git a/src/replay_clock/tests/condition_timeout.rs b/src/replay_clock/tests/condition_timeout.rs new file mode 100644 index 0000000..dff6346 --- /dev/null +++ b/src/replay_clock/tests/condition_timeout.rs @@ -0,0 +1,254 @@ +use super::*; + +const FIRED: &str = "2026-01-01T00:00:05.654321Z"; + +fn options(seconds: u64) -> ConditionWaitOptions { + ConditionWaitOptions::new("ready", "sha256:ready-v1").timeout(Duration::from_secs(seconds)) +} + +fn timeout(sequence: u64, ordinal: u64, seconds: u64, time: &str) -> Vec { + let mut payload = json!({"sequence":sequence,"condition_wait_id":format!("wait-{sequence}"),"condition_wait_occurrence_id":format!("rust:condition-wait:{ordinal}"),"condition_key":"ready","condition_definition_fingerprint":"sha256:ready-v1","timeout_seconds":seconds}); + let opened = event("ConditionWaitOpened", payload.clone(), START); + payload.as_object_mut().unwrap().remove("timeout_seconds"); + payload["timer_id"] = json!(format!("timeout-{sequence}")); + payload["timer_kind"] = json!("condition_timeout"); + payload["delay_seconds"] = json!(seconds); + payload["fire_at"] = json!("2026-01-01T00:00:30Z"); + let scheduled = event("TimerScheduled", payload.clone(), START); + if seconds == 0 { + payload.as_object_mut().unwrap().remove("fire_at"); + } + payload["fired_at"] = json!("2099-01-01T00:00:00Z"); + vec![opened, scheduled, event("TimerFired", payload, time)] +} + +fn observe(history: Vec, seconds: u64) -> Value { + let ctx = context(history); + let inner = ctx.clone(); + let output = ready( + &ctx, + Box::pin(async move { + let result = inner + .wait_condition(options(seconds), || { + panic!("persisted timeout must not evaluate a predicate") + }) + .await?; + assert_eq!(result, ConditionWaitResult::TimedOut); + let time = inner.replay_now().ok(); + assert_eq!(time, inner.replay_now().ok()); + AvroValue::from_serialize(&json!( + time.map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) + )) + }), + ) + .into_json() + .unwrap(); + assert!(ctx.take_commands().unwrap().is_empty()); + assert!(ctx.replay_now().is_err()); + output +} + +#[test] +fn replay_now_condition_fired_only_consumes_exact_event_with_cold_equal_and_backward_time() { + for seconds in [0, 30] { + for time in [START, FIRED, "2025-12-31T23:59:59.123456Z"] { + let mut history = vec![start()]; + history.extend(timeout(1, 0, seconds, time)); + assert_eq!(observe(history.clone(), seconds), json!(time)); + assert_eq!(observe(history.clone(), seconds), observe(history, seconds)); + } + } +} + +#[test] +fn replay_now_condition_timeout_ignores_payload_reuse_deadline_and_unconsumed_future_times() { + let mut history = timeout(1, 0, 30, FIRED); + history[2].payload["reused_recorded_at"] = json!("2099-01-01T00:00:00Z"); + history[2] + .raw + .insert("timestamp".into(), json!("2099-01-01T00:00:00Z")); + history.extend(timeout(2, 1, 30, "2099-01-01T00:00:00Z")); + assert_eq!(observe(history, 30), json!(FIRED)); + for time in [None, Some(Value::Null), Some(json!("bad")), Some(json!(42))] { + let mut history = timeout(1, 0, 30, FIRED); + match time { + Some(value) => { + history[2].raw.insert("recorded_at".into(), value); + } + None => { + history[2].raw.remove("recorded_at"); + } + } + history[2].raw.insert("created_at".into(), json!(FIRED)); + history[2].raw.insert("timestamp".into(), json!(FIRED)); + assert!(observe(history, 30).is_null()); + } +} + +#[test] +fn replay_now_condition_timeout_requires_complete_native_definition_and_explicit_counters() { + let original = timeout(1, 0, 30, FIRED); + for index in [1, 2] { + for field in [ + "condition_wait_occurrence_id", + "condition_key", + "condition_definition_fingerprint", + ] { + for replacement in [None, Some(json!("foreign")), Some(Value::Null)] { + let mut history = original.clone(); + match replacement { + Some(value) => { + history[index].payload[field] = value; + } + None => { + history[index] + .payload + .as_object_mut() + .unwrap() + .remove(field); + } + } + assert!(observe(history, 30).is_null()); + } + } + for sequence in [json!("1"), json!(2)] { + let mut history = original.clone(); + history[index].payload["sequence"] = sequence; + assert!(observe(history, 30).is_null()); + } + let mut history = original.clone(); + history[index].payload["workflow_sequence"] = json!(2); + assert!(observe(history, 30).is_null()); + } + let mut history = original; + history[0].payload["sequence"] = json!("1"); + assert!(observe(history, 30).is_null()); +} + +#[test] +fn replay_now_condition_timeout_preserves_legacy_timer_identity_and_duplicate_refusals() { + for mutation in ["duplicate", "timer_id", "delay"] { + let mut history = timeout(1, 0, 30, FIRED); + match mutation { + "duplicate" => history.push(history[2].clone()), + "timer_id" => history[2].payload["timer_id"] = json!("foreign"), + "delay" => history[2].payload["delay_seconds"] = json!(31), + _ => unreachable!(), + } + assert!(matches!( + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None), + Err(Error::NonDeterministicReplay(_)) + )); + } +} + +#[test] +fn replay_now_condition_timeout_refuses_cancelled_aliased_and_out_of_order_evidence() { + let original = timeout(1, 0, 30, FIRED); + let mut history = original.clone(); + let mut cancel = history[2].clone(); + cancel.event_type = "TimerCancelled".into(); + history.push(cancel); + assert!(observe(history, 30).is_null()); + let mut history = original.clone(); + let mut alias = history[2].clone(); + alias.payload["condition_wait_id"] = json!("foreign"); + history.push(alias); + assert!(observe(history, 30).is_null()); + let mut history = original.clone(); + history.swap(1, 2); + assert!(observe(history, 30).is_null()); + let mut history = original; + let mut other = history[0].clone(); + other.payload["sequence"] = json!(2); + other.payload["condition_wait_occurrence_id"] = json!("rust:condition-wait:1"); + history.push(other); + assert!(observe(history, 30).is_null()); +} + +#[test] +fn replay_now_condition_timeout_uses_final_physical_wait_not_an_orphaned_early_timeout() { + let first = timeout(1, 0, 30, "2099-01-01T00:00:00Z"); + let second = timeout(2, 0, 20, FIRED); + let history = vec![ + first[0].clone(), + first[1].clone(), + second[0].clone(), + first[2].clone(), + second[1].clone(), + second[2].clone(), + ]; + assert_eq!(observe(history.clone(), 30), json!(FIRED)); + let ctx = context(history[..4].to_vec()); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner.wait_condition(options(30), || Ok(false)).await?; + Ok(AvroValue::Null) + }); + assert!(Worker::poll_workflow_future(&ctx, &mut future) + .unwrap() + .is_pending()); + let _guard = Guard::enter(&ctx); + assert!(ctx.replay_now().is_err()); +} + +#[test] +fn replay_now_condition_timeout_group_stays_unavailable_after_actual_consumption() { + let mut history = timeout(1, 0, 30, FIRED); + for event in &mut history { + crate::apply_parallel_group_path( + event.payload.as_object_mut().unwrap(), + &[crate::parallel_group_entry(1, 1, 0, "condition")], + ); + } + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner + .parallel(vec![crate::ParallelOperation::condition( + options(30), + || Ok(false), + )]) + .await?; + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); +} + +#[tokio::test] +async fn replay_now_condition_fired_only_does_not_grant_snapshot_or_reconstructed_query_time() { + let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests"); + worker.register_replayed_workflow( + "timeout-clock", + || 0_u64, + |ctx, _, _| async move { + ctx.wait_condition(options(30), || Ok(false)).await?; + assert!(ctx.replay_now().is_err()); + Ok(Value::Null) + }, + ); + worker.register_query("timeout-clock", "snapshot", |ctx, _| async move { + Ok(json!(ctx.replay_now().is_err())) + }); + worker.register_replayed_query::( + "timeout-clock", + "reconstructed", + |ctx, _, _| async move { Ok(json!(ctx.replay_now().is_err())) }, + ); + for query in ["snapshot", "reconstructed"] { + let task:QueryTask=serde_json::from_value(json!({"query_task_id":"timeout-q","workflow_type":"timeout-clock","query_name":query,"payload_codec":DEFAULT_CODEC,"history_events":timeout(1,0,30,FIRED)})).unwrap(); + assert_eq!( + worker + .execute_query_task(task) + .await + .unwrap() + .into_json() + .unwrap(), + json!(true) + ); + } + assert_eq!(observe(timeout(1, 0, 30, FIRED), 30), json!(FIRED)); +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-condition-fired-only.json b/tests/fixtures/replay-regressions/business-replay-clock-condition-fired-only.json new file mode 100644 index 0000000..2940539 --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-condition-fired-only.json @@ -0,0 +1,86 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "rust-business-replay-clock-condition-fired-only", + "protocol_version": "1.20", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.business-replay-clock-condition-timeout", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "ConditionWaitOpened", + "payload": { + "sequence": 1, + "condition_wait_id": "native-shaped-condition", + "condition_wait_occurrence_id": "rust:condition-wait:0", + "condition_key": "ready", + "condition_definition_fingerprint": "sha256:ready-v1", + "timeout_seconds": 30 + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 1 + }, + { + "event_type": "TimerScheduled", + "payload": { + "sequence": 1, + "condition_wait_id": "native-shaped-condition", + "condition_wait_occurrence_id": "rust:condition-wait:0", + "condition_key": "ready", + "condition_definition_fingerprint": "sha256:ready-v1", + "timer_id": "native-shaped-timeout", + "timer_kind": "condition_timeout", + "delay_seconds": 30, + "fire_at": "2026-01-01T00:00:30.000000Z" + }, + "recorded_at": "2026-01-01T00:00:00.000001Z", + "sequence": 2 + }, + { + "event_type": "TimerFired", + "payload": { + "sequence": 1, + "condition_wait_id": "native-shaped-condition", + "condition_wait_occurrence_id": "rust:condition-wait:0", + "condition_key": "ready", + "condition_definition_fingerprint": "sha256:ready-v1", + "timer_id": "native-shaped-timeout", + "timer_kind": "condition_timeout", + "delay_seconds": 30, + "fire_at": "2026-01-01T00:00:30.000000Z", + "fired_at": "2026-01-01T00:00:30.123456Z" + }, + "recorded_at": "2026-01-01T00:00:30.654321Z", + "sequence": 3 + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "result": "TimedOut", + "time": "2026-01-01T00:00:30.654321Z", + "unchanged_prefix": true + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "result": "TimedOut", + "time": "2026-01-01T00:00:30.654321Z", + "unchanged_prefix": true + } + } + ], + "worker_task": { + "workflow_id": "condition-fired-clock-parent", + "payload_codec": "avro" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 86e3604..1e142ac 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -509,6 +509,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.business-replay-clock-child" | "corpus.business-replay-clock-signal" | "corpus.business-replay-clock-condition" + | "corpus.business-replay-clock-condition-timeout" | "corpus.business-replay-clock-parallel-activity" | "corpus.nexus-result" | "corpus.nexus-cancellation" @@ -632,6 +633,9 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< "corpus.business-replay-clock-parallel-activity" => { business_clock::register_parallel_activity(&mut worker, workflow_type) } + "corpus.business-replay-clock-condition-timeout" => { + business_clock::register_condition_timeout(&mut worker, workflow_type) + } "corpus.business-replay-clock-condition" => { business_clock::register_condition(&mut worker, workflow_type) } diff --git a/tests/replay_regression_corpus/business_clock.rs b/tests/replay_regression_corpus/business_clock.rs index fa1a39a..76afe3e 100644 --- a/tests/replay_regression_corpus/business_clock.rs +++ b/tests/replay_regression_corpus/business_clock.rs @@ -139,3 +139,23 @@ pub(super) fn register_parallel_activity(worker: &mut Worker, workflow_type: &st })) }); } + +pub(super) fn register_condition_timeout(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + let entry_unavailable = ctx.replay_now().is_err(); + let result = ctx + .wait_condition( + ConditionWaitOptions::new("ready", "sha256:ready-v1") + .timeout(Duration::from_secs(30)), + || Ok(false), + ) + .await?; + let time = ctx.replay_now().ok(); + Ok(json!({ + "entry_unavailable":entry_unavailable, + "result":format!("{result:?}"), + "time":time.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "unchanged_prefix":time.is_some() && time==ctx.replay_now().ok(), + })) + }); +}