diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md index 7397ef0..b9cb778 100644 --- a/docs/replay-clock-source.md +++ b/docs/replay-clock-source.md @@ -1,4 +1,4 @@ -# Additive replay clock: first Rust source cohort +# Additive replay clock: Rust source cohorts This source checkpoint implements `WorkflowContext::replay_now()` and `QueryContext::replay_now()` toward [the cross-SDK replay-clock feature](https://github.com/durable-workflow/.github/issues/215). @@ -21,12 +21,13 @@ counter identity to order emitted events. | Ordinary entry | Core uses original `WorkflowRun.started_at`; current Rust worker task has no such carrier | Unavailable, even if the start history row has an insertion timestamp | | Scalar remote activity | Existing validated command sequence/type/result correlation, then the selected terminal event | Selected recorded time; successful redrive reuse retains `reused_recorded_at`; the existing Core failure branch uses its event time | | Scalar plain timer | Existing validated `TimerScheduled`/`TimerFired` pair, exact sequence/ID/delay | Matched fire time, with recorded reuse provenance | -| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable | +| Scalar child result | One schedule, optional start and unique parent `ChildRunCompleted` / `ChildRunFailed` / `ChildRunCancelled` / `ChildRunTerminated`, with matching nonempty child call, instance, run and type identities | Parent resolution event time for both send and throw; child `closed_at` and `reused_recorded_at` are not this branch's clock authority | +| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child 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 supported boundary rules | | Direct update handler | Unique committed `UpdateAccepted` correlated to the handler's update ID | That event's validated time; missing/ambiguous acceptance is unavailable | | Executor condition predicate | Current executor exposes a full signal/update view without retaining the exact triggering consumed prefix | Unavailable inside the predicate; enclosing workflow scope is restored on return/error | | Inline side effect/version/memo/search/local result | User code can resume before its event is persisted | Unavailable on first execution and the corresponding cold path in this cohort; no implicit marker/yield/fence | -| Child/signal/condition result, grouped/selection/Nexus or cancellation callback | Their dedicated clock algorithms are outside this initial cohort | Explicitly unavailable; existing execution/cancellation behavior remains intact | +| Signal/condition result, grouped/selection/Nexus or cancellation callback | Their dedicated clock algorithms are outside this initial cohort | Explicitly unavailable; existing execution/cancellation behavior remains intact | The mutable workflow clock is bound only during handler construction and each future poll. A separate execution-local guard restores its predecessor on @@ -57,3 +58,22 @@ also consumed as source evidence, without inventing its ordinary start anchor. Normal current-head checks remain required. Published cross-SDK conformance, producer implementation/adoption, release and guide/localized delivery remain separate work on the owning issue. + +The scalar child cohort follows Core's persisted branches at +[`4b2cdc55` `WorkflowExecutor::resumeChildResolution`](https://github.com/durable-workflow/workflow/blob/4b2cdc5507120d9c213b0b02a85738bec96c574a/src/V2/Support/WorkflowExecutor.php#L1638) +and the corresponding `QueryStateReplayer` branch. Both completion and failure +resume at the selected parent resolution event's own `recorded_at`. Unlike the +successful activity/timer reuse path, they do not select `reused_recorded_at`. +The child row's `closed_at` is not a substitute. Rust consumes the existing +validated scalar command before advancing this clock. Identical duplicate +terminal payloads, missing/conflicting lifecycle identities, changing-run +retry/continuation lineage, out-of-order lifecycle or invalid/missing time keep +business time unavailable without changing legacy result/failure behavior. +A schedule without a resolution stays pending and does not advance time. + +Focused child controls exercise actual calls and caught terminal failures, +unchanged-prefix and cold replay, backward/precise times, future-row isolation, +identity/time refusals, snapshot/reconstructed query prefixes and interleaved +poll/error scope restoration. The separate child corpus consumer runs unchanged +against base and candidate production code, retaining an exact behavioral +negative rather than a compilation failure. diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index ad23d07..77837aa 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -68,6 +68,9 @@ }, { "glob": "src/cancellation_replay_clock.rs" + }, + { + "glob": "src/replay_clock.rs" } ] } diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py index 05135da..884bf91 100644 --- a/scripts/ci/test-validate-regression-corpus.py +++ b/scripts/ci/test-validate-regression-corpus.py @@ -627,6 +627,25 @@ def test_neutral_edit_inside_replay_function_is_related(self) -> None: self.assertTrue(self._guard_matches(REPLAY_GUARD)) + def test_business_clock_module_requires_replay_evidence(self) -> None: + source = "src/replay_clock.rs" + (self.root / source).write_text( + "fn child_boundary(history: &[usize]) -> Option { history.last().copied() }\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) + + # Reproduce the original module-coverage gap with the predecessor set. + 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/child.rs"}, guard) + for guard in guards + )) + def test_history_fixture_in_unchanged_test_module_is_not_production_replay(self) -> None: diff = """\ diff --git a/src/lib.rs b/src/lib.rs diff --git a/src/replay_clock.rs b/src/replay_clock.rs index a08db43..7f3dcc7 100644 --- a/src/replay_clock.rs +++ b/src/replay_clock.rs @@ -76,6 +76,54 @@ pub(super) struct Clock { boundaries: HashMap, } +enum BoundaryKind { + Activity, + Timer, + Child, +} + +fn child_boundary(history: &[HistoryEvent], indices: &[usize]) -> Option { + // A scalar call has one schedule, an optional start, then one resolution. + // Legacy replay can accept duplicate payloads, but their time is ambiguous. + let (&scheduled, &terminal) = match indices { + [scheduled, terminal] => (scheduled, terminal), + [scheduled, started, terminal] if history[*started].event_type == "ChildRunStarted" => { + (scheduled, terminal) + } + _ => return None, + }; + let schedule = &history[scheduled]; + if schedule.event_type != "ChildWorkflowScheduled" + || !matches!( + history[terminal].event_type.as_str(), + "ChildRunCompleted" | "ChildRunFailed" | "ChildRunCancelled" | "ChildRunTerminated" + ) + { + return None; + } + for key in [ + "child_call_id", + "child_workflow_instance_id", + "child_workflow_run_id", + "child_workflow_type", + ] { + let identity = schedule + .payload + .get(key)? + .as_str() + .filter(|id| !id.is_empty())?; + if indices + .iter() + .any(|&index| history[index].payload.get(key).and_then(Value::as_str) != Some(identity)) + { + // Changed-run retry/continuation needs its own lineage algorithm. + // Missing identities keep legacy result handling, without time. + return None; + } + } + Some(terminal) +} + impl Clock { pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self { let supported: HashMap<_, _> = commands @@ -87,7 +135,7 @@ impl Clock { parallel_group_path, .. } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { - Some((*sequence, true)) + Some((*sequence, BoundaryKind::Activity)) } RecordedCommand::Timer { sequence, @@ -95,31 +143,55 @@ impl Clock { parallel_group_path, .. } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { - Some((*sequence, false)) + Some((*sequence, BoundaryKind::Timer)) + } + RecordedCommand::ChildWorkflow { + sequence, + outcome: Some(_), + parallel_group_path, + .. + } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { + Some((*sequence, BoundaryKind::Child)) } _ => None, }) .collect(); let mut boundaries = HashMap::new(); + let mut children: HashMap> = HashMap::new(); for (index, event) in history.iter().enumerate() { let Some(sequence) = durable_event_sequence(event) else { continue; }; let matched = match supported.get(&sequence) { - Some(true) => matches!( + Some(BoundaryKind::Activity) => matches!( event.event_type.as_str(), "ActivityCompleted" | "ActivityFailed" | "ActivityTimedOut" | "ActivityCancelled" ), - Some(false) => event.event_type == "TimerFired" && !is_internal_timer_event(event), + Some(BoundaryKind::Timer) => { + event.event_type == "TimerFired" && !is_internal_timer_event(event) + } + Some(BoundaryKind::Child) => { + if event.event_type == "ChildWorkflowScheduled" + || event.event_type.starts_with("ChildRun") + { + children.entry(sequence).or_default().push(index); + } + false + } None => false, }; if matched { boundaries.entry(sequence).or_insert(index); } } + for (sequence, indices) in children { + if let Some(index) = child_boundary(history, &indices) { + boundaries.insert(sequence, index); + } + } // Index identities once. Timestamps are read only when that boundary // is actually consumed, never by scanning ahead from the accessor. Self { @@ -207,7 +279,7 @@ impl WorkflowContext { /// or invalid evidence returns [`Error::ReplayClockUnavailable`]. The clock /// is available only while this workflow's handler/future is executing. /// - /// This initial source cohort supports correlated scalar activity/timer + /// This source cohort supports correlated scalar activity/timer/child /// results and the explicit redrive start anchor. Ordinary start authority, /// inline metadata/local completions and grouped/other operation families /// remain unavailable. Cancellation's existing budget clock is unchanged. @@ -247,6 +319,10 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option Vec { + let mut payload = json!({ + "sequence":sequence, + "child_call_id":format!("call-{sequence}"), + "child_workflow_instance_id":format!("child-{sequence}"), + "child_workflow_run_id":format!("run-{sequence}"), + "child_workflow_type":"clock-child", + }); + let schedule = event("ChildWorkflowScheduled", payload.clone(), START); + let started = event("ChildRunStarted", payload.clone(), START); + if kind == "ChildRunCompleted" { + payload["result"] = + encode_typed_envelope(&AvroValue::String("child-result".into()), DEFAULT_CODEC) + .unwrap(); + } else { + payload["message"] = json!("child failure"); + } + // Core's scalar child branches use the parent resolution record, not the + // child's close time or the reused-time rule used for successful activities. + payload["closed_at"] = json!("2098-01-01T00:00:00Z"); + payload["reused_recorded_at"] = json!("2099-01-01T00:00:00Z"); + vec![schedule, started, event(kind, payload, time)] +} + +#[test] +fn replay_now_child_completion_and_caught_failure_cold_replay_exact_times() { + for failed_time in ["2025-12-31T23:59:59.123456Z", CHILD_TIME] { + let mut history = vec![start()]; + history.extend(child(1, "ChildRunCompleted", CHILD_TIME)); + history.extend(child(2, "ChildRunFailed", failed_time)); + let execute = || { + let ctx = context(history.clone()); + let inner = ctx.clone(); + let result = ready( + &ctx, + Box::pin(async move { + assert_eq!(read(&inner), START); + let result = inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + assert_eq!(result.child.run_id.as_deref(), Some("run-1")); + assert_eq!(result.result, json!("child-result")); + let completed = read(&inner); + assert_eq!(completed, read(&inner)); + assert_eq!(completed, CHILD_TIME); + let Err(Error::ChildWorkflowFailed(failure)) = inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await + else { + panic!("expected the recorded child failure"); + }; + assert_eq!(failure.kind, ChildWorkflowFailureKind::Failed); + assert_eq!(failure.child_workflow_run_id.as_deref(), Some("run-2")); + Ok(AvroValue::from_serialize( + &json!({"completed":completed,"failed":read(&inner)}), + )?) + }), + ); + assert_eq!(ctx.state.lock().unwrap().command_cursor, 2); + assert!(ctx.take_commands().unwrap().is_empty()); + assert!(ctx.replay_now().is_err()); + result + }; + let first = execute(); + assert_eq!(first, execute()); + assert_eq!(first.into_json().unwrap()["failed"], failed_time); + } +} + +#[test] +fn replay_now_child_terminal_categories_use_their_own_parent_record() { + for kind in [ + "ChildRunCompleted", + "ChildRunFailed", + "ChildRunCancelled", + "ChildRunTerminated", + ] { + let mut history = vec![start()]; + history.extend(child(1, kind, CHILD_TIME)); + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + let outcome = inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await; + match kind { + "ChildRunCompleted" => assert!(outcome.is_ok()), + _ => assert!(matches!(outcome, Err(Error::ChildWorkflowFailed(_)))), + } + assert_eq!(read(&inner), CHILD_TIME); + Ok(AvroValue::Null) + }), + ); + } +} + +#[test] +fn replay_now_child_missing_or_conflicting_identity_preserves_legacy_result_without_time() { + for field in [ + "child_call_id", + "child_workflow_instance_id", + "child_workflow_run_id", + ] { + for index in 0..3 { + for replacement in [ + None, + Some(json!("foreign-child")), + Some(json!("")), + Some(Value::Null), + Some(json!(17)), + ] { + let mut history = child(1, "ChildRunCompleted", CHILD_TIME); + match replacement { + Some(value) => { + history[index].payload[field] = value; + } + None => { + history[index] + .payload + .as_object_mut() + .unwrap() + .remove(field); + } + } + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert_eq!( + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]) + ) + .await? + .result, + json!("child-result") + ); + assert!(matches!( + inner.replay_now(), + Err(Error::ReplayClockUnavailable) + )); + Ok(AvroValue::Null) + }), + ); + assert!(ctx.take_commands().unwrap().is_empty()); + } + } + } + let mut missing_type = child(1, "ChildRunCompleted", CHILD_TIME); + for event in &mut missing_type { + event + .payload + .as_object_mut() + .unwrap() + .remove("child_workflow_type"); + } + let ctx = context(missing_type); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); +} + +#[test] +fn replay_now_child_invalid_time_and_duplicate_terminal_do_not_borrow_other_authority() { + for bad_time in [ + None, + Some(Value::Null), + Some(json!("invalid")), + Some(json!(42)), + ] { + let mut history = vec![start()]; + history.extend(child(1, "ChildRunCompleted", CHILD_TIME)); + match bad_time { + Some(value) => { + history[3].raw.insert("recorded_at".into(), value); + } + None => { + history[3].raw.remove("recorded_at"); + } + } + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); + } + let mut history = child(1, "ChildRunCompleted", CHILD_TIME); + history.remove(1); // Legacy duplicate delivery without a separate start record. + let mut duplicate = history.last().unwrap().clone(); + duplicate + .raw + .insert("recorded_at".into(), json!("2099-01-01T00:00:00Z")); + history.push(duplicate); + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); +} + +#[test] +fn replay_now_child_future_result_is_not_consumed_and_pending_schedule_has_no_time() { + let mut history = vec![start()]; + history.extend(child(1, "ChildRunCompleted", CHILD_TIME)); + history.extend(child(2, "ChildRunCompleted", "2099-01-01T00:00:00Z")); + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + assert_eq!(read(&inner), CHILD_TIME); + std::future::pending().await + }); + assert!(Worker::poll_workflow_future(&ctx, &mut future) + .unwrap() + .is_pending()); + assert_eq!(ctx.state.lock().unwrap().command_cursor, 1); + assert!(ctx.replay_now().is_err()); + let mut history = vec![start()]; + history.extend( + child(1, "ChildRunCompleted", CHILD_TIME) + .into_iter() + .take(2), + ); + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + panic!("a pending child must stay pending"); + }); + assert!(Worker::poll_workflow_future(&ctx, &mut future) + .unwrap() + .is_pending()); + let _guard = Guard::enter(&ctx); + assert_eq!(read(&ctx), START); + assert!(ctx.take_commands().unwrap().is_empty()); +} + +#[test] +fn replay_now_child_poll_scope_restores_outer_context_after_failure_and_interleaving() { + let outer = context(vec![start()]); + let _outer_guard = Guard::enter(&outer); + let mut futures = vec![]; + for time in [CHILD_TIME, "2026-01-01T00:00:02.654321Z"] { + let ctx = context(child(1, "ChildRunCompleted", time)); + let inner = ctx.clone(); + let future: WorkflowFuture = Box::pin(async move { + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + assert_eq!(read(&inner), time); + tokio::task::yield_now().await; + assert_eq!(read(&inner), time); + Err(Error::WorkerLoop("expected workflow failure".into())) + }); + futures.push((ctx, future)); + } + for (ctx, future) in &mut futures { + assert!(Worker::poll_workflow_future(ctx, future) + .unwrap() + .is_pending()); + assert!(ctx.replay_now().is_err()); + assert_eq!(read(&outer), START); + } + for (ctx, future) in &mut futures { + assert!(matches!( + Worker::poll_workflow_future(ctx, future).unwrap(), + Poll::Ready(Err(Error::WorkerLoop(_))) + )); + assert!(ctx.replay_now().is_err()); + assert_eq!(read(&outer), START); + } +} + +#[tokio::test] +async fn replay_now_child_snapshot_and_reconstructed_queries_use_their_actual_prefix() { + let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests"); + worker.register_replayed_workflow( + "clock-child-parent", + || 0_u64, + |ctx, _args, _state| async move { + ctx.start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + ctx.start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + Ok(Value::Null) + }, + ); + worker.register_query("clock-child-parent", "snapshot", |ctx, _args| async move { + Ok(json!(ctx.replay_now().ok().map( + |time| time.to_rfc3339_opts(chrono::SecondsFormat::Micros, true) + ))) + }); + worker.register_replayed_query::( + "clock-child-parent", + "reconstructed", + |ctx, _state, _args| async move { + Ok(json!(ctx.replay_now().ok().map( + |time| time.to_rfc3339_opts(chrono::SecondsFormat::Micros, true) + ))) + }, + ); + let mut history = child(1, "ChildRunCompleted", CHILD_TIME); + for (query, expected) in [ + ("snapshot", json!(CHILD_TIME)), + ("reconstructed", json!(CHILD_TIME)), + ] { + if query == "reconstructed" { + history.extend( + child(2, "ChildRunCompleted", "2099-01-01T00:00:00Z") + .into_iter() + .take(2), + ); + } + let task: QueryTask=serde_json::from_value(json!({"query_task_id":"child-query","workflow_type":"clock-child-parent","query_name":query,"payload_codec":DEFAULT_CODEC,"history_events":history})).unwrap(); + assert_eq!( + worker + .execute_query_task(task) + .await + .unwrap() + .into_json() + .unwrap(), + expected + ); + } + assert!(snapshot_time(&history, DEFAULT_CODEC).is_none()); +} + +#[test] +fn replay_now_child_lifecycle_order_and_command_correlation_are_required() { + let mut history = child(1, "ChildRunCompleted", CHILD_TIME); + history.swap(0, 2); + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert_eq!( + inner + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]) + ) + .await? + .result, + json!("child-result") + ); + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); + let mut history = vec![start()]; + history.extend(child(1, "ChildRunCompleted", CHILD_TIME)); + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert!(matches!( + inner + .start_child_workflow( + "foreign-type", + ChildWorkflowOptions::new("children"), + json!([]) + ) + .await, + Err(Error::NonDeterministicReplay(_)) + )); + assert_eq!(read(&inner), START); + assert_eq!(inner.state.lock().unwrap().command_cursor, 0); + Ok(AvroValue::Null) + }), + ); + for (field, value) in [ + ("sequence", json!(2)), + ("child_workflow_type", json!("foreign-type")), + ] { + let mut history = child(1, "ChildRunCompleted", CHILD_TIME); + history[2].payload[field] = value; + assert!( + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None).is_err() + ); + } +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-child-results.json b/tests/fixtures/replay-regressions/business-replay-clock-child-results.json new file mode 100644 index 0000000..2d9cf5a --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-child-results.json @@ -0,0 +1,107 @@ +{ + "$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-child-results", + "protocol_version": "1.19", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.business-replay-clock-child", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "workflow_type": "corpus.business-replay-clock-child" + }, + "recorded_at": "2026-01-01T00:00:00Z" + }, + { + "sequence": 2, + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 1, + "child_call_id": "clock-call-1", + "child_workflow_instance_id": "clock-child-1", + "child_workflow_run_id": "clock-child-run-1", + "child_workflow_type": "clock-child" + }, + "recorded_at": "2026-01-01T00:00:00Z" + }, + { + "sequence": 3, + "event_type": "ChildRunCompleted", + "payload": { + "sequence": 1, + "child_call_id": "clock-call-1", + "child_workflow_instance_id": "clock-child-1", + "child_workflow_run_id": "clock-child-run-1", + "child_workflow_type": "clock-child", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoQcmVjb3JkZWQ=" + } + }, + "recorded_at": "2026-01-01T00:00:05.654321Z" + }, + { + "sequence": 4, + "event_type": "ChildWorkflowScheduled", + "payload": { + "sequence": 2, + "child_call_id": "clock-call-2", + "child_workflow_instance_id": "clock-child-2", + "child_workflow_run_id": "clock-child-run-2", + "child_workflow_type": "clock-child" + }, + "recorded_at": "2026-01-01T00:00:00Z" + }, + { + "sequence": 5, + "event_type": "ChildRunFailed", + "payload": { + "sequence": 2, + "child_call_id": "clock-call-2", + "child_workflow_instance_id": "clock-child-2", + "child_workflow_run_id": "clock-child-run-2", + "child_workflow_type": "clock-child", + "message": "recorded child failure", + "failure_id": "clock-child-failure", + "failure_category": "child_workflow" + }, + "recorded_at": "2026-01-01T00:00:03.123456Z" + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "completed": "2026-01-01T00:00:05.654321Z", + "failed": "2026-01-01T00:00:03.123456Z", + "unchanged_prefix": true, + "result": "recorded", + "failure_reason": "child_workflow" + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "completed": "2026-01-01T00:00:05.654321Z", + "failed": "2026-01-01T00:00:03.123456Z", + "unchanged_prefix": true, + "result": "recorded", + "failure_reason": "child_workflow" + } + } + ], + "worker_task": { + "workflow_id": "child-clock-parent", + "payload_codec": "avro" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 5f428c3..d44fe2a 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -506,6 +506,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.child-policy-author" | "corpus.activity-policy-author" | "corpus.business-replay-clock" + | "corpus.business-replay-clock-child" | "corpus.nexus-result" | "corpus.nexus-cancellation" ) { @@ -622,6 +623,9 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< } match workflow_type { "corpus.business-replay-clock" => business_clock::register(&mut worker, workflow_type), + "corpus.business-replay-clock-child" => { + business_clock::register_child(&mut worker, workflow_type) + } "corpus.nexus-result" => { worker.register_workflow(workflow_type, |ctx, _| async move { ctx.call_nexus_service("endpoint", "service", "operation", json!(["world"])) diff --git a/tests/replay_regression_corpus/business_clock.rs b/tests/replay_regression_corpus/business_clock.rs index dc91bc7..9c9073e 100644 --- a/tests/replay_regression_corpus/business_clock.rs +++ b/tests/replay_regression_corpus/business_clock.rs @@ -32,3 +32,38 @@ pub(super) fn register(worker: &mut Worker, workflow_type: &str) { })) }); } + +pub(super) fn register_child(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + let entry_unavailable = ctx.replay_now().is_err(); + let child = ctx + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await?; + let completed = ctx.replay_now().ok(); + let outcome = ctx + .start_child_workflow( + "clock-child", + ChildWorkflowOptions::new("children"), + json!([]), + ) + .await; + let Err(Error::ChildWorkflowFailed(failure)) = outcome else { + return Err(Error::WorkerLoop( + "expected the recorded child failure".into(), + )); + }; + let failed = ctx.replay_now().ok(); + Ok(json!({ + "entry_unavailable":entry_unavailable, + "completed":completed.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "failed":failed.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "unchanged_prefix":failed.is_some() && failed==ctx.replay_now().ok(), + "result":child.result, + "failure_reason":failure.reason, + })) + }); +}