diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md new file mode 100644 index 0000000..7397ef0 --- /dev/null +++ b/docs/replay-clock-source.md @@ -0,0 +1,59 @@ +# Additive replay clock: first Rust source cohort + +This source checkpoint implements `WorkflowContext::replay_now()` and +`QueryContext::replay_now()` toward [the cross-SDK replay-clock feature](https://github.com/durable-workflow/.github/issues/215). +It does not qualify a published package or close that feature. Existing APIs, +UUID decisions, command/history shapes and the cancellation budget clock are +unchanged. Business decisions adopt the new accessor through ordinary workflow +versioning. + +The accessor returns a recorded UTC `chrono::DateTime`, or +`Error::ReplayClockUnavailable` when the current invocation has no supported, +validated time authority. It never substitutes wall time, a future history row, +a maximum over all events, a task ID or an observation timestamp. Sequential +consumption selects the matched event's time, including its precision; it does +not enforce monotonically increasing timestamps. Use stream/outbox/workflow +counter identity to order emitted events. + +| Boundary | Authority in this cohort | Availability | +| --- | --- | --- | +| Redrive entry | Unique `WorkflowStarted` with `recovery_kind=redrive` and valid `replayed_started_at` | Recorded original start | +| Ordinary entry | Core uses original `WorkflowRun.started_at`; current Rust worker task has no such carrier | Unavailable, even if the start history row has an insertion timestamp | +| Scalar remote activity | Existing validated command sequence/type/result correlation, then the selected terminal event | Selected recorded time; successful redrive reuse retains `reused_recorded_at`; the existing Core failure branch uses its event time | +| Scalar plain timer | Existing validated `TimerScheduled`/`TimerFired` pair, exact sequence/ID/delay | Matched fire time, with recorded reuse provenance | +| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable | +| Replayed-instance query | Clock of the actual reconstructed workflow prefix, rather than a later visible row | Same supported boundary rules | +| Direct update handler | Unique committed `UpdateAccepted` correlated to the handler's update ID | That event's validated time; missing/ambiguous acceptance is unavailable | +| Executor condition predicate | Current executor exposes a full signal/update view without retaining the exact triggering consumed prefix | Unavailable inside the predicate; enclosing workflow scope is restored on return/error | +| Inline side effect/version/memo/search/local result | User code can resume before its event is persisted | Unavailable on first execution and the corresponding cold path in this cohort; no implicit marker/yield/fence | +| Child/signal/condition result, grouped/selection/Nexus or cancellation callback | Their dedicated clock algorithms are outside this initial cohort | Explicitly unavailable; existing execution/cancellation behavior remains intact | + +The mutable workflow clock is bound only during handler construction and each +future poll. A separate execution-local guard restores its predecessor on +return/error; one workflow cannot read another workflow's active clock. Query +and update contexts own an immutable engine-correlated value, with no shared +ambient mutable state. Clock reads append no commands/history. + +The ordinary entry gap is [confirmed by Server](https://github.com/durable-workflow/server/issues/325#issuecomment-6671799265): +worker claims omit `run.started_at`; native start and history insertion sample +time separately. The existing describe endpoint needs exact namespace/workflow/run +binding; export summary time does not establish a worker anchor. Server owns a +later additive authoritative task carrier and its precision/missing contract. +This source does not add a wire field or extra describe/export RPC. + +An advancing time immediately after a newly authored inline marker/local result +also needs an explicitly selected opt-in persistence/resume boundary. It cannot +be manufactured from the existing same-pass result. Core/Server own that +producer boundary; SDK work remains for the other already-persisted operation +families and exact predicate-prefix retention. These are distinct from the +private operating-clock work in Rust #191/#192. + +Focused source controls exercise the actual Rust context, future, query/update +and predicate paths; advancing deadline/elapsed calculations; unchanged-prefix +reads; cold replay; backward/equal/precise times; redrive provenance; future-row +isolation; concurrent/nested/error scope restoration; unavailable inputs and +first-pass/cold marker handling. A retained published PHP activity history is +also consumed as source evidence, without inventing its ordinary start anchor. +Normal current-head checks remain required. Published cross-SDK conformance, +producer implementation/adoption, release and guide/localized delivery remain +separate work on the owning issue. diff --git a/src/lib.rs b/src/lib.rs index f47f490..61f7b3f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,6 +17,7 @@ mod nexus; pub mod pagination; mod query_authority; mod raw_request; +mod replay_clock; mod run_activities; mod runtime_payloads; mod runtime_uploads; @@ -275,6 +276,8 @@ pub use nexus::{ #[derive(Debug, Error)] #[non_exhaustive] pub enum Error { + #[error("replay time is unavailable without a supported authoritative consumed boundary")] + ReplayClockUnavailable, #[error("transport error: {0}")] Transport(#[from] reqwest::Error), #[error( @@ -7184,6 +7187,7 @@ pub struct QueryContext { workflow_input_avro_value: AvroValue, history_events: Arc>, signal_events: Arc>, + replay_time: Option>, } impl QueryContext { @@ -9613,13 +9617,14 @@ impl Worker { ) })?; let history_events = Arc::new(std::mem::take(&mut task.history_events)); - let context = QueryContext { + let mut context = QueryContext { workflow_id: task.workflow_id, run_id: task.run_id, workflow_type: task.workflow_type.clone(), run_status: task.run_status, workflow_input, workflow_input_avro_value: workflow_input_typed.clone(), + replay_time: replay_clock::snapshot_time(&history_events, &task.payload_codec), history_events: Arc::clone(&history_events), signal_events: Arc::new(signal_events), }; @@ -9676,16 +9681,20 @@ impl Worker { cancellation_scope_id: "root".into(), }; query_authority::before_handler(authority, &self.client).await?; - let mut invocation = replay(workflow_context.clone(), workflow_input_typed.clone()) - .map_err(|error| { - QueryTaskExecutionFailure::new( - "query_workflow_state_unavailable", - error.to_string(), - "QueryWorkflowStateUnavailable", - ) - })?; + let mut invocation = { + let _clock = replay_clock::Guard::enter(&workflow_context); + replay(workflow_context.clone(), workflow_input_typed.clone()) + } + .map_err(|error| { + QueryTaskExecutionFailure::new( + "query_workflow_state_unavailable", + error.to_string(), + "QueryWorkflowStateUnavailable", + ) + })?; let outcome = { let mut cx = TaskContext::from_waker(noop_waker_ref()); + let _clock = replay_clock::Guard::enter(&workflow_context); let _replay = cancellation_replay_clock::ReplayGuard::enter(&workflow_context) .map_err(|error| { QueryTaskExecutionFailure::new( @@ -9750,6 +9759,18 @@ impl Worker { ) })?; query_authority::before_handler(authority, &self.client).await?; + context.replay_time = workflow_context + .state + .lock() + .map_err(|_| { + QueryTaskExecutionFailure::new( + "query_workflow_state_unavailable", + "workflow state poisoned", + "QueryWorkflowStateUnavailable", + ) + })? + .replay_time() + .ok(); handler(context, state, args).map_err(|message| { QueryTaskExecutionFailure::new( "query_workflow_state_unavailable", @@ -9910,7 +9931,10 @@ impl Worker { state, cancellation_scope_id: "root".into(), }; - let future = (workflow.execute)(ctx.clone(), input)?; + let future = { + let _clock = replay_clock::Guard::enter(&ctx); + (workflow.execute)(ctx.clone(), input)? + }; Ok(PreparedWorkflowTask::Execution { task, context: ctx, @@ -9923,6 +9947,7 @@ impl Worker { future: &mut WorkflowFuture, ) -> Result>> { let mut cx = TaskContext::from_waker(noop_waker_ref()); + let _clock = replay_clock::Guard::enter(ctx); let _replay = cancellation_replay_clock::ReplayGuard::enter(ctx)?; Ok(future.as_mut().poll(&mut cx)) } @@ -10144,6 +10169,7 @@ impl Worker { run_status: Some("running".to_string()), workflow_input, workflow_input_avro_value, + replay_time: replay_clock::update_time(&task.history_events, update_id), history_events: Arc::new(task.history_events.clone()), signal_events: Arc::new(signal_events), }; @@ -11322,6 +11348,7 @@ impl WorkflowContext { if let Some(recorded) = state.recorded_commands.get(state.command_cursor).cloned() { return match recorded { RecordedCommand::SideEffect { sequence, value } => { + state.replay_clock.unavailable(); state.command_cursor += 1; value.deserialize().map_err(|error| { Error::NonDeterministicReplay(ReplayFailure::new( @@ -11345,6 +11372,7 @@ impl WorkflowContext { .lock() .map_err(|_| Error::WorkflowStatePoisoned)?; let result = encode_typed_envelope(&avro_value, &state.payload_codec)?; + state.replay_clock.unavailable(); state.commands.push(json!({ "type": "record_side_effect", "result": result, @@ -11365,6 +11393,7 @@ impl WorkflowContext { if let Some(recorded) = state.recorded_commands.get(state.command_cursor).cloned() { return match recorded { RecordedCommand::SideEffect { value, .. } => { + state.replay_clock.unavailable(); state.command_cursor += 1; Ok(value) } @@ -11379,6 +11408,7 @@ impl WorkflowContext { .lock() .map_err(|_| Error::WorkflowStatePoisoned)?; let result = encode_typed_envelope(&value, &state.payload_codec)?; + state.replay_clock.unavailable(); state.commands.push(json!({ "type": "record_side_effect", "result": result, @@ -11424,6 +11454,7 @@ impl WorkflowContext { state.workflow_stream_command_counter += 1; return match recorded { RecordedCommand::SideEffect { .. } => { + state.replay_clock.unavailable(); state.command_cursor += 1; Ok(()) } @@ -11453,6 +11484,7 @@ impl WorkflowContext { directive["max_pending_items"] = json!(max_pending_items); } let result = encode_typed_envelope(&AvroValue::Null, &state.payload_codec)?; + state.replay_clock.unavailable(); state.commands.push(json!({ "type": "record_side_effect", "result": result, @@ -11511,6 +11543,7 @@ impl WorkflowContext { state.workflow_stream_command_counter += 1; return match recorded { RecordedCommand::SideEffect { .. } => { + state.replay_clock.unavailable(); state.command_cursor += 1; Ok(()) } @@ -11532,6 +11565,7 @@ impl WorkflowContext { directive["retention_seconds"] = json!(retention_seconds); } let result = encode_typed_envelope(&AvroValue::Null, &state.payload_codec)?; + state.replay_clock.unavailable(); state.commands.push(json!({ "type": "record_side_effect", "result": result, @@ -11588,6 +11622,7 @@ impl WorkflowContext { ))); } } + state.replay_clock.unavailable(); state.command_cursor += 1; Ok(()) } @@ -11602,6 +11637,7 @@ impl WorkflowContext { if !attribute_types.is_empty() { command.insert("attribute_types".to_string(), json!(attribute_types)); } + state.replay_clock.unavailable(); state.commands.push(Value::Object(command)); Ok(()) } @@ -11676,6 +11712,7 @@ impl WorkflowContext { )?; // Older workers could record the same decision again. Keep // each physical boundary without skipping intervening work. + state.replay_clock.unavailable(); state.command_cursor += 1; return Ok(version); } @@ -11708,6 +11745,7 @@ impl WorkflowContext { max_supported, sequence, )?; + state.replay_clock.unavailable(); state.command_cursor += 1; state.version_markers.insert(change_id, (version, sequence)); Ok(version) @@ -11742,6 +11780,7 @@ impl WorkflowContext { } let version = max_supported; + state.replay_clock.unavailable(); state.commands.push(json!({ "type": "record_version_marker", "change_id": change_id, @@ -11791,6 +11830,7 @@ impl WorkflowContext { "recorded memo entries differ from the current workflow update", ))); } + state.replay_clock.unavailable(); state.command_cursor += 1; Ok(()) } @@ -11799,6 +11839,7 @@ impl WorkflowContext { } let entries_envelope = encode_typed_envelope(&entries, DEFAULT_CODEC)?; + state.replay_clock.unavailable(); state.commands.push(json!({ "type": "upsert_memo", "entries": entries_envelope, @@ -12010,6 +12051,7 @@ struct WorkflowState { cancellation_history: CancellationHistory, cancellation_consumed: bool, cancellation_clock: Option, + replay_clock: replay_clock::Clock, cancellation_shield_depth: u64, cancellation_delivery_enabled: bool, cancellation_delivery_intent: Option, @@ -12272,6 +12314,7 @@ impl WorkflowState { run_id, task_queue, payload_codec, + replay_clock: replay_clock::Clock::new(&history, &recorded_commands), history_events: Arc::new(history), history_budget: WorkflowHistoryBudget { event_count, @@ -13826,7 +13869,7 @@ impl ParallelCall { .state .lock() .map_err(|_| Error::WorkflowStatePoisoned)? - .advance_cancellation_sequences(&sequences, Some(failed.sequence)); + .consume_replay_group(&sequences, Some(failed.sequence)); return Poll::Ready(Err(Error::ParallelFailed(ParallelFailure { group_id, member_path: failed.member_path.clone(), @@ -13866,7 +13909,7 @@ impl ParallelCall { .state .lock() .map_err(|_| Error::WorkflowStatePoisoned)? - .advance_cancellation_sequences(&sequences, None); + .consume_replay_group(&sequences, None); Poll::Ready(Ok(match results { ParallelAvroResult::Group(results) => results, ParallelAvroResult::Activity(_) @@ -14512,7 +14555,7 @@ impl Future for SelectCall { .state .lock() .map_err(|_| Error::WorkflowStatePoisoned)? - .advance_cancellation_selection(&marker.selection_group_id); + .consume_replay_selection(&marker.selection_group_id); Poll::Ready(Ok(SelectionResult { key: winner.key.clone(), index: winner.index, @@ -14941,7 +14984,7 @@ impl Future for DurableOperationAwaitCall { Err(error) => return Poll::Ready(Err(error)), Ok(false) => {} Ok(true) => { - state.advance_cancellation_receipt( + state.consume_replay_receipt( &self.handle.selection_group_id, self.handle.base_sequence, ); @@ -14958,13 +15001,13 @@ impl Future for DurableOperationAwaitCall { } match recorded_selection_member_outcome(&state, &self.handle) { Ok(Some(result)) => { - state.advance_cancellation_handle(&self.handle, false); + state.consume_replay_handle(&self.handle, false); Poll::Ready(Ok(result)) } Ok(None) => Poll::Pending, Err(error) => { if !workflow_task_integrity_error(&error) { - state.advance_cancellation_handle(&self.handle, true); + state.consume_replay_handle(&self.handle, true); } Poll::Ready(Err(error)) } @@ -15046,7 +15089,7 @@ impl Future for CancelDurableOperationCall { match selection_cancellation_for_handle(&state, &self.handle) { Err(error) => return Poll::Ready(Err(error)), Ok(true) => { - state.advance_cancellation_receipt( + state.consume_replay_receipt( &self.handle.selection_group_id, self.handle.base_sequence, ); @@ -15217,6 +15260,7 @@ impl ActivityCall { if let Some(result) = &self.local_result { return match result.lock() { Ok(mut result) => result.take().map_or(Poll::Pending, |result| { + state.replay_clock.unavailable(); Poll::Ready(result.map_err(Error::ActivityFailed)) }), Err(_) => Poll::Ready(Err(Error::WorkflowStatePoisoned)), @@ -15406,7 +15450,10 @@ impl ActivityCall { } state.command_cursor += 1; if let Some(outcome) = outcome { - state.advance_cancellation_sequence(sequence, &self.parallel_group_path); + state.consume_replay_sequence(sequence, &self.parallel_group_path); + if self.local { + state.replay_clock.unavailable(); + } return Poll::Ready(outcome.map_err(Error::ActivityFailed)); } state.matched_recorded_pending = true; @@ -15604,7 +15651,7 @@ impl Future for TimerCall { } state.command_cursor += 1; if fired { - state.advance_cancellation_sequence(sequence, &self.parallel_group_path); + state.consume_replay_sequence(sequence, &self.parallel_group_path); return Poll::Ready(Ok(())); } state.matched_recorded_pending = true; @@ -15812,7 +15859,7 @@ impl Future for ConditionWaitCall { } result = recorded_result; if result.is_some() { - state.advance_cancellation_sequence(sequence, &self.parallel_group_path); + state.consume_replay_sequence(sequence, &self.parallel_group_path); } previous_sequence = Some(sequence); cursor += 1; @@ -15993,7 +16040,13 @@ impl ConditionWaitCall { .parallel_group_path .first() .is_some_and(|entry| entry.parallel_group_mode.as_deref() == Some("select")); - match (self.predicate)() { + let predicate_result = { + // This predicate inspects the full signal/update view; the current + // executor does not retain its exact triggering event prefix. + let _clock = replay_clock::Guard::unavailable(); + (self.predicate)() + }; + match predicate_result { Ok(true) if !selection_member || recorded_wait => { return Poll::Ready(Ok(ConditionWaitResult::Satisfied)) } @@ -16180,7 +16233,7 @@ impl ChildWorkflowCall { } state.command_cursor += 1; if let Some(outcome) = outcome { - state.advance_cancellation_sequence(sequence, &self.parallel_group_path); + state.consume_replay_sequence(sequence, &self.parallel_group_path); return Poll::Ready(outcome.map_err(Error::ChildWorkflowFailed)); } state.matched_recorded_pending = true; @@ -16415,7 +16468,7 @@ impl SignalCall { } state.command_cursor += 1; if let Some(value) = value { - state.advance_cancellation_sequence(sequence, &self.parallel_group_path); + state.consume_replay_sequence(sequence, &self.parallel_group_path); return Poll::Ready(Ok(value)); } if state @@ -16427,7 +16480,7 @@ impl SignalCall { .resume_signal .take() .expect("matching resume signal is present"); - state.unavailable_cancellation_boundary(&self.parallel_group_path); + state.consume_unproven_boundary(&self.parallel_group_path); return Poll::Ready(Ok(signal.arguments)); } @@ -16468,7 +16521,7 @@ impl SignalCall { .resume_signal .take() .expect("matching resume signal is present"); - state.unavailable_cancellation_boundary(&self.parallel_group_path); + state.consume_unproven_boundary(&self.parallel_group_path); return Poll::Ready(Ok(signal.arguments)); } diff --git a/src/nexus.rs b/src/nexus.rs index a6e8fc3..aea03ac 100644 --- a/src/nexus.rs +++ b/src/nexus.rs @@ -503,6 +503,7 @@ impl Future for NexusServiceCall { Err(error) => return Poll::Ready(Err(error)), } } + state.replay_clock.unavailable(); state.command_cursor += 1; if !self.options.completion_wait() { if let Some(admission) = call.admission { diff --git a/src/replay_clock.rs b/src/replay_clock.rs new file mode 100644 index 0000000..a08db43 --- /dev/null +++ b/src/replay_clock.rs @@ -0,0 +1,290 @@ +//! Additive business time, independent of cancellation's monotonic budget clock. + +use super::{ + durable_event_sequence, is_internal_timer_event, recorded_commands, DurableOperationHandle, + Error, HistoryEvent, ParallelGroupMetadata, QueryContext, RecordedCommand, Result, Value, + WorkflowContext, WorkflowIdentity, WorkflowState, +}; +use chrono::{DateTime, Utc}; +use std::{ + cell::RefCell, + collections::HashMap, + sync::{Arc, Mutex, Weak}, +}; + +thread_local! { + static ACTIVE: RefCell>>> = const { RefCell::new(None) }; +} + +pub(super) struct Guard(Option>>); + +impl Guard { + pub(super) fn enter(context: &WorkflowContext) -> Self { + Self(ACTIVE.with(|active| active.replace(Some(Arc::downgrade(&context.state))))) + } + + pub(super) fn unavailable() -> Self { + Self(ACTIVE.with(|active| active.replace(None))) + } +} + +impl Drop for Guard { + fn drop(&mut self) { + ACTIVE.with(|active| active.replace(self.0.take())); + } +} + +fn unavailable() -> Error { + Error::ReplayClockUnavailable +} + +fn parse(value: Option<&Value>) -> Option> { + DateTime::parse_from_rfc3339(value?.as_str()?) + .ok() + .map(|time| time.with_timezone(&Utc)) +} + +fn recorded_time(event: &HistoryEvent, reused: bool) -> Option> { + if reused && event.payload.get("reused_recorded_at").is_some() { + return parse(event.payload.get("reused_recorded_at")); + } + // A present but invalid field cannot borrow another event's time. + parse( + event + .raw + .get("timestamp") + .or_else(|| event.raw.get("recorded_at")), + ) +} + +fn start_time(history: &[HistoryEvent]) -> Option> { + let mut starts = history + .iter() + .filter(|event| event.event_type == "WorkflowStarted"); + let start = starts.next()?; + if starts.next().is_some() || start.payload["recovery_kind"] != "redrive" { + // Ordinary Core entry uses run.started_at, not the history insertion + // time. The current worker task has no authoritative carrier for it. + return None; + } + parse(start.payload.get("replayed_started_at")) +} + +#[derive(Debug)] +pub(super) struct Clock { + time: Option>, + boundaries: HashMap, +} + +impl Clock { + pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self { + let supported: HashMap<_, _> = commands + .iter() + .filter_map(|command| match command { + RecordedCommand::Activity { + sequence, + outcome: Some(_), + parallel_group_path, + .. + } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { + Some((*sequence, true)) + } + RecordedCommand::Timer { + sequence, + fired: true, + parallel_group_path, + .. + } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { + Some((*sequence, false)) + } + _ => None, + }) + .collect(); + let mut boundaries = HashMap::new(); + for (index, event) in history.iter().enumerate() { + let Some(sequence) = durable_event_sequence(event) else { + continue; + }; + let matched = match supported.get(&sequence) { + Some(true) => matches!( + event.event_type.as_str(), + "ActivityCompleted" + | "ActivityFailed" + | "ActivityTimedOut" + | "ActivityCancelled" + ), + Some(false) => event.event_type == "TimerFired" && !is_internal_timer_event(event), + None => false, + }; + if matched { + boundaries.entry(sequence).or_insert(index); + } + } + // Index identities once. Timestamps are read only when that boundary + // is actually consumed, never by scanning ahead from the accessor. + Self { + time: start_time(history), + boundaries, + } + } + + pub(super) fn unavailable(&mut self) { + self.time = None; + } +} + +impl WorkflowState { + pub(super) fn consume_replay_sequence( + &mut self, + sequence: u64, + path: &[ParallelGroupMetadata], + ) { + if path.is_empty() { + self.observe_replay_sequence(sequence); + } else { + self.replay_clock.unavailable(); + } + self.advance_cancellation_sequence(sequence, path); + } + + pub(super) fn consume_unproven_boundary(&mut self, path: &[ParallelGroupMetadata]) { + self.replay_clock.unavailable(); + self.unavailable_cancellation_boundary(path); + } + + pub(super) fn consume_replay_group(&mut self, sequences: &[u64], failed: Option) { + self.replay_clock.unavailable(); + self.advance_cancellation_sequences(sequences, failed); + } + + pub(super) fn consume_replay_selection(&mut self, group: &str) { + self.replay_clock.unavailable(); + self.advance_cancellation_selection(group); + } + + pub(super) fn consume_replay_receipt(&mut self, group: &str, sequence: u64) { + self.replay_clock.unavailable(); + self.advance_cancellation_receipt(group, sequence); + } + + pub(super) fn consume_replay_handle(&mut self, handle: &DurableOperationHandle, failed: bool) { + self.replay_clock.unavailable(); + self.advance_cancellation_handle(handle, failed); + } + + pub(super) fn replay_time(&self) -> Result> { + // Cancellation keeps its existing separate semantics. This first + // cohort does not claim business-clock correlation for its callbacks. + if self.cancellation_clock.is_some() { + return Err(unavailable()); + } + self.replay_clock.time.ok_or_else(unavailable) + } + + pub(super) fn observe_replay_sequence(&mut self, sequence: u64) { + self.replay_clock.time = self + .replay_clock + .boundaries + .get(&sequence) + .and_then(|index| self.history_events.get(*index)) + .and_then(|event| { + recorded_time( + event, + matches!( + event.event_type.as_str(), + "ActivityCompleted" | "TimerFired" + ), + ) + }); + } +} + +impl WorkflowContext { + /// Recorded business time at the last supported consumed replay boundary. + /// + /// Repeated reads at an unchanged prefix are equal. This never reads wall + /// time, creates a command, or selects an unconsumed future event. Missing + /// or invalid evidence returns [`Error::ReplayClockUnavailable`]. The clock + /// is available only while this workflow's handler/future is executing. + /// + /// This initial source cohort supports correlated scalar activity/timer + /// results and the explicit redrive start anchor. Ordinary start authority, + /// inline metadata/local completions and grouped/other operation families + /// remain unavailable. Cancellation's existing budget clock is unchanged. + /// 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> { + let active = ACTIVE.with(|active| active.borrow().as_ref().and_then(Weak::upgrade)); + if !active.is_some_and(|active| Arc::ptr_eq(&active, &self.state)) { + return Err(unavailable()); + } + self.state + .lock() + .map_err(|_| Error::WorkflowStatePoisoned)? + .replay_time() + } +} + +impl QueryContext { + /// Recorded time of this immutable, engine-correlated callback prefix. + /// + /// Missing, invalid or unproved boundary evidence is explicitly unavailable. + /// Reading this accessor does not change the snapshot or append history. + pub fn replay_now(&self) -> Result> { + self.replay_time.ok_or_else(unavailable) + } +} + +pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option> { + let event = history.last()?; + if event.event_type == "WorkflowStarted" { + return start_time(history); + } + if !matches!( + event.event_type.as_str(), + "ActivityCompleted" + | "ActivityFailed" + | "ActivityTimedOut" + | "ActivityCancelled" + | "TimerFired" + ) { + return None; + } + let commands = recorded_commands( + history, + codec, + WorkflowIdentity { + workflow_id: None, + run_id: None, + }, + ) + .ok()?; + Clock::new(history, &commands) + .boundaries + .get(&durable_event_sequence(event)?) + .and_then(|index| history.get(*index)) + .and_then(|event| { + recorded_time( + event, + matches!( + event.event_type.as_str(), + "ActivityCompleted" | "TimerFired" + ), + ) + }) +} + +pub(super) fn update_time(history: &[HistoryEvent], update_id: &str) -> Option> { + let mut events = history.iter().filter(|event| { + event.event_type == "UpdateAccepted" + && event.payload["update_id"].as_str() == Some(update_id) + }); + let event = events.next()?; + if events.next().is_some() { + return None; + } + recorded_time(event, true) +} + +#[cfg(test)] +mod tests; diff --git a/src/replay_clock/tests.rs b/src/replay_clock/tests.rs new file mode 100644 index 0000000..f8eca40 --- /dev/null +++ b/src/replay_clock/tests.rs @@ -0,0 +1,515 @@ +use super::*; +use crate::{ + decode_wire_value, encode_typed_envelope, json, AvroValue, Client, ConditionWaitOptions, + ConditionWaitResult, QueryTask, Worker, WorkflowFuture, WorkflowTask, DEFAULT_CODEC, +}; +use std::{task::Poll, time::Duration}; + +const START: &str = "2026-01-01T00:00:00.123456Z"; + +fn event(kind: &str, payload: Value, time: &str) -> HistoryEvent { + serde_json::from_value(json!({"event_type":kind,"payload":payload,"recorded_at":time})).unwrap() +} + +fn start() -> HistoryEvent { + event( + "WorkflowStarted", + json!({"recovery_kind":"redrive","replayed_started_at":START}), + "2026-02-01T00:00:00Z", + ) +} + +fn timer(sequence: u64, time: &str) -> Vec { + let payload = + json!({"sequence":sequence,"timer_id":format!("timer-{sequence}"),"delay_seconds":1}); + vec![ + event("TimerScheduled", payload.clone(), START), + event("TimerFired", payload, time), + ] +} + +fn context(history: Vec) -> WorkflowContext { + WorkflowContext { + state: Arc::new(Mutex::new( + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None).unwrap(), + )), + cancellation_scope_id: "root".into(), + } +} + +fn read(context: &WorkflowContext) -> String { + context + .replay_now() + .unwrap() + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true) +} + +fn ready(context: &WorkflowContext, mut future: WorkflowFuture) -> AvroValue { + match Worker::poll_workflow_future(context, &mut future).unwrap() { + Poll::Ready(result) => result.unwrap(), + Poll::Pending => panic!("expected ready replay"), + } +} + +#[test] +fn replay_now_deadline_loop_advances_and_cold_replays_identically() { + let mut history = vec![start()]; + for i in 1..=3 { + history.extend(timer(i, &format!("2026-01-01T00:00:0{i}.123456Z"))); + } + let execute = || { + let ctx = context(history.clone()); + let future_ctx = ctx.clone(); + ready(&ctx,Box::pin(async move { + let initial=future_ctx.replay_now()?; let deadline=initial+chrono::Duration::seconds(3); + let mut times=vec![]; + while future_ctx.replay_now()? assert_eq!(read(&inner), value), + None => assert!(matches!( + inner.replay_now(), + Err(Error::ReplayClockUnavailable) + )), + }; + Ok(AvroValue::Null) + }), + ); + } +} + +#[test] +fn replay_now_missing_start_time_is_not_history_insertion_or_wall_time() { + for history in [ + vec![], + vec![event("WorkflowStarted", json!({}), START)], + vec![event( + "WorkflowStarted", + json!({"recovery_kind":"redrive","replayed_started_at":"invalid"}), + START, + )], + ] { + let ctx = context(history); + let _guard = Guard::enter(&ctx); + assert!(matches!( + ctx.replay_now(), + Err(Error::ReplayClockUnavailable) + )); + } + let mut history = vec![start()]; + let mut events = timer(1, START); + events[1].raw.remove("recorded_at"); + history.extend(events); + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner.sleep(Duration::from_secs(1)).await?; + assert!(matches!( + inner.replay_now(), + Err(Error::ReplayClockUnavailable) + )); + Ok(AvroValue::Null) + }), + ); +} + +#[test] +fn replay_now_nested_contexts_and_error_exit_restore_scope() { + let a = context(vec![start()]); + let b = context(vec![start()]); + let _outer = Guard::enter(&a); + assert_eq!(read(&a), START); + assert!(b.replay_now().is_err()); + let result: Result<()> = (|| { + let _inner = Guard::enter(&b); + assert!(a.replay_now().is_err()); + assert_eq!(read(&b), START); + Err(Error::WorkerLoop("test error".into())) + })(); + assert!(result.is_err()); + assert_eq!(read(&a), START); + { + let _blocked = Guard::unavailable(); + assert!(a.replay_now().is_err()); + } + assert_eq!(read(&a), START); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn replay_now_concurrent_workflows_have_independent_scopes() { + let a = tokio::spawn(async { + let ctx = context(vec![start()]); + let inner = ctx.clone(); + let result = ready( + &ctx, + Box::pin(async move { Ok(AvroValue::from_serialize(&read(&inner))?) }), + ); + assert!(ctx.replay_now().is_err()); + result + }); + let b = tokio::spawn(async { + let mut start = start(); + start.payload["replayed_started_at"] = json!("2026-01-01T00:00:10Z"); + let ctx = context(vec![start]); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { Ok(AvroValue::from_serialize(&read(&inner))?) }), + ) + }); + assert_ne!(a.await.unwrap(), b.await.unwrap()); +} + +#[test] +fn replay_now_inline_marker_is_unavailable_on_first_pass_and_cold_replay() { + let recorded = event( + "SideEffectRecorded", + json!({"sequence":1,"result":encode_typed_envelope(&AvroValue::String("fixed".into()),DEFAULT_CODEC).unwrap()}), + "2026-01-01T00:00:01Z", + ); + for history in [vec![start()], vec![start(), recorded]] { + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + let value: String = inner.side_effect(|| "fixed".to_string())?; + assert_eq!(value, "fixed"); + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); + } +} + +#[test] +fn replay_now_legacy_version_default_keeps_same_prefix() { + let mut history = vec![start()]; + history.extend(timer(1, "2026-01-01T00:00:01Z")); + let ctx = context(history); + let _guard = Guard::enter(&ctx); + assert_eq!(ctx.get_version("old-patch", -1, 1).unwrap(), -1); + assert_eq!(read(&ctx), START); + assert!(ctx.state.lock().unwrap().commands.is_empty()); +} + +#[test] +fn replay_now_snapshot_prefix_and_update_handler_are_exact_and_immutable() { + let mut history = vec![start()]; + history.extend(timer(1, "2026-01-01T00:00:02Z")); + assert_eq!( + snapshot_time(&history, DEFAULT_CODEC) + .unwrap() + .timestamp_subsec_micros(), + 0 + ); + let saved = serde_json::to_value(&history).unwrap(); + let at = snapshot_time(&history, DEFAULT_CODEC); + assert_eq!(at, snapshot_time(&history, DEFAULT_CODEC)); + assert_eq!(saved, serde_json::to_value(&history).unwrap()); + history.push(event( + "UnknownFutureBoundary", + json!({}), + "2099-01-01T00:00:00Z", + )); + assert!(snapshot_time(&history, DEFAULT_CODEC).is_none()); + let accepted = event( + "UpdateAccepted", + json!({"update_id":"u1","update_name":"inspect"}), + START, + ); + assert_eq!( + update_time(&[accepted.clone()], "u1") + .unwrap() + .timestamp_subsec_micros(), + 123456 + ); + assert!(update_time(&[accepted.clone()], "u2").is_none()); + assert!(update_time(&[accepted.clone(), accepted], "u1").is_none()); +} + +#[tokio::test] +async fn replay_now_actual_query_and_update_callbacks_use_their_own_prefix() { + let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests"); + worker.register_workflow("clock", |_ctx, _args| async { Ok(Value::Null) }); + worker.register_query("clock", "inspect", |ctx, _args| async move { + Ok(json!(ctx + .replay_now() + .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) + .ok())) + }); + worker.register_update("clock", "inspect", |ctx, _args| async move { + Ok(json!(ctx + .replay_now() + .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) + .ok())) + }); + let mut history = vec![start()]; + history.extend(timer(1, "2026-01-01T00:00:02.456789Z")); + let query:QueryTask=serde_json::from_value(json!({"query_task_id":"q1","workflow_type":"clock","query_name":"inspect","payload_codec":DEFAULT_CODEC,"history_events":history})).unwrap(); + for _ in 0..2 { + assert_eq!( + worker + .execute_query_task(query.clone()) + .await + .unwrap() + .into_json() + .unwrap(), + json!("2026-01-01T00:00:02.456789Z") + ); + } + let accepted = event( + "UpdateAccepted", + json!({"update_id":"u1","update_name":"inspect"}), + START, + ); + let task:WorkflowTask=serde_json::from_value(json!({"task_id":"u1","workflow_type":"clock","payload_codec":DEFAULT_CODEC,"workflow_update_id":"u1","update_name":"inspect","history_events":[accepted]})).unwrap(); + let commands = worker.execute_update_task(&task, "u1").unwrap(); + assert_eq!(commands[0]["type"], "complete_update"); + assert_eq!( + decode_wire_value(&commands[0]["result"], DEFAULT_CODEC).unwrap(), + json!(START) + ); + let mut unbound = task; + unbound.history_events.clear(); + let commands = worker.execute_update_task(&unbound, "u1").unwrap(); + assert_eq!( + decode_wire_value(&commands[0]["result"], DEFAULT_CODEC).unwrap(), + Value::Null + ); +} + +#[test] +fn replay_now_direct_predicate_is_unavailable_and_restores_workflow_scope() { + let ctx = context(vec![start()]); + let inner = ctx.clone(); + let predicate = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + let before = inner.replay_now()?; + let result = inner + .wait_condition( + ConditionWaitOptions::new("ready", "clock-test"), + move || { + assert!(matches!( + predicate.replay_now(), + Err(Error::ReplayClockUnavailable) + )); + Ok(true) + }, + ) + .await?; + assert_eq!(result, ConditionWaitResult::Satisfied); + assert_eq!(before, inner.replay_now()?); + Ok(AvroValue::Null) + }), + ); + assert!(ctx.replay_now().is_err()); + assert!(ctx.state.lock().unwrap().commands.is_empty()); +} + +#[test] +fn replay_now_reads_real_published_activity_evidence_without_inventing_entry_time() { + let fixture: Value = serde_json::from_str(include_str!( + "../../tests/fixtures/replay-regressions/published-php-complete-explicit-backoff.json" + )) + .unwrap(); + let history: Vec = serde_json::from_value(fixture["history"].clone()).unwrap(); + let mut state = + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None).unwrap(); + assert!(state.replay_time().is_err()); + let sequence = state + .recorded_commands + .iter() + .find_map(|command| match command { + RecordedCommand::Activity { + sequence, + outcome: Some(_), + .. + } => Some(*sequence), + _ => None, + }) + .unwrap(); + state.observe_replay_sequence(sequence); + let actual = state.replay_time().unwrap(); + let selected = state + .history_events + .iter() + .find(|event| { + event.event_type == "ActivityCompleted" + && durable_event_sequence(event) == Some(sequence) + }) + .unwrap(); + assert_eq!(actual, recorded_time(selected, true).unwrap()); +} + +#[test] +fn replay_now_inline_local_result_has_no_persistence_time_on_either_path() { + let ctx = context(vec![start()]); + let inner = ctx.clone(); + let mut call = inner.local_activity("local", json!([])); + let outcome = Ok(AvroValue::String("local-result".into())); + call.local_result = Some(Arc::new(Mutex::new(Some(outcome)))); + ready( + &ctx, + Box::pin(async move { + assert_eq!(call.await?, json!("local-result")); + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); + let complete = event( + "ActivityCompleted", + json!({"sequence":1,"activity_type":"local","execution_mode":"local","result":encode_typed_envelope(&AvroValue::String("local-result".into()),DEFAULT_CODEC).unwrap()}), + "2026-01-01T00:00:04Z", + ); + let ctx = context(vec![start(), complete]); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert_eq!( + inner.local_activity("local", json!([])).await?, + json!("local-result") + ); + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); +} + +#[test] +fn replay_now_failure_uses_its_consumed_throw_time_and_equal_events_keep_order() { + let failure = event( + "ActivityFailed", + json!({"sequence":1,"activity_type":"a","message":"expected failure","reused_recorded_at":START}), + "2026-01-01T00:00:05.999999Z", + ); + let ctx = context(vec![start(), failure]); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert!(matches!( + inner.activity("a", json!([])).await, + Err(Error::ActivityFailed(_)) + )); + assert_eq!(read(&inner), "2026-01-01T00:00:05.999999Z"); + Ok(AvroValue::Null) + }), + ); + let mut history = vec![start()]; + history.extend(timer(1, START)); + history.extend(timer(2, START)); + let ctx = context(history); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + inner.sleep(Duration::from_secs(1)).await?; + assert_eq!(read(&inner), START); + inner.sleep(Duration::from_secs(1)).await?; + assert_eq!(read(&inner), START); + assert_eq!(inner.state.lock().unwrap().command_cursor, 2); + Ok(AvroValue::Null) + }), + ); +} + +#[tokio::test] +async fn replay_now_reconstructed_query_uses_consumed_prefix_not_snapshot_tail() { + let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests"); + worker.register_replayed_workflow( + "clock", + || 0_u64, + |ctx, _args, _state| async move { + ctx.sleep(Duration::from_secs(1)).await?; + ctx.sleep(Duration::from_secs(1)).await?; + Ok(Value::Null) + }, + ); + worker.register_replayed_query::( + "clock", + "inspect", + |ctx, _state, _args| async move { + Ok(json!(ctx + .replay_now()? + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true))) + }, + ); + let mut history = vec![start()]; + history.extend(timer(1, "2026-01-01T00:00:03.765432Z")); + history.push(event( + "TimerScheduled", + json!({"sequence":2,"timer_id":"timer-2","delay_seconds":1}), + "2099-01-01T00:00:00Z", + )); + let task:QueryTask=serde_json::from_value(json!({"query_task_id":"q1","workflow_type":"clock","query_name":"inspect","payload_codec":DEFAULT_CODEC,"history_events":history})).unwrap(); + assert_eq!( + worker + .execute_query_task(task) + .await + .unwrap() + .into_json() + .unwrap(), + json!("2026-01-01T00:00:03.765432Z") + ); +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-consumed-boundaries.json b/tests/fixtures/replay-regressions/business-replay-clock-consumed-boundaries.json new file mode 100644 index 0000000..0480da4 --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-consumed-boundaries.json @@ -0,0 +1,105 @@ +{ + "$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-consumed-boundaries", + "protocol_version": "1.19", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.business-replay-clock", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "workflow_class": "ClockWorkflow", + "workflow_type": "corpus.business-replay-clock", + "workflow_instance_id": "clock-replay-workflow", + "workflow_run_id": "regression-corpus-run", + "workflow_command_id": "clock-redrive-command", + "continued_from_run_id": "clock-source-run", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-01-01T00:00:00.123456Z", + "workflow_link_id": "clock-redrive-link" + }, + "recorded_at": "2026-02-01T00:00:00Z" + }, + { + "sequence": 2, + "event_type": "TimerScheduled", + "payload": { + "sequence": 1, + "timer_id": "clock-timer", + "delay_seconds": 1, + "fire_at": "2026-01-01T00:00:01.123456Z" + }, + "recorded_at": "2026-02-01T00:00:00Z" + }, + { + "sequence": 3, + "event_type": "TimerFired", + "payload": { + "sequence": 1, + "timer_id": "clock-timer", + "delay_seconds": 1, + "reused_recorded_at": "2026-01-01T00:00:01.123456Z", + "fire_at": "2026-01-01T00:00:01.123456Z", + "fired_at": "2026-01-01T00:00:01.123456Z", + "reused_from_run_id": "clock-source-run" + }, + "recorded_at": "2026-02-01T00:00:00Z" + }, + { + "sequence": 4, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 2, + "activity_type": "clock-activity", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoQcmVjb3JkZWQ=" + }, + "reused_recorded_at": "2026-01-01T00:00:05.123456Z", + "activity_execution_id": "clock-successor-activity", + "reused_activity_execution_id": "clock-source-activity", + "reused_from_run_id": "clock-source-run" + }, + "recorded_at": "2026-02-01T00:00:00Z" + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "start": "2026-01-01T00:00:00.123456Z", + "timer": "2026-01-01T00:00:01.123456Z", + "activity": "2026-01-01T00:00:05.123456Z", + "unchanged_prefix": true, + "elapsed_seconds": 5, + "result": "recorded" + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "start": "2026-01-01T00:00:00.123456Z", + "timer": "2026-01-01T00:00:01.123456Z", + "activity": "2026-01-01T00:00:05.123456Z", + "unchanged_prefix": true, + "elapsed_seconds": 5, + "result": "recorded" + } + } + ], + "worker_task": { + "workflow_id": "clock-replay-workflow", + "payload_codec": "avro" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index 3791a6a..5f428c3 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -24,6 +24,9 @@ use sha2::{Digest, Sha256}; const FIXTURE_SCHEMA: &str = "durable-workflow.replay-regression/v1"; +#[path = "replay_regression_corpus/business_clock.rs"] +mod business_clock; + #[path = "replay_regression_corpus/redrive_successor_strictness.rs"] mod redrive_successor_strictness; @@ -502,6 +505,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.redrive-successor-timer-cancellation" | "corpus.child-policy-author" | "corpus.activity-policy-author" + | "corpus.business-replay-clock" | "corpus.nexus-result" | "corpus.nexus-cancellation" ) { @@ -617,6 +621,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< .map_err(|error| format!("{fixture_id} sticky cache configuration failed: {error}"))?; } match workflow_type { + "corpus.business-replay-clock" => business_clock::register(&mut worker, workflow_type), "corpus.nexus-result" => { worker.register_workflow(workflow_type, |ctx, _| async move { ctx.call_nexus_service("endpoint", "service", "operation", json!(["world"])) @@ -1976,6 +1981,7 @@ fn redrive_successor_fixture_path(name: &str) -> PathBuf { /// The valid successor fixtures and the step each resumes at. const REDRIVE_SUCCESSOR_FIXTURES: &[(&str, u64)] = &[ + ("business-replay-clock-consumed-boundaries.json", 3), ("redrive-successor-reused-markers.json", 4), ("redrive-successor-reused-timer.json", 3), ("redrive-successor-reused-cancellation-clock.json", 3), diff --git a/tests/replay_regression_corpus/business_clock.rs b/tests/replay_regression_corpus/business_clock.rs new file mode 100644 index 0000000..dc91bc7 --- /dev/null +++ b/tests/replay_regression_corpus/business_clock.rs @@ -0,0 +1,34 @@ +use super::*; +use chrono::{DateTime, SecondsFormat, Utc}; + +// The same official consumer must compile against the pre-accessor SDK. +// Inherent methods win over this explicit unavailable baseline shim. +trait LegacyReplayClock { + fn replay_now(&self) -> Result, Error>; +} +impl LegacyReplayClock for WorkflowContext { + fn replay_now(&self) -> Result, Error> { + Err(Error::WorkerLoop( + "baseline SDK has no replay_now accessor".into(), + )) + } +} + +pub(super) fn register(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + assert!(LegacyReplayClock::replay_now(&ctx).is_err()); + let start = ctx.replay_now().ok(); + ctx.sleep(Duration::from_secs(1)).await?; + let timer = ctx.replay_now().ok(); + let result = ctx.activity("clock-activity", json!([])).await?; + let activity = ctx.replay_now().ok(); + Ok(json!({ + "start":start.map(|time| time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "timer":timer.map(|time| time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "activity":activity.map(|time| time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "unchanged_prefix":activity.is_some() && activity==ctx.replay_now().ok(), + "elapsed_seconds":start.zip(activity).map(|(start,end)| (end-start).num_seconds()), + "result":result, + })) + }); +}