From 8c2e217d3b1282669265d6743dd91c4853b3b38a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:09:58 +0000 Subject: [PATCH 1/3] Retry children and keep their timeout options as published PHP does Native refused retry_policy, execution_timeout_seconds and run_timeout_seconds on start_child_workflow. It now records the retry and timeout snapshots on the call, the parent's child events and the child's started event, gives each child run its deadlines, and retries a failed or timed-out child as a new run of the same child instance under a new link while the policy has attempts left and the failure is retryable (Workflow's startChildRetryIfAvailable). The retry keeps the execution deadline, starts its own run deadline and waits out the backoff. The child summary follows the call to its latest run, so the parent stays described as waiting for the child. A timed-out child's ChildRunFailed now names closed_reason timed_out, as PHP's does. The child-retries fixture records seven cases against published 2.11.7. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 3 +- .../2026-10-29-child-retries.md | 60 + docs/rust-server-port.md | 22 +- rust/src/runtime/child_retries.rs | 169 +++ rust/src/runtime/child_summary.rs | 73 +- rust/src/runtime/children.rs | 201 ++- rust/src/runtime/mod.rs | 1 + rust/tests/execution.rs | 5 +- rust/tests/execution/child_retries.rs | 152 ++ .../server-parity/child-retries-contract.mjs | 26 + .../server-parity/child-retries-probe.php | 190 +++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../Fixtures/ServerParity/child-retries.json | 1317 +++++++++++++++++ .../ServerParityChildRetriesContractTest.mjs | 62 + 15 files changed, 2238 insertions(+), 52 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-29-child-retries.md create mode 100644 rust/src/runtime/child_retries.rs create mode 100644 rust/tests/execution/child_retries.rs create mode 100644 scripts/conformance/server-parity/child-retries-contract.mjs create mode 100644 scripts/conformance/server-parity/child-retries-probe.php create mode 100644 tests/Fixtures/ServerParity/child-retries.json create mode 100644 tests/Unit/ServerParityChildRetriesContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index f461a6396..c0ee3b841 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -65,6 +65,7 @@ on: - tests/Unit/ServerParityManyUpdatesContractTest.mjs - tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs - tests/Unit/ServerParityChildDescriptionsContractTest.mjs + - tests/Unit/ServerParityChildRetriesContractTest.mjs - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs - tests/Unit/ServerParityQueryArgumentsContractTest.mjs - tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs @@ -1991,7 +1992,7 @@ jobs: bash scripts/ci/run-composer.sh --rm -v "$PWD:/source" -w /source -- \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs tests/Unit/ServerParityProbeNotesTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityConcurrentWaitsContractTest.mjs tests/Unit/ServerParityContinueTransfersContractTest.mjs tests/Unit/ServerParityChildDescriptionsContractTest.mjs tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs tests/Unit/ServerParityManyUpdatesContractTest.mjs tests/Unit/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.mjs tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs tests/Unit/ServerParityParentCloseCancellationContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityServiceEndpointsContractTest.mjs tests/Unit/ServerParityServiceCallsContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityAdapterCacheTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs tests/Unit/ServerParityProbeNotesTest.mjs tests/Unit/ServerParitySdkDirectionLanesTest.mjs tests/Unit/ServerParityCodecDeadlineCellsTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityFailureLocationTest.mjs tests/Unit/ServerParityFailureExcerptTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityConcurrentWaitsContractTest.mjs tests/Unit/ServerParityContinueTransfersContractTest.mjs tests/Unit/ServerParityChildDescriptionsContractTest.mjs tests/Unit/ServerParityChildRetriesContractTest.mjs tests/Unit/ServerParityWorkerErrorBodiesContractTest.mjs tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs tests/Unit/ServerParityManyUpdatesContractTest.mjs tests/Unit/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.mjs tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs tests/Unit/ServerParityParentCloseCancellationContractTest.mjs tests/Unit/ServerParityRecordLocalActivityContractTest.mjs tests/Unit/ServerParityActivityCancellationAckContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParityServiceEndpointsContractTest.mjs tests/Unit/ServerParityServiceCallsContractTest.mjs tests/Unit/ServerParityWorkflowImportContractTest.mjs tests/Unit/ServerParityWaterlineImportContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityOperatorDashboardTypesContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityClusterInfoContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityMessageStreamsContractTest.mjs tests/Unit/ServerParityCancellationScopesContractTest.mjs tests/Unit/ServerParityWorkflowPollRequestContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.mjs tests/Unit/ServerParityFixtureSelectionTest.mjs tests/Unit/ServerParityUpdateValidationRetryTest.mjs tests/Unit/ServerParityKilledAuthorPollTest.mjs tests/Unit/ServerParityImagePullTest.mjs tests/Unit/ServerParityAdapterCacheTest.mjs tests/Unit/ServerParityWorkerSessionsContractTest.mjs tests/Unit/ServerParityWorkerSessionActivitiesContractTest.mjs tests/Unit/ServerParityNexusOperationsContractTest.mjs tests/Unit/ServerParityBridgeAdaptersContractTest.mjs tests/Unit/ServerParityWorkflowStreamWritesContractTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md b/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md new file mode 100644 index 000000000..d6872bc56 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md @@ -0,0 +1,60 @@ +# October 29, child retry and timeout options + +- **Before.** Native refused `retry_policy`, `execution_timeout_seconds` and + `run_timeout_seconds` on `start_child_workflow` with `422 + unsupported_request_field`. Its child runs never had a budget or deadline. +- **Contract** (published Server 2.11.7, Workflow 2.15.7 + `ChildWorkflowRetryPolicy` and the bridge's `applyStartChildWorkflow` and + `startChildRetryIfAvailable`): + - **Recorded options.** The call, `ChildWorkflowScheduled`, + `ChildRunStarted` and the child's `WorkflowStarted` record a retry + snapshot (`snapshot_version`, `max_attempts` at least 1, + `backoff_seconds`, distinct trimmed `non_retryable_error_types`) and a + timeout snapshot. A policy with no known key records nothing. Each + timeout sets its own deadline on the child run. + - **Retry.** When a child run fails, or times out, and the call's policy + has attempts left (one per link of the call), the child instance starts + a new run under a new link. It keeps the call, the execution deadline + and the arguments, and gets a new run deadline. Its task is available + after the policy's backoff. The parent records `ChildRunStarted` with + `retry_attempt`, `retry_of_child_workflow_run_id` and + `retry_backoff_seconds`, without null fields, and keeps waiting. + - **No retry.** The failure is non-retryable when the run's failure is + marked so, or the policy names the type or class of the run's + `WorkflowFailed` event or of its failure. The parent then sees the + child fail, as it does when attempts run out. + - **Unchanged.** The parent's child wait keeps naming the first run. +- **Fixture.** `child-retries`, raw worker HTTP on protocol 1.20 in a fresh + namespace. Each case starts a parent with one child and plays the child's + attempts: + - a failure, then a completion, with a one-second backoff and both + timeouts; + - two failures with two attempts; + - a failure of a non-retryable type; + - a failure marked non-retryable; + - an empty policy; + - timeouts without a retry policy; + - two completions after a one-second run deadline. + + It records the answers, the parent's child events, the child instance's + runs and their started events, the parent's description while a retry is + pending, and the parent's resumption. IDs are labels and deadlines are + recorded by presence, with how a retry's deadlines relate to the first + run's. +- **Native** (`child_retries.rs`, `children.rs`, `child_summary.rs`): + - the three options are accepted and recorded as above; + - `resolve_child` retries a failed child before the parent sees it; + - the child summary follows a retried call to its latest run and link, + so the parent stays described as waiting for the child; + - a timed-out child's `ChildRunFailed` names `closed_reason: timed_out`, + as PHP's does, while the call keeps `failed`. Native named `failed` + for any timed-out child, retried or not; the fixture's first native + recording found it. +- **Tests.** `child_retries.rs` (native), and the snapshot, backoff and + non-retryable rules in `child_retries.rs` unit tests. +- **Not covered here.** + - Cooperative cancellation delivered to a retried child: native's + delivery expects one link per call and refuses it. This is from reading + the code; it is not recorded against PHP. + - Memos and search attributes inherited by child runs, retried or not + (a separate gap). diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 7a10f8229..b3a6c102c 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -599,10 +599,24 @@ contract fixture before its correction. cases against published 2.11.5 ([checkpoint](rust-server-port-checkpoints/2026-10-29-child-descriptions.md)). An open activity is likewise described ahead of open children - (October 29). A child with a retry or timeout policy, or a cancellation - policy other than `abandon` under a worker that declares cooperative - cancellation, is still described without its wait, pending its own - evidence. + (October 29). A child with a cancellation policy other than `abandon` + under a worker that declares cooperative cancellation is still described + without its wait, pending its own evidence. +- **Child retry and timeout options** (October 29, closed). Published PHP + accepts `retry_policy`, `execution_timeout_seconds` and + `run_timeout_seconds` on `start_child_workflow`, records them on the call + and the child's runs, and retries a failed or timed-out child as a new run + of the same child instance while attempts remain and the failure is + retryable (`startChildRetryIfAvailable`). Native refused the three fields + with `unsupported_request_field`. It now records and retries the same + way, and describes a parent waiting on a retried child as waiting for it. + The `child-retries` fixture records seven cases against published 2.11.7 + ([checkpoint](rust-server-port-checkpoints/2026-10-29-child-retries.md)). + Remaining, from reading the code rather than a recording: native's + cooperative cancellation delivery expects one link per call, so it + refuses one delivered to a retried child + (`cancellation_delivery_shape_mismatch`); and a retried child inherits no + memos or search attributes (below). - **Trace context (Core 2.10, published in Server 2.11.1).** Core adds nullable `observability_context` columns on runs, commands and tasks. Native has no carrier: it accepts and ignores a top-level carrier on diff --git a/rust/src/runtime/child_retries.rs b/rust/src/runtime/child_retries.rs new file mode 100644 index 000000000..ee7900f02 --- /dev/null +++ b/rust/src/runtime/child_retries.rs @@ -0,0 +1,169 @@ +//! A child's retry and timeout options (`ChildWorkflowRetryPolicy`): what +//! a `start_child_workflow` command records, and whether a failed child run +//! is retried as a new run of the same child instance. +use serde_json::{Value, json}; + +const SNAPSHOT_VERSION: i64 = 1; + +/// `snapshotExternal`: a missing or empty policy records nothing; any other +/// policy records its attempts, backoff and non-retryable error types. +pub(super) fn retry_snapshot(command: &Value) -> Option { + let policy = &command["retry_policy"]; + let empty = match policy { + Value::Object(map) => map.is_empty(), + Value::Array(list) => list.is_empty(), + _ => true, + }; + if empty { + return None; + } + Some(json!({ + "snapshot_version": SNAPSHOT_VERSION, + "max_attempts": max_attempts(policy), + "backoff_seconds": backoff(&policy["backoff_seconds"]), + "non_retryable_error_types": error_types(&policy["non_retryable_error_types"]), + })) +} + +/// `timeoutSnapshot`: recorded when either timeout is set. +pub(super) fn timeout_snapshot(execution: Option, run: Option) -> Option { + (execution.is_some() || run.is_some()).then(|| { + json!({ + "snapshot_version": SNAPSHOT_VERSION, + "execution_timeout_seconds": execution, + "run_timeout_seconds": run, + }) + }) +} + +/// A command's timeout, when it is an integer (`is_int`). +pub(super) fn timeout(command: &Value, field: &str) -> Option { + command[field].as_i64() +} + +pub(super) fn max_attempts(policy: &Value) -> i64 { + policy["max_attempts"].as_i64().map_or(1, |attempts| attempts.max(1)) +} + +/// The delay before the next attempt, after `completed` attempts: the +/// backoff list's entry for that attempt, or its last entry. +pub(super) fn backoff_seconds(policy: &Value, completed: i64) -> i64 { + let backoff = backoff(&policy["backoff_seconds"]); + if backoff.is_empty() { + return 0; + } + let index = usize::try_from((completed - 1).max(0)).unwrap_or(usize::MAX); + backoff[index.min(backoff.len() - 1)] +} + +/// `isNonRetryableFailure`: the run's first failure is marked +/// non-retryable, or the policy names the type or class of the terminal +/// `WorkflowFailed` event or the failure's class. +pub(super) fn non_retryable(policy: &Value, failure: Option<(bool, Option)>, failed_event: Option<&Value>) -> bool { + if failure.as_ref().is_some_and(|(flag, _)| *flag) { + return true; + } + let types = error_types(&policy["non_retryable_error_types"]); + if types.is_empty() { + return false; + } + let mut candidates = Vec::new(); + if let Some(event) = failed_event { + for key in ["exception_type", "exception_class"] { + if let Some(value) = event[key].as_str() { + candidates.push(value.to_owned()); + } + } + } + if let Some((_, Some(class))) = failure { + candidates.push(class); + } + candidates.iter().any(|candidate| !candidate.is_empty() && types.contains(candidate)) +} + +/// `normalizeBackoff`: integers and numeric strings, at least zero. +fn backoff(value: &Value) -> Vec { + let Some(entries) = list(value) else { + return Vec::new(); + }; + entries + .filter_map(|entry| match entry { + Value::Number(number) if number.is_i64() => number.as_i64(), + Value::String(text) => text.trim().parse::().ok().filter(|v| v.is_finite()).map(|v| v.trunc() as i64), + _ => None, + }) + .map(|seconds| seconds.max(0)) + .collect() +} + +/// `normalizeErrorTypes`: trimmed, non-empty, distinct strings. +fn error_types(value: &Value) -> Vec { + let Some(entries) = list(value) else { + return Vec::new(); + }; + let mut types: Vec = Vec::new(); + for entry in entries { + let text = entry.as_str().map(str::trim).unwrap_or_default(); + if !text.is_empty() && !types.iter().any(|known| known == text) { + types.push(text.to_owned()); + } + } + types +} + +/// A PHP array's values: a list, or an object's values in order. +fn list(value: &Value) -> Option + '_>> { + match value { + Value::Array(entries) => Some(Box::new(entries.iter())), + Value::Object(entries) => Some(Box::new(entries.values())), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_policy_is_recorded_as_php_snapshots_it() { + assert_eq!(retry_snapshot(&json!({})), None); + assert_eq!(retry_snapshot(&json!({"retry_policy": {}})), None); + assert_eq!(retry_snapshot(&json!({"retry_policy": []})), None); + assert_eq!( + retry_snapshot(&json!({"retry_policy": {"max_attempts": 3, "backoff_seconds": [2, "5", -1], + "non_retryable_error_types": [" Fatal ", "Fatal", "", 7]}})), + Some(json!({"snapshot_version": 1, "max_attempts": 3, "backoff_seconds": [2, 5, 0], + "non_retryable_error_types": ["Fatal"]})) + ); + assert_eq!( + retry_snapshot(&json!({"retry_policy": {"max_attempts": 0}})), + Some(json!({"snapshot_version": 1, "max_attempts": 1, "backoff_seconds": [], + "non_retryable_error_types": []})) + ); + assert_eq!(timeout_snapshot(None, None), None); + assert_eq!( + timeout_snapshot(None, Some(5)), + Some(json!({"snapshot_version": 1, "execution_timeout_seconds": null, "run_timeout_seconds": 5})) + ); + } + + #[test] + fn backoff_uses_the_entry_for_the_attempt_then_the_last() { + let policy = json!({"backoff_seconds": [1, 4]}); + assert_eq!(backoff_seconds(&policy, 1), 1); + assert_eq!(backoff_seconds(&policy, 2), 4); + assert_eq!(backoff_seconds(&policy, 7), 4); + assert_eq!(backoff_seconds(&json!({}), 1), 0); + } + + #[test] + fn a_failure_is_non_retryable_by_flag_or_named_type() { + let policy = json!({"non_retryable_error_types": ["Fatal"]}); + assert!(non_retryable(&json!({}), Some((true, None)), None)); + assert!(!non_retryable(&json!({}), Some((false, Some("Fatal".into()))), None)); + assert!(non_retryable(&policy, None, Some(&json!({"exception_type": "Fatal"})))); + assert!(non_retryable(&policy, None, Some(&json!({"exception_class": "Fatal"})))); + assert!(non_retryable(&policy, Some((false, Some("Fatal".into()))), None)); + assert!(!non_retryable(&policy, Some((false, Some("Boom".into()))), Some(&json!({"exception_type": "Boom"})))); + } +} diff --git a/rust/src/runtime/child_summary.rs b/rust/src/runtime/child_summary.rs index 284e092d3..65126f721 100644 --- a/rust/src/runtime/child_summary.rs +++ b/rust/src/runtime/child_summary.rs @@ -18,11 +18,17 @@ pub(super) struct ChildSummaryHistory { struct ExpectedChild { id: String, + /// The first run, which the parent's wait keeps naming. run: String, instance: String, label: String, started: bool, status: &'static str, + /// The current attempt's run, link and run number. A retry + /// (`ChildWorkflowRetryPolicy`) starts a new run under a new link. + latest_run: String, + latest_link: String, + run_number: i64, } impl ChildSummaryHistory { @@ -30,17 +36,16 @@ impl ChildSummaryHistory { let Some(sequence) = payload["sequence"].as_i64().filter(|v| *v > 0) else { return false; }; - let Some(identity) = ExpectedChild::identity(payload) else { - return false; - }; if event == "ChildWorkflowScheduled" { + let Some(identity) = ExpectedChild::identity(payload) else { + return false; + }; // Any number of children may be open at once, and the parent-close - // policy only decides what happens when the parent closes. + // policy only decides what happens when the parent closes. Retry + // and timeout policies only decide how the call's runs end. if self.calls.contains_key(&sequence) || self.calls.values().any(|child| child.id == identity.id) || payload["cancellation_policy"] != "abandon" - || !payload["retry_policy"].is_null() - || !payload["timeout_policy"].is_null() { return false; } @@ -50,25 +55,33 @@ impl ChildSummaryHistory { let Some(previous) = self.calls.get_mut(&sequence) else { return false; }; - if !previous.same_identity(&identity) || previous.status != "waiting" { + if !previous.same_call(payload) || previous.status != "waiting" { return false; } + let text = |key: &str| payload[key].as_str().filter(|v| !v.is_empty()).map(str::to_owned); match event { "ChildRunStarted" if !previous.started => { if payload["child_run_number"] != 1 - || [ - "execution_timeout_seconds", - "run_timeout_seconds", - "execution_deadline_at", - "run_deadline_at", - ] - .iter() - .any(|key| !payload[*key].is_null()) + || payload["workflow_link_id"] != previous.id.as_str() + || payload["child_workflow_run_id"] != previous.run.as_str() { return false; } previous.started = true; } + "ChildRunStarted" if payload["retry_attempt"].is_i64() => { + let (Some(run), Some(link), Some(number)) = + (text("child_workflow_run_id"), text("workflow_link_id"), payload["child_run_number"].as_i64()) + else { + return false; + }; + if number <= previous.run_number || run == previous.latest_run || link == previous.latest_link { + return false; + } + previous.latest_run = run; + previous.latest_link = link; + previous.run_number = number; + } "ChildRunCompleted" | "ChildRunCancelled" | "ChildRunTerminated" | "ChildRunFailed" if previous.started => { @@ -78,7 +91,11 @@ impl ChildSummaryHistory { "ChildRunTerminated" => "terminated", _ => "failed", }; - if payload["child_run_number"] != 1 || payload["child_status"] != status { + if payload["child_run_number"] != previous.run_number + || payload["child_status"] != status + || payload["workflow_link_id"] != previous.latest_link.as_str() + || payload["child_workflow_run_id"] != previous.latest_run.as_str() + { return false; } previous.status = status; @@ -113,14 +130,20 @@ impl ExpectedChild { label, started: false, status: "waiting", + latest_run: text("child_workflow_run_id")?, + latest_link: text("child_call_id")?, + run_number: 1, }) } - fn same_identity(&self, other: &Self) -> bool { - self.id == other.id - && self.run == other.run - && self.instance == other.instance - && self.label == other.label + /// The call's identity, which every run of it shares. + fn same_call(&self, payload: &Value) -> bool { + payload["child_call_id"] == self.id.as_str() + && payload["child_workflow_instance_id"] == self.instance.as_str() + && payload["child_workflow_type"] == self.label.as_str() + && payload["child_workflow_class"] == self.label.as_str() + && payload["parallel_group_id"].is_null() + && payload["parallel_group_path"].is_null() } fn wait_status(&self) -> &'static str { @@ -150,7 +173,7 @@ where history: &ChildSummaryHistory, summary: &mut ActiveSummary, ) -> Result { - let rows = Self::query("SELECT w.*,c.status AS call_status,c.resolved_child_run_id,c.resolved_child_instance_id,c.child_workflow_type,c.metadata AS call_metadata,l.id AS link_id,l.child_workflow_run_id AS link_run,l.child_workflow_instance_id AS link_instance FROM workflow_run_waits w LEFT JOIN workflow_child_calls c ON c.parent_workflow_run_id=w.workflow_run_id AND c.sequence=w.sequence LEFT JOIN workflow_links l ON l.parent_workflow_run_id=w.workflow_run_id AND l.sequence=w.sequence AND l.link_type='child_workflow' WHERE w.workflow_run_id=$1 AND w.kind='child' ORDER BY w.sequence") + let rows = Self::query("SELECT w.*,c.status AS call_status,c.resolved_child_run_id,c.resolved_child_instance_id,c.child_workflow_type,c.metadata AS call_metadata,l.id AS link_id,l.child_workflow_run_id AS link_run,l.child_workflow_instance_id AS link_instance FROM workflow_run_waits w LEFT JOIN workflow_child_calls c ON c.parent_workflow_run_id=w.workflow_run_id AND c.sequence=w.sequence LEFT JOIN workflow_links l ON l.parent_workflow_run_id=w.workflow_run_id AND l.sequence=w.sequence AND l.link_type='child_workflow' AND l.child_workflow_run_id=c.resolved_child_run_id WHERE w.workflow_run_id=$1 AND w.kind='child' ORDER BY w.sequence") .bind(run_id).fetch_all(&mut **tx).await?; if rows.len() != history.calls.len() { return Ok(false); @@ -266,14 +289,14 @@ where child.status }, )? - && matches("resolved_child_run_id", &child.run)? + && matches("resolved_child_run_id", &child.latest_run)? && matches("resolved_child_instance_id", &child.instance)? && matches("child_workflow_type", &child.label)? && DB::optional_document_row(row, "call_metadata")? .as_ref() .is_some_and(|v| v["child_call_id"] == child.id) - && matches("link_id", &child.id)? - && matches("link_run", &child.run)? + && matches("link_id", &child.latest_link)? + && matches("link_run", &child.latest_run)? && matches("link_instance", &child.instance)?) } } diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index 24f7933fb..bedfc8fb9 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -1,6 +1,8 @@ //! Sequential child outcomes and abandoned-parent cleanup retain the original call. -//! Retry, parallel groups and parent-close actions remain -//! unqualified and are refused at admission rather than silently discarded. +//! A retried child keeps the call under a new link (`child_retries`). +//! Parallel child groups remain unqualified and are refused at admission +//! rather than silently discarded. +use super::schedule_storage::iso8601; use super::transaction::Transaction; use super::{ Result, @@ -38,6 +40,9 @@ pub(super) fn validate_child(command: &Value) -> Result<()> { "queue", "parent_close_policy", "cancellation_policy", + "retry_policy", + "execution_timeout_seconds", + "run_timeout_seconds", ], )?; if text(command, "workflow_type")?.len() > 191 @@ -277,18 +282,28 @@ where // `ChildCallService`: a child keeps its parent's build pin. A child of // an unpinned parent is unpinned too (checked against published 2.11.1). let compatibility = DB::optional_string(parent, "compatibility")?.filter(|build| !build.trim().is_empty()); - // PHP children without explicit timeout options have nullable budgets - // and deadlines. Do not invent parent defaults for these runs. + // `applyStartChildWorkflow`: each deadline follows its own timeout. + // A child without timeout options has no budget or deadline, whatever + // its parent's. + let execution_timeout = super::child_retries::timeout(command, "execution_timeout_seconds"); + let run_timeout = super::child_retries::timeout(command, "run_timeout_seconds"); + let execution_deadline = execution_timeout.map(|seconds| started_at + chrono::Duration::seconds(seconds)); + let run_deadline = run_timeout.map(|seconds| started_at + chrono::Duration::seconds(seconds)); + let retry_policy = super::child_retries::retry_snapshot(command); + let timeout_policy = super::child_retries::timeout_snapshot(execution_timeout, run_timeout); Self::query("INSERT INTO workflow_instances(id,namespace,workflow_type,workflow_class,current_run_id,run_count,started_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,1,$6,$7,$8)") .bind(&child_instance).bind(namespace.as_deref()).bind(workflow_type).bind(workflow_type).bind(&child_run).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).execute(&mut **tx).await?; - Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,namespace,workflow_type,workflow_class,run_number,status,payload_codec,arguments,queue,started_at,created_at,updated_at,last_progress_at,compatibility) VALUES ($1,$2,$3,$4,$5,1,'pending','avro',$6,$7,$8,$9,$10,$11,$12)") - .bind(&child_run).bind(&child_instance).bind(namespace.as_deref()).bind(workflow_type).bind(workflow_type).bind(&arguments).bind(&queue).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(compatibility.as_deref()).execute(&mut **tx).await?; - Self::query("INSERT INTO workflow_child_calls(parent_workflow_run_id,parent_workflow_instance_id,sequence,child_workflow_type,child_workflow_class,resolved_child_instance_id,resolved_child_run_id,parent_close_policy,cancellation_propagation,queue,status,scheduled_at,started_at,arguments,metadata,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,($9<>'abandon'),$10,'started',$11,$12,$13,$14,$15,$16)") + Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,namespace,workflow_type,workflow_class,run_number,status,payload_codec,arguments,queue,started_at,created_at,updated_at,last_progress_at,compatibility,run_timeout_seconds,execution_deadline_at,run_deadline_at) VALUES ($1,$2,$3,$4,$5,1,'pending','avro',$6,$7,$8,$9,$10,$11,$12,NULLIF($13,0),$14,$15)") + .bind(&child_run).bind(&child_instance).bind(namespace.as_deref()).bind(workflow_type).bind(workflow_type).bind(&arguments).bind(&queue).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).bind(compatibility.as_deref()) + // Timeouts are at least one second; zero binds "no timeout". + .bind(run_timeout.unwrap_or(0)).bind(DB::optional_time(execution_deadline)).bind(DB::optional_time(run_deadline)).execute(&mut **tx).await?; + Self::query("INSERT INTO workflow_child_calls(parent_workflow_run_id,parent_workflow_instance_id,sequence,child_workflow_type,child_workflow_class,resolved_child_instance_id,resolved_child_run_id,parent_close_policy,cancellation_propagation,queue,status,scheduled_at,started_at,arguments,metadata,created_at,updated_at,retry_policy,timeout_policy) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,($9<>'abandon'),$10,'started',$11,$12,$13,$14,$15,$16,$17,$18)") .bind(&parent_run).bind(&parent_instance).bind(sequence).bind(workflow_type).bind(workflow_type).bind(&child_instance).bind(&child_run).bind(&policies.parent_close).bind(&policies.cancellation).bind(&queue).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)) - .bind(DB::document(&json!({"payload":arguments}))).bind(DB::document(&json!({"child_call_id":call,"attempt_count":1,"cancellation_policy":policies.cancellation}))).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).execute(&mut **tx).await?; + .bind(DB::document(&json!({"payload":arguments}))).bind(DB::document(&json!({"child_call_id":call,"attempt_count":1,"cancellation_policy":policies.cancellation}))).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)) + .bind(retry_policy.as_ref().map(DB::document)).bind(timeout_policy.as_ref().map(DB::document)).execute(&mut **tx).await?; Self::query("INSERT INTO workflow_links(id,link_type,sequence,parent_workflow_instance_id,parent_workflow_run_id,child_workflow_instance_id,child_workflow_run_id,is_primary_parent,parent_close_policy,created_at,updated_at) VALUES ($1,'child_workflow',$2,$3,$4,$5,$6,TRUE,$7,$8,$9)") .bind(&call).bind(sequence).bind(&parent_instance).bind(&parent_run).bind(&child_instance).bind(&child_run).bind(&policies.parent_close).bind(DB::bind_time(started_at)).bind(DB::bind_time(started_at)).execute(&mut **tx).await?; - let scheduled = json!({"sequence":sequence,"workflow_link_id":call,"child_call_id":call,"child_workflow_instance_id":child_instance,"child_workflow_run_id":child_run,"child_workflow_class":workflow_type,"child_workflow_type":workflow_type,"parent_close_policy":policies.parent_close,"cancellation_policy":policies.cancellation,"retry_policy":null,"timeout_policy":null}); + let scheduled = json!({"sequence":sequence,"workflow_link_id":call,"child_call_id":call,"child_workflow_instance_id":child_instance,"child_workflow_run_id":child_run,"child_workflow_class":workflow_type,"child_workflow_type":workflow_type,"parent_close_policy":policies.parent_close,"cancellation_policy":policies.cancellation,"retry_policy":retry_policy,"timeout_policy":timeout_policy}); Self::append( tx, &parent_run, @@ -298,15 +313,12 @@ where None, ) .await?; + let deadlines = json!({"execution_timeout_seconds":execution_timeout,"run_timeout_seconds":run_timeout, + "execution_deadline_at":execution_deadline.map(iso8601),"run_deadline_at":run_deadline.map(iso8601)}); let mut started = scheduled; started["child_run_number"] = json!(1); - for field in [ - "execution_timeout_seconds", - "run_timeout_seconds", - "execution_deadline_at", - "run_deadline_at", - ] { - started[field] = Value::Null; + for (field, value) in deadlines.as_object().expect("an object") { + started[field] = value.clone(); } Self::append( tx, @@ -317,7 +329,10 @@ where None, ) .await?; - let mut child_started = json!({"workflow_class":workflow_type,"workflow_type":workflow_type,"workflow_instance_id":child_instance,"workflow_run_id":child_run,"parent_workflow_instance_id":parent_instance,"parent_workflow_run_id":parent_run,"parent_sequence":sequence,"workflow_link_id":call,"child_call_id":call,"retry_policy":null,"timeout_policy":null,"execution_timeout_seconds":null,"run_timeout_seconds":null,"execution_deadline_at":null,"run_deadline_at":null}); + let mut child_started = json!({"workflow_class":workflow_type,"workflow_type":workflow_type,"workflow_instance_id":child_instance,"workflow_run_id":child_run,"parent_workflow_instance_id":parent_instance,"parent_workflow_run_id":parent_run,"parent_sequence":sequence,"workflow_link_id":call,"child_call_id":call,"retry_policy":retry_policy,"timeout_policy":timeout_policy}); + for (field, value) in deadlines.as_object().expect("an object") { + child_started[field] = value.clone(); + } Self::attach_workflow_contract(tx, &queue, workflow_type, None, &mut child_started).await?; Self::append(tx, &child_run, "WorkflowStarted", child_started, None, None).await?; let child_task = Self::create_task(tx, &child_run, &queue, "workflow", json!({})).await?; @@ -390,7 +405,10 @@ where // event names the continue-as-new link and the original call, and the // parent's link to it is a new one (`applyContinueAsNew`). let continued = original["continued_from_run_id"].is_string(); - let call_id = if continued { + // A retried child keeps its first call too; its own link is new + // (`startChildRetryIfAvailable`). + let retried = original["retry_of_child_workflow_run_id"].is_string(); + let call_id = if continued || retried { original["child_call_id"].as_str().map(str::to_owned).ok_or_else(|| refuse(StatusCode::CONFLICT, "child_call_mismatch"))? } else { link_id.clone() @@ -399,7 +417,7 @@ where || original["parent_workflow_instance_id"] != DB::string(link, "parent_workflow_instance_id")? || original["parent_sequence"] != sequence - || (!continued && original["workflow_link_id"] != call_id) + || (!continued && original["workflow_link_id"] != link_id) || original["child_call_id"] != call_id || original["workflow_instance_id"] != DB::string(child, "workflow_instance_id")? || original["workflow_run_id"] != child_run @@ -426,6 +444,13 @@ where { return Err(refuse(StatusCode::CONFLICT, "child_call_mismatch")); } + // `dispatchParentResumeTasksForRun`: a failed child is retried, when + // its policy allows, before the parent sees it fail. + if matches!(resolution, ChildResolution::Failed(_)) + && Self::start_child_retry(tx, &parent, link, call, child, sequence, &call_id).await? + { + return Ok(()); + } let (status, event_type) = match resolution { ChildResolution::Completed(_) => ("completed", "ChildRunCompleted"), ChildResolution::Cancelled(_) => ("cancelled", "ChildRunCancelled"), @@ -442,7 +467,13 @@ where .await?; } let parent_resume = Self::child_parent_resume(tx, &parent).await?; - let mut payload = json!({"sequence":sequence,"workflow_link_id":link_id,"child_call_id":call_id,"child_workflow_instance_id":DB::string(child,"workflow_instance_id")?,"child_workflow_run_id":child_run,"child_workflow_class":DB::string(child,"workflow_class")?,"child_workflow_type":DB::string(child,"workflow_type")?,"child_run_number":DB::number(child,"run_number")?,"child_status":status,"closed_reason":status,"closed_at":closed_at}); + // The event names the child run's closed reason: a run that timed + // out failed with `timed_out`. The call keeps its status. + let closed_reason = match resolution { + ChildResolution::Failed(failure) if failure["timeout_kind"].is_string() => "timed_out", + _ => status, + }; + let mut payload = json!({"sequence":sequence,"workflow_link_id":link_id,"child_call_id":call_id,"child_workflow_instance_id":DB::string(child,"workflow_instance_id")?,"child_workflow_run_id":child_run,"child_workflow_class":DB::string(child,"workflow_class")?,"child_workflow_type":DB::string(child,"workflow_type")?,"child_run_number":DB::number(child,"run_number")?,"child_status":status,"closed_reason":closed_reason,"closed_at":closed_at}); match resolution { ChildResolution::Completed(output) => { payload["output"] = wire(output); @@ -482,4 +513,134 @@ where .bind(task_id).bind(task_status).bind(&parent_run).bind(format!("child:{call_id}")).execute(&mut **tx).await?; Ok(()) } + /// `startChildRetryIfAvailable`: when the call's retry policy has + /// attempts left and the failure is retryable, the child instance starts + /// a new run after the policy's backoff, and the parent keeps waiting on + /// the same call. Returns whether a retry started. + async fn start_child_retry( + tx: &mut Transaction<'_, DB>, + parent: &DB::Row, + link: &DB::Row, + call: &DB::Row, + failed: &DB::Row, + sequence: i64, + call_id: &str, + ) -> Result { + let parent_run = DB::string(parent, "id")?; + let failed_run = DB::string(failed, "id")?; + // The policy the parent recorded when it scheduled the call, or the + // call's own when that event is in an earlier run. + let mut scheduled = Value::Null; + for row in Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='ChildWorkflowScheduled' ORDER BY sequence") + .bind(&parent_run).fetch_all(&mut **tx).await? + { + let payload = DB::document_row(&row, "payload")?; + if payload["sequence"] == sequence { + scheduled = payload; + break; + } + } + let policy = if scheduled["retry_policy"].is_object() || scheduled["retry_policy"].is_array() { + scheduled["retry_policy"].clone() + } else { + DB::optional_document_row(call, "retry_policy")? + .filter(|policy| policy.is_object() || policy.is_array()) + .unwrap_or(Value::Null) + }; + if policy.is_null() { + return Ok(false); + } + let attempts = Self::scalar("SELECT COUNT(*) FROM workflow_links WHERE parent_workflow_run_id=$1 AND sequence=$2 AND link_type='child_workflow'") + .bind(&parent_run).bind(sequence).fetch_one(&mut **tx).await?; + if attempts >= super::child_retries::max_attempts(&policy) { + return Ok(false); + } + let failure = Self::query("SELECT CASE WHEN non_retryable THEN 1 ELSE 0 END AS non_retryable_flag,exception_class FROM workflow_failures WHERE workflow_run_id=$1 ORDER BY created_at,id LIMIT 1") + .bind(&failed_run).fetch_optional(&mut **tx).await?; + let failure = match failure { + Some(row) => Some((DB::number(&row, "non_retryable_flag")? != 0, DB::optional_string(&row, "exception_class")?)), + None => None, + }; + let failed_event = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowFailed' ORDER BY sequence DESC LIMIT 1") + .bind(&failed_run).fetch_optional(&mut **tx).await?; + let failed_event = match failed_event { + Some(row) => Some(DB::document_row(&row, "payload")?), + None => None, + }; + if super::child_retries::non_retryable(&policy, failure, failed_event.as_ref()) { + return Ok(false); + } + let timeout_policy = if scheduled["timeout_policy"].is_object() { scheduled["timeout_policy"].clone() } else { Value::Null }; + let execution_timeout = timeout_policy["execution_timeout_seconds"].as_i64(); + let run_timeout = timeout_policy["run_timeout_seconds"].as_i64(); + let at = now(); + let backoff = super::child_retries::backoff_seconds(&policy, attempts); + let available_at = at + chrono::Duration::seconds(backoff); + let instance = DB::string(failed, "workflow_instance_id")?; + let run_number = Self::scalar("SELECT COALESCE(MAX(run_number),0) FROM workflow_runs WHERE workflow_instance_id=$1") + .bind(&instance).fetch_one(&mut **tx).await? + 1; + // The execution deadline spans every attempt; each run gets its own. + let execution_deadline = DB::optional_instant(failed, "execution_deadline_at")? + .or(execution_timeout.map(|seconds| at + chrono::Duration::seconds(seconds))); + let run_deadline = run_timeout.map(|seconds| at + chrono::Duration::seconds(seconds)); + let retry_run = id(); + Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,run_number,workflow_class,workflow_type,namespace,business_key,visibility_labels,status,compatibility,payload_codec,arguments,run_timeout_seconds,execution_deadline_at,run_deadline_at,connection,queue,started_at,last_progress_at,created_at,updated_at) SELECT $1,workflow_instance_id,$2,workflow_class,workflow_type,namespace,business_key,visibility_labels,'pending',compatibility,payload_codec,arguments,NULLIF($3,0),$4,$5,connection,queue,$6,$7,$8,$9 FROM workflow_runs WHERE id=$10") + .bind(&retry_run).bind(run_number).bind(run_timeout.unwrap_or(0)).bind(DB::optional_time(execution_deadline)).bind(DB::optional_time(run_deadline)) + .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(&failed_run) + .execute(&mut **tx).await?; + Self::query("UPDATE workflow_instances SET current_run_id=$1,run_count=CASE WHEN run_count>$2 THEN run_count ELSE $2 END,updated_at=$3 WHERE id=$4") + .bind(&retry_run).bind(run_number).bind(DB::bind_time(at)).bind(&instance).execute(&mut **tx).await?; + let parent_close = DB::optional_string(link, "parent_close_policy")?.unwrap_or_else(|| "abandon".into()); + let retry_link = id(); + Self::query("INSERT INTO workflow_links(id,link_type,sequence,parent_workflow_instance_id,parent_workflow_run_id,child_workflow_instance_id,child_workflow_run_id,is_primary_parent,parent_close_policy,created_at,updated_at) VALUES ($1,'child_workflow',$2,$3,$4,$5,$6,TRUE,$7,$8,$9)") + .bind(&retry_link).bind(sequence).bind(DB::string(parent, "workflow_instance_id")?).bind(&parent_run).bind(&instance).bind(&retry_run) + .bind(&parent_close).bind(DB::bind_time(at)).bind(DB::bind_time(at)).execute(&mut **tx).await?; + let workflow_type = DB::string(failed, "workflow_type")?; + let workflow_class = DB::string(failed, "workflow_class")?; + let deadlines = json!({"execution_timeout_seconds":execution_timeout,"run_timeout_seconds":run_timeout, + "execution_deadline_at":execution_deadline.map(iso8601),"run_deadline_at":run_deadline.map(iso8601)}); + let retry = json!({"retry_attempt":attempts + 1,"retry_of_child_workflow_run_id":failed_run}); + let mut started = json!({"sequence":sequence,"workflow_link_id":retry_link,"child_call_id":call_id, + "child_workflow_instance_id":instance,"child_workflow_run_id":retry_run,"child_workflow_class":workflow_class, + "child_workflow_type":workflow_type,"child_run_number":run_number,"parent_close_policy":parent_close, + "retry_backoff_seconds":backoff,"retry_policy":policy,"timeout_policy":timeout_policy}); + for source in [&retry, &deadlines] { + for (field, value) in source.as_object().expect("an object") { + started[field] = value.clone(); + } + } + // PHP records the parent's event without its null fields. + if let Some(fields) = started.as_object_mut() { + fields.retain(|_, value| !value.is_null()); + } + Self::append(tx, &parent_run, "ChildRunStarted", started, None, None).await?; + let mut child_started = json!({"workflow_class":workflow_class,"workflow_type":workflow_type,"workflow_instance_id":instance, + "workflow_run_id":retry_run,"parent_workflow_instance_id":DB::string(parent, "workflow_instance_id")?, + "parent_workflow_run_id":parent_run,"parent_sequence":sequence,"workflow_link_id":retry_link,"child_call_id":call_id, + "retry_policy":policy,"timeout_policy":timeout_policy}); + for source in [&retry, &deadlines] { + for (field, value) in source.as_object().expect("an object") { + child_started[field] = value.clone(); + } + } + let queue = DB::string(failed, "queue")?; + Self::attach_workflow_contract(tx, &queue, &workflow_type, None, &mut child_started).await?; + Self::append(tx, &retry_run, "WorkflowStarted", child_started, None, None).await?; + let task = Self::create_task(tx, &retry_run, &queue, "workflow", json!({})).await?; + if backoff > 0 { + Self::query("UPDATE workflow_tasks SET available_at=$1 WHERE id=$2") + .bind(DB::bind_time(available_at)).bind(&task).execute(&mut **tx).await?; + } + // `markChildCallRetryStarted`. + let mut metadata = DB::optional_document_row(call, "metadata")?.filter(Value::is_object).unwrap_or_else(|| json!({})); + metadata["child_call_id"] = json!(call_id); + metadata["attempt_count"] = json!(attempts + 1); + metadata["last_retry_of_child_workflow_run_id"] = json!(failed_run); + metadata["last_retry_backoff_seconds"] = json!(backoff); + Self::query("UPDATE workflow_child_calls SET resolved_child_instance_id=$1,resolved_child_run_id=$2,status='started',started_at=$3,metadata=$4,updated_at=$5 WHERE id=$6") + .bind(&instance).bind(&retry_run).bind(DB::bind_time(at)).bind(DB::document(&metadata)).bind(DB::bind_time(at)) + .bind(DB::number(call, "id")?).execute(&mut **tx).await?; + tx.summary_changed(&parent_run); + Ok(true) + } } diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 1af7651f5..272112ec8 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -42,6 +42,7 @@ mod cancellation; mod cancellation_cascade; mod cancellation_scopes; mod scope_checkpoint; +mod child_retries; mod child_summary; mod child_cancellation; mod children; diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index db2b7a7f8..ccf91e368 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -354,6 +354,8 @@ mod start_search_attributes; mod worker_error_bodies; #[path = "execution/child_descriptions.rs"] mod child_descriptions; +#[path = "execution/child_retries.rs"] +mod child_retries; enum TestDatabase { Sqlite(tempfile::TempDir), @@ -1314,9 +1316,8 @@ async fn sequential_children_commit_original_links_results_and_one_parent_resump let refused = finish_task(&app_a, &parent_task, json!([cooperative])).await; assert_eq!(refused.0, StatusCode::CONFLICT, "{}", refused.1); assert_eq!(refused.1["reason"], "child_cancellation_policy_not_supported"); + // Retry and timeout options are served (`child_retries`). for (field, option) in [ - ("retry_policy", json!({"maximum_attempts":2})), - ("run_timeout_seconds", json!(10)), ("queue", json!("")), ("parallel_group_path", json!([])), ] { diff --git a/rust/tests/execution/child_retries.rs b/rust/tests/execution/child_retries.rs new file mode 100644 index 000000000..4ac531eb8 --- /dev/null +++ b/rust/tests/execution/child_retries.rs @@ -0,0 +1,152 @@ +use super::*; + +// Published Workflow retries a failed child as a new run of the same child +// instance while its retry policy has attempts left +// (`startChildRetryIfAvailable`). The parent keeps waiting on the same call +// and is resumed once a run of the child resolves it. The child-retries +// fixture records the same cases against published Server 2.11.7. + +fn child(options: Value) -> Value { + let mut command = json!({"type":"start_child_workflow","workflow_type":"child","arguments":envelope(Payload::Array(vec![]))}); + for (field, value) in options.as_object().unwrap() { + command[field] = value.clone(); + } + command +} + +fn fail(non_retryable: bool) -> Value { + json!([{"type":"fail_workflow","message":"attempt failed","exception_type":"Boom","exception_class":"Boom","non_retryable":non_retryable}]) +} + +fn child_events(events: &Value) -> Vec { + events.as_array().unwrap().iter().filter(|event| event["event_type"].as_str().unwrap().starts_with("Child")).cloned().collect() +} + +async fn parent_with_child(app: &Router, id: &str, options: Value) -> Value { + register(app, "worker", json!(["echo", "child"]), json!([])).await; + let parent = start(app, id).await; + let task = poll(app, "worker", "workflow").await; + assert_eq!(task["workflow_type"], "echo", "{task}"); + let (status, body) = finish_task(app, &task, json!([child(options)])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + parent +} + +async fn child_task(app: &Router) -> Value { + let task = poll(app, "worker", "workflow").await; + assert_eq!(task["workflow_type"], "child", "{task}"); + task +} + +#[tokio::test] +async fn a_failed_child_is_retried_as_a_new_run_of_the_same_call() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let parent = parent_with_child(&app, "retried", json!({"retry_policy":{"max_attempts":3,"backoff_seconds":[0]}, + "execution_timeout_seconds":600,"run_timeout_seconds":300})).await; + let first = child_task(&app).await; + let (status, body) = finish_task(&app, &first, fail(false)).await; + assert_eq!((status, &body["run_status"]), (StatusCode::OK, &json!("failed")), "{body}"); + + let (_, described) = request(&app, "GET", "/api/workflows/retried", Value::Null).await; + assert_eq!((described["status"].clone(), described["wait_kind"].clone()), (json!("waiting"), json!("child")), "{described}"); + let events = child_events(&run_history(&app, &json!("retried"), &parent["run_id"]).await); + let kinds: Vec<_> = events.iter().map(|event| event["event_type"].clone()).collect(); + assert_eq!(kinds, [json!("ChildWorkflowScheduled"), json!("ChildRunStarted"), json!("ChildRunStarted")]); + let (scheduled, retry) = (&events[0]["payload"], &events[2]["payload"]); + assert_eq!(scheduled["retry_policy"], json!({"snapshot_version":1,"max_attempts":3,"backoff_seconds":[0],"non_retryable_error_types":[]})); + assert_eq!(scheduled["timeout_policy"], json!({"snapshot_version":1,"execution_timeout_seconds":600,"run_timeout_seconds":300})); + assert_eq!((retry["child_run_number"].clone(), retry["retry_attempt"].clone(), retry["retry_backoff_seconds"].clone()), + (json!(2), json!(2), json!(0)), "{retry}"); + assert_eq!(retry["child_call_id"], scheduled["child_call_id"]); + assert_ne!(retry["workflow_link_id"], scheduled["workflow_link_id"]); + assert_eq!(retry["retry_of_child_workflow_run_id"], first["run_id"]); + assert_eq!(retry["execution_deadline_at"], events[1]["payload"]["execution_deadline_at"], "the execution deadline spans attempts"); + assert!(retry.get("cancellation_policy").is_none(), "{retry}"); + + let second = child_task(&app).await; + assert_eq!(second["run_id"], retry["child_workflow_run_id"]); + let (status, body) = finish_task(&app, &second, json!([{"type":"complete_workflow","result":envelope(Payload::Array(vec![]))}])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let resumed = poll(&app, "worker", "workflow").await; + assert_eq!(resumed["workflow_type"], "echo", "{resumed}"); + let events = child_events(&run_history(&app, &json!("retried"), &parent["run_id"]).await); + let completed = &events.last().unwrap()["payload"]; + assert_eq!(events.last().unwrap()["event_type"], "ChildRunCompleted"); + assert_eq!((completed["child_run_number"].clone(), completed["workflow_link_id"].clone()), (json!(2), retry["workflow_link_id"].clone())); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn a_child_fails_its_parent_when_attempts_run_out_or_the_failure_is_not_retryable() { + for (id, options, attempts, non_retryable) in [ + ("exhausted", json!({"retry_policy":{"max_attempts":2}}), 2, false), + ("flagged", json!({"retry_policy":{"max_attempts":3}}), 1, true), + ("named", json!({"retry_policy":{"max_attempts":3,"non_retryable_error_types":["Boom"]}}), 1, false), + ("empty", json!({"retry_policy":{}}), 1, false), + ("unknown-keys", json!({"retry_policy":{"maximum_attempts":2}}), 1, false), + ] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let parent = parent_with_child(&app, id, options).await; + for _ in 0..attempts { + let task = child_task(&app).await; + let (status, body) = finish_task(&app, &task, fail(non_retryable)).await; + assert_eq!(status, StatusCode::OK, "{id}: {body}"); + } + let resumed = poll(&app, "worker", "workflow").await; + assert_eq!(resumed["workflow_type"], "echo", "{id}: {resumed}"); + let events = child_events(&run_history(&app, &json!(id), &parent["run_id"]).await); + let last = events.last().unwrap(); + assert_eq!(last["event_type"], "ChildRunFailed", "{id}"); + assert_eq!(last["payload"]["child_run_number"], attempts, "{id}"); + assert_eq!(events.iter().filter(|event| event["event_type"] == "ChildRunStarted").count(), attempts as usize, "{id}"); + runtime.close().await; + database.remove().await; + } +} + +#[tokio::test] +async fn a_backoff_delays_the_retried_run() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let parent = parent_with_child(&app, "delayed", json!({"retry_policy":{"max_attempts":2,"backoff_seconds":[3600]}})).await; + let first = child_task(&app).await; + let (status, body) = finish_task(&app, &first, fail(false)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!(poll(&app, "worker", "workflow").await.is_null(), "the retry waits out its backoff"); + let events = child_events(&run_history(&app, &json!("delayed"), &parent["run_id"]).await); + let instance = events[0]["payload"]["child_workflow_instance_id"].as_str().unwrap().to_owned(); + let (_, runs) = request(&app, "GET", &format!("/api/workflows/{instance}/runs"), Value::Null).await; + let statuses: Vec<_> = runs["runs"].as_array().unwrap().iter().map(|run| (run["run_number"].clone(), run["status"].clone())).collect(); + assert!(statuses.contains(&(json!(2), json!("pending"))), "{runs}"); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn a_timed_out_child_is_retried_and_its_last_timeout_fails_the_parent() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let parent = parent_with_child(&app, "timed-out", json!({"retry_policy":{"max_attempts":2},"run_timeout_seconds":1})).await; + for _ in 0..2 { + let task = child_task(&app).await; + tokio::time::sleep(std::time::Duration::from_millis(1200)).await; + let (status, body) = finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Array(vec![]))}])).await; + assert_eq!((status, &body["reason"]), (StatusCode::CONFLICT, &json!("run_timed_out")), "{body}"); + } + let resumed = poll(&app, "worker", "workflow").await; + assert_eq!(resumed["workflow_type"], "echo", "{resumed}"); + let events = child_events(&run_history(&app, &json!("timed-out"), &parent["run_id"]).await); + let failed = &events.last().unwrap()["payload"]; + assert_eq!(events.last().unwrap()["event_type"], "ChildRunFailed"); + assert_eq!((failed["child_run_number"].clone(), failed["child_status"].clone(), failed["closed_reason"].clone(), failed["failure_category"].clone()), + (json!(2), json!("failed"), json!("timed_out"), json!("timeout")), "{failed}"); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/child-retries-contract.mjs b/scripts/conformance/server-parity/child-retries-contract.mjs new file mode 100644 index 000000000..e0ee98386 --- /dev/null +++ b/scripts/conformance/server-parity/child-retries-contract.mjs @@ -0,0 +1,26 @@ +import assert from 'node:assert/strict'; + +// Reviewed child retry and timeout options (child-retries-probe.php), +// recorded against published Server 2.11.7. The fixture's `expected` holds +// the whole observation of each case; IDs are labels in order of first +// appearance and deadlines are recorded by presence. +export function checkChildRetries(fixture, observation) { + const definition = fixture.child_retries; + const state = observation.child_retries; + const projection = {scope: 'child_retries', cases: Object.keys(definition.cases).length}; + assert.ok(state && typeof state === 'object', 'actual child retries observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}, + 'embedded child retries explicitly inapplicable'); + return projection; + } + assert.equal(state.applicable, true, 'HTTP targets record child retries'); + assert.deepStrictEqual(Object.keys(definition.expected).sort(), Object.keys(definition.cases).sort(), + 'every child retries case has an expectation'); + for (const [name, expected] of Object.entries(definition.expected)) { + assert.deepStrictEqual(state[name], expected, `child retries ${name}`); + } + assert.deepStrictEqual(Object.keys(state).sort(), ['applicable', ...Object.keys(definition.expected)].sort(), + 'child retries cases'); + return projection; +} diff --git a/scripts/conformance/server-parity/child-retries-probe.php b/scripts/conformance/server-parity/child-retries-probe.php new file mode 100644 index 000000000..86cecaeba --- /dev/null +++ b/scripts/conformance/server-parity/child-retries-probe.php @@ -0,0 +1,190 @@ +createNamespace($namespace, 'child retries λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new ChildRetriesHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $headers = array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', 'X-Durable-Workflow-Protocol-Version' => '1.20', + ]); + $raw = static function (string $method, string $path, ?array $body = null) use ($http, $headers): array { + $response = $http->request($method, $path, ['headers' => $headers, + ...($body === null ? [] : ['json' => $body === [] ? (object) [] : $body])]); + + return ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + }; + $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + $answer = static fn (array $answer): array => ['status' => $answer['status'], + 'reason' => $answer['response']['reason'] ?? null, 'run_status' => $answer['response']['run_status'] ?? null, + 'outcome' => $answer['response']['outcome'] ?? null, 'recorded' => $answer['response']['recorded'] ?? null]; + $state = ['applicable' => true]; + foreach ($definition['cases'] as $case => $plan) { + $labels = []; + $label = static function (mixed $id) use (&$labels): mixed { + if (! is_string($id) || $id === '') { + return $id; + } + $key = strtolower($id); + $labels[$key] ??= '{id:'.(count($labels) + 1).'}'; + + return $labels[$key]; + }; + $id = $workflowId.'-cr-'.$case; + $queue = $id.'-queue'; + $worker = $id.'-worker'; + $registered = $raw('POST', 'worker/register', ['worker_id' => $worker, 'task_queue' => $queue, 'runtime' => 'php', + 'sdk_version' => 'raw-http', 'supported_workflow_types' => [$definition['workflow_type'], $definition['child_type']], + 'supported_activity_types' => [], + 'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', 'sticky_execution'], + ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http'])]); + $started = $raw('POST', 'workflows', ['workflow_id' => $id, 'workflow_type' => $definition['workflow_type'], + 'task_queue' => $queue, 'input' => $empty]); + if ($registered['status'] !== 201 || $started['status'] !== 201) { + throw new RuntimeException('Child retries setup: '.json_encode([$registered, $started])); + } + $parentRun = $started['response']['run_id']; + // A ready task is handed to a poll that starts after it is available, + // so each poll is short and repeated. + $poll = static function (string $type) use ($raw, $worker, $queue, $case): array { + for ($attempt = 0; $attempt < 10; $attempt++) { + $task = $raw('POST', 'worker/workflow-tasks/poll', ['worker_id' => $worker, 'task_queue' => $queue, + 'timeout_seconds' => 1])['response']['task'] ?? null; + if (is_array($task)) { + if (($task['workflow_type'] ?? null) !== $type) { + throw new RuntimeException("Child retries {$case}: expected a {$type} task, got ".json_encode($task['workflow_type'] ?? null)); + } + + return $task; + } + } + throw new RuntimeException("Child retries {$case}: no {$type} task"); + }; + $complete = static fn (array $task, array $commands): array => $raw('POST', + 'worker/workflow-tasks/'.$task['task_id'].'/complete', ['lease_owner' => $task['lease_owner'], + 'workflow_task_attempt' => $task['workflow_task_attempt'], 'commands' => $commands]); + $parentEvents = static function () use ($raw, $id, $parentRun): array { + $events = $raw('GET', 'workflows/'.rawurlencode($id).'/runs/'.$parentRun.'/history')['response']['events'] ?? []; + + return array_values(array_filter($events, static fn (array $event): bool => str_starts_with($event['event_type'], 'Child'))); + }; + $described = static fn (): array => array_intersect_key($raw('GET', 'workflows/'.rawurlencode($id))['response'], + array_flip($definition['described_fields'])); + + $observed = []; + $parentTask = $poll($definition['workflow_type']); + $observed['start'] = $answer($complete($parentTask, [['type' => 'start_child_workflow', + 'workflow_type' => $definition['child_type'], 'arguments' => $empty, ...$plan['options']]])); + $observed['attempts'] = []; + foreach ($plan['attempts'] as $index => $step) { + if ($index > 0 && isset($plan['retry_wait_seconds'])) { + usleep((int) ($plan['retry_wait_seconds'] * 1_000_000)); + } + $task = $poll($definition['child_type']); + $attempt = ['run_id' => $label($task['run_id'] ?? null)]; + if (isset($step['fail'])) { + $attempt['answer'] = $answer($complete($task, [['type' => 'fail_workflow', ...$step['fail']]])); + } else { + if (isset($step['late_seconds'])) { + usleep((int) ($step['late_seconds'] * 1_000_000)); + } + $attempt['answer'] = $answer($complete($task, [['type' => 'complete_workflow', 'result' => $empty]])); + } + $observed['attempts'][] = $attempt; + if ($index === 0 && count($plan['attempts']) > 1) { + $observed['described_during_retry'] = $described(); + } + } + + $events = $parentEvents(); + $observed['parent_events'] = array_map(static function (array $event) use ($definition, $label): array { + $projected = ['event_type' => $event['event_type']]; + foreach ($definition['event_fields'] as $field) { + if (array_key_exists($field, $event['payload'])) { + $projected[$field] = childRetriesValue($field, $event['payload'][$field], $label); + } + } + + return $projected; + }, $events); + // How each later run's deadlines relate to the first run's: the + // execution deadline spans every attempt, and each run's deadline + // starts with it. + $runs = array_values(array_filter($events, static fn (array $event): bool => $event['event_type'] === 'ChildRunStarted')); + $observed['retry_deadlines'] = array_map(static fn (array $event): array => [ + 'execution_deadline_same' => ($event['payload']['execution_deadline_at'] ?? null) === ($runs[0]['payload']['execution_deadline_at'] ?? null), + 'run_deadline_not_earlier' => ($event['payload']['run_deadline_at'] ?? '') >= ($runs[0]['payload']['run_deadline_at'] ?? ''), + ], array_slice($runs, 1)); + + $child = $events[0]['payload']['child_workflow_instance_id'] ?? null; + if (! is_string($child)) { + throw new RuntimeException("Child retries {$case}: no child instance"); + } + $listed = $raw('GET', 'workflows/'.rawurlencode($child).'/runs')['response']['runs'] ?? []; + usort($listed, static fn (array $a, array $b): int => ($a['run_number'] ?? 0) <=> ($b['run_number'] ?? 0)); + $observed['child_runs'] = array_map(static fn (array $run): array => ['run_id' => $label($run['run_id'] ?? null), + 'run_number' => $run['run_number'] ?? null, 'status' => $run['status'] ?? null, + 'is_current_run' => $run['is_current_run'] ?? null], $listed); + $observed['child_started'] = array_map(static function (array $run) use ($raw, $child, $definition, $label): array { + $history = $raw('GET', 'workflows/'.rawurlencode($child).'/runs/'.$run['run_id'].'/history')['response']['events'] ?? []; + $started = array_values(array_filter($history, static fn (array $event): bool => $event['event_type'] === 'WorkflowStarted'))[0]['payload'] ?? []; + $projected = []; + foreach ($definition['started_fields'] as $field) { + if (array_key_exists($field, $started)) { + $projected[$field] = childRetriesValue($field, $started[$field], $label); + } + } + + return $projected; + }, $listed); + + // The parent is resumed once the call resolves; its task ends the + // parent, so nothing is left for later fixtures' maintenance passes. + $resumed = $poll($definition['workflow_type']); + $observed['parent_resumed'] = $answer($complete($resumed, [['type' => 'complete_workflow', 'result' => $empty]])); + $state[$case] = $observed; + } + + return $state; +} + +/** IDs as labels, deadlines by presence, and a timeout message without its time. */ +function childRetriesValue(string $field, mixed $value, Closure $label): mixed +{ + if (in_array($field, ['execution_deadline_at', 'run_deadline_at'], true)) { + return is_string($value) ? '{deadline}' : $value; + } + if ($field === 'message' && is_string($value)) { + return preg_replace('/\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d+)?(?:Z|[+-]\d\d:\d\d)/', '{time}', $value); + } + if (str_ends_with($field, '_id') || in_array($field, ['workflow_link_id', 'child_call_id'], true)) { + return $label($value); + } + + return $value; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index c43cb3587..8b083a34b 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -19,6 +19,7 @@ import {checkLargeCommandBatches} from './large-command-batches-contract.mjs'; import {checkManyUpdates} from './many-updates-contract.mjs'; import {checkWorkerErrorBodies} from './worker-error-bodies-contract.mjs'; import {checkChildDescriptions} from './child-descriptions-contract.mjs'; +import {checkChildRetries} from './child-retries-contract.mjs'; import {checkStartSearchAttributes} from './start-search-attributes-contract.mjs'; import {checkQueryArgumentRouting} from './query-arguments-contract.mjs'; import {checkQueryContractArguments} from './query-contract-arguments-contract.mjs'; @@ -183,6 +184,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const manyUpdates = fixture.many_updates ? checkManyUpdates(fixture, observation) : null; const workerErrorBodies = fixture.worker_error_bodies ? checkWorkerErrorBodies(fixture, observation) : null; const childDescriptions = fixture.child_descriptions ? checkChildDescriptions(fixture, observation) : null; + const childRetries = fixture.child_retries ? checkChildRetries(fixture, observation) : null; const startSearchAttributes = fixture.start_search_attributes ? checkStartSearchAttributes(fixture, observation) : null; const queryArguments = fixture.query_argument_routing ? checkQueryArgumentRouting(fixture, observation) : null; const queryContractArguments = fixture.query_contract_arguments ? checkQueryContractArguments(fixture, observation) : null; @@ -765,6 +767,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(manyUpdates ? {many_updates: manyUpdates} : {}), ...(workerErrorBodies ? {worker_error_bodies: workerErrorBodies} : {}), ...(childDescriptions ? {child_descriptions: childDescriptions} : {}), + ...(childRetries ? {child_retries: childRetries} : {}), ...(startSearchAttributes ? {start_search_attributes: startSearchAttributes} : {}), ...(queryArguments ? {query_argument_routing: queryArguments} : {}), ...(queryContractArguments ? {query_contract_arguments: queryContractArguments} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 1d2df32a6..91ffb46b1 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -437,6 +437,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $manyUpdates = isset($fixture['many_updates']) ? finishHttpManyUpdates($client, $fixture, $workflowId, $url) : null; $workerErrorBodies = isset($fixture['worker_error_bodies']) ? finishHttpWorkerErrorBodies($client, $fixture, $workflowId, $url) : null; $childDescriptions = isset($fixture['child_descriptions']) ? finishHttpChildDescriptions($client, $fixture, $workflowId, $url) : null; + $childRetries = isset($fixture['child_retries']) ? finishHttpChildRetries($client, $fixture, $workflowId, $url) : null; $startSearchAttributes = isset($fixture['start_search_attributes']) ? finishHttpStartSearchAttributes($client, $fixture, $workflowId, $url) : null; $queryArgumentRouting = isset($fixture['query_argument_routing']) ? finishHttpQueryArgumentRouting($client, $fixture, $workflowId, $url) : null; $queryContractArguments = isset($fixture['query_contract_arguments']) ? finishHttpQueryContractArguments($client, $fixture, $workflowId, $url) : null; @@ -572,6 +573,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'many_updates' => $manyUpdates, 'worker_error_bodies' => $workerErrorBodies, 'child_descriptions' => $childDescriptions, + 'child_retries' => $childRetries, 'start_search_attributes' => $startSearchAttributes, 'query_argument_routing' => $queryArgumentRouting, 'query_contract_arguments' => $queryContractArguments, @@ -910,6 +912,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/many-updates-probe.php'; require __DIR__.'/worker-error-bodies-probe.php'; require __DIR__.'/child-descriptions-probe.php'; + require __DIR__.'/child-retries-probe.php'; require __DIR__.'/start-search-attributes-probe.php'; require __DIR__.'/query-arguments-probe.php'; require __DIR__.'/query-contract-arguments-probe.php'; @@ -1114,6 +1117,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['child_descriptions'])) { $observation['child_descriptions'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } + if ($mode === 'embedded' && isset($fixture['child_retries'])) { + $observation['child_retries'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; + } if ($mode === 'embedded' && isset($fixture['start_search_attributes'])) { $observation['start_search_attributes'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_search_attribute_registry']; } diff --git a/tests/Fixtures/ServerParity/child-retries.json b/tests/Fixtures/ServerParity/child-retries.json new file mode 100644 index 000000000..2ee991c28 --- /dev/null +++ b/tests/Fixtures/ServerParity/child-retries.json @@ -0,0 +1,1317 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "child-retries", + "review_state": "reviewed", + "contract": "A child's retry and timeout options (published Server 2.11.7 with Workflow 2.15.7 ChildWorkflowRetryPolicy and the bridge's startChildRetryIfAvailable), over raw worker HTTP on protocol 1.20 in a fresh namespace. Each case starts a parent with one start_child_workflow command carrying retry_policy, execution_timeout_seconds or run_timeout_seconds, then fails, completes or late-completes the child's runs. The call, ChildWorkflowScheduled, ChildRunStarted and the child's WorkflowStarted record a retry snapshot and a timeout snapshot; a policy with no known key records nothing. A failed or timed-out child run is retried as a new run of the same child instance under a new link while the policy has attempts left and the failure is retryable; the retry keeps the call and the execution deadline, takes a new run deadline, and waits out the backoff, and the parent stays waiting for the child. A failure marked non-retryable or of a type the policy names, or the last attempt's failure, fails the parent's call. IDs are labels in order of first appearance and deadlines are recorded by presence. Embedded applications have no worker protocol.", + "workflow_type": "parity.v1.echo", + "input": { + "message": "child retries λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "child retries λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "child_retries": { + "workflow_type": "parity.v1.child_retries", + "child_type": "parity.v1.child_retries_child", + "empty_blob": "wwHioz3/VYAiNwwA", + "described_fields": [ + "status", + "status_bucket", + "is_terminal", + "resume_step_sequence", + "recovery_kind", + "wait_kind", + "wait_reason" + ], + "event_fields": [ + "sequence", + "workflow_link_id", + "child_call_id", + "child_workflow_instance_id", + "child_workflow_run_id", + "child_workflow_type", + "child_run_number", + "parent_close_policy", + "cancellation_policy", + "retry_attempt", + "retry_of_child_workflow_run_id", + "retry_backoff_seconds", + "retry_policy", + "timeout_policy", + "execution_timeout_seconds", + "run_timeout_seconds", + "execution_deadline_at", + "run_deadline_at", + "child_status", + "closed_reason", + "failure_id", + "failure_category", + "exception_type", + "exception_class", + "message" + ], + "started_fields": [ + "workflow_type", + "workflow_instance_id", + "workflow_run_id", + "parent_workflow_instance_id", + "parent_workflow_run_id", + "parent_sequence", + "workflow_link_id", + "child_call_id", + "retry_attempt", + "retry_of_child_workflow_run_id", + "retry_policy", + "timeout_policy", + "execution_timeout_seconds", + "run_timeout_seconds", + "execution_deadline_at", + "run_deadline_at" + ], + "cases": { + "retry_then_complete": { + "options": { + "retry_policy": { + "max_attempts": 3, + "backoff_seconds": [ + 1 + ] + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "retry_wait_seconds": 1.5, + "attempts": [ + { + "fail": { + "message": "attempt failed", + "exception_type": "Boom", + "exception_class": "Boom" + } + }, + { + "complete": true + } + ] + }, + "attempts_exhausted": { + "options": { + "retry_policy": { + "max_attempts": 2 + } + }, + "attempts": [ + { + "fail": { + "message": "attempt failed", + "exception_type": "Boom", + "exception_class": "Boom" + } + }, + { + "fail": { + "message": "attempt failed", + "exception_type": "Boom", + "exception_class": "Boom" + } + } + ] + }, + "non_retryable_type": { + "options": { + "retry_policy": { + "max_attempts": 3, + "non_retryable_error_types": [ + "Fatal" + ] + } + }, + "attempts": [ + { + "fail": { + "message": "fatal attempt", + "exception_type": "Fatal", + "exception_class": "FatalError" + } + } + ] + }, + "non_retryable_flag": { + "options": { + "retry_policy": { + "max_attempts": 3 + } + }, + "attempts": [ + { + "fail": { + "message": "final attempt", + "exception_type": "Boom", + "exception_class": "Boom", + "non_retryable": true + } + } + ] + }, + "empty_policy": { + "options": { + "retry_policy": {} + }, + "attempts": [ + { + "fail": { + "message": "attempt failed", + "exception_type": "Boom", + "exception_class": "Boom" + } + } + ] + }, + "timeouts_only": { + "options": { + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "attempts": [ + { + "complete": true + } + ] + }, + "run_timeout_retried": { + "options": { + "retry_policy": { + "max_attempts": 2 + }, + "run_timeout_seconds": 1 + }, + "attempts": [ + { + "late_seconds": 1.5 + }, + { + "late_seconds": 1.5 + } + ] + } + }, + "expected": { + "retry_then_complete": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 200, + "reason": null, + "run_status": "failed", + "outcome": "completed", + "recorded": true + } + }, + { + "run_id": "{id:2}", + "answer": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + } + ], + "described_during_retry": { + "status": "waiting", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "child", + "wait_reason": "Waiting for child workflow parity.v1.child_retries_child" + }, + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [ + 1 + ], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + } + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [ + 1 + ], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300, + "execution_deadline_at": "{deadline}", + "run_deadline_at": "{deadline}" + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:2}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 2, + "parent_close_policy": "abandon", + "retry_attempt": 2, + "retry_of_child_workflow_run_id": "{id:1}", + "retry_backoff_seconds": 1, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [ + 1 + ], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300, + "execution_deadline_at": "{deadline}", + "run_deadline_at": "{deadline}" + }, + { + "event_type": "ChildRunCompleted", + "sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:2}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 2, + "child_status": "completed", + "closed_reason": "completed" + } + ], + "retry_deadlines": [ + { + "execution_deadline_same": true, + "run_deadline_not_earlier": true + } + ], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "failed", + "is_current_run": false + }, + { + "run_id": "{id:2}", + "run_number": 2, + "status": "completed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:4}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:6}", + "parent_workflow_run_id": "{id:7}", + "parent_sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [ + 1 + ], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300, + "execution_deadline_at": "{deadline}", + "run_deadline_at": "{deadline}" + }, + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:4}", + "workflow_run_id": "{id:2}", + "parent_workflow_instance_id": "{id:6}", + "parent_workflow_run_id": "{id:7}", + "parent_sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "retry_attempt": 2, + "retry_of_child_workflow_run_id": "{id:1}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [ + 1 + ], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300, + "execution_deadline_at": "{deadline}", + "run_deadline_at": "{deadline}" + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + }, + "attempts_exhausted": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 200, + "reason": null, + "run_status": "failed", + "outcome": "completed", + "recorded": true + } + }, + { + "run_id": "{id:2}", + "answer": { + "status": 200, + "reason": null, + "run_status": "failed", + "outcome": "completed", + "recorded": true + } + } + ], + "described_during_retry": { + "status": "waiting", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "child", + "wait_reason": "Waiting for child workflow parity.v1.child_retries_child" + }, + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:2}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 2, + "parent_close_policy": "abandon", + "retry_attempt": 2, + "retry_of_child_workflow_run_id": "{id:1}", + "retry_backoff_seconds": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + } + }, + { + "event_type": "ChildRunFailed", + "sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:2}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 2, + "child_status": "failed", + "closed_reason": "failed", + "failure_id": "{id:6}", + "failure_category": "application", + "exception_type": "Boom", + "exception_class": "Boom", + "message": "attempt failed" + } + ], + "retry_deadlines": [ + { + "execution_deadline_same": true, + "run_deadline_not_earlier": true + } + ], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "failed", + "is_current_run": false + }, + { + "run_id": "{id:2}", + "run_number": 2, + "status": "failed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:4}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:7}", + "parent_workflow_run_id": "{id:8}", + "parent_sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + }, + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:4}", + "workflow_run_id": "{id:2}", + "parent_workflow_instance_id": "{id:7}", + "parent_workflow_run_id": "{id:8}", + "parent_sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "retry_attempt": 2, + "retry_of_child_workflow_run_id": "{id:1}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + }, + "non_retryable_type": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 200, + "reason": null, + "run_status": "failed", + "outcome": "completed", + "recorded": true + } + } + ], + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [], + "non_retryable_error_types": [ + "Fatal" + ] + }, + "timeout_policy": null + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [], + "non_retryable_error_types": [ + "Fatal" + ] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + }, + { + "event_type": "ChildRunFailed", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "child_status": "failed", + "closed_reason": "failed", + "failure_id": "{id:4}", + "failure_category": "application", + "exception_type": "Fatal", + "exception_class": "FatalError", + "message": "fatal attempt" + } + ], + "retry_deadlines": [], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "failed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:3}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:5}", + "parent_workflow_run_id": "{id:6}", + "parent_sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [], + "non_retryable_error_types": [ + "Fatal" + ] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + }, + "non_retryable_flag": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 200, + "reason": null, + "run_status": "failed", + "outcome": "completed", + "recorded": true + } + } + ], + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + }, + { + "event_type": "ChildRunFailed", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "child_status": "failed", + "closed_reason": "failed", + "failure_id": "{id:4}", + "failure_category": "application", + "exception_type": "Boom", + "exception_class": "Boom", + "message": "final attempt" + } + ], + "retry_deadlines": [], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "failed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:3}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:5}", + "parent_workflow_run_id": "{id:6}", + "parent_sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 3, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + }, + "empty_policy": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 200, + "reason": null, + "run_status": "failed", + "outcome": "completed", + "recorded": true + } + } + ], + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": null, + "timeout_policy": null + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": null, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + }, + { + "event_type": "ChildRunFailed", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "child_status": "failed", + "closed_reason": "failed", + "failure_id": "{id:4}", + "failure_category": "application", + "exception_type": "Boom", + "exception_class": "Boom", + "message": "attempt failed" + } + ], + "retry_deadlines": [], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "failed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:3}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:5}", + "parent_workflow_run_id": "{id:6}", + "parent_sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "retry_policy": null, + "timeout_policy": null, + "execution_timeout_seconds": null, + "run_timeout_seconds": null, + "execution_deadline_at": null, + "run_deadline_at": null + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + }, + "timeouts_only": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + } + ], + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": null, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + } + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": null, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300, + "execution_deadline_at": "{deadline}", + "run_deadline_at": "{deadline}" + }, + { + "event_type": "ChildRunCompleted", + "sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "child_workflow_instance_id": "{id:3}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "child_status": "completed", + "closed_reason": "completed" + } + ], + "retry_deadlines": [], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "completed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:3}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:4}", + "parent_workflow_run_id": "{id:5}", + "parent_sequence": 1, + "workflow_link_id": "{id:2}", + "child_call_id": "{id:2}", + "retry_policy": null, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300 + }, + "execution_timeout_seconds": 600, + "run_timeout_seconds": 300, + "execution_deadline_at": "{deadline}", + "run_deadline_at": "{deadline}" + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + }, + "run_timeout_retried": { + "start": { + "status": 200, + "reason": null, + "run_status": "waiting", + "outcome": "completed", + "recorded": true + }, + "attempts": [ + { + "run_id": "{id:1}", + "answer": { + "status": 409, + "reason": "run_timed_out", + "run_status": "failed", + "outcome": "completed", + "recorded": false + } + }, + { + "run_id": "{id:2}", + "answer": { + "status": 409, + "reason": "run_timed_out", + "run_status": "failed", + "outcome": "completed", + "recorded": false + } + } + ], + "described_during_retry": { + "status": "waiting", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "child", + "wait_reason": "Waiting for child workflow parity.v1.child_retries_child" + }, + "parent_events": [ + { + "event_type": "ChildWorkflowScheduled", + "sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1 + } + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:1}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 1, + "parent_close_policy": "abandon", + "cancellation_policy": "abandon", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1 + }, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1, + "execution_deadline_at": null, + "run_deadline_at": "{deadline}" + }, + { + "event_type": "ChildRunStarted", + "sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:2}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 2, + "parent_close_policy": "abandon", + "retry_attempt": 2, + "retry_of_child_workflow_run_id": "{id:1}", + "retry_backoff_seconds": 0, + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1 + }, + "run_timeout_seconds": 1, + "run_deadline_at": "{deadline}" + }, + { + "event_type": "ChildRunFailed", + "sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "child_workflow_instance_id": "{id:4}", + "child_workflow_run_id": "{id:2}", + "child_workflow_type": "parity.v1.child_retries_child", + "child_run_number": 2, + "child_status": "failed", + "closed_reason": "timed_out", + "failure_id": "{id:6}", + "failure_category": "timeout", + "exception_class": "Workflow\\V2\\Exceptions\\WorkflowTimeoutException", + "message": "Workflow run deadline expired at {time}." + } + ], + "retry_deadlines": [ + { + "execution_deadline_same": true, + "run_deadline_not_earlier": true + } + ], + "child_runs": [ + { + "run_id": "{id:1}", + "run_number": 1, + "status": "failed", + "is_current_run": false + }, + { + "run_id": "{id:2}", + "run_number": 2, + "status": "failed", + "is_current_run": true + } + ], + "child_started": [ + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:4}", + "workflow_run_id": "{id:1}", + "parent_workflow_instance_id": "{id:7}", + "parent_workflow_run_id": "{id:8}", + "parent_sequence": 1, + "workflow_link_id": "{id:3}", + "child_call_id": "{id:3}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1 + }, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1, + "execution_deadline_at": null, + "run_deadline_at": "{deadline}" + }, + { + "workflow_type": "parity.v1.child_retries_child", + "workflow_instance_id": "{id:4}", + "workflow_run_id": "{id:2}", + "parent_workflow_instance_id": "{id:7}", + "parent_workflow_run_id": "{id:8}", + "parent_sequence": 1, + "workflow_link_id": "{id:5}", + "child_call_id": "{id:3}", + "retry_attempt": 2, + "retry_of_child_workflow_run_id": "{id:1}", + "retry_policy": { + "snapshot_version": 1, + "max_attempts": 2, + "backoff_seconds": [], + "non_retryable_error_types": [] + }, + "timeout_policy": { + "snapshot_version": 1, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1 + }, + "execution_timeout_seconds": null, + "run_timeout_seconds": 1, + "execution_deadline_at": null, + "run_deadline_at": "{deadline}" + } + ], + "parent_resumed": { + "status": 200, + "reason": null, + "run_status": "completed", + "outcome": "completed", + "recorded": true + } + } + } + } +} diff --git a/tests/Unit/ServerParityChildRetriesContractTest.mjs b/tests/Unit/ServerParityChildRetriesContractTest.mjs new file mode 100644 index 000000000..d7b43f731 --- /dev/null +++ b/tests/Unit/ServerParityChildRetriesContractTest.mjs @@ -0,0 +1,62 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkChildRetries} from '../../scripts/conformance/server-parity/child-retries-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/child-retries.json', import.meta.url), 'utf8')); +const expected = fixture.child_retries.expected; +const observation = () => structuredClone({mode: 'http', child_retries: {applicable: true, ...expected}}); +const cases = Object.keys(fixture.child_retries.cases).length; +const events = name => expected[name].parent_events.map(event => event.event_type); + +test('the published answers pass', () => { + assert.deepStrictEqual(checkChildRetries(fixture, observation()), {scope: 'child_retries', cases}); +}); + +test('embedded applications have no worker protocol', () => { + assert.deepStrictEqual(checkChildRetries(fixture, + {mode: 'embedded', child_retries: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), + {scope: 'child_retries', cases}); +}); + +test('a failed or timed-out child is retried while attempts remain and the failure is retryable', () => { + const started = ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunStarted']; + assert.deepStrictEqual(events('retry_then_complete'), [...started, 'ChildRunCompleted']); + assert.deepStrictEqual(events('attempts_exhausted'), [...started, 'ChildRunFailed']); + assert.deepStrictEqual(events('run_timeout_retried'), [...started, 'ChildRunFailed']); + for (const name of ['non_retryable_type', 'non_retryable_flag', 'empty_policy']) { + assert.deepStrictEqual(events(name), ['ChildWorkflowScheduled', 'ChildRunStarted', 'ChildRunFailed'], name); + } + const retry = expected.retry_then_complete.parent_events[2]; + assert.deepStrictEqual([retry.child_run_number, retry.retry_attempt, retry.retry_backoff_seconds], [2, 2, 1]); + assert.equal(retry.child_call_id, expected.retry_then_complete.parent_events[0].child_call_id, 'the retry keeps the call'); + assert.notEqual(retry.workflow_link_id, retry.child_call_id, 'the retry has its own link'); + assert.equal(expected.retry_then_complete.described_during_retry.wait_kind, 'child'); + assert.deepStrictEqual(expected.retry_then_complete.retry_deadlines, + [{execution_deadline_same: true, run_deadline_not_earlier: true}]); + assert.deepStrictEqual(expected.run_timeout_retried.attempts.map(attempt => attempt.answer.reason), + ['run_timed_out', 'run_timed_out']); +}); + +test('an empty policy records nothing and timeouts alone record no retry policy', () => { + assert.equal(expected.empty_policy.parent_events[0].retry_policy, null); + assert.equal(expected.timeouts_only.parent_events[0].retry_policy, null); + assert.deepStrictEqual(expected.timeouts_only.parent_events[0].timeout_policy, + {snapshot_version: 1, execution_timeout_seconds: 600, run_timeout_seconds: 300}); +}); + +for (const [name, mutate] of [ + ['a refused child option', state => { state.retry_then_complete.start = {status: 422, reason: 'unsupported_request_field', + run_status: null, outcome: null, recorded: null}; }], + ['a child failed without its retry', state => { state.attempts_exhausted.parent_events.splice(2, 1); }], + ['a retry of a non-retryable type', state => { state.non_retryable_type.child_runs.push({run_id: '{id:9}', run_number: 2, + status: 'pending', is_current_run: true}); }], + ['a retry that renews the execution deadline', state => { + state.retry_then_complete.retry_deadlines[0].execution_deadline_same = false; }], + ['a parent described without a wait during the retry', state => { + state.retry_then_complete.described_during_retry.wait_kind = null; }], + ['an unexpected case', state => { state.extra = {}; }], +]) test(`the contract rejects ${name}`, () => { + const raw = observation(); mutate(raw.child_retries); + assert.throws(() => checkChildRetries(fixture, raw)); +}); From f6c58c798477ad822c30cf1e4556489baa274853 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:29:30 +0000 Subject: [PATCH 2/3] Start a child retry on MySQL and MariaDB too The retry read its next run number from MAX() of an unsigned MySQL column, which does not decode as a signed integer, and repeated a placeholder in one statement, which MySQL's statement templates refuse. Read the latest run as a row and bind the run count twice. The child-retries fixture passes on MariaDB 10.11, PostgreSQL 17 and SQLite. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/children.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index bedfc8fb9..db6a18ade 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -577,8 +577,11 @@ where let backoff = super::child_retries::backoff_seconds(&policy, attempts); let available_at = at + chrono::Duration::seconds(backoff); let instance = DB::string(failed, "workflow_instance_id")?; - let run_number = Self::scalar("SELECT COALESCE(MAX(run_number),0) FROM workflow_runs WHERE workflow_instance_id=$1") - .bind(&instance).fetch_one(&mut **tx).await? + 1; + // The latest run as a row: an aggregate of MySQL's unsigned column + // does not decode as a signed integer. + let latest = Self::query("SELECT run_number FROM workflow_runs WHERE workflow_instance_id=$1 ORDER BY run_number DESC LIMIT 1") + .bind(&instance).fetch_one(&mut **tx).await?; + let run_number = DB::number(&latest, "run_number")? + 1; // The execution deadline spans every attempt; each run gets its own. let execution_deadline = DB::optional_instant(failed, "execution_deadline_at")? .or(execution_timeout.map(|seconds| at + chrono::Duration::seconds(seconds))); @@ -588,8 +591,9 @@ where .bind(&retry_run).bind(run_number).bind(run_timeout.unwrap_or(0)).bind(DB::optional_time(execution_deadline)).bind(DB::optional_time(run_deadline)) .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(&failed_run) .execute(&mut **tx).await?; - Self::query("UPDATE workflow_instances SET current_run_id=$1,run_count=CASE WHEN run_count>$2 THEN run_count ELSE $2 END,updated_at=$3 WHERE id=$4") - .bind(&retry_run).bind(run_number).bind(DB::bind_time(at)).bind(&instance).execute(&mut **tx).await?; + // Each placeholder appears once, in order, as MySQL's templates require. + Self::query("UPDATE workflow_instances SET current_run_id=$1,run_count=CASE WHEN run_count>$2 THEN run_count ELSE $3 END,updated_at=$4 WHERE id=$5") + .bind(&retry_run).bind(run_number).bind(run_number).bind(DB::bind_time(at)).bind(&instance).execute(&mut **tx).await?; let parent_close = DB::optional_string(link, "parent_close_policy")?.unwrap_or_else(|| "abandon".into()); let retry_link = id(); Self::query("INSERT INTO workflow_links(id,link_type,sequence,parent_workflow_instance_id,parent_workflow_run_id,child_workflow_instance_id,child_workflow_run_id,is_primary_parent,parent_close_policy,created_at,updated_at) VALUES ($1,'child_workflow',$2,$3,$4,$5,$6,TRUE,$7,$8,$9)") From eb7abf8d87938658dce3ac72bfa3b4c0d27df50a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 22:05:42 +0000 Subject: [PATCH 3/3] Refuse a child deadline or backoff too far away for a timestamp Security's review of #869 found that an admitted i64::MAX child timeout or backoff could overflow chrono's date arithmetic. A child timeout or backoff entry too far away is now refused at admission (422 invalid_child_timeout or invalid_child_retry_policy), as a timer delay or activity backoff is, and every child deadline uses checked arithmetic. Cooperative cancellation delivered to a retried child stays refused and is tracked in #873. Co-Authored-By: Claude Opus 5.5 --- .../2026-10-29-child-retries.md | 15 ++++++++-- docs/rust-server-port.md | 17 ++++++----- rust/src/runtime/child_retries.rs | 28 +++++++++++++++++ rust/src/runtime/children.rs | 30 +++++++++++++++---- rust/tests/execution/child_retries.rs | 23 ++++++++++++++ 5 files changed, 97 insertions(+), 16 deletions(-) diff --git a/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md b/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md index d6872bc56..77d3a1b03 100644 --- a/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md +++ b/docs/rust-server-port-checkpoints/2026-10-29-child-retries.md @@ -52,9 +52,20 @@ recording found it. - **Tests.** `child_retries.rs` (native), and the snapshot, backoff and non-retryable rules in `child_retries.rs` unit tests. +- **Bounds** (Security review). A child timeout or backoff entry too far + away for a timestamp is refused at admission with `422 + invalid_child_timeout` or `invalid_child_retry_policy`, and every child + deadline is computed with checked arithmetic, so an extreme value cannot + overflow. PHP's Carbon wraps such a value instead; native refuses a timer + delay or activity backoff the same way. +- **CI.** The first hosted run failed the MariaDB corpus: the retry read + `MAX(run_number)`, which MySQL returns unsigned, and repeated a + placeholder. Both are fixed. Locally, the fixture passes on MariaDB 10.11, + PostgreSQL 17 and SQLite. - **Not covered here.** - Cooperative cancellation delivered to a retried child: native's - delivery expects one link per call and refuses it. This is from reading - the code; it is not recorded against PHP. + delivery expects one link per call and refuses it with `409 + cancellation_delivery_shape_mismatch`. Security's review confirmed this + from the source. A fix needs a PHP recording first: #873. - Memos and search attributes inherited by child runs, retried or not (a separate gap). diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index b3a6c102c..245928f88 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -612,11 +612,12 @@ contract fixture before its correction. way, and describes a parent waiting on a retried child as waiting for it. The `child-retries` fixture records seven cases against published 2.11.7 ([checkpoint](rust-server-port-checkpoints/2026-10-29-child-retries.md)). - Remaining, from reading the code rather than a recording: native's - cooperative cancellation delivery expects one link per call, so it - refuses one delivered to a retried child - (`cancellation_delivery_shape_mismatch`); and a retried child inherits no - memos or search attributes (below). + A child timeout or backoff too far away for a timestamp is refused + (`invalid_child_timeout`, `invalid_child_retry_policy`), as a timer delay + is. Remaining: native's cooperative cancellation delivery expects one + link per call, so it refuses one delivered to a retried child + (`cancellation_delivery_shape_mismatch`; #873, pending a PHP recording); + and a retried child inherits no memos or search attributes (below). - **Trace context (Core 2.10, published in Server 2.11.1).** Core adds nullable `observability_context` columns on runs, commands and tasks. Native has no carrier: it accepts and ignores a top-level carrier on @@ -673,9 +674,9 @@ contract fixture before its correction. - A field for a feature native cannot run, such as `worker_session`, is refused with `unsupported_request_field`. Fields PHP ignores are dropped as PHP drops them. - - A timer delay, retry backoff or wait timeout too large for a timestamp - is refused on native. PHP accepts it, and wraps a timer's fire time into - the past, so the timer fires at once. + - A timer delay, retry backoff, child timeout or wait timeout too large + for a timestamp is refused on native. PHP accepts it, and wraps a + timer's fire time into the past, so the timer fires at once. - **Search attribute limits.** Over 100 attributes or 64 KB per run, PHP raises an uncaught `InvalidArgumentException` while applying the command. Native refuses the completion with `422 search_attribute_limit_exceeded` diff --git a/rust/src/runtime/child_retries.rs b/rust/src/runtime/child_retries.rs index ee7900f02..6f5b48680 100644 --- a/rust/src/runtime/child_retries.rs +++ b/rust/src/runtime/child_retries.rs @@ -1,6 +1,9 @@ //! A child's retry and timeout options (`ChildWorkflowRetryPolicy`): what //! a `start_child_workflow` command records, and whether a failed child run //! is retried as a new run of the same child instance. +use super::{Result, refuse}; +use axum::http::StatusCode; +use chrono::{DateTime, Utc}; use serde_json::{Value, json}; const SNAPSHOT_VERSION: i64 = 1; @@ -41,6 +44,21 @@ pub(super) fn timeout(command: &Value, field: &str) -> Option { command[field].as_i64() } +/// `at` plus a non-negative number of seconds, when the result is a +/// timestamp. +pub(super) fn after(at: DateTime, seconds: i64) -> Option> { + (seconds >= 0).then_some(())?; + chrono::Duration::try_seconds(seconds).and_then(|delta| at.checked_add_signed(delta)) +} + +/// A child run's deadline for an optional timeout; one too far away is +/// refused (`validate_child` refuses it first). +pub(super) fn deadline(at: DateTime, seconds: Option) -> Result>> { + seconds + .map(|seconds| after(at, seconds).ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_child_timeout"))) + .transpose() +} + pub(super) fn max_attempts(policy: &Value) -> i64 { policy["max_attempts"].as_i64().map_or(1, |attempts| attempts.max(1)) } @@ -147,6 +165,16 @@ mod tests { ); } + #[test] + fn a_deadline_too_far_away_is_refused_rather_than_overflowing() { + let at = chrono::Utc::now(); + assert!(after(at, i64::MAX).is_none()); + assert!(after(at, -1).is_none()); + assert_eq!(after(at, 5), Some(at + chrono::Duration::seconds(5))); + assert!(deadline(at, Some(i64::MAX)).is_err()); + assert_eq!(deadline(at, None).unwrap(), None); + } + #[test] fn backoff_uses_the_entry_for_the_attempt_then_the_last() { let policy = json!({"backoff_seconds": [1, 4]}); diff --git a/rust/src/runtime/children.rs b/rust/src/runtime/children.rs index db6a18ade..9062a738a 100644 --- a/rust/src/runtime/children.rs +++ b/rust/src/runtime/children.rs @@ -59,6 +59,21 @@ pub(super) fn validate_child(command: &Value) -> Result<()> { } policies(command)?; envelope(command, "arguments")?; + // A deadline or backoff too far away for a timestamp is refused, as a + // timer delay or an activity backoff is (PHP's Carbon wraps it). + let now = now(); + for field in ["execution_timeout_seconds", "run_timeout_seconds"] { + if let Some(seconds) = super::child_retries::timeout(command, field) + && super::child_retries::after(now, seconds).is_none() + { + return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_child_timeout")); + } + } + if let Some(backoff) = command["retry_policy"]["backoff_seconds"].as_array() + && backoff.iter().any(|entry| entry.as_i64().is_some_and(|seconds| super::child_retries::after(now, seconds).is_none())) + { + return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_child_retry_policy")); + } Ok(()) } @@ -287,8 +302,8 @@ where // its parent's. let execution_timeout = super::child_retries::timeout(command, "execution_timeout_seconds"); let run_timeout = super::child_retries::timeout(command, "run_timeout_seconds"); - let execution_deadline = execution_timeout.map(|seconds| started_at + chrono::Duration::seconds(seconds)); - let run_deadline = run_timeout.map(|seconds| started_at + chrono::Duration::seconds(seconds)); + let execution_deadline = super::child_retries::deadline(started_at, execution_timeout)?; + let run_deadline = super::child_retries::deadline(started_at, run_timeout)?; let retry_policy = super::child_retries::retry_snapshot(command); let timeout_policy = super::child_retries::timeout_snapshot(execution_timeout, run_timeout); Self::query("INSERT INTO workflow_instances(id,namespace,workflow_type,workflow_class,current_run_id,run_count,started_at,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,1,$6,$7,$8)") @@ -575,7 +590,8 @@ where let run_timeout = timeout_policy["run_timeout_seconds"].as_i64(); let at = now(); let backoff = super::child_retries::backoff_seconds(&policy, attempts); - let available_at = at + chrono::Duration::seconds(backoff); + let available_at = super::child_retries::after(at, backoff) + .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_child_retry_policy"))?; let instance = DB::string(failed, "workflow_instance_id")?; // The latest run as a row: an aggregate of MySQL's unsigned column // does not decode as a signed integer. @@ -583,9 +599,11 @@ where .bind(&instance).fetch_one(&mut **tx).await?; let run_number = DB::number(&latest, "run_number")? + 1; // The execution deadline spans every attempt; each run gets its own. - let execution_deadline = DB::optional_instant(failed, "execution_deadline_at")? - .or(execution_timeout.map(|seconds| at + chrono::Duration::seconds(seconds))); - let run_deadline = run_timeout.map(|seconds| at + chrono::Duration::seconds(seconds)); + let execution_deadline = match DB::optional_instant(failed, "execution_deadline_at")? { + Some(deadline) => Some(deadline), + None => super::child_retries::deadline(at, execution_timeout)?, + }; + let run_deadline = super::child_retries::deadline(at, run_timeout)?; let retry_run = id(); Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,run_number,workflow_class,workflow_type,namespace,business_key,visibility_labels,status,compatibility,payload_codec,arguments,run_timeout_seconds,execution_deadline_at,run_deadline_at,connection,queue,started_at,last_progress_at,created_at,updated_at) SELECT $1,workflow_instance_id,$2,workflow_class,workflow_type,namespace,business_key,visibility_labels,'pending',compatibility,payload_codec,arguments,NULLIF($3,0),$4,$5,connection,queue,$6,$7,$8,$9 FROM workflow_runs WHERE id=$10") .bind(&retry_run).bind(run_number).bind(run_timeout.unwrap_or(0)).bind(DB::optional_time(execution_deadline)).bind(DB::optional_time(run_deadline)) diff --git a/rust/tests/execution/child_retries.rs b/rust/tests/execution/child_retries.rs index 4ac531eb8..179c63fc9 100644 --- a/rust/tests/execution/child_retries.rs +++ b/rust/tests/execution/child_retries.rs @@ -150,3 +150,26 @@ async fn a_timed_out_child_is_retried_and_its_last_timeout_fails_the_parent() { runtime.close().await; database.remove().await; } + +#[tokio::test] +async fn a_child_deadline_or_backoff_too_far_away_is_refused_not_overflowed() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "worker", json!(["echo", "child"]), json!([])).await; + start(&app, "extreme").await; + let task = poll(&app, "worker", "workflow").await; + for (options, reason) in [ + (json!({"execution_timeout_seconds": i64::MAX}), "invalid_child_timeout"), + (json!({"run_timeout_seconds": i64::MAX}), "invalid_child_timeout"), + (json!({"retry_policy": {"max_attempts": 2, "backoff_seconds": [1, i64::MAX]}}), "invalid_child_retry_policy"), + ] { + let (status, body) = finish_task(&app, &task, json!([child(options.clone())])).await; + assert_eq!((status, &body["reason"]), (StatusCode::UNPROCESSABLE_ENTITY, &json!(reason)), "{options}: {body}"); + } + // The task is still open and the runtime still serves it. + let (status, body) = finish_task(&app, &task, json!([child(json!({"run_timeout_seconds": 60}))])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + runtime.close().await; + database.remove().await; +}