From 706868b584799a51230615c2e79edeac8d945556 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:54:32 -0600 Subject: [PATCH 1/3] Child start turn: a fresh parent turn when a child start commits (#966) start_child_workflow takes `await`: `result` (the default, unchanged) or `started`. With `started`, the completion that commits ChildWorkflowScheduled and ChildRunStarted also creates one fresh parent workflow task, while the child keeps running. Its poll carries the committed child identity from history, and the child's resolution later gives the parent its usual turn. This is the runtime side of sdk-php#185. - One turn per completion. None when the completion closes the parent, or when the parent already has a ready or leased workflow task. - The turn's payload names `workflow_wait_kind: "child_started"` with the first child's call id, initial run, sequence and ChildRunStarted. - Duplicates stay under the immutable task receipt: no second child and no second turn. - A cancellation-scope checkpoint, which keeps its claim, refuses the flag before any child starts (child_start_turn_not_at_checkpoint). - An unknown value, or `await` on another command, is refused with 422. The normalizer and the child field allowlist now keep the field. - Advertised as `child_workflow_start_turn: true` in cluster info and server_capabilities. The cluster-capabilities parity fixture gains a reviewed `native_additions` map for members published PHP never had; its contract refuses an addition that shadows a published member. docs/contracts/child-start-turn.md records the contract. Tests: child_start_turn.rs (flagged turn, duplicate, child continue-as-new resolving the original call, default task count, several children, closing completion, refusals, advertisement) and a checkpoint case. They pass on SQLite, PostgreSQL and MySQL; the full SQLite suite passes apart from the macOS-only payload_discovery hostname check. Negative control: suppressing the turn fails both turn tests. Co-Authored-By: Claude Opus 5.5 --- docs/contracts/child-start-turn.md | 66 +++++++ rust/src/runtime/children.rs | 49 ++++- rust/src/runtime/cluster_info.rs | 5 +- rust/src/runtime/completion_rules.rs | 14 +- rust/src/runtime/scope_checkpoint.rs | 8 + rust/src/runtime/store.rs | 23 ++- rust/tests/execution.rs | 2 + rust/tests/execution/child_start_turn.rs | 177 ++++++++++++++++++ rust/tests/execution/scope_checkpoints.rs | 27 +++ .../cluster-capabilities-contract.mjs | 11 +- .../ServerParity/cluster-capabilities.json | 6 + ...rParityClusterCapabilitiesContractTest.mjs | 14 ++ 12 files changed, 393 insertions(+), 9 deletions(-) create mode 100644 docs/contracts/child-start-turn.md create mode 100644 rust/tests/execution/child_start_turn.rs 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..8a17517a4 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,38 @@ 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 or the parent already has a + /// workflow task waiting for it. Its poll carries the committed + /// `ChildWorkflowScheduled` and `ChildRunStarted` history. + 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?; + let waiting = Self::scalar("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status IN ('ready','leased') AND id<>$2") + .bind(run_id) + .bind(task_id) + .fetch_one(&mut **tx) + .await?; + if open == 0 || waiting != 0 { + 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 567ca3483..7b0a47ca8 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 8b4208e66..53614cb6b 100644 --- a/rust/src/runtime/scope_checkpoint.rs +++ b/rust/src/runtime/scope_checkpoint.rs @@ -275,6 +275,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, @@ -285,6 +292,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 3d15633f6..91343be96 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -3205,6 +3205,7 @@ where Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false, commands).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, @@ -3215,6 +3216,7 @@ where external: &external, local: &local, exceptions: &exceptions, + start_turn: &start_turn, }; let order = (0..commands.len()) .filter(|index| !is_terminal(&commands[*index])) @@ -3312,6 +3314,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. @@ -3354,7 +3362,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(), @@ -3369,7 +3377,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; @@ -5014,6 +5027,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> { @@ -5138,7 +5154,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 618a946c7..93be75b40 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -368,6 +368,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..6ed7249ca --- /dev/null +++ b/rust/tests/execution/child_start_turn.rs @@ -0,0 +1,177 @@ +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) { + register(app, "worker", json!(["echo"]), json!([])).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; +} 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 = []; }], From bee92fe3f7d343032677388aa8ada8a11f0969fc Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 2 Nov 2026 21:08:54 -0600 Subject: [PATCH 2/3] Child start turn: only a ready root task stands in for the turn Security found that any ready or leased workflow task suppressed the start turn. An update accepted during the parent's claim gets its own task at that completion, and since it doesn't replay the workflow, the promised turn was lost. child_start_turn now follows child_parent_resume: only a ready workflow task without workflow_update_id can carry the committed start history. An update task or a leased task, whose worker may have read history before the start, doesn't replace the turn. Counterfactuals: - an update accepted during the claim keeps its task, and the parent gets a separate root turn with the child's identity; - a ready signal task carries the start history, with no second turn; - a leased task beside the claim doesn't stand in for the turn; - a child that finishes before the turn is claimed resolves in that same turn, with no extra task and no stuck parent. Negative controls: the old predicate fails the update and leased cases. Current main (3.0.2) is merged in. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/children.rs | 21 +++-- rust/tests/execution/child_start_turn.rs | 111 ++++++++++++++++++++++- 2 files changed, 124 insertions(+), 8 deletions(-) diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index 8a17517a4..b38184cd4 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -405,9 +405,11 @@ where /// 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 or the parent already has a - /// workflow task waiting for it. Its poll carries the committed - /// `ChildWorkflowScheduled` and `ChildRunStarted` history. + /// 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, @@ -419,13 +421,18 @@ where .bind(run_id) .fetch_one(&mut **tx) .await?; - let waiting = Self::scalar("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status IN ('ready','leased') AND id<>$2") + 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_one(&mut **tx) + .fetch_all(&mut **tx) .await?; - if open == 0 || waiting != 0 { - return Ok(None); + 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) } diff --git a/rust/tests/execution/child_start_turn.rs b/rust/tests/execution/child_start_turn.rs index 6ed7249ca..ede1acc4d 100644 --- a/rust/tests/execution/child_start_turn.rs +++ b/rust/tests/execution/child_start_turn.rs @@ -30,7 +30,8 @@ fn child(mode: Option<&str>) -> Value { /// `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) { - register(app, "worker", json!(["echo"]), json!([])).await; + // 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}"); @@ -175,3 +176,111 @@ async fn an_unknown_await_is_refused_before_any_child_and_the_capability_is_adve runtime.close().await; database.remove().await; } + +async fn update_task_count(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' AND payload LIKE '%workflow_update_id%'")).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!(update_task_count(&database, &started["run_id"]).await >= 1); + 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; +} From 37797d8b8ee31cbb8dba965ebbe422d5de664e92 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Mon, 2 Nov 2026 21:20:17 -0600 Subject: [PATCH 3/3] child start turn test: drop the json LIKE count The update counterfactual counted update tasks with payload LIKE, which PostgreSQL rejects on its json column (the same portability failure #969 fixed). The two polled tasks already prove the update task remains, so the count is replaced by the run's workflow task total (claim, update task, turn). The child tests pass on PostgreSQL 17 (65) and SQLite. Co-Authored-By: Claude Opus 5.5 --- rust/tests/execution/child_start_turn.rs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/rust/tests/execution/child_start_turn.rs b/rust/tests/execution/child_start_turn.rs index ede1acc4d..021b748f5 100644 --- a/rust/tests/execution/child_start_turn.rs +++ b/rust/tests/execution/child_start_turn.rs @@ -177,11 +177,6 @@ async fn an_unknown_await_is_refused_before_any_child_and_the_capability_is_adve database.remove().await; } -async fn update_task_count(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' AND payload LIKE '%workflow_update_id%'")).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. @@ -205,7 +200,7 @@ async fn an_update_task_does_not_replace_the_start_turn() { 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!(update_task_count(&database, &started["run_id"]).await >= 1); + assert_eq!(parent_tasks(&database, &started["run_id"]).await, 3, "the claim, the update task and the turn"); runtime.close().await; database.remove().await; }