diff --git a/rust/src/runtime/redrive.rs b/rust/src/runtime/redrive.rs index 3a60fd234..d0dd0febd 100644 --- a/rust/src/runtime/redrive.rs +++ b/rust/src/runtime/redrive.rs @@ -1,8 +1,8 @@ -//! Redriving a failed run, as published PHP's `DefaultWorkflowControlPlane:: -//! redrive` and `FailedRunRedrivePlan` do: a run that failed right after an -//! activity failed continues in a new run whose history reuses the completed -//! activities, side effects and version markers before that step, so its -//! workflow resumes at the failed step. +//! Failed-run redrive follows Core's `DefaultWorkflowControlPlane::redrive` +//! and `FailedRunRedrivePlan`: a run that failed right after an activity +//! failed continues in a new run whose history reuses completed activities, +//! side effects, version markers and fired plain timers before that step. +//! The workflow resumes at the failed step. use super::laravel_rules::{self, Rule}; use super::{Result, RuntimeError, backend::Backend, refuse, store::{Store, id, now}, transaction::Transaction}; use axum::http::StatusCode; @@ -25,13 +25,21 @@ fn iso(at: DateTime) -> String { at.format("%Y-%m-%dT%H:%M:%S+00:00").to_string() } +/// Replay clocks retain the persisted instant's precision. Deadline formatting +/// remains separate; rounding a clock changes workflow decisions on replay. +fn replay_time(at: DateTime) -> String { + at.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, false) +} + /// One step the new run reuses: an `ActivityCompleted`, or a marker step /// (`SideEffectRecorded`, `VersionMarkerRecorded`, `MemoUpserted`, -/// `SearchAttributesUpserted`) without its source task and command. +/// `SearchAttributesUpserted`), or a fired plain timer without its source +/// identity, task and command. Timers retain both halves of their history pair. struct Reused { kind: String, payload: Value, recorded_at: DateTime, + scheduled_payload: Option, } /// The marker steps `FailedRunRedrivePlan::forRun` reuses as recorded. @@ -43,6 +51,13 @@ fn external(result: &Value) -> bool { || result.as_str().is_some_and(|blob| blob.starts_with(super::external_resolution::STORED_PREFIX)) } +/// Core's `isPlainTimer`: group, scope, cleanup and timeout timers have their +/// own outstanding dependencies and cannot be copied as an ordinary sleep. +fn plain_timer(payload: &Value) -> bool { + payload.is_object() && ["timer_kind", "cancellation_cleanup", "cancellation_scope_id", "parallel_group_id", + "condition_wait_id", "signal_wait_id"].iter().all(|field| payload[*field].is_null()) +} + /// What an accepted redrive continues from. struct Decision { resume: i64, @@ -59,6 +74,7 @@ fn plan(status: &str, events: &[(String, Value, DateTime)]) -> std::result: let mut failure: Option<&Value> = None; let mut steps: BTreeMap)>> = BTreeMap::new(); let mut markers: BTreeMap)> = BTreeMap::new(); + let mut timers: BTreeMap)>> = BTreeMap::new(); let mut last_activity: Option<(&str, &Value)> = None; for (kind, payload, recorded_at) in events { if failure.is_some() { @@ -83,6 +99,15 @@ fn plan(status: &str, events: &[(String, Value, DateTime)]) -> std::result: } markers.insert(sequence, (kind, payload, *recorded_at)); } + "TimerScheduled" | "TimerFired" => { + let Some(sequence) = payload["sequence"].as_i64().filter(|sequence| started && *sequence >= 1) else { + return Err("invalid_step_sequence"); + }; + if !plain_timer(payload) { + return Err("unsupported_timer"); + } + timers.entry(sequence).or_default().push((kind, payload, *recorded_at)); + } "ActivityScheduled" | "ActivityStarted" | "ActivityHeartbeatRecorded" | "ActivityCompleted" | "ActivityFailed" => { if payload["execution_mode"] == "local" || payload["local_activity"] == true { return Err("unsupported_local_activity"); @@ -105,14 +130,17 @@ fn plan(status: &str, events: &[(String, Value, DateTime)]) -> std::result: if failure["failed_step_kind"] != "activity" { return Err("unrecorded_failure_boundary"); } - // A marker takes its own step before the failed one. - if markers.keys().any(|sequence| *sequence >= boundary) { + // Reused markers and timers each take one step before the failed activity. + if markers.keys().chain(timers.keys()).any(|sequence| *sequence >= boundary) { return Err("invalid_failed_step"); } + if timers.keys().any(|sequence| steps.contains_key(sequence) || markers.contains_key(sequence)) { + return Err("duplicate_step_sequence"); + } if markers.keys().any(|sequence| steps.contains_key(sequence)) { return Err("duplicate_step_sequence"); } - if (steps.len() + markers.len()) as i64 != boundary { + if (steps.len() + markers.len() + timers.len()) as i64 != boundary { return Err("activity_sequence_gap"); } let mut reused = Vec::new(); @@ -125,7 +153,27 @@ fn plan(status: &str, events: &[(String, Value, DateTime)]) -> std::result: fields.remove("task"); fields.remove("command"); } - reused.push(Reused { kind: (*kind).to_owned(), payload, recorded_at: *recorded_at }); + reused.push(Reused { kind: (*kind).to_owned(), payload, recorded_at: *recorded_at, scheduled_payload: None }); + continue; + } + if let Some(events) = timers.get(&sequence) { + let [("TimerScheduled", scheduled, _), ("TimerFired", fired, recorded_at)] = events.as_slice() else { + return Err("incomplete_timer_step"); + }; + if !scheduled["timer_id"].is_string() || scheduled["timer_id"] != fired["timer_id"] { + return Err("incomplete_timer_step"); + } + let mut scheduled = (*scheduled).clone(); + let mut fired = (*fired).clone(); + for payload in [&mut scheduled, &mut fired] { + if let Some(fields) = payload.as_object_mut() { + for key in ["task", "command", "timer_id"] { + fields.remove(key); + } + } + } + reused.push(Reused { kind: "TimerFired".to_owned(), payload: fired, recorded_at: *recorded_at, + scheduled_payload: Some(scheduled) }); continue; } let Some(events) = steps.get(&sequence) else { @@ -170,7 +218,7 @@ fn plan(status: &str, events: &[(String, Value, DateTime)]) -> std::result: if !payload["result"]["blob"].is_string() { return Err("missing_activity_result"); } - reused.push(Reused { kind: "ActivityCompleted".to_owned(), payload: payload.clone(), recorded_at }); + reused.push(Reused { kind: "ActivityCompleted".to_owned(), payload: payload.clone(), recorded_at, scheduled_payload: None }); } match last_activity { Some(("ActivityFailed", payload)) if payload["sequence"].as_i64() == Some(boundary) => Ok((boundary, reused)), @@ -367,7 +415,7 @@ where Self::append(&mut tx, &run_id, "StartAccepted", json!({"workflow_command_id":command_id,"workflow_instance_id":workflow_id, "workflow_run_id":run_id,"workflow_class":workflow_type,"workflow_type":workflow_type,"outcome":"redriven"}), None, Some(&command_id)).await?; let source_started = DB::optional_instant(&source, "started_at")?; - let replayed = started["replayed_started_at"].as_str().map(str::to_owned).or_else(|| source_started.map(iso)); + let replayed = started["replayed_started_at"].as_str().map(str::to_owned).or_else(|| source_started.map(replay_time)); if !started.is_object() { started = json!({}); } @@ -379,11 +427,24 @@ where started[key] = value; } Self::append(&mut tx, &run_id, "WorkflowStarted", started, None, Some(&command_id)).await?; - // `FailedRunRedrivePlan::successorEvents`: each reused step in step + // Core's redrive writer copies each reused step in step // order. A step reused more than once keeps the time it first ran. for step in reused { let payload = &step.payload; - let recorded_at = payload["reused_recorded_at"].as_str().map(str::to_owned).unwrap_or_else(|| iso(step.recorded_at)); + let recorded_at = payload["reused_recorded_at"].as_str().map(str::to_owned).unwrap_or_else(|| replay_time(step.recorded_at)); + if let Some(mut scheduled) = step.scheduled_payload { + // This timer already fired. Reuse its two events with one new + // identity, without inserting a timer row or runnable task. + let timer_id = id(); + scheduled["timer_id"] = json!(timer_id); + Self::append(&mut tx, &run_id, "TimerScheduled", scheduled, None, None).await?; + let mut fired = step.payload; + fired["timer_id"] = json!(timer_id); + fired["reused_from_run_id"] = json!(source_id); + fired["reused_recorded_at"] = json!(recorded_at); + Self::append(&mut tx, &run_id, "TimerFired", fired, None, None).await?; + continue; + } if step.kind != "ActivityCompleted" { let mut marker = step.payload.clone(); marker["reused_from_run_id"] = json!(source_id); @@ -439,7 +500,7 @@ mod tests { assert_eq!(plan("failed", &unbounded).err(), Some("unrecorded_failure_boundary")); let mut timer = history.clone(); timer.insert(2, event("TimerScheduled", json!({"sequence":1}))); - assert_eq!(plan("failed", &timer).err(), Some("unsupported_history")); + assert_eq!(plan("failed", &timer).err(), Some("duplicate_step_sequence")); let mut external = history.clone(); external[3] = event("ActivityCompleted", json!({"sequence":1,"activity_type":"a","result":{"codec":"avro","external_payload":{}}})); assert_eq!(plan("failed", &external).err(), Some("external_activity_result_not_supported")); @@ -480,4 +541,78 @@ mod tests { missing.remove(3); assert_eq!(plan("failed", &missing).err(), Some("activity_sequence_gap")); } + + fn fired_timer_history() -> Vec<(String, Value, DateTime)> { + vec![event("WorkflowStarted", json!({})), + event("ActivityCompleted", json!({"sequence":1,"activity_type":"a","result":{"blob":"x"}})), + event("TimerScheduled", json!({"sequence":2,"timer_id":"source-timer","delay_seconds":5, + "task":{"id":"scheduled-task"},"command":{"id":"scheduled-command"}})), + event("TimerFired", json!({"sequence":2,"timer_id":"source-timer","delay_seconds":5, + "task":{"id":"fired-task"},"command":{"id":"fired-command"}})), + event("ActivityFailed", json!({"sequence":3,"activity_type":"b"})), + event("WorkflowFailed", json!({"failed_step_sequence":3,"failed_step_kind":"activity"}))] + } + + #[test] + fn plan_reuses_a_complete_plain_timer_in_step_order() { + let history = fired_timer_history(); + let original = history.clone(); + let (resume, reused) = plan("failed", &history).unwrap(); + assert_eq!((resume, reused.len()), (3, 2)); + assert_eq!(reused[0].kind, "ActivityCompleted"); + assert_eq!(reused[1].kind, "TimerFired"); + assert_eq!(reused[1].payload, json!({"sequence":2,"delay_seconds":5})); + assert_eq!(reused[1].scheduled_payload, Some(json!({"sequence":2,"delay_seconds":5}))); + assert_eq!(reused[1].recorded_at, history[3].2); + assert_eq!(history, original); + } + + #[test] + fn plan_refuses_non_plain_or_incomplete_timers_like_core() { + let history = fired_timer_history(); + for index in [2, 3] { + for field in ["timer_kind", "cancellation_cleanup", "cancellation_scope_id", "parallel_group_id", + "condition_wait_id", "signal_wait_id"] { + for value in [json!(false), json!(""), json!("condition")] { + let mut changed = history.clone(); + changed[index].1[field] = value; + assert_eq!(plan("failed", &changed).err(), Some("unsupported_timer"), "{field}"); + } + let mut changed = history.clone(); + changed[index].1[field] = Value::Null; + assert!(plan("failed", &changed).is_ok(), "null {field}"); + } + let mut missing = history.clone(); + missing.remove(index); + assert_eq!(plan("failed", &missing).err(), Some("incomplete_timer_step")); + let mut duplicate = history.clone(); + duplicate.insert(index, history[index].clone()); + assert_eq!(plan("failed", &duplicate).err(), Some("incomplete_timer_step")); + for value in [Value::Null, json!(5), json!("another-timer")] { + let mut changed = history.clone(); + changed[index].1["timer_id"] = value; + assert_eq!(plan("failed", &changed).err(), Some("incomplete_timer_step")); + } + for value in [Value::Null, json!(0), json!("2")] { + let mut changed = history.clone(); + changed[index].1["sequence"] = value; + assert_eq!(plan("failed", &changed).err(), Some("invalid_step_sequence")); + } + } + let mut reversed = history.clone(); + reversed.swap(2, 3); + assert_eq!(plan("failed", &reversed).err(), Some("incomplete_timer_step")); + for (sequence, expected) in [(1, "duplicate_step_sequence"), (3, "invalid_failed_step")] { + let mut changed = history.clone(); + changed[2].1["sequence"] = json!(sequence); + changed[3].1["sequence"] = json!(sequence); + assert_eq!(plan("failed", &changed).err(), Some(expected)); + } + let mut collision = history.clone(); + collision.insert(2, event("VersionMarkerRecorded", json!({"sequence":2}))); + assert_eq!(plan("failed", &collision).err(), Some("duplicate_step_sequence")); + let mut gap = history.clone(); + gap.remove(1); + assert_eq!(plan("failed", &gap).err(), Some("activity_sequence_gap")); + } } diff --git a/rust/tests/execution/redrive.rs b/rust/tests/execution/redrive.rs index d5c1d6411..497300070 100644 --- a/rust/tests/execution/redrive.rs +++ b/rust/tests/execution/redrive.rs @@ -19,6 +19,137 @@ pub(super) async fn failed_at_second_activity(app: &Router, workflow: &str) -> V failed_with(app, workflow, json!({"input":envelope(Payload::Array(vec![Payload::Long(9007199254740993)]))}), vec![]).await } +// Core workflow#992: a fired plain timer is replayed as a completed prefix +// step. The successor must neither wait again nor borrow the redrive's clock. +#[tokio::test] +async fn redrive_reuses_fired_timer_identity_and_original_clock_without_timer_work() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let registered = request(&app, "POST", "/api/worker/register", json!({"worker_id":"worker","task_queue":"test", + "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":["a","b"], + "workflow_definition_fingerprints":{"echo":FINGERPRINT}})).await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + let source = start(&app, "rd-timer").await; + let source_run = source["run_id"].as_str().unwrap(); + let task = poll(&app, "worker", "workflow").await; + assert_eq!(finish_task(&app, &task, json!([{"type":"schedule_activity","activity_type":"a", + "arguments":envelope(Payload::Array(vec![]))}])).await.0, StatusCode::OK); + complete_activity(&app, &poll(&app, "worker", "activity").await).await; + let task = poll(&app, "worker", "workflow").await; + assert_eq!(finish_task(&app, &task, json!([{"type":"start_timer","delay_seconds":1}])).await.0, StatusCode::OK); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15); + loop { + let history = run_history(&app, &source["workflow_id"], &source["run_id"]).await; + if kinds(&history).iter().any(|kind| kind == "TimerFired") { + break; + } + assert!(tokio::time::Instant::now() < deadline, "timer did not fire: {history}"); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + let task = poll(&app, "worker", "workflow").await; + assert_eq!(finish_task(&app, &task, json!([{"type":"schedule_activity","activity_type":"b", + "arguments":envelope(Payload::Array(vec![]))}])).await.0, StatusCode::OK); + fail_redrive_activity(&app, 3).await; + // Consumed clocks use full event precision; the initial anchor uses the + // persisted run start. Pin distinct ordered times after executing the + // real timer and activity lifecycle so those authorities cannot alias. + let recorded = run_history(&app, &source["workflow_id"], &source["run_id"]).await; + let clock_base = chrono::DateTime::parse_from_rfc3339("2026-11-03T00:00:00.123456+00:00").unwrap(); + for event in recorded.as_array().unwrap() { + let sequence = event["sequence"].as_i64().unwrap(); + let at = clock_base + chrono::Duration::seconds(sequence); + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_history_events SET recorded_at='{}' WHERE workflow_run_id='{source_run}' AND sequence={sequence}", + at.format("%Y-%m-%d %H:%M:%S%.6f")))).await; + } + let original_start = clock_base - chrono::Duration::seconds(1); + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_runs SET started_at='{}' WHERE id='{source_run}'", + original_start.format("%Y-%m-%d %H:%M:%S%.6f")))).await; + runtime.close().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let original = run_history(&app, &source["workflow_id"], &source["run_id"]).await; + let timer = |history: &Value, kind: &str| history.as_array().unwrap().iter() + .find(|event| event["event_type"] == kind).unwrap().clone(); + let original_fired = timer(&original, "TimerFired"); + let clock = |value: &Value| chrono::DateTime::parse_from_rfc3339(value.as_str().expect("public replay clock timestamp")).unwrap(); + let original_clock = clock(&original_fired["timestamp"]); + assert_eq!(original_clock.timestamp_subsec_micros(), 123456); + let counts = table_counts(&database).await; + let mut predecessor = source_run.to_owned(); + let mut previous_timer = original_fired["payload"]["timer_id"].clone(); + + // A second redrive changes the immediate predecessor, while retaining the + // first execution's time. Each new run gets one fresh shared timer ID. + for _ in 0..2 { + let (_, described) = request(&app, "GET", &format!("/api/workflows/rd-timer/runs/{predecessor}"), Value::Null).await; + assert_eq!(redrive_report(&described), (json!(true), Value::Null)); + let (status, accepted) = request(&app, "POST", &format!("/api/workflows/rd-timer/runs/{predecessor}/redrive"), json!({})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{accepted}"); + assert_eq!(accepted["resume_step_sequence"], 3); + let run = accepted["run_id"].as_str().unwrap(); + let history = run_history(&app, &source["workflow_id"], &accepted["run_id"]).await; + assert_eq!(kinds(&history), ["StartAccepted", "WorkflowStarted", "ActivityCompleted", "TimerScheduled", "TimerFired"]); + let scheduled = timer(&history, "TimerScheduled")["payload"].clone(); + let fired = timer(&history, "TimerFired")["payload"].clone(); + assert!(scheduled["timer_id"].is_string()); + assert_ne!(scheduled["timer_id"], previous_timer); + assert_eq!(scheduled["timer_id"], fired["timer_id"]); + assert_eq!(fired["reused_from_run_id"], predecessor); + assert_eq!(clock(&fired["reused_recorded_at"]), original_clock); + assert_eq!(clock(&history[1]["payload"]["replayed_started_at"]), original_start); + assert_ne!(original_start, clock(&timer(&original, "WorkflowStarted")["timestamp"])); + assert_eq!(clock(&history[2]["payload"]["reused_recorded_at"]), clock(&timer(&original, "ActivityCompleted")["timestamp"])); + for (kind, actual) in [("TimerScheduled", &scheduled), ("TimerFired", &fired)] { + let mut expected = timer(&original, kind)["payload"].clone(); + for field in ["task", "command", "timer_id"] { + expected.as_object_mut().unwrap().remove(field); + if field != "timer_id" { + assert!(actual.get(field).is_none(), "{kind}: {actual}"); + } + } + let mut actual = actual.clone(); + for field in ["timer_id", "reused_from_run_id", "reused_recorded_at"] { + actual.as_object_mut().unwrap().remove(field); + } + assert_eq!(actual, expected, "{kind}"); + } + // Physical timer inventories stay unchanged by redrive, on every backend. + let after = table_counts(&database).await; + let timer_count = |rows: &[(String, i64)]| rows.iter().find(|(name, _)| name == "workflow_run_timers") + .expect("the source catalog contains workflow_run_timers").1; + assert_eq!(timer_count(&after), timer_count(&counts)); + for sql in [format!("SELECT COUNT(*) FROM workflow_run_timers WHERE workflow_run_id='{run}'"), + format!("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id='{run}' AND task_type='timer'")] { + assert_eq!(task_retry::rows(&database, Box::leak(sql.into_boxed_str())).await, 0); + } + assert_eq!(run_history(&app, &source["workflow_id"], &source["run_id"]).await, original); + let task = poll(&app, "worker", "workflow").await; + assert_eq!(task["run_id"], accepted["run_id"]); + assert_eq!(clock(&task["replay_clock"]["started_at"]), original_start); + assert_eq!(finish_task(&app, &task, json!([{"type":"schedule_activity","activity_type":"b", + "arguments":envelope(Payload::Array(vec![]))}])).await.0, StatusCode::OK); + previous_timer = fired["timer_id"].clone(); + predecessor = run.to_owned(); + fail_redrive_activity(&app, 3).await; + } + runtime.close().await; + database.remove().await; +} + +async fn fail_redrive_activity(app: &Router, sequence: i64) { + let activity = poll(app, "worker", "activity").await; + assert_eq!(activity["activity_type"], "b", "{activity}"); + let failed = request(app, "POST", &format!("/api/worker/activity-tasks/{}/fail", activity["task_id"].as_str().unwrap()), + json!({"lease_owner":activity["lease_owner"],"activity_attempt_id":activity["activity_attempt_id"], + "failure":{"class":"App\\Boom","message":"boom"}})).await; + assert_eq!(failed.0, StatusCode::OK, "{}", failed.1); + let task = poll(app, "worker", "workflow").await; + let done = finish_task(app, &task, json!([{"type":"fail_workflow","message":"gave up","failed_step_sequence":sequence, + "failed_activity_execution_id":activity["activity_execution_id"]}])).await; + assert_eq!(done.1["run_status"], "failed", "{}", done.1); +} + // Published PHP 2.11.1's redrive continues a run that failed at an activity // step in a new run: its history reuses the completed activities before the // step, so the worker reschedules only the failed one. @@ -265,7 +396,7 @@ async fn redrive_reuses_side_effects_and_version_markers_at_their_recorded_time( let source = failed_with(&app, "rd-markers", json!({}), markers).await; let source_run = source["run_id"].as_str().unwrap().to_owned(); // The source's steps ran earlier than the redrive. - database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_history_events SET recorded_at='2026-11-03 00:00:00.000000' WHERE workflow_run_id='{source_run}' AND event_type IN ('SideEffectRecorded','VersionMarkerRecorded')"))).await; + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_history_events SET recorded_at='2026-11-03 00:00:00.654321' WHERE workflow_run_id='{source_run}' AND event_type IN ('SideEffectRecorded','VersionMarkerRecorded')"))).await; let source_history = run_history(&app, &json!("rd-markers"), &json!(source_run)).await; let recorded = |history: &Value, kind: &str| history.as_array().unwrap().iter().find(|event| event["event_type"] == kind).unwrap()["payload"].clone(); let path = format!("/api/workflows/rd-markers/runs/{source_run}"); @@ -277,7 +408,7 @@ async fn redrive_reuses_side_effects_and_version_markers_at_their_recorded_time( let run = redriven["run_id"].as_str().unwrap().to_owned(); let history = run_history(&app, &json!("rd-markers"), &json!(run)).await; assert_eq!(kinds(&history), ["StartAccepted", "WorkflowStarted", "SideEffectRecorded", "VersionMarkerRecorded", "ActivityCompleted"]); - let first_ran = json!("2026-11-03T00:00:00+00:00"); + let first_ran = json!("2026-11-03T00:00:00.654321+00:00"); for (kind, sequence) in [("SideEffectRecorded", 1), ("VersionMarkerRecorded", 2)] { let mut reused = recorded(&history, kind); assert_eq!((reused["sequence"].clone(), reused["reused_from_run_id"].clone(), reused["reused_recorded_at"].clone()),