diff --git a/docs/contracts/child-start-turn.md b/docs/contracts/child-start-turn.md new file mode 100644 index 000000000..a30d9f6de --- /dev/null +++ b/docs/contracts/child-start-turn.md @@ -0,0 +1,66 @@ +# Child start turn + +A workflow can start a child, get its committed identity in a fresh parent +turn while the child still runs, do other work, and later receive the child's +result (Server #966, the runtime side of sdk-php#185). No activity, timer, +signal or query is needed to get that turn. This is a native capability; +published PHP Server 2.x has no such mode. Workflow Core parity is +workflow#989. + +## Command + +`start_child_workflow` takes an optional `await`: + +- `result` (the default, and the behavior when the field is absent): the + parent's next turn comes when the child resolves, or from other work. The + completion creates only the child's task. +- `started`: the completion that commits `ChildWorkflowScheduled` and + `ChildRunStarted` also creates one new parent workflow task, with a fresh + task id, lease and claim. Its poll carries both committed events, so the + child call id, the child instance id and the initial child run id come from + canonical history. The child keeps running, and its resolution gives the + parent its usual later turn. + +Any other value, or `await` on another command type, is refused with 422 and +the validation message `await must be result or started.` (or the field-scope +message) before anything is written. + +## Turn rules + +- One turn per completion, however many children in it ask for one. +- No turn when the same completion closes the parent (complete, fail or + continue as new), and none when the parent already has a ready or leased + workflow task, which delivers the same history. +- The task payload names the wait `workflow_wait_kind: "child_started"`, with + `child_call_id`, `child_workflow_run_id`, `resume_source_kind: + "child_workflow_run"`, `resume_source_id`, `workflow_sequence` and + `workflow_event_type: "ChildRunStarted"` of the first such child. +- The completion stays under its immutable task receipt: a duplicate + completion, redelivery or retry of the original task starts no second child + and creates no second turn, and the original claim stays closed. +- A cancellation-scope checkpoint keeps the parent's claim, so it refuses a + child start with `await: "started"` (`child_start_turn_not_at_checkpoint`) + before any child starts. +- Parent-close policy (Abandon included), cancellation propagation, deadlines, + retries, child continue-as-new (`child_call_id` stays; the initial run id is + frozen in `ChildRunStarted`) and namespace isolation are unchanged. + +## Advertisement + +Native advertises `child_workflow_start_turn: true` in cluster info +`capabilities` and in worker `server_capabilities`. A runtime that does not +advertise it (native 3.0.x, PHP 2.x) ignores unknown command fields, so an SDK +must check the capability and refuse `await: "started"` itself before +submitting the child. + +Tests (`tests/execution/child_start_turn.rs`, +`tests/execution/scope_checkpoints.rs`) cover: + +- the flagged start, with its fresh claim and committed identity, the + duplicate completion, and child continue-as-new resolving the original call; +- the default and explicit `result` with an unchanged task count; +- several children in one completion, and a closing completion; +- refusal of an unknown value and of the checkpoint case; +- the advertisement. + +They run on SQLite, PostgreSQL and MySQL. diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index ae3c33a9d..b38184cd4 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -16,6 +16,19 @@ use chrono::{DateTime, Utc}; use serde_json::{Value, json}; use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; +/// `start_child_workflow.await` (Server #966): the default waits for the +/// child's result; `started` also gives the parent a turn when the start +/// commits. +pub(super) const AWAIT_RESULT: &str = "result"; +pub(super) const AWAIT_STARTED: &str = "started"; + +/// A started child: its first workflow task, and the start turn payload a +/// parent gets when it asked for one. +pub(super) struct StartedChild { + pub(super) task: String, + pub(super) turn: Value, +} + #[derive(Clone, Copy)] pub(super) enum ChildResolution<'a> { Completed(&'a str), @@ -43,6 +56,7 @@ pub(super) fn validate_child(command: &Value) -> Result<()> { "retry_policy", "execution_timeout_seconds", "run_timeout_seconds", + "await", ], )?; if text(command, "workflow_type")?.len() > 191 @@ -297,7 +311,7 @@ where task_id: &str, command: &Value, sequence: i64, - ) -> Result { + ) -> Result { let parent_run = DB::string(parent, "id")?; let parent_instance = DB::string(parent, "workflow_instance_id")?; // Published PHP starts a child whatever else the parent has open: @@ -382,7 +396,45 @@ where .bind(&parent_run) .execute(&mut **tx) .await?; - Ok(child_task) + // What a start turn names: this call, frozen at its first run. + let turn = json!({"workflow_wait_kind":"child_started","child_call_id":call,"child_workflow_run_id":child_run, + "resume_source_kind":"child_workflow_run","resume_source_id":child_run,"workflow_sequence":sequence, + "workflow_event_type":"ChildRunStarted"}); + Ok(StartedChild { task: child_task, turn }) + } + + /// The parent's start turn (Server #966): one fresh workflow task, created + /// with the completion that committed an `await: "started"` child start, + /// unless that completion closed the parent. As for a child's resolution + /// (`child_parent_resume`), only a ready workflow task that replays the + /// parent can carry the committed `ChildWorkflowScheduled` and + /// `ChildRunStarted` history instead: an update task does not replay the + /// workflow, and a leased task may have read history before the start. + pub(super) async fn child_start_turn( + tx: &mut Transaction<'_, DB>, + run_id: &str, + task_id: &str, + queue: &str, + turn: Value, + ) -> Result> { + let open = Self::scalar("SELECT COUNT(*) FROM workflow_runs WHERE id=$1 AND status IN ('pending','running','waiting')") + .bind(run_id) + .fetch_one(&mut **tx) + .await?; + if open == 0 { + return Ok(None); + } + let ready = Self::query("SELECT payload FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status='ready' AND id<>$2 ORDER BY id LIMIT 1001") + .bind(run_id) + .bind(task_id) + .fetch_all(&mut **tx) + .await?; + for task in &ready { + if DB::document_row(task, "payload")?["workflow_update_id"].as_str().is_none() { + return Ok(None); + } + } + Self::create_task(tx, run_id, queue, "workflow", turn).await.map(Some) } pub(super) async fn complete_child( diff --git a/rust/src/runtime/cluster_info.rs b/rust/src/runtime/cluster_info.rs index 17bcea912..fa5617c5f 100644 --- a/rust/src/runtime/cluster_info.rs +++ b/rust/src/runtime/cluster_info.rs @@ -61,6 +61,7 @@ pub(super) fn capabilities() -> Value { "activity_retry_policy", "child_workflow_retry_policy", "child_workflow_timeouts", + "child_workflow_start_turn", "parent_close_policy", "non_retryable_failures", "worker_sessions", @@ -152,7 +153,9 @@ mod tests { "workflow_update_runtime_contract", "waterline_v1_projection", "embedded_v2_import"] { assert!(caps.get(name).is_some(), "{name}"); } - assert_eq!(caps.as_object().unwrap().len(), 56); + // PHP's names plus native's child start turn (Server #966). + assert_eq!(caps.as_object().unwrap().len(), 57); + assert_eq!(caps["child_workflow_start_turn"], json!(true)); assert_eq!(caps["response_compression"], json!([])); assert_eq!(caps["timer_runtime_contract"], json!(false)); assert_eq!(caps["principal_attribution_contract"], json!(true)); diff --git a/rust/src/runtime/completion_rules.rs b/rust/src/runtime/completion_rules.rs index 8d9be4fc7..7b834f8f2 100644 --- a/rust/src/runtime/completion_rules.rs +++ b/rust/src/runtime/completion_rules.rs @@ -494,6 +494,14 @@ fn controller_scopes(commands: &[Value]) -> Vec<(String, Vec)> { add(at(field), format!("{field} is only supported for schedule_activity commands.")); } } + // Native's child start turn (Server #966): `result`, the default, or + // `started`. + if kind == "start_child_workflow" + && has_value(command, "await") + && !matches!(command["await"].as_str(), Some(super::children::AWAIT_RESULT | super::children::AWAIT_STARTED)) + { + add(at("await"), "await must be result or started.".into()); + } if has_value(command, "worker_session") && kind != "schedule_activity" { add(at("worker_session"), "worker_session is only supported for schedule_activity commands.".into()); } @@ -576,6 +584,8 @@ const FIELD_SCOPES: &[(&str, &[&str], &str)] = &[ "failed_step_sequence identifies an uncaught activity failure and only applies to a fail_workflow command."), ("failed_activity_execution_id", &["fail_workflow"], "failed_activity_execution_id identifies an uncaught activity failure and only applies to a fail_workflow command."), + ("await", &["start_child_workflow"], + "await chooses whether a child start gives the parent its next turn when the start commits (started) or when the child resolves (result) and only applies to a start_child_workflow command."), ("parent_close_policy", &["start_child_workflow"], "parent_close_policy declares how a child workflow reacts when its parent closes and only applies to a start_child_workflow command."), ("cancellation_policy", &["start_child_workflow", "schedule_activity"], @@ -760,7 +770,7 @@ const READS: &[(&str, &[&str])] = &[ "cancellation_policy", "worker_session"]), ("start_timer", &["delay_seconds", "cancellation_cleanup"]), ("start_child_workflow", &["workflow_type", "arguments", "payload_codec", "connection", "queue", "retry_policy", - "execution_timeout_seconds", "run_timeout_seconds", "parent_close_policy", "cancellation_policy"]), + "execution_timeout_seconds", "run_timeout_seconds", "parent_close_policy", "cancellation_policy", "await"]), ("continue_as_new", &["arguments", "payload_codec", "workflow_type", "queue"]), ("complete_update", &["update_id", "result", "payload_codec"]), ("fail_update", &["update_id", "message", "exception_class", "exception_type", "non_retryable"]), @@ -785,7 +795,7 @@ const SCOPED_FIELDS: &[&str] = &["retry_policy", "start_to_close_timeout", "sche "non_retryable", "exception", "failed_step_sequence", "failed_activity_execution_id", "timeout_kind", "parent_close_policy", "cancellation_scope_id", "cancellation_cleanup", "delay_seconds", "timeout_seconds", "payload_codec", "entries", "principal_subject", "principal_method", "principal_roles", "principal_tenant", - "principal_claims"]; + "principal_claims", "await"]; /// `optionalRetryPolicy` reads these keys and ignores the rest. const RETRY_POLICY_FIELDS: &[&str] = &["max_attempts", "backoff_seconds", "non_retryable_error_types"]; diff --git a/rust/src/runtime/scope_checkpoint.rs b/rust/src/runtime/scope_checkpoint.rs index 07da56cd1..d24d1eca7 100644 --- a/rust/src/runtime/scope_checkpoint.rs +++ b/rust/src/runtime/scope_checkpoint.rs @@ -276,6 +276,13 @@ store_methods! { } Err(error) => return Err(error), } + // A checkpoint keeps the parent's claim, so a child start turn has + // nothing to resume; it belongs to the task's completion (Server #966). + if commands.iter().any(|command| command["type"] == "start_child_workflow" && command["await"] == super::children::AWAIT_STARTED) { + tx.rollback().await?; + return refusal("child_start_turn_not_at_checkpoint"); + } + let start_turn = std::sync::Mutex::new(None); let apply = super::store::Apply { run: &run, task: &task, @@ -286,6 +293,7 @@ store_methods! { external, local, exceptions, + start_turn: &start_turn, }; let mut created = Vec::new(); for (index, command) in commands.iter().enumerate() { diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 816028756..dafab86b0 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -3188,6 +3188,7 @@ where Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false, commands, signal).await?; // Published parseCommands applies every other command first, whatever // the terminal command's position in the batch. + let start_turn = std::sync::Mutex::new(None); let apply = Apply { run: &run, task: &task, @@ -3198,6 +3199,7 @@ where external: &external, local: &local, exceptions: &exceptions, + start_turn: &start_turn, }; let order = (0..commands.len()) .filter(|index| !is_terminal(&commands[*index])) @@ -3295,6 +3297,12 @@ where if let Some(next) = Self::next_message_task(&mut tx, &run_id, &DB::string(&run, "queue")?, attempted, false).await? { created.push(next); } + let start_turn = start_turn.into_inner().unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(turn) = start_turn + && let Some(next) = Self::child_start_turn(&mut tx, &run_id, task_id, &DB::string(&run, "queue")?, turn).await? + { + created.push(next); + } Self::record_completion(&mut tx, task_id, &json!({COMPLETION_RECEIPT: {"request": body, "created_task_ids": created}})).await?; // PHP's executor leaves every run a completed task did not close // waiting, even when the task recorded only markers or side effects. @@ -3337,7 +3345,7 @@ where sequence_ref: &mut i64, created: &mut Vec, ) -> Result<()> { - let Apply { run, task, task_id, run_id, attempt, limits, external, local, exceptions } = *apply; + let Apply { run, task, task_id, run_id, attempt, limits, external, local, exceptions, start_turn } = *apply; let mut sequence = *sequence_ref; if matches!( command["type"].as_str(), @@ -3352,7 +3360,12 @@ where } else if command["type"] == "start_child_workflow" { Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Children, &limits, task_id, attempt).await?; sequence += 1; - created.push(Self::start_child(tx, &run, task_id, command, sequence).await?); + let started = Self::start_child(tx, &run, task_id, command, sequence).await?; + created.push(started.task); + if command["await"] == super::children::AWAIT_STARTED { + let mut turn = start_turn.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + turn.get_or_insert(started.turn); + } } else if command["type"] == super::local_activities::COMMAND { // `applyRecordLocalActivity`: one authored sequence. sequence += 1; @@ -5071,6 +5084,9 @@ pub(super) struct Apply<'a, R> { pub(super) external: &'a super::external_resolution::PreparedCommands, pub(super) local: &'a std::collections::BTreeMap, pub(super) exceptions: &'a std::collections::BTreeMap, + /// The first child this batch started with `await: "started"`: the + /// parent's start turn (Server #966). + pub(super) start_turn: &'a std::sync::Mutex>, } impl Clone for Apply<'_, R> { @@ -5195,7 +5211,8 @@ pub(crate) fn capabilities() -> Value { "sticky_execution": false, "local_activities": super::local_activities::manifest(), "message_streams": super::message_streams::worker_capability(), "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(), "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true, - "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true}); + "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true, + "child_workflow_start_turn": true}); value["runtime_external_payload_transport"] = super::payload_discovery::transport_manifest(); value["supported_workflow_task_commands"] .as_array_mut() diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c7b547bd4..fed2ee077 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -370,6 +370,8 @@ mod key_order; mod activity_result_repeats; #[path = "execution/task_fairness.rs"] mod task_fairness; +#[path = "execution/child_start_turn.rs"] +mod child_start_turn; enum TestDatabase { Sqlite(tempfile::TempDir), diff --git a/rust/tests/execution/child_start_turn.rs b/rust/tests/execution/child_start_turn.rs new file mode 100644 index 000000000..021b748f5 --- /dev/null +++ b/rust/tests/execution/child_start_turn.rs @@ -0,0 +1,281 @@ +use super::*; + +// Server #966: `start_child_workflow.await: "started"` gives the parent one +// fresh turn in the transaction that commits the child start, while the +// child is still open. The default, `result`, keeps today's task cost. + +async fn count(database: &TestDatabase, sql: String) -> i64 { + task_retry::rows(database, Box::leak(sql.into_boxed_str())).await +} + +async fn parent_tasks(database: &TestDatabase, run: &Value) -> i64 { + let run = run.as_str().unwrap(); + count(database, format!("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id='{run}' AND task_type='workflow'")).await +} + +async fn child_calls(database: &TestDatabase, run: &Value) -> i64 { + let run = run.as_str().unwrap(); + count(database, format!("SELECT COUNT(*) FROM workflow_child_calls WHERE parent_workflow_run_id='{run}'")).await +} + +fn child(mode: Option<&str>) -> Value { + let mut command = json!({"type":"start_child_workflow","workflow_type":"child","queue":"children", + "arguments":envelope(Payload::Array(vec![]))}); + if let Some(mode) = mode { + command["await"] = json!(mode); + } + command +} + +/// `worker` takes parents on the default queue and `child-worker` children +/// on their own, so a parent poll never claims a child. +async fn workers(app: &Router) { + // The parent worker also takes the `set` update and `finish` signal. + update_worker(app, "worker").await; + let (status, body) = request(app, "POST", "/api/worker/register", json!({"worker_id":"child-worker","task_queue":"children", + "runtime":"php","supported_workflow_types":["child"],"supported_activity_types":[]})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +async fn child_poll(app: &Router) -> Value { + let (status, body) = request(app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":"child-worker","task_queue":"children","timeout_seconds":0})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + body["task"].clone() +} + +fn events<'a>(task: &'a Value, kind: &str) -> Vec<&'a Value> { + task["history_events"].as_array().unwrap().iter().filter(|event| event["event_type"] == kind).collect() +} + +#[tokio::test] +async fn a_started_child_gives_the_parent_a_fresh_turn_while_it_runs() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + let started = start(&app, "turning").await; + let first = poll(&app, "worker", "workflow").await; + let (status, body) = finish_task(&app, &first, json!([child(Some("started"))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["created_task_ids"].as_array().unwrap().len(), 2, "the child's task and the parent's turn: {body}"); + assert_eq!(body["server_capabilities"]["child_workflow_start_turn"], true, "{body}"); + + // The turn is a fresh claim on the parent, with the committed identity in + // its history; the child has not run yet. + let turn = poll(&app, "worker", "workflow").await; + assert_eq!((turn["workflow_id"].clone(), turn["run_id"].clone()), (json!("turning"), started["run_id"].clone()), "{turn}"); + assert_ne!(turn["task_id"], first["task_id"]); + let scheduled = events(&turn, "ChildWorkflowScheduled"); + let run_started = events(&turn, "ChildRunStarted"); + assert_eq!((scheduled.len(), run_started.len()), (1, 1), "{turn}"); + let call = run_started[0]["payload"]["child_call_id"].clone(); + let initial_run = run_started[0]["payload"]["child_workflow_run_id"].clone(); + assert_eq!(scheduled[0]["payload"]["child_call_id"], call); + assert!(events(&turn, "ChildRunCompleted").is_empty()); + + // A duplicate of the original completion starts no second child and + // creates no second turn. + let (status, duplicate) = finish_task(&app, &first, json!([child(Some("started"))])).await; + assert_eq!((status, duplicate["recorded"].clone(), duplicate["reason"].clone()), + (StatusCode::OK, json!(false), json!("already_completed")), "{duplicate}"); + assert_eq!((child_calls(&database, &started["run_id"]).await, parent_tasks(&database, &started["run_id"]).await), (1, 2)); + + // The turn does other work; the parent then waits on the child. + let (status, body) = finish_task(&app, &turn, json!([{"type":"record_side_effect","result":envelope(Payload::Long(5))}])).await; + assert_eq!((status, body["run_status"].clone()), (StatusCode::OK, json!("waiting")), "{body}"); + assert!(poll(&app, "worker", "workflow").await.is_null(), "no further parent turn before the child resolves"); + + // The child continues as new and then completes: the call keeps its + // identity, and the parent's later turn carries the result. + let child_first = child_poll(&app).await; + assert_eq!(child_first["run_id"], initial_run, "{child_first}"); + let (status, body) = finish_task(&app, &child_first, json!([{"type":"continue_as_new"}])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let child_next = child_poll(&app).await; + assert_ne!(child_next["run_id"], initial_run); + let (status, body) = finish_task(&app, &child_next, json!([{"type":"complete_workflow","result":envelope(Payload::Long(7))}])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let resumed = poll(&app, "worker", "workflow").await; + assert_eq!(resumed["workflow_id"], "turning", "{resumed}"); + let completed = events(&resumed, "ChildRunCompleted"); + assert_eq!(completed.len(), 1, "{resumed}"); + assert_eq!(completed[0]["payload"]["child_call_id"], call); + assert_eq!(events(&resumed, "ChildRunStarted")[0]["payload"]["child_workflow_run_id"], initial_run, "the initial run id is frozen"); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn the_default_start_adds_no_parent_task() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + for (workflow, mode) in [("unflagged", None), ("result", Some("result"))] { + let started = start(&app, workflow).await; + let first = poll(&app, "worker", "workflow").await; + assert_eq!(first["workflow_id"], workflow); + let (status, body) = finish_task(&app, &first, json!([child(mode)])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["created_task_ids"].as_array().unwrap().len(), 1, "{workflow}: only the child's task: {body}"); + assert_eq!(parent_tasks(&database, &started["run_id"]).await, 1, "{workflow}"); + assert!(poll(&app, "worker", "workflow").await.is_null(), "{workflow}: no parent turn"); + } + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn one_turn_for_several_started_children_and_none_when_the_parent_closes() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + + let several = start(&app, "several").await; + let first = poll(&app, "worker", "workflow").await; + let (status, body) = finish_task(&app, &first, json!([child(Some("started")), child(Some("started"))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["created_task_ids"].as_array().unwrap().len(), 3, "two children and one turn: {body}"); + assert_eq!((child_calls(&database, &several["run_id"]).await, parent_tasks(&database, &several["run_id"]).await), (2, 2)); + let turn = poll(&app, "worker", "workflow").await; + assert_eq!(events(&turn, "ChildRunStarted").len(), 2, "{turn}"); + assert_eq!(turn["workflow_id"], "several"); + let (status, _) = finish_task(&app, &turn, json!([{"type":"complete_workflow","result":envelope(Payload::Long(1))}])).await; + assert_eq!(status, StatusCode::OK); + + let closing = start(&app, "closing").await; + let first = poll(&app, "worker", "workflow").await; + assert_eq!(first["workflow_id"], "closing"); + let (status, body) = finish_task(&app, &first, + json!([child(Some("started")), {"type":"complete_workflow","result":envelope(Payload::Long(1))}])).await; + assert_eq!((status, body["run_status"].clone()), (StatusCode::OK, json!("completed")), "{body}"); + assert_eq!(parent_tasks(&database, &closing["run_id"]).await, 1, "no turn for a closed parent"); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn an_unknown_await_is_refused_before_any_child_and_the_capability_is_advertised() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + let started = start(&app, "refused").await; + let first = poll(&app, "worker", "workflow").await; + for command in [child(Some("eventually")), json!({"type":"start_timer","delay_seconds":5,"await":"started"})] { + let (status, body) = finish_task(&app, &first, json!([command])).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert!(body.to_string().contains("commands.0.await"), "{body}"); + } + assert_eq!(child_calls(&database, &started["run_id"]).await, 0); + let (status, info) = request(&app, "GET", "/api/cluster/info", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{info}"); + assert_eq!(info["capabilities"]["child_workflow_start_turn"], true, "{info}"); + runtime.close().await; + database.remove().await; +} + +// An update accepted while the parent's task is claimed gets its own task +// at that task's completion. It does not replay the workflow, so it cannot +// stand in for the start turn: the parent gets both. +#[tokio::test] +async fn an_update_task_does_not_replace_the_start_turn() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + let started = start(&app, "updated-turn").await; + let first = poll(&app, "worker", "workflow").await; + let (status, update) = request(&app, "POST", "/api/workflows/updated-turn/update/set", json!({"input":[5]})).await; + assert!(status.is_success(), "{update}"); + let (status, body) = finish_task(&app, &first, json!([child(Some("started"))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let mut turns = Vec::new(); + for _ in 0..2 { + turns.push(poll(&app, "worker", "workflow").await); + } + let root: Vec<&Value> = turns.iter().filter(|task| task["workflow_update_id"].is_null()).collect(); + assert_eq!(root.len(), 1, "one root start turn beside the update task: {turns:?}"); + assert_eq!(events(root[0], "ChildRunStarted").len(), 1, "{}", root[0]); + assert!(turns.iter().any(|task| task["workflow_update_id"].is_string()), "the update task remains: {turns:?}"); + assert_eq!(parent_tasks(&database, &started["run_id"]).await, 3, "the claim, the update task and the turn"); + runtime.close().await; + database.remove().await; +} + +// A ready root task, here a declared signal's, replays the parent with the +// committed start in its history, so it carries the turn and none is added. +#[tokio::test] +async fn a_ready_root_task_carries_the_start_turn() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + let started = start(&app, "signalled-turn").await; + let first = poll(&app, "worker", "workflow").await; + let (status, body) = request(&app, "POST", "/api/workflows/signalled-turn/signal/finish", json!({"input":[1]})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let (status, body) = finish_task(&app, &first, json!([child(Some("started"))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(parent_tasks(&database, &started["run_id"]).await, 2, "the signal's task carries the turn: {body}"); + let carried = poll(&app, "worker", "workflow").await; + assert_eq!(carried["run_id"], started["run_id"]); + assert_eq!(events(&carried, "ChildRunStarted").len(), 1, "{carried}"); + assert!(poll(&app, "worker", "workflow").await.is_null(), "no second root turn"); + runtime.close().await; + database.remove().await; +} + +// A leased task's worker may have read history before the start, so it is +// not taken as the turn. +#[tokio::test] +async fn a_leased_task_does_not_replace_the_start_turn() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + let started = start(&app, "leased-turn").await; + let first = poll(&app, "worker", "workflow").await; + let run = started["run_id"].as_str().unwrap(); + // Another root task of the run, claimed elsewhere before the start. + database.execute(sqlx::AssertSqlSafe(format!( + "INSERT INTO workflow_tasks(id,workflow_run_id,namespace,task_type,status,payload,queue,lease_owner,lease_expires_at,attempt_count,available_at,created_at,updated_at) \ + SELECT '01M6FZZZZZZZZZZZZZZZZZZZZZ',workflow_run_id,namespace,'workflow','leased',payload,queue,'elsewhere','2037-01-01 00:00:00',1,available_at,created_at,updated_at \ + FROM workflow_tasks WHERE id='{}'", first["task_id"].as_str().unwrap()))).await; + let (status, body) = finish_task(&app, &first, json!([child(Some("started"))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(parent_tasks(&database, &started["run_id"]).await, 3, "the claim, the leased task and a fresh turn: {body}"); + let turn = poll(&app, "worker", "workflow").await; + assert_eq!(turn["run_id"], json!(run), "a fresh start turn beside the leased task: {body}"); + assert_ne!(turn["task_id"], "01M6FZZZZZZZZZZZZZZZZZZZZZ", "not the leased task"); + assert_eq!(events(&turn, "ChildRunStarted").len(), 1, "{turn}"); + runtime.close().await; + database.remove().await; +} + +// The child finishes before the start turn is claimed: the ready turn +// carries the resolution too, with no second task and no stuck parent. +#[tokio::test] +async fn a_child_finishing_before_the_turn_is_claimed_resolves_in_that_turn() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + workers(&app).await; + let started = start(&app, "fast-child").await; + let first = poll(&app, "worker", "workflow").await; + let (status, body) = finish_task(&app, &first, json!([child(Some("started"))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let child_task = child_poll(&app).await; + let (status, body) = finish_task(&app, &child_task, json!([{"type":"complete_workflow","result":envelope(Payload::Long(3))}])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(parent_tasks(&database, &started["run_id"]).await, 2, "the ready turn is the resume too"); + let turn = poll(&app, "worker", "workflow").await; + assert_eq!((events(&turn, "ChildRunStarted").len(), events(&turn, "ChildRunCompleted").len()), (1, 1), "{turn}"); + let (status, body) = finish_task(&app, &turn, json!([{"type":"complete_workflow","result":envelope(Payload::Long(3))}])).await; + assert_eq!((status, body["run_status"].clone()), (StatusCode::OK, json!("completed")), "{body}"); + assert!(poll(&app, "worker", "workflow").await.is_null()); + runtime.close().await; + database.remove().await; +} diff --git a/rust/tests/execution/scope_checkpoints.rs b/rust/tests/execution/scope_checkpoints.rs index b126a28a9..315283cd0 100644 --- a/rust/tests/execution/scope_checkpoints.rs +++ b/rust/tests/execution/scope_checkpoints.rs @@ -305,3 +305,30 @@ async fn a_command_after_an_update_completion_takes_the_following_sequence() { "{update} {marker_event}"); } } + +// A checkpoint keeps its claim, so a child start turn (Server #966) has no +// turn to give there: the prefix is refused before any child starts. The +// same start without the flag still checkpoints. +#[tokio::test] +async fn a_prefix_refuses_a_child_start_turn() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "scope-worker", json!(["cooperative_cancellation"])).await; + let run = start(&app, "scope-child-turn").await; + let task = claim(&app, "scope-worker").await; + let child = |mode: Option<&str>| { + let mut command = json!({"type":"start_child_workflow","workflow_type":"child","arguments":envelope(Payload::Array(vec![]))}); + if let Some(mode) = mode { + command["await"] = json!(mode); + } + command + }; + let (status, refused) = checkpoint(&app, &task, + json!({"checkpoint_id":"turn","start_sequence":1,"commands":[child(Some("started"))]}), "1.20").await; + assert_eq!((status, refused["reason"].clone()), (StatusCode::CONFLICT, json!("child_start_turn_not_at_checkpoint")), "{refused}"); + assert!(!history_types(&run_history(&app, &run["workflow_id"], &run["run_id"]).await).contains(&"ChildWorkflowScheduled".to_owned())); + let (status, receipt) = checkpoint(&app, &task, + json!({"checkpoint_id":"turn","start_sequence":1,"commands":[child(Some("result"))]}), "1.20").await; + assert_eq!(status, StatusCode::OK, "{receipt}"); + assert!(history_types(&run_history(&app, &run["workflow_id"], &run["run_id"]).await).contains(&"ChildWorkflowScheduled".to_owned())); +} diff --git a/scripts/conformance/server-parity/cluster-capabilities-contract.mjs b/scripts/conformance/server-parity/cluster-capabilities-contract.mjs index 458289de7..3dbfa5380 100644 --- a/scripts/conformance/server-parity/cluster-capabilities-contract.mjs +++ b/scripts/conformance/server-parity/cluster-capabilities-contract.mjs @@ -5,7 +5,8 @@ import assert from 'node:assert/strict'; // against published Server 2.11.7. Published PHP answers `expected` exactly. // Native (`implementation: "rust"`) answers it with the fixture's reviewed // `native_differences`, each naming a member native does not serve or does -// not publish, and names the members it leaves out in `native_unqualified`. +// not publish, and its reviewed `native_additions`, each a member published +// PHP never had, and names the members it leaves out in `native_unqualified`. export function expectedFor(definition, implementation) { const expected = structuredClone(definition.expected); if (implementation !== 'rust') return expected; @@ -16,6 +17,14 @@ export function expectedFor(definition, implementation) { assert.ok(Object.hasOwn(parent, last), `reviewed native difference ${path} names a published member`); if (value === '{absent}') delete parent[last]; else parent[last] = value; } + // Reviewed members only native serves, which published PHP never had. + for (const [path, {value}] of Object.entries(definition.native_additions ?? {})) { + const keys = path.split('.'); + const last = keys.pop(); + const parent = keys.reduce((node, key) => node[key], expected); + assert.ok(!Object.hasOwn(parent, last), `reviewed native addition ${path} is not a published member`); + parent[last] = value; + } return expected; } diff --git a/tests/Fixtures/ServerParity/cluster-capabilities.json b/tests/Fixtures/ServerParity/cluster-capabilities.json index 18d877b22..059b1e675 100644 --- a/tests/Fixtures/ServerParity/cluster-capabilities.json +++ b/tests/Fixtures/ServerParity/cluster-capabilities.json @@ -337,6 +337,12 @@ ], "reason": "Native names each member it leaves out." } + }, + "native_additions": { + "capabilities.child_workflow_start_turn": { + "value": true, + "reason": "Native serves start_child_workflow await: \"started\", a parent turn when the child start commits (Server #966); published PHP 2.x has no such mode." + } } } } diff --git a/tests/Unit/ServerParityClusterCapabilitiesContractTest.mjs b/tests/Unit/ServerParityClusterCapabilitiesContractTest.mjs index 224abe143..c32cd6dfa 100644 --- a/tests/Unit/ServerParityClusterCapabilitiesContractTest.mjs +++ b/tests/Unit/ServerParityClusterCapabilitiesContractTest.mjs @@ -34,7 +34,21 @@ test('every reviewed native difference names a published member and gives a reas assert.doesNotThrow(() => expectedFor(definition, 'rust')); }); +test('every reviewed native addition is a member published PHP lacks and gives a reason', () => { + for (const [path, addition] of Object.entries(definition.native_additions)) { + const member = path.split('.').pop(); + assert.ok(!Object.hasOwn(definition.expected.capabilities, member), path); + assert.ok(typeof addition.reason === 'string' && addition.reason.length > 10, path); + } + assert.equal(expectedFor(definition, 'rust').capabilities.child_workflow_start_turn, true); + assert.ok(!Object.hasOwn(expectedFor(definition, null).capabilities, 'child_workflow_start_turn'), 'published PHP never had it'); + const shadowing = structuredClone(definition); + shadowing.native_additions = {'capabilities.child_workflows': {value: false, reason: 'not an addition at all'}}; + assert.throws(() => expectedFor(shadowing, 'rust'), /is not a published member/); +}); + for (const [name, mutate] of [ + ['native without its child start turn', state => { delete state.capabilities.child_workflow_start_turn; }], ['native copying a manifest flag it does not publish', state => { state.capabilities.timer_runtime_contract = true; }], ['native advertising response compression', state => { state.capabilities.response_compression = ['gzip']; }], ['native without its unqualified list', state => { state.native_unqualified = []; }],