diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md index b9cb778..e35dac2 100644 --- a/docs/replay-clock-source.md +++ b/docs/replay-clock-source.md @@ -22,12 +22,13 @@ counter identity to order emitted events. | 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 | | Scalar child result | One schedule, optional start and unique parent `ChildRunCompleted` / `ChildRunFailed` / `ChildRunCancelled` / `ChildRunTerminated`, with matching nonempty child call, instance, run and type identities | Parent resolution event time for both send and throw; child `closed_at` and `reused_recorded_at` are not this branch's clock authority | -| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable | +| Scalar persisted signal delivery | One untimed `SignalWaitOpened` followed by a unique `SignalApplied`, matching nonempty wait ID/name and explicit command sequence, with the persisted value consumed | Applied event time, preserving valid `reused_recorded_at` exactly as Core signal resume does | +| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child/signal 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 | -| 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 | +| Condition result, timed signal wait, 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 @@ -77,3 +78,21 @@ identity/time refusals, snapshot/reconstructed query prefixes and interleaved poll/error scope restoration. The separate child corpus consumer runs unchanged against base and candidate production code, retaining an exact behavioral negative rather than a compilation failure. + +The scalar persisted-signal cohort follows Core +[`4b2cdc55` `WorkflowExecutor::resumeSignal`](https://github.com/durable-workflow/workflow/blob/4b2cdc5507120d9c213b0b02a85738bec96c574a/src/V2/Support/WorkflowExecutor.php#L1434), +which supplies `RedriveReplayClock::recordedAt` for the matched `SignalApplied`. +Rust consumes that recorded wait by command sequence and signal name. Business +time additionally requires matching nonempty persisted wait IDs and explicit +numeric command counters, the canonical persisted value, one ordered open/apply +pair and no timed-wait evidence. Missing/conflicting identity, duplicate/order +ambiguity, invalid/missing time or a timed wait leaves time unavailable while +preserving existing signal results, commands and cancellation behavior. + +A visible `SignalReceived` is not application. Neither an unconsumed future row +nor task-resume arguments without the committed applied carrier advance this +clock. Core query reconstruction has a separate committed received-event legacy +fallback; this cohort does not invent that authority for Rust or read a mutable +projection as persisted time. No new observation envelope, event, wire field, +marker, RPC or inline persistence fence is introduced. Reconstructed/snapshot +queries retain the existing actually consumed/inspected prefix rules. diff --git a/src/replay_clock.rs b/src/replay_clock.rs index 7f3dcc7..92b12ad 100644 --- a/src/replay_clock.rs +++ b/src/replay_clock.rs @@ -1,9 +1,9 @@ //! 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, + durable_event_sequence, is_internal_timer_event, is_recorded_signal_wait_event, + recorded_commands, DurableOperationHandle, Error, HistoryEvent, ParallelGroupMetadata, + QueryContext, RecordedCommand, Result, Value, WorkflowContext, WorkflowIdentity, WorkflowState, }; use chrono::{DateTime, Utc}; use std::{ @@ -80,6 +80,46 @@ enum BoundaryKind { Activity, Timer, Child, + Signal, +} + +fn signal_boundary(history: &[HistoryEvent], indices: &[usize], sequence: u64) -> Option { + let [opened, applied] = indices else { + return None; + }; + let open = &history[*opened]; + let apply = &history[*applied]; + if open.event_type != "SignalWaitOpened" + || apply.event_type != "SignalApplied" + || open + .payload + .get("timeout_seconds") + .is_some_and(|value| !value.is_null()) + || apply.payload.get("value").is_none_or(Value::is_null) + { + return None; + } + for event in [open, apply] { + 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)) + { + return None; + } + } + for key in ["signal_wait_id", "signal_name"] { + let identity = open + .payload + .get(key)? + .as_str() + .filter(|value| !value.is_empty())?; + if apply.payload.get(key).and_then(Value::as_str) != Some(identity) { + return None; + } + } + Some(*applied) } fn child_boundary(history: &[HistoryEvent], indices: &[usize]) -> Option { @@ -153,11 +193,20 @@ impl Clock { } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { Some((*sequence, BoundaryKind::Child)) } + RecordedCommand::SignalWait { + sequence, + value: Some(_), + parallel_group_path, + .. + } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => { + Some((*sequence, BoundaryKind::Signal)) + } _ => None, }) .collect(); let mut boundaries = HashMap::new(); let mut children: HashMap> = HashMap::new(); + let mut signals: HashMap> = HashMap::new(); for (index, event) in history.iter().enumerate() { let Some(sequence) = durable_event_sequence(event) else { continue; @@ -181,6 +230,14 @@ impl Clock { } false } + Some(BoundaryKind::Signal) => { + if is_recorded_signal_wait_event(event) + || (event.event_type.starts_with("Timer") && is_internal_timer_event(event)) + { + signals.entry(sequence).or_default().push(index); + } + false + } None => false, }; if matched { @@ -192,6 +249,11 @@ impl Clock { boundaries.insert(sequence, index); } } + for (sequence, indices) in signals { + if let Some(index) = signal_boundary(history, &indices, sequence) { + boundaries.insert(sequence, index); + } + } // Index identities once. Timestamps are read only when that boundary // is actually consumed, never by scanning ahead from the accessor. Self { @@ -264,7 +326,7 @@ impl WorkflowState { event, matches!( event.event_type.as_str(), - "ActivityCompleted" | "TimerFired" + "ActivityCompleted" | "TimerFired" | "SignalApplied" ), ) }); @@ -279,7 +341,7 @@ 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 + /// This source cohort supports correlated scalar activity/timer/child/signal /// results and the explicit redrive start anchor. Ordinary start authority, /// inline metadata/local completions and grouped/other operation families /// remain unavailable. Cancellation's existing budget clock is unchanged. @@ -323,6 +385,7 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option Option Vec { + let identity = json!({"sequence":sequence,"signal_name":"ready","signal_wait_id":format!("wait-{sequence}")}); + let mut applied = identity.clone(); + applied["workflow_command_id"] = json!(format!("signal-command-{sequence}")); + applied["value"] = encode_typed_envelope( + &AvroValue::Array(vec![AvroValue::String(format!("delivery-{sequence}"))]), + DEFAULT_CODEC, + ) + .unwrap(); + vec![ + event("SignalWaitOpened", identity, START), + event("SignalApplied", applied, time), + ] +} + +fn time(ctx: &WorkflowContext) -> Option { + ctx.replay_now() + .ok() + .map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Micros, true)) +} + +fn observe(history: Vec) -> Value { + let ctx = context(history); + let inner = ctx.clone(); + let output = ready( + &ctx, + Box::pin(async move { + let arguments = inner.wait_signal("ready").await?; + let observed = time(&inner); + assert_eq!(observed, time(&inner)); + Ok(AvroValue::from_serialize( + &json!({"arguments":arguments,"time":observed}), + )?) + }), + ) + .into_json() + .unwrap(); + assert!(ctx.take_commands().unwrap().is_empty()); + assert!(ctx.replay_now().is_err()); + output +} + +#[test] +fn replay_now_signal_consumption_cold_replays_backward_and_equal_times() { + for last in ["2025-12-31T23:59:59.123456Z", SIGNAL_TIME] { + let mut history = vec![start()]; + history.extend(signal(1, SIGNAL_TIME)); + history.extend(signal(2, last)); + let execute = || { + let ctx = context(history.clone()); + let inner = ctx.clone(); + let result = ready( + &ctx, + Box::pin(async move { + assert_eq!(read(&inner), START); + assert_eq!(inner.wait_signal("ready").await?, vec![json!("delivery-1")]); + assert_eq!(read(&inner), SIGNAL_TIME); + assert_eq!(read(&inner), read(&inner)); + assert_eq!(inner.wait_signal("ready").await?, vec![json!("delivery-2")]); + Ok(AvroValue::from_serialize(&time(&inner))?) + }), + ) + .into_json() + .unwrap(); + assert_eq!(ctx.state.lock().unwrap().command_cursor, 2); + assert!(ctx.take_commands().unwrap().is_empty()); + assert!(ctx.replay_now().is_err()); + result + }; + assert_eq!(execute(), json!(last)); + assert_eq!(execute(), execute()); + } +} + +#[test] +fn replay_now_signal_future_and_buffered_rows_are_not_consumption() { + let received = event( + "SignalReceived", + json!({"signal_name":"ready","signal_wait_id":"unconsumed","arguments":encode_typed_envelope(&AvroValue::String("buffered".into()),DEFAULT_CODEC).unwrap()}), + "2099-01-01T00:00:00Z", + ); + let mut buffered = signal(3, "2099-01-01T00:00:00Z").remove(1); + buffered.payload.as_object_mut().unwrap().remove("sequence"); + buffered.payload["signal_wait_id"] = json!("signal-command:buffered"); + // Its outer history order must not become an authored wait sequence. + buffered.raw.insert("sequence".into(), json!(1)); + let ctx = context(vec![start(), received.clone(), buffered.clone()]); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner.wait_signal("ready").await?; + panic!("unapplied receipt is not a committed wait result") + }); + assert!(Worker::poll_workflow_future(&ctx, &mut future) + .unwrap() + .is_pending()); + { + let _guard = Guard::enter(&ctx); + assert_eq!(read(&ctx), START); + } + assert_eq!(ctx.take_commands().unwrap()[0]["type"], "open_signal_wait"); + let mut history = vec![start(), received, buffered]; + history.extend(signal(1, SIGNAL_TIME)); + history.extend(signal(2, "2099-01-01T00:00:00Z")); + let ctx = context(history); + let inner = ctx.clone(); + let mut future: WorkflowFuture = Box::pin(async move { + inner.wait_signal("ready").await?; + assert_eq!(read(&inner), SIGNAL_TIME); + std::future::pending().await + }); + assert!(Worker::poll_workflow_future(&ctx, &mut future) + .unwrap() + .is_pending()); + assert_eq!(ctx.state.lock().unwrap().command_cursor, 1); + assert!(ctx.replay_now().is_err()); +} + +#[test] +fn replay_now_signal_identity_and_explicit_counter_are_required_without_changing_results() { + for index in 0..2 { + for replacement in [ + None, + Some(Value::Null), + Some(json!("")), + Some(json!(17)), + Some(json!("other-wait")), + ] { + let mut history = signal(1, SIGNAL_TIME); + match replacement { + Some(value) => { + history[index].payload["signal_wait_id"] = value; + } + None => { + history[index] + .payload + .as_object_mut() + .unwrap() + .remove("signal_wait_id"); + } + } + let output = observe(history); + assert_eq!(output["arguments"], json!(["delivery-1"])); + assert!(output["time"].is_null()); + } + let mut history = signal(1, SIGNAL_TIME); + history[index] + .payload + .as_object_mut() + .unwrap() + .remove("sequence"); + history[index].raw.insert("sequence".into(), json!(1)); + assert!(observe(history)["time"].is_null()); + let mut history = signal(1, SIGNAL_TIME); + history[index].payload["workflow_sequence"] = json!(2); + assert!(observe(history)["time"].is_null()); + let mut history = signal(1, SIGNAL_TIME); + history[index] + .payload + .as_object_mut() + .unwrap() + .remove("signal_name"); + history[index] + .raw + .insert("signal_name".into(), json!("ready")); + assert!(observe(history)["time"].is_null()); + } +} + +#[test] +fn replay_now_signal_time_and_reuse_do_not_borrow_unrelated_authority() { + for bad in [ + None, + Some(Value::Null), + Some(json!("invalid")), + Some(json!(42)), + ] { + let mut history = vec![start()]; + history.extend(signal(1, SIGNAL_TIME)); + match bad { + Some(value) => { + history[2].raw.insert("recorded_at".into(), value); + } + None => { + history[2].raw.remove("recorded_at"); + } + } + assert!(observe(history)["time"].is_null()); + } + let mut history = signal(1, SIGNAL_TIME); + history[1].payload["reused_recorded_at"] = json!("2025-12-01T00:00:00.123456Z"); + assert_eq!( + observe(history.clone())["time"], + json!("2025-12-01T00:00:00.123456Z") + ); + assert_eq!( + snapshot_time(&history, DEFAULT_CODEC) + .unwrap() + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true), + "2025-12-01T00:00:00.123456Z" + ); + history[1].payload["reused_recorded_at"] = json!("invalid"); + assert!(observe(history)["time"].is_null()); + let mut history = signal(1, SIGNAL_TIME); + history[1].raw.insert("timestamp".into(), json!("invalid")); + assert!(observe(history)["time"].is_null()); +} + +#[test] +fn replay_now_signal_ambiguous_order_payload_and_timed_shapes_are_unavailable() { + let mut history = signal(1, SIGNAL_TIME); + history.swap(0, 1); + assert!(observe(history)["time"].is_null()); + for kind in ["TimerScheduled", "TimerFired", "TimerCancelled"] { + let mut history = signal(1, SIGNAL_TIME); + history.push(event(kind,json!({"sequence":1,"timer_kind":"signal_timeout","signal_wait_id":"wait-1","timer_id":"signal-timeout"}),"2099-01-01T00:00:00Z")); + assert!(observe(history)["time"].is_null()); + } + let mut history = signal(1, SIGNAL_TIME); + history[0].payload["timeout_seconds"] = json!(5); + assert!(observe(history)["time"].is_null()); + let mut history = signal(1, SIGNAL_TIME); + history[1].payload.as_object_mut().unwrap().remove("value"); + let result = observe(history); + assert_eq!(result["arguments"], json!([])); + assert!(result["time"].is_null()); + for duplicate in [0, 1] { + let mut history = signal(1, SIGNAL_TIME); + history.push(history[duplicate].clone()); + assert!( + WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None).is_err() + ); + } + let mut history = signal(1, SIGNAL_TIME); + history[1].payload["signal_name"] = json!("different"); + assert!(WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None).is_err()); +} + +#[test] +fn replay_now_signal_task_resume_without_applied_record_stays_unavailable() { + for opened in [false, true] { + let mut history = vec![start()]; + if opened { + history.push(signal(1, SIGNAL_TIME).remove(0)); + } + let ctx = context(history); + ctx.state.lock().unwrap().resume_signal = Some(ResumeSignal { + signal_name: "ready".into(), + arguments: vec![AvroValue::String("task-delivery".into())], + }); + let inner = ctx.clone(); + ready( + &ctx, + Box::pin(async move { + assert_eq!( + inner.wait_signal("ready").await?, + vec![json!("task-delivery")] + ); + assert!(inner.replay_now().is_err()); + Ok(AvroValue::Null) + }), + ); + } +} + +#[test] +fn replay_now_signal_interleaved_error_polls_restore_the_outer_scope() { + let outer = context(vec![start()]); + let _guard = Guard::enter(&outer); + let mut tasks = vec![]; + for expected in [SIGNAL_TIME, "2026-01-01T00:00:02.123456Z"] { + let ctx = context(signal(1, expected)); + let inner = ctx.clone(); + let future: WorkflowFuture = Box::pin(async move { + inner.wait_signal("ready").await?; + assert_eq!(read(&inner), expected); + tokio::task::yield_now().await; + assert_eq!(read(&inner), expected); + Err(Error::WorkerLoop("expected error after signal".into())) + }); + tasks.push((ctx, future)); + } + for (ctx, future) in &mut tasks { + assert!(Worker::poll_workflow_future(ctx, future) + .unwrap() + .is_pending()); + assert!(ctx.replay_now().is_err()); + assert_eq!(read(&outer), START); + } + for (ctx, future) in &mut tasks { + assert!(matches!( + Worker::poll_workflow_future(ctx, future).unwrap(), + Poll::Ready(Err(Error::WorkerLoop(_))) + )); + assert!(ctx.replay_now().is_err()); + assert_eq!(read(&outer), START); + } +} + +#[tokio::test] +async fn replay_now_signal_queries_use_the_selected_existing_prefix() { + let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests"); + worker.register_replayed_workflow( + "signal-clock", + || 0_u64, + |ctx, _, _| async move { + ctx.wait_signal("ready").await?; + ctx.wait_signal("ready").await?; + Ok(Value::Null) + }, + ); + worker.register_query("signal-clock", "snapshot", |ctx, _| async move { + Ok(json!(ctx + .replay_now()? + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true))) + }); + worker.register_replayed_query::( + "signal-clock", + "reconstructed", + |ctx, _, _| async move { + Ok(json!(ctx + .replay_now()? + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true))) + }, + ); + let mut history = signal(1, SIGNAL_TIME); + for query in ["snapshot", "reconstructed"] { + if query == "reconstructed" { + history.push(signal(2, "2099-01-01T00:00:00Z").remove(0)); + } + let task:QueryTask=serde_json::from_value(json!({"query_task_id":"signal-q","workflow_type":"signal-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!(SIGNAL_TIME) + ); + } + assert!(snapshot_time(&history, DEFAULT_CODEC).is_none()); +} diff --git a/tests/fixtures/replay-regressions/business-replay-clock-signal-results.json b/tests/fixtures/replay-regressions/business-replay-clock-signal-results.json new file mode 100644 index 0000000..ecfa7cb --- /dev/null +++ b/tests/fixtures/replay-regressions/business-replay-clock-signal-results.json @@ -0,0 +1,116 @@ +{ + "$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-signal-results", + "protocol_version": "1.20", + "bindings": [ + "rust" + ], + "workflow": { + "type": "corpus.business-replay-clock-signal", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "workflow_type": "corpus.business-replay-clock-signal" + }, + "recorded_at": "2026-01-01T00:00:00Z" + }, + { + "sequence": 2, + "event_type": "SignalWaitOpened", + "payload": { + "sequence": 1, + "signal_name": "ready", + "signal_wait_id": "clock-signal-wait-1" + }, + "recorded_at": "2026-01-01T00:00:00Z" + }, + { + "sequence": 3, + "event_type": "SignalApplied", + "payload": { + "sequence": 1, + "signal_name": "ready", + "signal_wait_id": "clock-signal-wait-1", + "workflow_command_id": "clock-signal-command-1", + "signal_id": "clock-signal-1", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoKR3JhY2U=" + } + }, + "recorded_at": "2026-01-01T00:00:05.654321Z" + }, + { + "sequence": 4, + "event_type": "SignalWaitOpened", + "payload": { + "sequence": 2, + "signal_name": "ready", + "signal_wait_id": "clock-signal-wait-2" + }, + "recorded_at": "2026-01-01T00:00:00Z" + }, + { + "sequence": 5, + "event_type": "SignalApplied", + "payload": { + "sequence": 2, + "signal_name": "ready", + "signal_wait_id": "clock-signal-wait-2", + "workflow_command_id": "clock-signal-command-2", + "signal_id": "clock-signal-2", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoKR3JhY2U=" + } + }, + "recorded_at": "2026-01-01T00:00:03.123456Z" + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "first_time": "2026-01-01T00:00:05.654321Z", + "second_time": "2026-01-01T00:00:03.123456Z", + "unchanged_prefix": true, + "arguments": [ + [ + "Grace" + ], + [ + "Grace" + ] + ] + } + }, + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "entry_unavailable": true, + "first_time": "2026-01-01T00:00:05.654321Z", + "second_time": "2026-01-01T00:00:03.123456Z", + "unchanged_prefix": true, + "arguments": [ + [ + "Grace" + ], + [ + "Grace" + ] + ] + } + } + ], + "worker_task": { + "workflow_id": "signal-clock-parent", + "payload_codec": "avro" + } +} diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs index d44fe2a..58a8e07 100644 --- a/tests/replay_regression_corpus.rs +++ b/tests/replay_regression_corpus.rs @@ -507,6 +507,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< | "corpus.activity-policy-author" | "corpus.business-replay-clock" | "corpus.business-replay-clock-child" + | "corpus.business-replay-clock-signal" | "corpus.nexus-result" | "corpus.nexus-cancellation" ) { @@ -623,6 +624,9 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result< } match workflow_type { "corpus.business-replay-clock" => business_clock::register(&mut worker, workflow_type), + "corpus.business-replay-clock-signal" => { + business_clock::register_signal(&mut worker, workflow_type) + } "corpus.business-replay-clock-child" => { business_clock::register_child(&mut worker, workflow_type) } diff --git a/tests/replay_regression_corpus/business_clock.rs b/tests/replay_regression_corpus/business_clock.rs index 9c9073e..d15984d 100644 --- a/tests/replay_regression_corpus/business_clock.rs +++ b/tests/replay_regression_corpus/business_clock.rs @@ -67,3 +67,20 @@ pub(super) fn register_child(worker: &mut Worker, workflow_type: &str) { })) }); } + +pub(super) fn register_signal(worker: &mut Worker, workflow_type: &str) { + worker.register_workflow(workflow_type, |ctx, _| async move { + let entry_unavailable = ctx.replay_now().is_err(); + let first_arguments = ctx.wait_signal("ready").await?; + let first_time = ctx.replay_now().ok(); + let second_arguments = ctx.wait_signal("ready").await?; + let second_time = ctx.replay_now().ok(); + Ok(json!({ + "entry_unavailable":entry_unavailable, + "first_time":first_time.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "second_time":second_time.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)), + "unchanged_prefix":second_time.is_some() && second_time==ctx.replay_now().ok(), + "arguments":[first_arguments,second_arguments], + })) + }); +}