diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md index 6ca3e58..058f6b7 100644 --- a/docs/replay-clock-source.md +++ b/docs/replay-clock-source.md @@ -24,12 +24,13 @@ counter identity to order emitted events. | 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 | | 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 | | 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 supported boundary rules | +| 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 | | 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 | -| Signal timeout 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 timeout result, other 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 @@ -131,3 +132,37 @@ time invalidates the consumed boundary instead of borrowing the previous one. Grouped conditions and their dedicated prefix algorithms remain outside this scalar cohort. Canonical Rust occurrence fields are existing parser authority; this source adds no occurrence field or producer representation to Core. + +Persisted parallel Activity `all` follows [Core's descriptor aggregation](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L1916) +and [the accepted cross-SDK decision](https://github.com/durable-workflow/.github/issues/215#issuecomment-6677422213). +Only consumed members of that authored group participate. Each needs a unique +ordered schedule/terminal lifecycle, explicit numeric counter, exact type, +execution identity and complete matching group path. Nested all-activity paths +retain authored descriptor order. Duplicate or ambiguous evidence preserves +legacy results but cannot supply time. + +Success folds members in that order, comparing Carbon milliseconds and replacing +only on a strictly greater value. The first descriptor wins an equal-millisecond +tie, retaining its full recorded microsecond precision. Each consumed success +must have represented `recorded_at` or Core's actual `created_at` fallback. +Scalar redrive reuse timestamps, a later stream row, unconsumed groups and an +unrepresented activity row never supply authority. Empty groups add no boundary; +pending groups and other operation families have no aggregate clock here. + +Failure preserves Rust's existing lowest-member exception. Time is available +only when that member also wins Core's earliest-millisecond/member-index rule; +it is reconstructed with Core's millisecond precision. Conflicting multifailure +selection remains the [OPEN compatibility/versioning proposal](https://github.com/durable-workflow/.github/issues/215#issuecomment-6677866585), +without rewriting legacy outcomes. Grouped snapshot and reconstructed-query time +remain unavailable: Core's successful query group-send supplies no aggregate. +A later consumed supported scalar boundary can establish query time again. + +The measured reference in `tests/fixtures/replay-clock-carbon-reference.json` +uses PHP 8.4.27 and Carbon 3.15.1 at `ce0e5636`. Its +`Timestamp.php` SHA-256 is `57b2a235656f8edad7c333087a64e45491adaaf1b01ac9c40e53185b73db356d`. +`getTimestampMs()` rounds to nearest milliseconds, with halfway values away from +zero, including negative epochs. `createFromTimestampMs()` reconstructs those +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. diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index a4e9103..66fc1d8 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -74,6 +74,9 @@ }, { "glob": "src/replay_clock/condition.rs" + }, + { + "glob": "src/replay_clock/parallel_activity.rs" } ] } diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py index 51d7b1b..9739021 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_parallel_activity_clock_module_requires_replay_evidence(self) -> None: + source = "src/replay_clock/parallel_activity.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/parallel_activity.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/lib.rs b/src/lib.rs index 61f7b3f..2e6a323 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9682,7 +9682,7 @@ impl Worker { }; query_authority::before_handler(authority, &self.client).await?; let mut invocation = { - let _clock = replay_clock::Guard::enter(&workflow_context); + let _clock = replay_clock::Guard::query(&workflow_context); replay(workflow_context.clone(), workflow_input_typed.clone()) } .map_err(|error| { @@ -9694,7 +9694,7 @@ impl Worker { })?; let outcome = { let mut cx = TaskContext::from_waker(noop_waker_ref()); - let _clock = replay_clock::Guard::enter(&workflow_context); + let _clock = replay_clock::Guard::query(&workflow_context); let _replay = cancellation_replay_clock::ReplayGuard::enter(&workflow_context) .map_err(|error| { QueryTaskExecutionFailure::new( diff --git a/src/replay_clock.rs b/src/replay_clock.rs index fc7b44f..4df81d2 100644 --- a/src/replay_clock.rs +++ b/src/replay_clock.rs @@ -7,32 +7,46 @@ use super::{ }; use chrono::{DateTime, Utc}; use std::{ - cell::RefCell, + cell::{Cell, RefCell}, collections::HashMap, sync::{Arc, Mutex, Weak}, }; mod condition; +mod parallel_activity; thread_local! { static ACTIVE: RefCell>>> = const { RefCell::new(None) }; + static QUERY_REPLAY: Cell = const { Cell::new(false) }; } -pub(super) struct Guard(Option>>); +pub(super) struct Guard(Option>>, bool); impl Guard { pub(super) fn enter(context: &WorkflowContext) -> Self { - Self(ACTIVE.with(|active| active.replace(Some(Arc::downgrade(&context.state))))) + Self::scope(Some(Arc::downgrade(&context.state)), false) + } + + pub(super) fn query(context: &WorkflowContext) -> Self { + Self::scope(Some(Arc::downgrade(&context.state)), true) } pub(super) fn unavailable() -> Self { - Self(ACTIVE.with(|active| active.replace(None))) + Self::scope(None, false) + } + + fn scope(context: Option>>, query: bool) -> Self { + Self( + ACTIVE.with(|active| active.replace(context)), + QUERY_REPLAY.with(|active| active.replace(query)), + ) } } impl Drop for Guard { fn drop(&mut self) { ACTIVE.with(|active| active.replace(self.0.take())); + QUERY_REPLAY.with(|active| active.set(self.1)); } } @@ -76,6 +90,7 @@ fn start_time(history: &[HistoryEvent]) -> Option> { pub(super) struct Clock { time: Option>, boundaries: HashMap, + parallel_activities: parallel_activity::Members, } enum BoundaryKind { @@ -365,6 +380,7 @@ impl Clock { Self { time: start_time(history), boundaries, + parallel_activities: parallel_activity::Members::new(history, commands), } } @@ -393,7 +409,17 @@ impl WorkflowState { } pub(super) fn consume_replay_group(&mut self, sequences: &[u64], failed: Option) { - self.replay_clock.unavailable(); + // Core's query group-send branch supplies no aggregate time. Do not + // transfer executor authority into reconstructed query execution. + self.replay_clock.time = if QUERY_REPLAY.with(Cell::get) { + None + } else { + self.replay_clock.parallel_activities.consumed_time( + &self.history_events, + sequences, + failed, + ) + }; self.advance_cancellation_sequences(sequences, failed); } @@ -448,8 +474,8 @@ impl WorkflowContext { /// is available only while this workflow's handler/future is executing. /// /// This source cohort supports correlated scalar activity/timer/child/signal - /// results and the explicit redrive start anchor. Ordinary start authority, - /// inline metadata/local completions and grouped/other operation families + /// results, persisted Activity all groups, and the explicit redrive start anchor. Ordinary start authority, + /// inline metadata/local completions and other grouped operation families /// remain unavailable. Cancellation's existing budget clock is unchanged. /// Migrate business decisions through normal workflow versioning. Event /// timestamps do not provide a total order; retain stream/counter identity. diff --git a/src/replay_clock/parallel_activity.rs b/src/replay_clock/parallel_activity.rs new file mode 100644 index 0000000..51b551a --- /dev/null +++ b/src/replay_clock/parallel_activity.rs @@ -0,0 +1,207 @@ +use super::{ + durable_event_sequence, parse, DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value, +}; +use crate::{recorded_parallel_group_path, ParallelGroupMetadata}; +use std::collections::HashSet; + +#[derive(Debug)] +struct Member { + path: Vec, + execution: String, + terminal: Option, + failed: bool, +} + +#[derive(Debug)] +pub(super) struct Members(HashMap); + +fn all_activity(path: &[ParallelGroupMetadata]) -> bool { + !path.is_empty() + && path.iter().all(|entry| { + entry.parallel_group_kind == "activity" + && entry + .parallel_group_mode + .as_deref() + .is_none_or(|mode| mode == "all") + && entry.selection_member_key.is_none() + && entry.selection_member_index.is_none() + && entry.selection_member_base_sequence.is_none() + && entry.selection_member_size.is_none() + && entry.selection_member_kind.is_none() + }) +} + +fn member( + history: &[HistoryEvent], + indices: &[usize], + sequence: u64, + path: &[ParallelGroupMetadata], +) -> Option { + let events: Vec<_> = indices.iter().map(|&index| &history[index]).collect(); + let schedules: Vec<_> = events + .iter() + .filter(|event| event.event_type == "ActivityScheduled") + .collect(); + let [schedule] = schedules.as_slice() else { + return None; + }; + let execution = schedule + .payload + .get("activity_execution_id")? + .as_str() + .filter(|s| !s.is_empty())?; + let activity = schedule + .payload + .get("activity_type")? + .as_str() + .filter(|s| !s.is_empty())?; + let mut terminal = None; + for &index in indices { + let event = &history[index]; + if event.payload.get("sequence").and_then(Value::as_u64) != Some(sequence) + || event + .payload + .get("workflow_sequence") + .is_some_and(|value| value.as_u64() != Some(sequence)) + || event + .payload + .get("activity_execution_id") + .and_then(Value::as_str) + != Some(execution) + || event.payload.get("activity_type").and_then(Value::as_str) != Some(activity) + || recorded_parallel_group_path(&[event], sequence) + .ok() + .flatten() + .as_deref() + != Some(path) + { + return None; + } + if matches!( + event.event_type.as_str(), + "ActivityCompleted" | "ActivityFailed" | "ActivityTimedOut" | "ActivityCancelled" + ) { + // Legacy identical delivery may retain the result, but two rows + // cannot certify a unique consumed event's time. + if terminal.replace(index).is_some() || index <= *indices.first()? { + return None; + } + } + } + if history[*indices.first()?].event_type != "ActivityScheduled" { + return None; + } + Some(Member { + path: path.to_vec(), + execution: execution.into(), + terminal, + failed: terminal.is_some_and(|index| history[index].event_type != "ActivityCompleted"), + }) +} + +// Carbon 3.15.1 Timestamp::getTimestampMs calls round(microseconds / 1000), +// not truncation. Only accept the range where its float microsecond input is +// exact, and the persisted microsecond precision the Core carrier represents. +pub(super) fn milliseconds(time: DateTime) -> Option { + let micros = time.timestamp_micros(); + if micros.unsigned_abs() > (1_u64 << 53) || time.timestamp_subsec_nanos() % 1000 != 0 { + return None; + } + Some((micros as f64 / 1000.0).round() as i64) +} + +fn time(event: &HistoryEvent) -> Option> { + match event.raw.get("recorded_at") { + None | Some(Value::Null) => parse(event.raw.get("created_at")), + present => parse(present), + } +} + +impl Members { + pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self { + let mut events: HashMap> = HashMap::new(); + for (index, event) in history.iter().enumerate() { + if event.event_type.starts_with("Activity") { + if let Some(sequence) = durable_event_sequence(event) { + events.entry(sequence).or_default().push(index); + } + } + } + Self( + commands + .iter() + .filter_map(|command| { + let RecordedCommand::Activity { + sequence, + parallel_group_path: Some(path), + .. + } = command + else { + return None; + }; + if !all_activity(path) { + return None; + } + member(history, events.get(sequence)?, *sequence, path) + .map(|member| (*sequence, member)) + }) + .collect(), + ) + } + + pub(super) fn consumed_time( + &self, + history: &[HistoryEvent], + sequences: &[u64], + failed: Option, + ) -> Option> { + let root = self.0.get(sequences.first()?)?.path.first()?; + if root.parallel_group_base_sequence != *sequences.first()? + || root.parallel_group_size != sequences.len() + { + return None; + } + let mut executions = HashSet::new(); + let mut success: Option<(i64, DateTime)> = None; + let mut failure: Option<(i64, u64)> = None; + for (index, sequence) in sequences.iter().enumerate() { + let member = self.0.get(sequence)?; + let mut expected = root.clone(); + expected.parallel_group_index = index; + if member.path.first()? != &expected + || root + .parallel_group_base_sequence + .checked_add(index as u64)? + != *sequence + || !executions.insert(&member.execution) + { + return None; + } + if member.failed { + let ms = milliseconds(time(history.get(member.terminal?)?)?)?; + // Descriptor iteration retains the first member on a tie. + if failure.is_none_or(|(current, _)| ms < current) { + failure = Some((ms, *sequence)); + } + } else if failed.is_none() { + let time = time(history.get(member.terminal?)?)?; + let ms = milliseconds(time)?; + if success.is_none_or(|(current, _)| ms > current) { + success = Some((ms, time)); + } + } + } + match failed { + Some(selected) => { + let (ms, core_selected) = failure?; + // Keep the legacy exception even when Core selects a different + // failed member. That compatibility gap has no clock authority. + (selected == core_selected) + .then(|| DateTime::from_timestamp_millis(ms)) + .flatten() + } + None if failure.is_none() => success.map(|(_, time)| time), + None => None, + } + } +} diff --git a/src/replay_clock/tests.rs b/src/replay_clock/tests.rs index b1b7064..45cecf2 100644 --- a/src/replay_clock/tests.rs +++ b/src/replay_clock/tests.rs @@ -518,3 +518,5 @@ mod child; mod signal; mod condition; + +mod parallel_activity; diff --git a/src/replay_clock/tests/parallel_activity.rs b/src/replay_clock/tests/parallel_activity.rs new file mode 100644 index 0000000..4e97494 --- /dev/null +++ b/src/replay_clock/tests/parallel_activity.rs @@ -0,0 +1,379 @@ +use super::*; +use crate::{ParallelOperation, ParallelResult}; + +fn operations(nested: bool) -> Vec { + let a = ParallelOperation::activity("clock-a", json!([])); + let b = ParallelOperation::activity("clock-b", json!([])); + if nested { + vec![a, ParallelOperation::group(vec![b])] + } else { + vec![a, b] + } +} + +fn group(base: u64, kinds: [&str; 2], times: [&str; 2], nested: bool) -> Vec { + let descriptors = crate::parallel_descriptors(operations(nested), base).unwrap(); + let mut schedules = Vec::new(); + let mut terminals = Vec::new(); + for (index, descriptor) in descriptors.iter().enumerate() { + let sequence = base + index as u64; + let mut payload = json!({"sequence":sequence,"activity_execution_id":format!("execution-{sequence}"),"activity_type":if index==0 {"clock-a"} else {"clock-b"}}); + crate::apply_parallel_group_path(payload.as_object_mut().unwrap(), &descriptor.group_path); + schedules.push(event("ActivityScheduled", payload.clone(), START)); + if !kinds[index].is_empty() { + if kinds[index] == "ActivityCompleted" { + payload["result"] = encode_typed_envelope( + &AvroValue::String(format!("result-{index}")), + DEFAULT_CODEC, + ) + .unwrap(); + } else { + payload["message"] = json!(format!("failed-{index}")); + } + terminals.push(event(kinds[index], payload, times[index])); + } + } + schedules.extend(terminals); + schedules +} + +fn observe(history: Vec, nested: bool) -> Value { + let ctx = context(history); + let inner = ctx.clone(); + let output = ready(&ctx,Box::pin(async move { + let result = inner.parallel(operations(nested)).await; + let failed_index = match result { + Ok(results) => { + assert_eq!(results[0], ParallelResult::Activity(json!("result-0"))); + None + }, + Err(Error::ParallelFailed(failure)) => Some(failure.member_path), + other => panic!("unexpected group outcome: {other:?}"), + }; + let time=inner.replay_now().ok(); + assert_eq!(time,inner.replay_now().ok()); + AvroValue::from_serialize(&json!({"time":time.map(|t|t.to_rfc3339_opts(chrono::SecondsFormat::Micros,true)),"failed":failed_index})) + })).into_json().unwrap(); + assert!(ctx.take_commands().unwrap().is_empty()); + assert!(ctx.replay_now().is_err()); + output +} + +#[test] +fn replay_now_group_success_uses_core_ms_comparator_first_tie_and_full_precision() { + for nested in [false, true] { + for (times, expected) in [ + ( + ["2026-01-01T00:00:05.123100Z", "2026-01-01T00:00:05.123400Z"], + "2026-01-01T00:00:05.123100Z", + ), + ( + ["2026-01-01T00:00:05.123499Z", "2026-01-01T00:00:05.123500Z"], + "2026-01-01T00:00:05.123500Z", + ), + ( + ["1969-12-31T23:59:59.999500Z", "1969-12-31T23:59:59.999501Z"], + "1969-12-31T23:59:59.999501Z", + ), + ( + ["2026-01-01T00:00:05.123400Z", "2026-01-01T00:00:03.123499Z"], + "2026-01-01T00:00:05.123400Z", + ), + ] { + let mut history = vec![start()]; + history.extend(group(1, ["ActivityCompleted"; 2], times, nested)); + // Stream order is not descriptor order or business-time order. + history.swap(3, 4); + let first = observe(history.clone(), nested); + assert_eq!(first["time"], json!(expected)); + assert_eq!(first, observe(history, nested)); + } + } +} + +#[test] +fn replay_now_group_failure_preserves_legacy_winner_and_core_ms_reconstruction() { + for kind in ["ActivityFailed", "ActivityTimedOut", "ActivityCancelled"] { + let history = group( + 1, + [kind, "ActivityCompleted"], + ["2026-01-01T00:00:05.123500Z", "2099-01-01T00:00:00Z"], + false, + ); + let output = observe(history.clone(), false); + assert_eq!(output["time"], json!("2026-01-01T00:00:05.124000Z")); + assert_eq!(output["failed"], json!([0])); + assert_eq!(output, observe(history, false)); + } + for (times, expected) in [ + ( + ["1970-01-01T00:00:00.000500Z", "1970-01-01T00:00:00.001000Z"], + Some("1970-01-01T00:00:00.001000Z"), + ), + ( + ["1969-12-31T23:59:59.999500Z", "1970-01-01T00:00:00Z"], + Some("1969-12-31T23:59:59.999000Z"), + ), + (["2026-01-01T00:00:10Z", "2026-01-01T00:00:05Z"], None), + ] { + let output = observe(group(1, ["ActivityFailed"; 2], times, false), false); + assert_eq!( + output["failed"], + json!([0]), + "clock must not rewrite legacy failure selection" + ); + assert_eq!(output["time"], json!(expected)); + } +} + +#[test] +fn replay_now_group_time_is_scoped_to_consumed_group_not_future_or_reuse() { + let mut history = group( + 1, + ["ActivityCompleted"; 2], + ["2026-01-01T00:00:05.123100Z", "2026-01-01T00:00:05.123400Z"], + false, + ); + for event in &mut history[2..] { + event.payload["reused_recorded_at"] = json!("2099-01-01T00:00:00Z"); + } + history.extend(group( + 3, + ["ActivityCompleted"; 2], + ["2099-01-01T00:00:00Z"; 2], + false, + )); + assert_eq!( + observe(history, false)["time"], + json!("2026-01-01T00:00:05.123100Z") + ); +} + +#[test] +fn replay_now_group_requires_unique_numeric_correlated_lifecycle_even_if_legacy_result_survives() { + let original = group(1, ["ActivityCompleted"; 2], [START; 2], false); + for index in 0..4 { + for field in ["activity_execution_id", "activity_type"] { + let mut history = original.clone(); + history[index] + .payload + .as_object_mut() + .unwrap() + .remove(field); + // Legacy can infer a missing identity from the other lifecycle row. + assert!(observe(history, false)["time"].is_null()); + } + let mut history = original.clone(); + history[index].payload["sequence"] = json!(if index % 2 == 0 { "1" } else { "2" }); + assert!(observe(history, false)["time"].is_null()); + let mut history = original.clone(); + history[index].payload["workflow_sequence"] = json!(99); + assert!(observe(history, false)["time"].is_null()); + } + let mut history = original.clone(); + let mut duplicate = history[2].clone(); + duplicate.payload["sequence"] = json!("1"); + history.push(duplicate); + // String and numeric counters share the legacy index. Neither may hide a + // duplicate terminal row from the stricter business clock. + assert!(WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None).is_err()); + let mut history = original.clone(); + history.push(history[2].clone()); + assert!(observe(history, false)["time"].is_null()); + let mut history = original; + history.swap(0, 2); + assert!(observe(history, false)["time"].is_null()); +} + +#[test] +fn replay_now_group_requires_all_success_times_and_never_substitutes_other_authority() { + for value in [ + None, + Some(Value::Null), + Some(json!("invalid")), + Some(json!(42)), + ] { + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + match value { + Some(value) => { + history[2].raw.insert("recorded_at".into(), value); + } + None => { + history[2].raw.remove("recorded_at"); + } + } + history[2].payload["reused_recorded_at"] = json!(START); + assert!(observe(history, false)["time"].is_null()); + } + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + history[2].raw.remove("recorded_at"); + history[2] + .raw + .insert("created_at".into(), json!("2026-01-01T00:00:10.123456Z")); + assert_eq!( + observe(history, false)["time"], + json!("2026-01-01T00:00:10.123456Z") + ); + for time in ["2500-01-01T00:00:00Z", "2026-01-01T00:00:00.123456789Z"] { + assert!( + observe(group(1, ["ActivityCompleted"; 2], [time; 2], false), false)["time"].is_null() + ); + } +} + +#[test] +fn replay_now_group_pending_and_empty_paths_do_not_invent_an_aggregate() { + let ctx = context(group(1, ["ActivityCompleted", ""], [START; 2], false)); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner.parallel(operations(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()); + drop(_guard); + let ctx = context(vec![start()]); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner.parallel(vec![]).await?; + assert_eq!(read(&inner), START); + Ok(AvroValue::Null) + }), + ); +} + +#[tokio::test] +async fn replay_now_group_query_does_not_borrow_executor_aggregate() { + let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests"); + worker.register_replayed_workflow( + "group-clock", + || 0_u64, + |ctx, _, _| async move { + ctx.parallel(operations(false)).await?; + assert!( + ctx.replay_now().is_err(), + "query reconstruction has no Core group send time" + ); + ctx.sleep(Duration::from_secs(1)).await?; + Ok(Value::Null) + }, + ); + worker.register_query("group-clock", "snapshot", |ctx, _| async move { + Ok(json!(ctx.replay_now().is_err())) + }); + worker.register_replayed_query::( + "group-clock", + "reconstructed", + |ctx, _, _| async move { Ok(json!(ctx.replay_now().is_err())) }, + ); + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + for query in ["snapshot", "reconstructed"] { + if query == "reconstructed" { + history.push(event( + "TimerScheduled", + json!({"sequence":3,"timer_id":"later","delay_seconds":1}), + "2099-01-01T00:00:00Z", + )); + } + let task:QueryTask=serde_json::from_value(json!({"query_task_id":"group-q","workflow_type":"group-clock","query_name":query,"payload_codec":DEFAULT_CODEC,"history_events":history})).unwrap(); + assert_eq!( + worker + .execute_query_task(task) + .await + .unwrap() + .into_json() + .unwrap(), + json!(true) + ); + } + // Query execution must restore the ordinary executor's ambient mode. + assert!( + observe(group(1, ["ActivityCompleted"; 2], [START; 2], false), false)["time"].is_string() + ); +} + +#[test] +fn replay_now_group_carbon_reference_matches_measured_rounding_and_reconstruction() { + let reference: Value = serde_json::from_str(include_str!( + "../../../tests/fixtures/replay-clock-carbon-reference.json" + )) + .unwrap(); + for row in reference["vectors"].as_array().unwrap() { + let time = DateTime::parse_from_rfc3339(row["input"].as_str().unwrap()) + .unwrap() + .with_timezone(&Utc); + let ms = super::super::parallel_activity::milliseconds(time).unwrap(); + assert_eq!(json!(ms), row["milliseconds"]); + assert_eq!( + DateTime::::from_timestamp_millis(ms) + .unwrap() + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true), + row["reconstructed"].as_str().unwrap() + ); + } +} + +#[test] +fn replay_now_group_refuses_reused_execution_identity_and_keeps_legacy_group_shape_checks() { + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + for event in &mut history { + event.payload["activity_execution_id"] = json!("one-foreign-execution"); + } + assert!(observe(history, false)["time"].is_null()); + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + history[2].payload["activity_execution_id"] = json!("wrong-execution"); + assert!(observe(history, false)["time"].is_null()); + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + for event in &mut history { + event.payload["parallel_group_index"] = json!(0); + event.payload["parallel_group_path"][0]["parallel_group_index"] = json!(0); + } + assert!(matches!( + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None), + Err(Error::NonDeterministicReplay(_)) + )); +} + +#[test] +fn replay_now_mixed_activity_timer_group_does_not_borrow_activity_only_aggregation() { + let operations = || { + vec![ + ParallelOperation::activity("clock-a", json!([])), + ParallelOperation::timer(Duration::from_secs(1)), + ] + }; + let descriptors = crate::parallel_descriptors(operations(), 1).unwrap(); + let mut history = group(1, ["ActivityCompleted"; 2], [START; 2], false); + for index in [0, 2] { + crate::apply_parallel_group_path( + history[index].payload.as_object_mut().unwrap(), + &descriptors[0].group_path, + ); + } + for (index, kind) in [(1, "TimerScheduled"), (3, "TimerFired")] { + history[index] = event( + kind, + json!({"sequence":2,"timer_id":"mixed-timer","delay_seconds":1}), + START, + ); + crate::apply_parallel_group_path( + history[index].payload.as_object_mut().unwrap(), + &descriptors[1].group_path, + ); + } + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert_eq!(inner.parallel(operations()).await?.len(), 2); + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); +} diff --git a/tests/fixtures/replay-clock-carbon-reference.json b/tests/fixtures/replay-clock-carbon-reference.json new file mode 100644 index 0000000..328f4d7 --- /dev/null +++ b/tests/fixtures/replay-clock-carbon-reference.json @@ -0,0 +1,93 @@ +{ + "php": "8.4.27", + "carbon": "3.15.1", + "source": "ce0e5636fae7d1e99856eb70cae4c719a284aa08", + "timestamp_source_sha256": "57b2a235656f8edad7c333087a64e45491adaaf1b01ac9c40e53185b73db356d", + "vectors": [ + { + "input": "1970-01-01T00:00:00.000000Z", + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000499Z", + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000500Z", + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1970-01-01T00:00:00.000501Z", + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1969-12-31T23:59:59.999501Z", + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1969-12-31T23:59:59.999500Z", + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T23:59:59.999499Z", + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T23:59:59.998500Z", + "milliseconds": -2, + "reconstructed": "1969-12-31T23:59:59.998000Z" + }, + { + "input": "2026-01-01T00:00:00.123100Z", + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123400Z", + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123499Z", + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123500Z", + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "input": "2026-01-01T00:00:00.123501Z", + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "input": "2026-01-01T00:00:00.999500Z", + "milliseconds": 1767225601000, + "reconstructed": "2026-01-01T00:00:01.000000Z" + }, + { + "input": "1900-01-01T00:00:00.123499Z", + "milliseconds": -2208988799877, + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "input": "1900-01-01T00:00:00.123500Z", + "milliseconds": -2208988799877, + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T01:30:00.123500+01:30", + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + } + ] +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-parallel-activity.json b/tests/fixtures/replay-regressions/business-replay-clock-parallel-activity.json new file mode 100644 index 0000000..3f01d6b --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-parallel-activity.json @@ -0,0 +1,251 @@ +{ + "$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-parallel-activity", + "protocol_version": "1.20", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.business-replay-clock-parallel-activity", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "clock-a", + "activity_execution_id": "clock-execution-1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "clock-b", + "activity_execution_id": "clock-execution-2", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 2 + }, + { + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "clock-a", + "activity_execution_id": "clock-execution-1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoGb25l" + } + }, + "recorded_at": "2026-01-01T00:00:05.123100Z", + "sequence": 3 + }, + { + "event_type": "ActivityCompleted", + "payload": { + "sequence": 2, + "activity_type": "clock-b", + "activity_execution_id": "clock-execution-2", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoGb25l" + } + }, + "recorded_at": "2026-01-01T00:00:05.123400Z", + "sequence": 4 + }, + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 3, + "activity_type": "clock-a", + "activity_execution_id": "clock-execution-3", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 5 + }, + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 4, + "activity_type": "clock-b", + "activity_execution_id": "clock-execution-4", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 6 + }, + { + "event_type": "ActivityFailed", + "payload": { + "sequence": 3, + "activity_type": "clock-a", + "activity_execution_id": "clock-execution-3", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "message": "recorded group failure" + }, + "recorded_at": "2026-01-01T00:00:03.123500Z", + "sequence": 7 + }, + { + "event_type": "ActivityCompleted", + "payload": { + "sequence": 4, + "activity_type": "clock-b", + "activity_execution_id": "clock-execution-4", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoGb25l" + } + }, + "recorded_at": "2099-01-01T00:00:00Z", + "sequence": 8 + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "success_time": "2026-01-01T00:00:05.123100Z", + "failure_time": "2026-01-01T00:00:03.124000Z", + "failure_member": [ + 0 + ], + "unchanged_prefix": true + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "success_time": "2026-01-01T00:00:05.123100Z", + "failure_time": "2026-01-01T00:00:03.124000Z", + "failure_member": [ + 0 + ], + "unchanged_prefix": true + } + } + ], + "worker_task": { + "workflow_id": "parallel-activity-clock-parent", + "payload_codec": "avro" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 89becb4..86e3604 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-parallel-activity" | "corpus.nexus-result" | "corpus.nexus-cancellation" ) { @@ -628,6 +629,9 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< "corpus.business-replay-clock-signal" => { business_clock::register_signal(&mut worker, workflow_type) } + "corpus.business-replay-clock-parallel-activity" => { + business_clock::register_parallel_activity(&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 9933a42..fa1a39a 100644 --- a/tests/replay_regression_corpus/business_clock.rs +++ b/tests/replay_regression_corpus/business_clock.rs @@ -112,3 +112,30 @@ pub(super) fn register_condition(worker: &mut Worker, workflow_type: &str) { })) }); } + +pub(super) fn register_parallel_activity(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + let operations = || { + vec![ + ParallelOperation::activity("clock-a", json!([])), + ParallelOperation::activity("clock-b", json!([])), + ] + }; + let entry_unavailable = ctx.replay_now().is_err(); + ctx.parallel(operations()).await?; + let success = ctx.replay_now().ok(); + let Err(Error::ParallelFailed(failure)) = ctx.parallel(operations()).await else { + return Err(Error::WorkerLoop( + "expected recorded parallel failure".into(), + )); + }; + let failed = ctx.replay_now().ok(); + Ok(json!({ + "entry_unavailable":entry_unavailable, + "success_time":success.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "failure_time":failed.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "failure_member":failure.member_path, + "unchanged_prefix":failed.is_some() && failed==ctx.replay_now().ok(), + })) + }); +}