diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md index ec97d86..a952dad 100644 --- a/docs/replay-clock-source.md +++ b/docs/replay-clock-source.md @@ -24,8 +24,9 @@ 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 | +| Persisted remote Activity all group | Exact authored group/path and each consumed member lifecycle/type/sequence/execution identity | Canonical v1 millisecond aggregate on success, full chosen precision; exact agreed failure at v1 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 | +| Persisted plain Timer all group | Exact authored paths/counters and one matching schedule/fire per member, with unique physical timer IDs | Canonical v1 consumed-group millisecond fold, first descriptor on ties, full selected event precision; no grouped query 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 | @@ -141,7 +142,7 @@ 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 +Success folds members in that order, comparing canonical v1 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. @@ -150,8 +151,8 @@ 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 +only when that member also wins the canonical v1 earliest-millisecond/member-index rule; +it is reconstructed with canonical v1 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. @@ -160,12 +161,12 @@ 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. +These are attributed legacy PHP 8.4/Carbon measurements, not a portable numeric +contract. The same Carbon version can produce different rounding results under +other PHP runtimes. Their original inputs and outputs remain unchanged. +The separately labeled `tests/fixtures/replay-clock-canonical-v1.json` defines +the new accessor's exact integer rule. Historical source qualifications are +retained with their stated scope; they are not universal Core parity evidence. The persisted fired-only condition-timeout cohort follows Core [`resumeTimedOutCondition`](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L895). @@ -193,3 +194,43 @@ 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. + +Persisted plain Timer `all` follows Core +[`dispatchParallelTimer`](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L2007). +Each authored descriptor needs one ordered schedule/fire, matching nonempty timer +ID, numeric command counters, delay and complete group path. Internal or unknown +timer kinds, cancellation, duplicates, aliases or unsupported time evidence do +not supply an aggregate. Nested all-timer paths preserve descriptor order. + +The fold uses the same canonical v1 integer comparator as Activity groups. It selects +only consumed group members, keeps the first descriptor on a rounded-ms tie, +and returns that member's full represented precision. Raw `recorded_at` or Core's +represented `created_at` fallback supplies time; scalar reuse helpers and payload +fired/deadline timestamps do not. Pending/empty behavior and grouped query +unavailability retain the existing rules. Mixed groups and other group families +remain outside this bounded source mapping. No marker, wire or producer change +is introduced, and no installed/native or published conformance is inferred. + +The unreleased accessor's [canonical v1 numeric decision](https://github.com/durable-workflow/.github/issues/215#issuecomment-6679568700) +uses an authoritative instant's exact signed UTC microsecond count `U`: +`M = sign(U) * ((abs(U) + 500) div 1000)`. All operations are checked integers; +there is no float conversion or runtime `round()`. A half millisecond rounds +away from zero on both sides of the epoch. Unsupported precision or overflow +is unavailable. This removes the earlier source implementation's float-range +restriction for valid representable instants, without changing scalar event +times or rewriting legacy group results, selected errors, cancellation or history. + +Success still compares only validated consumed authored members, preserves the +first member on an equal `M`, and returns that member's original full-precision +instant. Qualified failure reconstruction uses the selected integer `M` only +when it agrees with the unchanged legacy winner. The multi-failure policy gap +remains OPEN. The canonical table covers signed boundaries, offsets and large +representable dates; a new real worker fixture independently contrasts the old +source's unavailable large epochs against the exact v1 result. + +Core remains the identity/boundary/order reference. Existing Core legacy +`Workflow::now()` values are unchanged and are not claimed numerically equivalent +on every supported PHP runtime. A matching explicitly versioned additive-clock +source/context on Core and published cross-SDK conformance remain required. +Adoption uses ordinary workflow versioning. No automatic marker, run mutation, +new public API name, live wire advertisement or release is introduced here. diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json index b46ba91..535993e 100644 --- a/regression-corpus-policy.json +++ b/regression-corpus-policy.json @@ -80,6 +80,12 @@ }, { "glob": "src/replay_clock/condition_timeout.rs" + }, + { + "glob": "src/replay_clock/parallel_timer.rs" + }, + { + "glob": "src/replay_clock/parallel.rs" } ] } diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py index c77d21f..283d96f 100644 --- a/scripts/ci/test-validate-regression-corpus.py +++ b/scripts/ci/test-validate-regression-corpus.py @@ -697,6 +697,40 @@ def matches(guard: dict[str, object]) -> bool: for guard in guards )) + def test_parallel_timer_clock_module_requires_replay_evidence(self) -> None: + source = "src/replay_clock/parallel_timer.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_timer.rs"}, guard) + for guard in guards + )) + + def test_parallel_common_clock_module_requires_replay_evidence(self) -> None: + source = "src/replay_clock/parallel.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.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 1910428..29d6c5b 100644 --- a/src/replay_clock.rs +++ b/src/replay_clock.rs @@ -14,7 +14,9 @@ use std::{ mod condition; mod condition_timeout; +mod parallel; mod parallel_activity; +mod parallel_timer; thread_local! { static ACTIVE: RefCell>>> = const { RefCell::new(None) }; @@ -112,6 +114,7 @@ pub(super) struct Clock { time: Option>, boundaries: HashMap, parallel_activities: parallel_activity::Members, + parallel_timers: parallel_timer::Members, } enum BoundaryKind { @@ -405,6 +408,7 @@ impl Clock { time: start_time(history), boundaries, parallel_activities: parallel_activity::Members::new(history, commands), + parallel_timers: parallel_timer::Members::new(history, commands), } } @@ -438,11 +442,16 @@ impl WorkflowState { 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.replay_clock + .parallel_activities + .consumed_time(&self.history_events, sequences, failed) + .or_else(|| { + self.replay_clock.parallel_timers.consumed_time( + &self.history_events, + sequences, + failed, + ) + }) }; self.advance_cancellation_sequences(sequences, failed); } @@ -489,10 +498,11 @@ impl WorkflowContext { /// or invalid evidence returns [`Error::ReplayClockUnavailable`]. The clock /// is available only while this workflow's handler/future is executing. /// - /// This source cohort supports correlated scalar activity/timer/child/signal - /// 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. + /// Supported boundaries are correlated scalar activity, timer, child, signal + /// and condition results, persisted Activity and Timer all groups, and the + /// explicit redrive start anchor. Ordinary start authority, inline metadata + /// and local completions, and other grouped 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. pub fn replay_now(&self) -> Result> { diff --git a/src/replay_clock/parallel.rs b/src/replay_clock/parallel.rs new file mode 100644 index 0000000..1ea49f3 --- /dev/null +++ b/src/replay_clock/parallel.rs @@ -0,0 +1,59 @@ +//! Portable v1 group time. Legacy Carbon observations remain separate evidence. +use super::{durable_event_sequence, parse, DateTime, HashMap, HistoryEvent, Utc, Value}; +use crate::ParallelGroupMetadata; + +pub(super) fn all_path(path: &[ParallelGroupMetadata], kind: &str) -> bool { + !path.is_empty() + && path.iter().all(|entry| { + entry.parallel_group_kind == kind + && 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() + }) +} + +pub(super) fn round_microseconds(micros: i64) -> Option { + let magnitude = micros.unsigned_abs().checked_add(500)? / 1000; + let milliseconds = i64::try_from(magnitude).ok()?; + if micros < 0 { + milliseconds.checked_neg() + } else { + Some(milliseconds) + } +} + +pub(super) fn milliseconds(time: DateTime) -> Option { + if time.timestamp_subsec_nanos() % 1000 != 0 { + return None; + } + let micros = time + .timestamp() + .checked_mul(1_000_000)? + .checked_add(i64::from(time.timestamp_subsec_micros()))?; + round_microseconds(micros) +} + +pub(super) fn event_time(event: &HistoryEvent) -> Option> { + match event.raw.get("recorded_at") { + None | Some(Value::Null) => parse(event.raw.get("created_at")), + present => parse(present), + } +} + +pub(super) fn event_indices(history: &[HistoryEvent], prefix: &str) -> HashMap> { + let mut events: HashMap> = HashMap::new(); + for (index, event) in history.iter().enumerate() { + if event.event_type.starts_with(prefix) { + if let Some(sequence) = durable_event_sequence(event) { + events.entry(sequence).or_default().push(index); + } + } + } + events +} diff --git a/src/replay_clock/parallel_activity.rs b/src/replay_clock/parallel_activity.rs index 51b551a..07568a9 100644 --- a/src/replay_clock/parallel_activity.rs +++ b/src/replay_clock/parallel_activity.rs @@ -1,6 +1,5 @@ -use super::{ - durable_event_sequence, parse, DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value, -}; +use super::parallel::{all_path, event_indices, event_time, milliseconds}; +use super::{DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value}; use crate::{recorded_parallel_group_path, ParallelGroupMetadata}; use std::collections::HashSet; @@ -15,22 +14,6 @@ struct Member { #[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], @@ -99,34 +82,9 @@ fn member( }) } -// 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); - } - } - } + let events = event_indices(history, "Activity"); Self( commands .iter() @@ -139,7 +97,7 @@ impl Members { else { return None; }; - if !all_activity(path) { + if !all_path(path, "activity") { return None; } member(history, events.get(sequence)?, *sequence, path) @@ -178,13 +136,13 @@ impl Members { return None; } if member.failed { - let ms = milliseconds(time(history.get(member.terminal?)?)?)?; + let ms = milliseconds(event_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 time = event_time(history.get(member.terminal?)?)?; let ms = milliseconds(time)?; if success.is_none_or(|(current, _)| ms > current) { success = Some((ms, time)); @@ -193,10 +151,10 @@ impl Members { } match failed { Some(selected) => { - let (ms, core_selected) = failure?; - // Keep the legacy exception even when Core selects a different + let (ms, canonical_selected) = failure?; + // Keep the legacy exception even when v1 selects a different // failed member. That compatibility gap has no clock authority. - (selected == core_selected) + (selected == canonical_selected) .then(|| DateTime::from_timestamp_millis(ms)) .flatten() } diff --git a/src/replay_clock/parallel_timer.rs b/src/replay_clock/parallel_timer.rs new file mode 100644 index 0000000..81bc4a8 --- /dev/null +++ b/src/replay_clock/parallel_timer.rs @@ -0,0 +1,136 @@ +use super::parallel::{all_path, event_indices, event_time, milliseconds}; +use super::{DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value}; +use crate::{recorded_parallel_group_path, ParallelGroupMetadata}; +use std::collections::HashSet; + +#[derive(Debug)] +struct Member { + path: Vec, + timer: String, + fired: usize, +} + +#[derive(Debug)] +pub(super) struct Members(HashMap); + +fn member( + history: &[HistoryEvent], + indices: &[usize], + sequence: u64, + delay: u64, + path: &[ParallelGroupMetadata], +) -> Option { + let [scheduled, fired] = indices else { + return None; + }; + if 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 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("timer_id").and_then(Value::as_str) != Some(timer) + || event.payload.get("delay_seconds").and_then(Value::as_u64) != Some(delay) + || event + .payload + .get("timer_kind") + .is_some_and(|value| !value.is_null()) + || crate::is_internal_timer_event(event) + || recorded_parallel_group_path(&[event], sequence) + .ok() + .flatten() + .as_deref() + != Some(path) + { + return None; + } + } + Some(Member { + path: path.to_vec(), + timer: timer.into(), + fired: *fired, + }) +} + +impl Members { + pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self { + let events = event_indices(history, "Timer"); + Self( + commands + .iter() + .filter_map(|command| { + let RecordedCommand::Timer { + sequence, + delay_seconds, + fired: true, + parallel_group_path: Some(path), + } = command + else { + return None; + }; + if !all_path(path, "timer") { + return None; + } + member( + history, + events.get(sequence)?, + *sequence, + *delay_seconds, + path, + ) + .map(|member| (*sequence, member)) + }) + .collect(), + ) + } + + pub(super) fn consumed_time( + &self, + history: &[HistoryEvent], + sequences: &[u64], + failed: Option, + ) -> Option> { + if failed.is_some() { + return None; + } + 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 timers = HashSet::new(); + let mut selected: Option<(i64, DateTime)> = 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 + || !timers.insert(&member.timer) + { + return None; + } + let time = event_time(history.get(member.fired)?)?; + let ms = milliseconds(time)?; + if selected.is_none_or(|(current, _)| ms > current) { + selected = Some((ms, time)); + } + } + selected.map(|(_, time)| time) + } +} diff --git a/src/replay_clock/tests.rs b/src/replay_clock/tests.rs index fe88fc2..a60a6c8 100644 --- a/src/replay_clock/tests.rs +++ b/src/replay_clock/tests.rs @@ -522,3 +522,7 @@ mod condition; mod parallel_activity; mod condition_timeout; + +mod parallel_timer; + +mod parallel; diff --git a/src/replay_clock/tests/parallel.rs b/src/replay_clock/tests/parallel.rs new file mode 100644 index 0000000..b1ae1cb --- /dev/null +++ b/src/replay_clock/tests/parallel.rs @@ -0,0 +1,52 @@ +use super::*; + +#[test] +fn replay_clock_canonical_v1_uses_exact_signed_microseconds_across_epoch_boundaries_and_offsets() { + let table: Value = serde_json::from_str(include_str!( + "../../../tests/fixtures/replay-clock-canonical-v1.json" + )) + .unwrap(); + for row in table["vectors"].as_array().unwrap() { + let time = DateTime::parse_from_rfc3339(row["input"].as_str().unwrap()) + .unwrap() + .with_timezone(&Utc); + assert_eq!( + time.timestamp_micros().to_string(), + row["utc_microseconds"].as_str().unwrap() + ); + let ms = super::super::parallel::milliseconds(time).unwrap(); + assert_eq!( + ms.to_string(), + row["milliseconds"].as_str().unwrap(), + "{}", + row["id"] + ); + assert_eq!( + DateTime::::from_timestamp_millis(ms) + .unwrap() + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true), + row["reconstructed"].as_str().unwrap() + ); + } +} + +#[test] +fn replay_clock_canonical_v1_handles_signed_integer_limits_without_float_or_abs_overflow() { + use super::super::parallel::round_microseconds; + for (micro, ms) in [ + (-501, -1), + (-500, -1), + (-499, 0), + (499, 0), + (500, 1), + (501, 1), + (i64::MAX, 9_223_372_036_854_776), + (i64::MIN, -9_223_372_036_854_776), + ] { + assert_eq!(round_microseconds(micro), Some(ms)); + } + let too_precise = DateTime::parse_from_rfc3339("2026-01-01T00:00:00.123456789Z") + .unwrap() + .with_timezone(&Utc); + assert!(super::super::parallel::milliseconds(too_precise).is_none()); +} diff --git a/src/replay_clock/tests/parallel_activity.rs b/src/replay_clock/tests/parallel_activity.rs index 4e97494..1a1f3cb 100644 --- a/src/replay_clock/tests/parallel_activity.rs +++ b/src/replay_clock/tests/parallel_activity.rs @@ -214,7 +214,7 @@ fn replay_now_group_requires_all_success_times_and_never_substitutes_other_autho 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"] { + for time in ["2026-01-01T00:00:00.123456789Z"] { assert!( observe(group(1, ["ActivityCompleted"; 2], [time; 2], false), false)["time"].is_null() ); @@ -299,6 +299,8 @@ async fn replay_now_group_query_does_not_borrow_executor_aggregate() { #[test] fn replay_now_group_carbon_reference_matches_measured_rounding_and_reconstruction() { + // These attributed PHP 8.4/Carbon observations coincide with v1 on these + // inputs. They are not universal runtime or producer-context authority. let reference: Value = serde_json::from_str(include_str!( "../../../tests/fixtures/replay-clock-carbon-reference.json" )) @@ -307,7 +309,7 @@ fn replay_now_group_carbon_reference_matches_measured_rounding_and_reconstructio let time = DateTime::parse_from_rfc3339(row["input"].as_str().unwrap()) .unwrap() .with_timezone(&Utc); - let ms = super::super::parallel_activity::milliseconds(time).unwrap(); + let ms = super::super::parallel::milliseconds(time).unwrap(); assert_eq!(json!(ms), row["milliseconds"]); assert_eq!( DateTime::::from_timestamp_millis(ms) diff --git a/src/replay_clock/tests/parallel_timer.rs b/src/replay_clock/tests/parallel_timer.rs new file mode 100644 index 0000000..2ce3724 --- /dev/null +++ b/src/replay_clock/tests/parallel_timer.rs @@ -0,0 +1,259 @@ +use super::*; +use crate::{ParallelOperation, ParallelResult}; + +fn operations(nested: bool) -> Vec { + let timer = || ParallelOperation::timer(Duration::from_secs(1)); + if nested { + vec![timer(), ParallelOperation::group(vec![timer()])] + } else { + vec![timer(), timer()] + } +} + +fn group(base: u64, times: [&str; 2], nested: bool) -> Vec { + let descriptors = crate::parallel_descriptors(operations(nested), base).unwrap(); + let mut schedules = Vec::new(); + let mut fires = Vec::new(); + for (index, descriptor) in descriptors.iter().enumerate() { + let sequence = base + index as u64; + let mut payload = json!({"sequence":sequence,"timer_id":format!("timer-{sequence}"),"delay_seconds":1,"fire_at":"2099-01-01T00:00:00Z"}); + crate::apply_parallel_group_path(payload.as_object_mut().unwrap(), &descriptor.group_path); + schedules.push(event("TimerScheduled", payload.clone(), START)); + payload["fired_at"] = json!("2099-01-01T00:00:01Z"); + fires.push(event("TimerFired", payload, times[index])); + } + schedules.extend(fires); + 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 values = inner.parallel(operations(nested)).await?; + assert_eq!(values[0], ParallelResult::Timer); + assert_eq!( + values[1], + if nested { + ParallelResult::Group(vec![ParallelResult::Timer]) + } else { + ParallelResult::Timer + } + ); + 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_timer_all_uses_descriptor_order_core_ms_ties_and_full_precision_on_cold_replay() { + 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", + ), + ( + ["2025-12-31T23:59:59.654321Z", "2025-12-31T23:59:58.654321Z"], + "2025-12-31T23:59:59.654321Z", + ), + ] { + let mut history = vec![start()]; + history.extend(group(1, times, nested)); + history.swap(3, 4); + assert_eq!(observe(history.clone(), nested), json!(expected)); + assert_eq!(observe(history.clone(), nested), observe(history, nested)); + } + } +} + +#[test] +fn replay_now_timer_all_never_borrows_reuse_payload_or_other_groups_future_times() { + let mut history = group(1, [START; 2], false); + for event in &mut history[2..] { + event.payload["reused_recorded_at"] = json!("2099-01-01T00:00:00Z"); + event + .raw + .insert("timestamp".into(), json!("2099-01-01T00:00:00Z")); + } + history.extend(group(3, ["invalid-future-time"; 2], false)); + assert_eq!(observe(history, false), json!(START)); +} + +#[test] +fn replay_now_timer_all_missing_or_invalid_time_has_no_scalar_fallback() { + for value in [None, Some(Value::Null), Some(json!("bad")), Some(json!(42))] { + let mut history = group(1, [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).is_null()); + } + let mut history = group(1, [START; 2], false); + history[2].raw.remove("recorded_at"); + history[2] + .raw + .insert("created_at".into(), json!("2026-01-01T00:00:10.654321Z")); + assert_eq!( + observe(history, false), + json!("2026-01-01T00:00:10.654321Z") + ); + for time in ["2026-01-01T00:00:00.123456789Z"] { + assert!(observe(group(1, [time; 2], false), false).is_null()); + } +} + +#[test] +fn replay_now_timer_all_requires_typed_counter_unique_identity_and_complete_plain_timer_evidence() { + let original = group(1, [START; 2], false); + for index in 0..4 { + let sequence = if index % 2 == 0 { 1 } else { 2 }; + let mut history = original.clone(); + history[index].payload["sequence"] = json!(sequence.to_string()); + assert!(observe(history, false).is_null()); + let mut history = original.clone(); + history[index] + .payload + .as_object_mut() + .unwrap() + .remove("sequence"); + history[index] + .raw + .insert("sequence".into(), json!(sequence)); + assert!(observe(history, false).is_null()); + let mut history = original.clone(); + history[index].payload["workflow_sequence"] = json!(99); + assert!(observe(history, false).is_null()); + let mut history = original.clone(); + history[index].payload["timer_kind"] = json!("unknown-kind"); + assert!(observe(history, false).is_null()); + } + let mut history = original.clone(); + for event in &mut history { + event.payload["timer_id"] = json!("one-aliased-timer"); + } + assert!(observe(history, false).is_null()); + let mut history = original.clone(); + let mut cancel = history[2].clone(); + cancel.event_type = "TimerCancelled".into(); + history.push(cancel); + assert!(observe(history, false).is_null()); + let mut history = original; + history.swap(0, 2); + assert!(observe(history, false).is_null()); +} + +#[test] +fn replay_now_timer_all_keeps_legacy_duplicate_and_mismatched_timer_refusals() { + for change in ["duplicate", "identity", "delay"] { + let mut history = group(1, [START; 2], false); + match change { + "duplicate" => history.push(history[2].clone()), + "identity" => history[2].payload["timer_id"] = json!("foreign"), + "delay" => history[2].payload["delay_seconds"] = json!(2), + _ => unreachable!(), + } + assert!(matches!( + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None), + Err(Error::NonDeterministicReplay(_)) + )); + } +} + +#[test] +fn replay_now_pending_timer_group_has_no_aggregate_and_empty_group_keeps_previous_prefix() { + let mut history = group(1, [START; 2], false); + history.pop(); + let ctx = context(history); + 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()); + } + 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_timer_groups_do_not_supply_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( + "timer-group-clock", + || 0_u64, + |ctx, _, _| async move { + ctx.parallel(operations(false)).await?; + assert!(ctx.replay_now().is_err()); + ctx.sleep(Duration::from_secs(1)).await?; + Ok(Value::Null) + }, + ); + worker.register_query("timer-group-clock", "snapshot", |ctx, _| async move { + Ok(json!(ctx.replay_now().is_err())) + }); + worker.register_replayed_query::( + "timer-group-clock", + "reconstructed", + |ctx, _, _| async move { Ok(json!(ctx.replay_now().is_err())) }, + ); + let mut history = group(1, [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":"timer-group-q","workflow_type":"timer-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) + ); + } + assert_eq!(observe(group(1, [START; 2], false), false), json!(START)); +} diff --git a/tests/fixtures/replay-clock-canonical-v1.json b/tests/fixtures/replay-clock-canonical-v1.json new file mode 100644 index 0000000..1aae5d7 --- /dev/null +++ b/tests/fixtures/replay-clock-canonical-v1.json @@ -0,0 +1,189 @@ +{ + "schema": "durable-workflow.replay-clock-quantization/v1", + "arithmetic": "sign(U) * ((abs(U) + 500) div 1000)", + "integer_fields": "decimal strings preserve exact values across JSON runtimes", + "vectors": [ + { + "id": "canonical-v1-01", + "input": "1970-01-01T00:00:00.000000Z", + "utc_microseconds": "0", + "milliseconds": "0", + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "id": "canonical-v1-02", + "input": "1970-01-01T00:00:00.000499Z", + "utc_microseconds": "499", + "milliseconds": "0", + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "id": "canonical-v1-03", + "input": "1970-01-01T00:00:00.000500Z", + "utc_microseconds": "500", + "milliseconds": "1", + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "id": "canonical-v1-04", + "input": "1970-01-01T00:00:00.000501Z", + "utc_microseconds": "501", + "milliseconds": "1", + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "id": "canonical-v1-05", + "input": "1969-12-31T23:59:59.999501Z", + "utc_microseconds": "-499", + "milliseconds": "0", + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "id": "canonical-v1-06", + "input": "1969-12-31T23:59:59.999500Z", + "utc_microseconds": "-500", + "milliseconds": "-1", + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "id": "canonical-v1-07", + "input": "1969-12-31T23:59:59.999499Z", + "utc_microseconds": "-501", + "milliseconds": "-1", + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "id": "canonical-v1-08", + "input": "2026-01-01T00:00:00.123400Z", + "utc_microseconds": "1767225600123400", + "milliseconds": "1767225600123", + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-09", + "input": "2026-01-01T00:00:00.123499Z", + "utc_microseconds": "1767225600123499", + "milliseconds": "1767225600123", + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-10", + "input": "2026-01-01T00:00:00.123500Z", + "utc_microseconds": "1767225600123500", + "milliseconds": "1767225600124", + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-11", + "input": "2026-01-01T00:00:00.123501Z", + "utc_microseconds": "1767225600123501", + "milliseconds": "1767225600124", + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-12", + "input": "2026-01-01T01:30:00.123500+01:30", + "utc_microseconds": "1767225600123500", + "milliseconds": "1767225600124", + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-13", + "input": "2026-01-01T00:00:00.123500-00:00", + "utc_microseconds": "1767225600123500", + "milliseconds": "1767225600124", + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-14", + "input": "1900-01-01T00:00:00.123499Z", + "utc_microseconds": "-2208988799876501", + "milliseconds": "-2208988799877", + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-15", + "input": "1900-01-01T00:00:00.123500Z", + "utc_microseconds": "-2208988799876500", + "milliseconds": "-2208988799877", + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-16", + "input": "1900-01-01T00:00:00.123501Z", + "utc_microseconds": "-2208988799876499", + "milliseconds": "-2208988799876", + "reconstructed": "1900-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-17", + "input": "1000-01-01T00:00:00.123499Z", + "utc_microseconds": "-30610223999876501", + "milliseconds": "-30610223999877", + "reconstructed": "1000-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-18", + "input": "1000-01-01T00:00:00.123500Z", + "utc_microseconds": "-30610223999876500", + "milliseconds": "-30610223999877", + "reconstructed": "1000-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-19", + "input": "1000-01-01T00:00:00.123501Z", + "utc_microseconds": "-30610223999876499", + "milliseconds": "-30610223999876", + "reconstructed": "1000-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-20", + "input": "3000-01-01T00:00:00.123499Z", + "utc_microseconds": "32503680000123499", + "milliseconds": "32503680000123", + "reconstructed": "3000-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-21", + "input": "3000-01-01T00:00:00.123500Z", + "utc_microseconds": "32503680000123500", + "milliseconds": "32503680000124", + "reconstructed": "3000-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-22", + "input": "3000-01-01T00:00:00.123501Z", + "utc_microseconds": "32503680000123501", + "milliseconds": "32503680000124", + "reconstructed": "3000-01-01T00:00:00.124000Z" + }, + { + "id": "canonical-v1-23", + "input": "0001-01-01T00:00:00.000499Z", + "utc_microseconds": "-62135596799999501", + "milliseconds": "-62135596800000", + "reconstructed": "0001-01-01T00:00:00.000000Z" + }, + { + "id": "canonical-v1-24", + "input": "0001-01-01T00:00:00.000500Z", + "utc_microseconds": "-62135596799999500", + "milliseconds": "-62135596800000", + "reconstructed": "0001-01-01T00:00:00.000000Z" + }, + { + "id": "canonical-v1-25", + "input": "9999-01-01T00:00:00.123499Z", + "utc_microseconds": "253370764800123499", + "milliseconds": "253370764800123", + "reconstructed": "9999-01-01T00:00:00.123000Z" + }, + { + "id": "canonical-v1-26", + "input": "9999-01-01T00:00:00.123500Z", + "utc_microseconds": "253370764800123500", + "milliseconds": "253370764800124", + "reconstructed": "9999-01-01T00:00:00.124000Z" + } + ] +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.json b/tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.json new file mode 100644 index 0000000..01fbf87 --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.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-canonical-v1-large-epochs", + "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": "3000-01-01T00:00:05.123499Z", + "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": "3000-01-01T00:00:05.123500Z", + "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": "1000-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": "3000-01-01T00:00:00Z", + "sequence": 8 + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "success_time": "3000-01-01T00:00:05.123500Z", + "failure_time": "1000-01-01T00:00:03.123000Z", + "failure_member": [ + 0 + ], + "unchanged_prefix": true + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "success_time": "3000-01-01T00:00:05.123500Z", + "failure_time": "1000-01-01T00:00:03.123000Z", + "failure_member": [ + 0 + ], + "unchanged_prefix": true + } + } + ], + "worker_task": { + "workflow_id": "parallel-activity-clock-parent", + "payload_codec": "avro" + } +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json b/tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json new file mode 100644 index 0000000..fd9454c --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json @@ -0,0 +1,140 @@ +{ + "$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-timer", + "protocol_version": "1.20", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.business-replay-clock-parallel-timer", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "TimerScheduled", + "payload": { + "sequence": 1, + "timer_id": "clock-timer-1", + "delay_seconds": 1, + "fire_at": "2026-01-01T00:00:01Z", + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 1 + }, + { + "event_type": "TimerScheduled", + "payload": { + "sequence": 2, + "timer_id": "clock-timer-2", + "delay_seconds": 1, + "fire_at": "2026-01-01T00:00:01Z", + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "recorded_at": "2026-01-01T00:00:00Z", + "sequence": 2 + }, + { + "event_type": "TimerFired", + "payload": { + "sequence": 1, + "timer_id": "clock-timer-1", + "delay_seconds": 1, + "fire_at": "2026-01-01T00:00:01Z", + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "recorded_at": "2026-01-01T00:00:05.123100Z", + "sequence": 3 + }, + { + "event_type": "TimerFired", + "payload": { + "sequence": 2, + "timer_id": "clock-timer-2", + "delay_seconds": 1, + "fire_at": "2026-01-01T00:00:01Z", + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-timers:1:2", + "parallel_group_kind": "timer", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "recorded_at": "2026-01-01T00:00:05.123400Z", + "sequence": 4 + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "members": 2, + "time": "2026-01-01T00:00:05.123100Z", + "unchanged_prefix": true + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "members": 2, + "time": "2026-01-01T00:00:05.123100Z", + "unchanged_prefix": true + } + } + ], + "worker_task": { + "workflow_id": "timer-group-clock-parent", + "payload_codec": "avro" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 1e142ac..8af64a6 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -511,6 +511,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.business-replay-clock-condition" | "corpus.business-replay-clock-condition-timeout" | "corpus.business-replay-clock-parallel-activity" + | "corpus.business-replay-clock-parallel-timer" | "corpus.nexus-result" | "corpus.nexus-cancellation" ) { @@ -630,6 +631,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-timer" => { + business_clock::register_parallel_timer(&mut worker, workflow_type) + } "corpus.business-replay-clock-parallel-activity" => { business_clock::register_parallel_activity(&mut worker, workflow_type) } diff --git a/tests/replay_regression_corpus/business_clock.rs b/tests/replay_regression_corpus/business_clock.rs index 76afe3e..40833a5 100644 --- a/tests/replay_regression_corpus/business_clock.rs +++ b/tests/replay_regression_corpus/business_clock.rs @@ -159,3 +159,12 @@ pub(super) fn register_condition_timeout(worker: &mut Worker, workflow_type: &st })) }); } + +pub(super) fn register_parallel_timer(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + let entry_unavailable=ctx.replay_now().is_err(); + let values=ctx.parallel(vec![ParallelOperation::timer(Duration::from_secs(1)),ParallelOperation::timer(Duration::from_secs(1))]).await?; + let time=ctx.replay_now().ok(); + Ok(json!({"entry_unavailable":entry_unavailable,"members":values.len(),"time":time.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)),"unchanged_prefix":time.is_some() && time==ctx.replay_now().ok()})) + }); +}