From 8273d4ab33ceeffbb0c09c34c22c26234397854d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 01:08:45 +0000 Subject: [PATCH 1/8] Rust #325: accept completions and parallel groups over 100 Published PHP 2.11.5 accepts a workflow task completion with any non-empty number of commands; each command meets the run's structural limits instead (2,000 pending activities and timers by default). Native refused more than 100 with invalid_commands, and its timer, activity and external-wait summaries gave up past 100 open waits, so such a run was described without its wait. Remove both command caps and the three summary bounds. PHP bounds a parallel group's declared size by the command batch size (1,000 by default), checked in WorkflowCommandNormalizer's preflightParallelMetadata. WorkerController runs it after the transport rules and before any guard, and it collects one of three validation_failed messages per command across the batch. Native capped groups at 100 and answered any invalid metadata with development_parallel_group_unqualified. Port the preflight into completion_rules and share one COMMAND_BATCH_SIZE constant. The large-command-batches fixture records 150 timers, 150 activities, a 150-member group and a refused 1,001-member group; published PHP and native both pass it. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 3 +- .../2026-10-29-large-command-batches.md | 60 ++++++ docs/rust-server-port.md | 23 ++- rust/src/runtime/activity_summary.rs | 7 +- rust/src/runtime/all_groups.rs | 2 +- rust/src/runtime/cancellation_scopes.rs | 3 +- rust/src/runtime/completion_rules.rs | 132 ++++++++++++- rust/src/runtime/cooperative.rs | 3 +- rust/src/runtime/external_resolution.rs | 2 +- rust/src/runtime/external_wait_summary.rs | 7 +- rust/src/runtime/memo_limits.rs | 6 + rust/src/runtime/store.rs | 4 +- rust/src/runtime/timer_summary.rs | 6 +- rust/tests/execution.rs | 3 + rust/tests/execution/large_command_batches.rs | 82 ++++++++ .../conformance/server-parity/contract.mjs | 3 + .../large-command-batches-contract.mjs | 32 +++ .../large-command-batches-probe.php | 110 +++++++++++ scripts/conformance/server-parity/probe.php | 6 + .../ServerParity/large-command-batches.json | 183 ++++++++++++++++++ ...rParityLargeCommandBatchesContractTest.mjs | 55 ++++++ 21 files changed, 709 insertions(+), 23 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md create mode 100644 rust/tests/execution/large_command_batches.rs create mode 100644 scripts/conformance/server-parity/large-command-batches-contract.mjs create mode 100644 scripts/conformance/server-parity/large-command-batches-probe.php create mode 100644 tests/Fixtures/ServerParity/large-command-batches.json create mode 100644 tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 1b77ab365..ec1c39f85 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -59,6 +59,7 @@ on: - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs - tests/Unit/ServerParityConcurrentWaitsContractTest.mjs - tests/Unit/ServerParityContinueTransfersContractTest.mjs + - tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs - tests/Unit/ServerParityQueryArgumentsContractTest.mjs - tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs @@ -1985,7 +1986,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/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/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/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/ServerParityLargeCommandBatchesContractTest.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-large-command-batches.md b/docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md new file mode 100644 index 000000000..390eab79b --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md @@ -0,0 +1,60 @@ +# October 29, completions with more than 100 commands, and groups over 100 members + +- **Before.** + - Native refused a workflow task completion with more than 100 commands + with `422 invalid_commands`. The cap was in two places: the completion + itself (`store.rs`) and the preparation of its external command + payloads (`external_resolution.rs`). + - The timer, activity and external-wait summaries gave up past 100 open + waits (`timer_summary.rs`, `activity_summary.rs`, + `external_wait_summary.rs`). A run with more than 100 was therefore + described without its wait. +- **Contract** (published 2.11.5; `WorkerController::completeWorkflowTask`, + Core 2.15.1 `WorkflowCommandNormalizer`, `StructuralLimits`): + - `commands` must be a non-empty array, with no count limit of its own. + `StructuralLimits::guardCommandBatchSize` is used only by the embedded + executor. + - Each command meets the run's structural limits instead: pending + activities and timers, 2,000 each by default. + - The run is described as waiting on its first open wait, however many + are open. +- **Fixture.** `large-command-batches`, raw worker HTTP on protocol 1.20 in + a fresh namespace. Each case completes a run's first workflow task with + 150 `start_timer` or 150 `schedule_activity` commands. It records: + - the completion answer (200, run waiting, 150 created tasks); + - the run's description (`wait_kind` timer or activity, with its reason); + - the history counts by type, read in pages. + + A 1,001-command case is left out: published PHP answered + `409 stale_worker_registration` for it on SQLite, so it gives no usable + reference. +- **Native.** + - Both command caps are removed; a completion must still be non-empty. + - The three summaries no longer stop at 100 open waits. +- **Parallel groups.** + - **Before.** Native capped a group's declared size at 100 members + (`all_groups.rs`). It answered any invalid group metadata with + `422 development_parallel_group_unqualified`. + - **Contract.** `WorkflowCommandNormalizer::preflightParallelMetadata` + runs in `WorkerController` after the transport rules and integer + normalization, before the controller's scopes and any ownership guard. + It casts digit strings in the parallel integer fields, refuses parallel + fields on a non-leaf command, and checks each leaf's metadata + (`optionalParallelMetadata`). It collects errors across the batch, with + one of three messages. A group's declared size is bounded by + `StructuralLimits::commandBatchSizeLimit` (1,000 by default). + - **Fixture.** An all group of 150 activities, each with a one-entry + `parallel_group_path`, is accepted like the plain batch. A group + declaring 1,001 members (two commands) is refused + `422 validation_failed`, with an error on each command, and records + nothing. + - **Native.** `completion_rules.rs` ports the preflight at the same point + in `checked`. The batch-size limit is one shared constant + (`memo_limits::COMMAND_BATCH_SIZE`), used by the group, cooperative + cancellation and cancellation scope checks. A valid group that native + does not run (for example a child group) is still refused later with + `development_parallel_group_unqualified`. +- **Tests.** `large_command_batches.rs` (150 timers and 150 activities + accepted, the run described as waiting on its wait; a 150-member group + accepted and a 1,001-member group refused on both commands). Both fail on + the previous source. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 2e574aa10..7c045ddcd 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -303,9 +303,12 @@ contract fixture before its correction. of the Workflow package's `DW_V2_LIMIT_*` limits: the memo size, the command batch size as a bound on parallel group metadata, the pending signal and update counts with their payload size at intake, and the - payload size of an activity's result. Native enforces all but the command - batch size, and still caps parallel groups at 100 members and a run at 64 - updates over its lifetime. #612 tracks the rest + payload size of an activity's result. Native enforces all four; the + command batch size is fixed at its default of 1,000, since native refuses + to start with another value. A completion itself has no command-count + limit in either (October 29: native's 100-command and 100-member group + caps are removed; the 64-update lifetime cap went on October 23). #612 + tracks the rest ([checkpoint](rust-server-port-checkpoints/2026-10-23-structural-limits.md)). - **Completion budget, ahead of published PHP.** Native applies Phase A of [Workflow #930](https://github.com/durable-workflow/workflow/issues/930) @@ -561,6 +564,20 @@ contract fixture before its correction. fixture records twenty-two cases against published 2.11.5 ([checkpoint](rust-server-port-checkpoints/2026-10-28-query-contract-arguments.md)). +- **Completions with more than 100 commands** (October 29, closed). + Published PHP accepts any non-empty workflow task completion from an + external worker; each command meets the run's structural limits (pending + activities and timers) instead. Native refused more than 100 commands + with `invalid_commands`. Its timer, activity and external-wait summaries + also stopped at 100 open waits, so a run with more than 100 was described + without its wait. Native also capped a parallel group at 100 members, + where PHP bounds it by the command batch size (1,000). It refused an + invalid group as `development_parallel_group_unqualified`, where PHP's + preflight answers `validation_failed` with an error on each command. All + of these are removed or ported. The `large-command-batches` fixture records + 150 timers, 150 activities, a 150-member group and a 1,001-member group + against published 2.11.5 + ([checkpoint](rust-server-port-checkpoints/2026-10-29-large-command-batches.md)). - **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/activity_summary.rs b/rust/src/runtime/activity_summary.rs index e88e5f026..10bf8ffba 100644 --- a/rust/src/runtime/activity_summary.rs +++ b/rust/src/runtime/activity_summary.rs @@ -99,8 +99,7 @@ impl ActivitySummaryHistory { }, }; if event == "ActivityScheduled" { - if self.open.len() >= 100 - || self.open.contains_key(id) + if self.open.contains_key(id) || attempts != 0 || next.attempt_id.is_some() { @@ -213,9 +212,9 @@ where history: &ActivitySummaryHistory, summary: &mut ActiveSummary, ) -> Result { - let rows = Self::query("SELECT w.*,a.id AS activity_id,a.sequence AS activity_sequence,a.activity_type,a.status AS activity_status,a.attempt_count,a.current_attempt_id,a.parallel_group_path AS activity_group_path,q.id AS matched_task_id,q.task_type AS matched_task_type,q.payload AS matched_task_payload FROM workflow_run_waits w LEFT JOIN activity_executions a ON a.id=w.resume_source_id AND a.workflow_run_id=w.workflow_run_id LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND w.kind='activity' AND w.status='open' ORDER BY w.sequence,w.opened_at,w.resume_source_id LIMIT 101") + let rows = Self::query("SELECT w.*,a.id AS activity_id,a.sequence AS activity_sequence,a.activity_type,a.status AS activity_status,a.attempt_count,a.current_attempt_id,a.parallel_group_path AS activity_group_path,q.id AS matched_task_id,q.task_type AS matched_task_type,q.payload AS matched_task_payload FROM workflow_run_waits w LEFT JOIN activity_executions a ON a.id=w.resume_source_id AND a.workflow_run_id=w.workflow_run_id LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND w.kind='activity' AND w.status='open' ORDER BY w.sequence,w.opened_at,w.resume_source_id") .bind(run_id).fetch_all(&mut **tx).await?; - if rows.len() != history.open.len() || rows.len() > 100 || !history.local.is_empty() { + if rows.len() != history.open.len() || !history.local.is_empty() { return Ok(false); } let mut selected_activity_task = false; diff --git a/rust/src/runtime/all_groups.rs b/rust/src/runtime/all_groups.rs index 3dcd27839..c7cc4834e 100644 --- a/rust/src/runtime/all_groups.rs +++ b/rust/src/runtime/all_groups.rs @@ -18,7 +18,7 @@ fn entry(value: &Value, leaf: &str) -> Result { .ok_or_else(invalid)?; let size = value["parallel_group_size"] .as_i64() - .filter(|size| (1..=100).contains(size)) + .filter(|size| (1..=super::memo_limits::COMMAND_BATCH_SIZE).contains(size)) .ok_or_else(invalid)?; let index = value["parallel_group_index"] .as_i64() diff --git a/rust/src/runtime/cancellation_scopes.rs b/rust/src/runtime/cancellation_scopes.rs index 37e71d1f3..eb6c37e7e 100644 --- a/rust/src/runtime/cancellation_scopes.rs +++ b/rust/src/runtime/cancellation_scopes.rs @@ -34,8 +34,7 @@ const DELIVERY_SCHEMA: &str = "durable-workflow.cancellation-scope-delivery/v1"; const SCOPED_CONTEXT_SCHEMA: &str = "durable-workflow.scoped-cancellation-context/v1"; const ACTIVITY_SCOPE_SCHEMA: &str = "durable-workflow.activity-scope-cancellation/v1"; const TIMER_SCOPE_SCHEMA: &str = "durable-workflow.scoped-timer-cancellation/v1"; -/// PHP's default `command_batch_size` structural limit. -const SPAN_LIMIT: i64 = 1000; +use super::memo_limits::COMMAND_BATCH_SIZE as SPAN_LIMIT; /// The command types whose `cancellation_scope_id` records membership. const MEMBER_COMMANDS: &[&str] = &["schedule_activity", "start_timer", "start_child_workflow", "open_signal_wait", "open_condition_wait"]; /// The command types the admission guard lets carry a scope. diff --git a/rust/src/runtime/completion_rules.rs b/rust/src/runtime/completion_rules.rs index 6c8382191..2e91bec62 100644 --- a/rust/src/runtime/completion_rules.rs +++ b/rust/src/runtime/completion_rules.rs @@ -13,7 +13,7 @@ //! integers, so `"7"` is a valid delay. use super::laravel_validation::{Rule, default_message, integer, passes, present}; use super::Result; -use serde_json::Value; +use serde_json::{Value, json}; #[derive(Clone, Copy)] enum Check { @@ -407,6 +407,14 @@ pub(super) fn checked(body: Value) -> Result { } } } + // `WorkflowCommandNormalizer::preflightParallelMetadata`, which PHP + // runs here, before the controller's scopes and any ownership guard. + if let Some(commands) = body["commands"].as_array_mut() { + commands.iter_mut().for_each(parallel_integers); + if let Some(error) = super::laravel_validation::failure_messages(parallel_preflight(commands)) { + return Err(error); + } + } if let Some(commands) = body["commands"].as_array() && let Some(error) = super::laravel_validation::failure_messages(controller_scopes(commands)) { @@ -415,6 +423,49 @@ pub(super) fn checked(body: Value) -> Result { Ok(body) } +/// `PARALLEL_METADATA_INTEGER_FIELDS`. +const PARALLEL_INTEGER_FIELDS: &[&str] = &["parallel_group_base_sequence", "parallel_group_size", "parallel_group_index", + "selection_member_index", "selection_member_base_sequence", "selection_member_size"]; + +/// `normalizeParallelMetadataIntegerFields`: a digit string becomes an +/// integer, at the top level and in each path entry. +fn parallel_integers(command: &mut Value) { + let normalize = |fields: &mut serde_json::Map| { + for field in PARALLEL_INTEGER_FIELDS { + if let Some(value) = fields.get_mut(*field) { + whole(value); + } + } + }; + let Some(fields) = command.as_object_mut() else { + return; + }; + normalize(fields); + if let Some(Value::Array(path)) = fields.get_mut("parallel_group_path") { + path.iter_mut().filter_map(Value::as_object_mut).for_each(normalize); + } +} + +/// `preflightParallelMetadata`'s errors across the batch: parallel fields on +/// a command that is not a parallel leaf, and a leaf's invalid metadata. +fn parallel_preflight(commands: &[Value]) -> Vec<(String, Vec)> { + let mut found = Vec::new(); + for (index, command) in commands.iter().enumerate() { + let Some(kind) = command.get("type").and_then(Value::as_str) else { + continue; + }; + if PARALLEL_LEAVES.contains(&kind) { + found.extend(parallel_metadata_error(index, kind, command)); + continue; + } + for field in PARALLEL_FIELDS.iter().filter(|field| has_value(command, field)) { + found.push((format!("commands.{index}.{field}"), + vec![format!("{field} is only supported for activity, timer, and child-workflow scheduling commands.")])); + } + } + found +} + fn has_value(command: &Value, field: &str) -> bool { command.get(field).is_some_and(|value| !value.is_null()) } @@ -602,6 +653,85 @@ pub(super) fn normalizer_checks(index: usize, command: &Value) -> Result<()> { } } +/// `WorkflowCommandNormalizer::optionalParallelMetadata`'s refusal for a +/// leaf command: incomplete top-level metadata or path, an invalid path +/// entry, or an innermost entry that differs from the top level. +fn parallel_metadata_error(index: usize, kind: &str, command: &Value) -> Option<(String, Vec)> { + let leaf = match kind { + "schedule_activity" => "activity", + "start_timer" => "timer", + "start_child_workflow" => "child", + "open_condition_wait" => "condition", + "open_signal_wait" => "signal", + _ => return None, + }; + if !PARALLEL_FIELDS.iter().any(|field| command.get(*field).is_some()) { + return None; + } + let refuse = |field: String, message: &str| Some((field, vec![message.to_owned()])); + let top = parallel_entry(command, leaf); + let path = command.get("parallel_group_path").and_then(Value::as_array).filter(|path| !path.is_empty()); + let (Some(top), Some(path)) = (top, path) else { + return refuse(format!("commands.{index}.parallel_group_path"), + "Parallel workflow commands require complete top-level metadata and a non-empty group path."); + }; + let mut innermost = None; + for (position, entry) in path.iter().enumerate() { + innermost = parallel_entry(entry, leaf); + if innermost.is_none() { + return refuse(format!("commands.{index}.parallel_group_path.{position}"), + "Each parallel group path entry must contain valid identity, kind, base, size, and index fields."); + } + } + (innermost != Some(top)).then(|| (format!("commands.{index}.parallel_group_path"), + vec!["The innermost parallel group path entry must match the top-level parallel metadata.".to_owned()])) +} + +/// `parallelMetadataEntry`: the entry's normalized fields, or `None` when +/// PHP refuses it. A group's size is bounded by the command batch size. +fn parallel_entry(value: &Value, leaf: &str) -> Option> { + let text = |field: &str| value.get(field).and_then(Value::as_str); + let int = |field: &str| value.get(field).filter(|value| value.is_i64() || value.is_u64()).and_then(Value::as_i64); + let id = text("parallel_group_id").filter(|id| !id.is_empty())?; + let kind = text("parallel_group_kind").filter(|kind| *kind == leaf || *kind == "mixed")?; + let base = int("parallel_group_base_sequence").filter(|base| *base >= 1)?; + let size = int("parallel_group_size").filter(|size| (1..=super::memo_limits::COMMAND_BATCH_SIZE).contains(size))?; + let index = int("parallel_group_index").filter(|index| (0..size).contains(index))?; + let mode = match value.get("parallel_group_mode") { + None | Some(Value::Null) => "all", + Some(mode) => mode.as_str().filter(|mode| matches!(*mode, "all" | "select"))?, + }; + let prefix = match (mode, kind) { + ("select", _) => "select-calls", + (_, "activity") => "parallel-activities", + (_, "child") => "parallel-children", + (_, "timer") => "parallel-timers", + _ => "parallel-calls", + }; + if id != format!("{prefix}:{base}:{size}") { + return None; + } + let mut entry = vec![("parallel_group_id", json!(id)), ("parallel_group_kind", json!(kind))]; + if mode == "select" { + let key = value.get("selection_member_key") + .filter(|key| key.as_str().is_some_and(|key| !key.is_empty()) || key.as_u64().is_some())?; + let member_index = int("selection_member_index").filter(|index| *index >= 0)?; + let member_base = int("selection_member_base_sequence").filter(|member| (base..base + size).contains(member))?; + let member_size = int("selection_member_size").filter(|member| *member >= 1 && member_base + member <= base + size)?; + let member_kind = text("selection_member_kind") + .filter(|kind| matches!(*kind, "activity" | "child" | "timer" | "signal" | "condition" | "group"))?; + entry.extend([("parallel_group_mode", json!("select")), ("parallel_group_base_sequence", json!(base)), + ("parallel_group_size", json!(size)), ("parallel_group_index", json!(index)), + ("selection_member_key", key.clone()), ("selection_member_index", json!(member_index)), + ("selection_member_base_sequence", json!(member_base)), ("selection_member_size", json!(member_size)), + ("selection_member_kind", json!(member_kind))]); + } else { + entry.extend([("parallel_group_base_sequence", json!(base)), ("parallel_group_size", json!(size)), + ("parallel_group_index", json!(index))]); + } + Some(entry) +} + /// The payload fields whose bare string PHP's normalizer takes as the blob /// itself, in the run's codec (`resolveCommandPayloadWithCodec`). const BLOB_FIELDS: &[(&str, &str)] = &[ diff --git a/rust/src/runtime/cooperative.rs b/rust/src/runtime/cooperative.rs index 8a3fe513d..c4b59dcb9 100644 --- a/rust/src/runtime/cooperative.rs +++ b/rust/src/runtime/cooperative.rs @@ -102,8 +102,7 @@ pub(super) async fn encode_command(value: Value, semaphore: Arc) -> R })? } -/// PHP's default `command_batch_size` structural limit. -const PARALLEL_SPAN_LIMIT: i64 = 1000; +use super::memo_limits::COMMAND_BATCH_SIZE as PARALLEL_SPAN_LIMIT; /// The reasons `CooperativeCancellationDelivery::validate` returns. /// `DefaultWorkflowTaskBridge::deliverCancellation` answers each with its diff --git a/rust/src/runtime/external_resolution.rs b/rust/src/runtime/external_resolution.rs index eb6b762f6..455a9b1ff 100644 --- a/rust/src/runtime/external_resolution.rs +++ b/rust/src/runtime/external_resolution.rs @@ -68,7 +68,7 @@ impl Runtime { pub(super) async fn prepare_command_payloads(&self, body: &Value) -> Result { let commands = body["commands"] .as_array() - .filter(|commands| !commands.is_empty() && commands.len() <= 100) + .filter(|commands| !commands.is_empty()) .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_commands"))?; let selected: Vec<_> = commands .iter() diff --git a/rust/src/runtime/external_wait_summary.rs b/rust/src/runtime/external_wait_summary.rs index 06abf03fa..6ff758254 100644 --- a/rust/src/runtime/external_wait_summary.rs +++ b/rust/src/runtime/external_wait_summary.rs @@ -97,7 +97,7 @@ impl ExternalWaitHistory { ) else { return false; }; - if self.open.len() >= 100 || self.open.contains_key(id) { + if self.open.contains_key(id) { return false; } self.open.insert( @@ -146,7 +146,6 @@ impl ExternalWaitHistory { || wait.sequence != sequence || wait.timeout != Some(delay) || wait.timer.is_some() - || self.timers.len() >= 100 || self.timers.contains_key(timer_id) || super::wait_timeouts::wait_fields(payload) != wait.identity { @@ -224,9 +223,9 @@ where history: &ExternalWaitHistory, summary: &mut ActiveSummary, ) -> Result { - let rows = Self::query("SELECT w.*,t.id AS signal_timer_id,t.status AS signal_timer_status,t.sequence AS signal_timer_sequence,t.delay_seconds AS signal_timer_delay,t.fire_at AS signal_timer_fire_at,q.id AS signal_task_id,q.task_type AS signal_task_type,q.payload AS signal_task_payload FROM workflow_run_waits w LEFT JOIN workflow_run_timers t ON t.id=w.resume_source_id AND t.workflow_run_id=w.workflow_run_id AND w.resume_source_kind='timer' LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND (w.status='open' OR (w.status='resolved' AND w.source_status='timed_out')) AND w.kind IN ('condition','signal') ORDER BY w.sequence,w.wait_id LIMIT 101") + let rows = Self::query("SELECT w.*,t.id AS signal_timer_id,t.status AS signal_timer_status,t.sequence AS signal_timer_sequence,t.delay_seconds AS signal_timer_delay,t.fire_at AS signal_timer_fire_at,q.id AS signal_task_id,q.task_type AS signal_task_type,q.payload AS signal_task_payload FROM workflow_run_waits w LEFT JOIN workflow_run_timers t ON t.id=w.resume_source_id AND t.workflow_run_id=w.workflow_run_id AND w.resume_source_kind='timer' LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND (w.status='open' OR (w.status='resolved' AND w.source_status='timed_out')) AND w.kind IN ('condition','signal') ORDER BY w.sequence,w.wait_id") .bind(run_id).fetch_all(&mut **tx).await?; - if rows.len() != history.open.len() || rows.len() > 100 { + if rows.len() != history.open.len() { return Ok(false); } let mut condition = None; diff --git a/rust/src/runtime/memo_limits.rs b/rust/src/runtime/memo_limits.rs index d29d02596..b9213fe69 100644 --- a/rust/src/runtime/memo_limits.rs +++ b/rust/src/runtime/memo_limits.rs @@ -80,6 +80,12 @@ fn configured(suffix: &str, default: usize, invalid: &'static str) -> Result, ) -> Result { let transition_codec_for_streams = transition_codec.clone(); + // `commands` is required and non-empty, with no count limit of its + // own: each command meets the run's structural limits instead. let commands = body["commands"] .as_array() - .filter(|c| !c.is_empty() && c.len() <= 100) + .filter(|c| !c.is_empty()) .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_commands"))?; super::workflow_failures::validate_batch(commands)?; // Validate the entire batch before mutation. Unknown commands and diff --git a/rust/src/runtime/timer_summary.rs b/rust/src/runtime/timer_summary.rs index 7600be2ff..9fdeecd6d 100644 --- a/rust/src/runtime/timer_summary.rs +++ b/rust/src/runtime/timer_summary.rs @@ -44,7 +44,7 @@ impl TimerSummaryHistory { ) else { return false; }; - if kind != "TimerScheduled" || self.open.len() >= 100 || self.open.contains_key(id) { + if kind != "TimerScheduled" || self.open.contains_key(id) { return false; } self.open.insert( @@ -81,9 +81,9 @@ where history: &TimerSummaryHistory, summary: &mut ActiveSummary, ) -> Result { - let rows = Self::query("SELECT w.*,t.id AS timer_id,t.sequence AS timer_sequence,t.status AS timer_status,t.delay_seconds,t.fire_at,q.id AS matched_task_id,q.task_type AS matched_task_type,q.payload AS matched_task_payload FROM workflow_run_waits w LEFT JOIN workflow_run_timers t ON t.id=w.resume_source_id AND t.workflow_run_id=w.workflow_run_id LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND w.kind='timer' AND w.status='open' ORDER BY w.sequence,w.deadline_at,w.opened_at,w.resume_source_id LIMIT 101") + let rows = Self::query("SELECT w.*,t.id AS timer_id,t.sequence AS timer_sequence,t.status AS timer_status,t.delay_seconds,t.fire_at,q.id AS matched_task_id,q.task_type AS matched_task_type,q.payload AS matched_task_payload FROM workflow_run_waits w LEFT JOIN workflow_run_timers t ON t.id=w.resume_source_id AND t.workflow_run_id=w.workflow_run_id LEFT JOIN workflow_tasks q ON q.id=w.task_id AND q.workflow_run_id=w.workflow_run_id WHERE w.workflow_run_id=$1 AND w.kind='timer' AND w.status='open' ORDER BY w.sequence,w.deadline_at,w.opened_at,w.resume_source_id") .bind(run_id).fetch_all(&mut **tx).await?; - if rows.len() != history.open.len() || rows.len() > 100 { + if rows.len() != history.open.len() { return Ok(false); } let mut selected_timer_task = false; diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 14e67cc6d..2319e19b0 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -334,6 +334,9 @@ mod worker_sessions; #[path = "execution/workflow_import_fences.rs"] mod workflow_import_fences; +#[path = "execution/large_command_batches.rs"] +mod large_command_batches; + #[path = "execution/query_argument_routing.rs"] mod query_argument_routing; diff --git a/rust/tests/execution/large_command_batches.rs b/rust/tests/execution/large_command_batches.rs new file mode 100644 index 000000000..ffdfbffa7 --- /dev/null +++ b/rust/tests/execution/large_command_batches.rs @@ -0,0 +1,82 @@ +use super::*; + +// Published Server 2.11.5 accepts a workflow task completion with any +// non-empty number of commands from an external worker; each command meets +// the run's structural limits (pending activities and timers) instead. The +// large-command-batches fixture records 150 timers and 150 activities in one +// completion, a 150-member parallel group and a refused 1,001-member group. + +#[tokio::test] +async fn a_completion_may_carry_more_than_a_hundred_commands() { + for (workflow, command, event) in [ + ("many-timers", json!({"type":"start_timer","delay_seconds":3600}), "TimerScheduled"), + ("many-activities", json!({"type":"schedule_activity","activity_type":"activity-echo","arguments":envelope(Payload::Array(vec![]))}), + "ActivityScheduled"), + ] { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "worker", json!(["echo"]), json!(["activity-echo"])).await; + let started = start(&app, workflow).await; + let task = poll(&app, "worker", "workflow").await; + let (status, body) = finish_task(&app, &task, json!(vec![command; 150])).await; + assert_eq!(status, StatusCode::OK, "{workflow}: {body}"); + assert_eq!((body["outcome"].clone(), body["run_status"].clone()), (json!("completed"), json!("waiting")), "{workflow}"); + assert_eq!(body["created_task_ids"].as_array().map(Vec::len), Some(150), "{workflow}"); + // The run is described as waiting on its first open wait, as PHP + // summarizes every open wait. + let (_, described) = request(&app, "GET", &format!("/api/workflows/{workflow}"), Value::Null).await; + let kind = if event == "TimerScheduled" { "timer" } else { "activity" }; + assert_eq!((described["status"].clone(), described["wait_kind"].clone()), (json!("waiting"), json!(kind)), "{workflow}: {described}"); + let run = started["run_id"].as_str().unwrap(); + assert_eq!(task_retry::rows(&database, Box::leak(format!( + "SELECT COUNT(*) FROM workflow_history_events WHERE workflow_run_id='{run}' AND event_type='{event}'").into_boxed_str())).await, 150, "{workflow}"); + runtime.close().await; + database.remove().await; + } +} + +fn group_member(size: usize, index: usize) -> Value { + let entry = json!({"parallel_group_id":format!("parallel-activities:1:{size}"),"parallel_group_kind":"activity", + "parallel_group_base_sequence":1,"parallel_group_size":size,"parallel_group_index":index}); + let mut command = json!({"type":"schedule_activity","activity_type":"activity-echo","arguments":envelope(Payload::Array(vec![]))}); + command.as_object_mut().unwrap().extend(entry.as_object().unwrap().clone()); + command["parallel_group_path"] = json!([entry]); + command +} + +// A parallel group's declared size is bounded by the command batch size +// (1,000 by default), not by 100. PHP's preflight refuses a larger group on +// every command before any guard, and records nothing. +#[tokio::test] +async fn a_parallel_group_may_hold_up_to_the_command_batch_size() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "worker", json!(["echo"]), json!(["activity-echo"])).await; + let over = start(&app, "group-over-limit").await; + let task = poll(&app, "worker", "workflow").await; + let (status, body) = finish_task(&app, &task, json!([group_member(1001, 0), group_member(1001, 1)])).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::UNPROCESSABLE_ENTITY, json!("validation_failed")), "{body}"); + let message = "Parallel workflow commands require complete top-level metadata and a non-empty group path."; + assert_eq!(body["message"], format!("{message} (and 1 more error)")); + assert_eq!(body["errors"], json!({"commands.0.parallel_group_path":[message],"commands.1.parallel_group_path":[message]})); + let run = over["run_id"].as_str().unwrap(); + assert_eq!(task_retry::rows(&database, Box::leak(format!( + "SELECT COUNT(*) FROM workflow_history_events WHERE workflow_run_id='{run}' AND event_type='ActivityScheduled'").into_boxed_str())).await, 0); + + let started = start(&app, "group-of-150").await; + let task = poll(&app, "worker", "workflow").await; + assert_eq!(task["workflow_id"], "group-of-150", "{task}"); + let commands: Vec<_> = (0..150).map(|index| group_member(150, index)).collect(); + let (status, body) = finish_task(&app, &task, json!(commands)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["created_task_ids"].as_array().map(Vec::len), Some(150)); + let (_, described) = request(&app, "GET", "/api/workflows/group-of-150", Value::Null).await; + assert_eq!((described["status"].clone(), described["wait_kind"].clone()), (json!("waiting"), json!("activity")), "{described}"); + let run = started["run_id"].as_str().unwrap(); + assert_eq!(task_retry::rows(&database, Box::leak(format!( + "SELECT COUNT(*) FROM workflow_history_events WHERE workflow_run_id='{run}' AND event_type='ActivityScheduled' AND payload LIKE '%parallel-activities:1:150%'").into_boxed_str())).await, 150); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 14da87169..e3d86efa5 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -15,6 +15,7 @@ import {checkStorageTest} from './storage-test-contract.mjs'; import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; import {checkConcurrentWaits} from './concurrent-waits-contract.mjs'; import {checkContinueTransfers} from './continue-transfers-contract.mjs'; +import {checkLargeCommandBatches} from './large-command-batches-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'; @@ -175,6 +176,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const activityLeaseRenewal = fixture.activity_lease_renewal ? checkActivityLeaseRenewal(fixture, observation, workflowId) : null; const concurrentWaits = fixture.concurrent_waits ? checkConcurrentWaits(fixture, observation) : null; const continueTransfers = fixture.continue_transfers ? checkContinueTransfers(fixture, observation) : null; + const largeCommandBatches = fixture.large_command_batches ? checkLargeCommandBatches(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; @@ -753,6 +755,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(activityLeaseRenewal ? {activity_lease_renewal: activityLeaseRenewal} : {}), ...(concurrentWaits ? {concurrent_waits: concurrentWaits} : {}), ...(continueTransfers ? {continue_transfers: continueTransfers} : {}), + ...(largeCommandBatches ? {large_command_batches: largeCommandBatches} : {}), ...(startSearchAttributes ? {start_search_attributes: startSearchAttributes} : {}), ...(queryArguments ? {query_argument_routing: queryArguments} : {}), ...(queryContractArguments ? {query_contract_arguments: queryContractArguments} : {}), diff --git a/scripts/conformance/server-parity/large-command-batches-contract.mjs b/scripts/conformance/server-parity/large-command-batches-contract.mjs new file mode 100644 index 000000000..88bfd4478 --- /dev/null +++ b/scripts/conformance/server-parity/large-command-batches-contract.mjs @@ -0,0 +1,32 @@ +import assert from 'node:assert/strict'; + +// Reviewed workflow task completions with more than 100 commands +// (large-command-batches-probe.php), recorded against published Server +// 2.11.5: an external worker's completion of 150 timers or 150 activities is +// accepted in one batch, creates a task for each, and leaves the run waiting +// on its first open wait. So is an all group of 150 activities. A group +// declaring 1,001 members, above the command batch size, is refused before +// any guard with an error on each command's parallel_group_path. The +// fixture's `expected` holds the whole observation. +export function checkLargeCommandBatches(fixture, observation) { + const definition = fixture.large_command_batches; + const state = observation.large_command_batches; + const projection = {scope: 'large_command_batches', runs: Object.keys(definition.expected).length}; + assert.ok(state && typeof state === 'object', 'actual large command batches observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}, + 'embedded large command batches explicitly inapplicable'); + return projection; + } + assert.equal(state.applicable, true, 'HTTP targets record large command batches'); + for (const [run, steps] of Object.entries(definition.expected)) { + assert.ok(state[run] && typeof state[run] === 'object', `large command batches ${run} recorded`); + for (const [step, expected] of Object.entries(steps)) { + assert.deepStrictEqual(state[run][step], expected, `large command batches ${run} ${step}`); + } + assert.deepStrictEqual(Object.keys(state[run]).sort(), Object.keys(steps).sort(), `large command batches ${run} steps`); + } + assert.deepStrictEqual(Object.keys(state).sort(), ['applicable', ...Object.keys(definition.expected)].sort(), + 'large command batches runs'); + return projection; +} diff --git a/scripts/conformance/server-parity/large-command-batches-probe.php b/scripts/conformance/server-parity/large-command-batches-probe.php new file mode 100644 index 000000000..9b7d5db2a --- /dev/null +++ b/scripts/conformance/server-parity/large-command-batches-probe.php @@ -0,0 +1,110 @@ +createNamespace($namespace, 'large command batches λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new LargeCommandBatchesHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 60, '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)]; + }; + $type = $definition['workflow_type']; + $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']]; + + $state = ['applicable' => true]; + foreach ($definition['cases'] as $case => $shape) { + $id = $workflowId.'-lcb-'.$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' => [$type], + 'supported_activity_types' => [$definition['activity_type']], + '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' => $type, 'task_queue' => $queue, + 'input' => $empty]); + if ($registered['status'] !== 201 || $started['status'] !== 201) { + throw new RuntimeException('Large command batches setup: '.json_encode([$registered, $started])); + } + $task = $raw('POST', 'worker/workflow-tasks/poll', ['worker_id' => $worker, 'task_queue' => $queue, + 'timeout_seconds' => 5])['response']['task'] ?? null; + if (! is_array($task)) { + throw new RuntimeException('Large command batches: no workflow task for '.$case); + } + $commands = []; + for ($index = 0; $index < $shape['commands']; $index++) { + $commands[] = $shape['kind'] === 'timers' + ? ['type' => 'start_timer', 'delay_seconds' => 3600] + : ['type' => 'schedule_activity', 'activity_type' => $definition['activity_type'], 'arguments' => $empty]; + } + // An all group of activities declares itself on every member, with + // a one-group path. A group larger than the limit is declared by the + // batch's members only. + if (isset($shape['group_size'])) { + foreach ($commands as $index => $command) { + $entry = ['parallel_group_id' => 'parallel-activities:1:'.$shape['group_size'], + 'parallel_group_kind' => 'activity', 'parallel_group_base_sequence' => 1, + 'parallel_group_size' => $shape['group_size'], 'parallel_group_index' => $index]; + $commands[$index] += [...$entry, 'parallel_group_path' => [$entry]]; + } + } + $completed = $raw('POST', 'worker/workflow-tasks/'.$task['task_id'].'/complete', ['lease_owner' => $task['lease_owner'], + 'workflow_task_attempt' => $task['workflow_task_attempt'], 'commands' => $commands]); + $observed = ['completed' => ['status' => $completed['status'], 'outcome' => $completed['response']['outcome'] ?? null, + 'reason' => $completed['response']['reason'] ?? null, 'run_status' => $completed['response']['run_status'] ?? null, + 'message' => $completed['response']['message'] ?? null, + 'errors' => $completed['response']['errors'] ?? null, + 'created_tasks' => count($completed['response']['created_task_ids'] ?? [])]]; + // The run as described after the batch: its waits and liveness. + $described = $raw('GET', 'workflows/'.rawurlencode($id))['response']; + $observed['described'] = array_intersect_key($described, array_flip($definition['described_fields'])); + $counts = []; + $groups = []; + $token = null; + do { + $page = $raw('GET', 'workflows/'.rawurlencode($id).'/runs/'.$started['response']['run_id'].'/history?page_size=1000' + .($token === null ? '' : '&next_page_token='.rawurlencode($token)))['response']; + foreach ($page['events'] ?? [] as $event) { + $counts[$event['event_type']] = ($counts[$event['event_type']] ?? 0) + 1; + if (is_string($event['payload']['parallel_group_id'] ?? null)) { + $groups[$event['payload']['parallel_group_id']] = ($groups[$event['payload']['parallel_group_id']] ?? 0) + 1; + } + } + $token = $page['next_page_token'] ?? null; + } while (is_string($token) && $token !== ''); + ksort($counts); + $observed['events'] = $counts; + $observed['parallel_groups'] = $groups; + $state[$case] = $observed; + } + + return $state; +} diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index a1b65b9ea..4fbbaa6fc 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -433,6 +433,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $activityLeaseRenewal = isset($fixture['activity_lease_renewal']) ? finishHttpActivityLeaseRenewal($client, $fixture, $workflowId, $url) : null; $concurrentWaits = isset($fixture['concurrent_waits']) ? finishHttpConcurrentWaits($client, $fixture, $workflowId, $url) : null; $continueTransfers = isset($fixture['continue_transfers']) ? finishHttpContinueTransfers($client, $fixture, $workflowId, $url) : null; + $largeCommandBatches = isset($fixture['large_command_batches']) ? finishHttpLargeCommandBatches($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; @@ -564,6 +565,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'activity_lease_renewal' => $activityLeaseRenewal, 'concurrent_waits' => $concurrentWaits, 'continue_transfers' => $continueTransfers, + 'large_command_batches' => $largeCommandBatches, 'start_search_attributes' => $startSearchAttributes, 'query_argument_routing' => $queryArgumentRouting, 'query_contract_arguments' => $queryContractArguments, @@ -898,6 +900,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/activity-lease-renewal-probe.php'; require __DIR__.'/concurrent-waits-probe.php'; require __DIR__.'/continue-transfers-probe.php'; + require __DIR__.'/large-command-batches-probe.php'; require __DIR__.'/start-search-attributes-probe.php'; require __DIR__.'/query-arguments-probe.php'; require __DIR__.'/query-contract-arguments-probe.php'; @@ -1090,6 +1093,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['continue_transfers'])) { $observation['continue_transfers'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } + if ($mode === 'embedded' && isset($fixture['large_command_batches'])) { + $observation['large_command_batches'] = ['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/large-command-batches.json b/tests/Fixtures/ServerParity/large-command-batches.json new file mode 100644 index 000000000..b410a72fd --- /dev/null +++ b/tests/Fixtures/ServerParity/large-command-batches.json @@ -0,0 +1,183 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "large-command-batches", + "review_state": "reviewed", + "contract": "Workflow task completions with more than 100 commands (published Server 2.11.5 WorkerController::completeWorkflowTask with Workflow 2.15.1 WorkflowCommandNormalizer), over raw worker HTTP on protocol 1.20 in a fresh namespace. An external worker's completion has no command-count limit of its own; each command meets the run's structural limits instead. Completing a run's first workflow task with 150 start_timer commands, or 150 schedule_activity commands, is accepted (200 completed, run waiting) with a created task for each, records 150 TimerScheduled or ActivityScheduled events, and describes the run as waiting on its first timer or activity. An all group of 150 activities, each declaring the group with a one-entry parallel_group_path, is accepted the same way. A group declaring 1,001 members (above the default 1,000-command batch size that bounds a declared group) is refused 422 validation_failed before any guard, with an error on each command's parallel_group_path, and records nothing. Embedded applications have no worker protocol.", + "workflow_type": "parity.v1.echo", + "input": { + "message": "large command batches λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "large command batches λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "large_command_batches": { + "workflow_type": "parity.v1.large_command_batches", + "activity_type": "parity.v1.large_command_batches_activity", + "empty_blob": "wwHioz3/VYAiNwwA", + "cases": { + "timers": { + "kind": "timers", + "commands": 150 + }, + "activities": { + "kind": "activities", + "commands": 150 + }, + "activity_group": { + "kind": "activities", + "commands": 150, + "group_size": 150 + }, + "group_over_limit": { + "kind": "activities", + "commands": 2, + "group_size": 1001 + } + }, + "expected": { + "timers": { + "completed": { + "status": 200, + "outcome": "completed", + "reason": null, + "run_status": "waiting", + "message": null, + "errors": null, + "created_tasks": 150 + }, + "described": { + "status": "waiting", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "timer", + "wait_reason": "Waiting for timer" + }, + "events": { + "StartAccepted": 1, + "TimerScheduled": 150, + "WorkflowStarted": 1 + }, + "parallel_groups": [] + }, + "activities": { + "completed": { + "status": 200, + "outcome": "completed", + "reason": null, + "run_status": "waiting", + "message": null, + "errors": null, + "created_tasks": 150 + }, + "described": { + "status": "waiting", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "activity", + "wait_reason": "Waiting for activity parity.v1.large_command_batches_activity" + }, + "events": { + "ActivityScheduled": 150, + "StartAccepted": 1, + "WorkflowStarted": 1 + }, + "parallel_groups": [] + }, + "activity_group": { + "completed": { + "status": 200, + "outcome": "completed", + "reason": null, + "run_status": "waiting", + "message": null, + "errors": null, + "created_tasks": 150 + }, + "described": { + "status": "waiting", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "activity", + "wait_reason": "Waiting for activity parity.v1.large_command_batches_activity" + }, + "events": { + "ActivityScheduled": 150, + "StartAccepted": 1, + "WorkflowStarted": 1 + }, + "parallel_groups": { + "parallel-activities:1:150": 150 + } + }, + "group_over_limit": { + "completed": { + "status": 422, + "outcome": null, + "reason": "validation_failed", + "run_status": null, + "message": "Parallel workflow commands require complete top-level metadata and a non-empty group path. (and 1 more error)", + "errors": { + "commands.0.parallel_group_path": [ + "Parallel workflow commands require complete top-level metadata and a non-empty group path." + ], + "commands.1.parallel_group_path": [ + "Parallel workflow commands require complete top-level metadata and a non-empty group path." + ] + }, + "created_tasks": 0 + }, + "described": { + "status": "pending", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "workflow-task", + "wait_reason": "Workflow task leased to worker" + }, + "events": { + "StartAccepted": 1, + "WorkflowStarted": 1 + }, + "parallel_groups": [] + } + }, + "described_fields": [ + "status", + "status_bucket", + "wait_kind", + "wait_reason", + "is_terminal", + "resume_step_sequence", + "recovery_kind" + ] + } +} diff --git a/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs b/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs new file mode 100644 index 000000000..7eff343d7 --- /dev/null +++ b/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs @@ -0,0 +1,55 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkLargeCommandBatches} from '../../scripts/conformance/server-parity/large-command-batches-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/large-command-batches.json', import.meta.url), 'utf8')); +const expected = fixture.large_command_batches.expected; +const observation = () => structuredClone({mode: 'http', large_command_batches: {applicable: true, ...expected}}); + +test('the published answers pass', () => { + assert.deepStrictEqual(checkLargeCommandBatches(fixture, observation()), {scope: 'large_command_batches', runs: 4}); +}); + +test('embedded applications have no worker protocol', () => { + assert.deepStrictEqual(checkLargeCommandBatches(fixture, + {mode: 'embedded', large_command_batches: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), + {scope: 'large_command_batches', runs: 4}); +}); + +test('150 commands are accepted in one completion', () => { + for (const [name, event] of [['timers', 'TimerScheduled'], ['activities', 'ActivityScheduled']]) { + assert.deepStrictEqual([expected[name].completed.status, expected[name].completed.created_tasks, expected[name].events[event]], + [200, 150, 150], name); + } +}); + +test('a 150-member group is accepted, and a 1,001-member group is refused on every command', () => { + const group = expected.activity_group; + assert.deepStrictEqual([group.completed.status, group.completed.created_tasks, group.parallel_groups], + [200, 150, {'parallel-activities:1:150': 150}]); + const over = expected.group_over_limit.completed; + assert.deepStrictEqual([over.status, over.reason, Object.keys(over.errors)], + [422, 'validation_failed', ['commands.0.parallel_group_path', 'commands.1.parallel_group_path']]); + assert.deepStrictEqual(expected.group_over_limit.events, {StartAccepted: 1, WorkflowStarted: 1}); +}); + +for (const [name, mutate] of [ + ['the former 100-command refusal', state => { + state.timers.completed = {status: 422, outcome: null, reason: 'invalid_commands', run_status: null, message: 'invalid_commands', + errors: null, created_tasks: 0}; + state.timers.events = {StartAccepted: 1, WorkflowStarted: 1}; + }], + ['a truncated batch', state => { state.activities.events.ActivityScheduled = 100; }], + ['a run described without its wait', state => { state.timers.described.wait_kind = null; }], + ['the former 100-member group refusal', state => { + state.activity_group.completed = {status: 422, outcome: null, reason: 'development_parallel_group_unqualified', + run_status: null, message: 'development_parallel_group_unqualified', errors: null, created_tasks: 0}; + }], + ['an over-limit group refused on its first command only', state => { + delete state.group_over_limit.completed.errors['commands.1.parallel_group_path']; + }], +]) test(`the contract rejects ${name}`, () => { + const raw = observation(); mutate(raw.large_command_batches); + assert.throws(() => checkLargeCommandBatches(fixture, raw)); +}); From d7098c6fa176931cc20e0428b51fe6ba6049738d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:36:25 +0000 Subject: [PATCH 2/8] Rust #325: keep the bounded-update test's worker active updates_have_no_lifetime_cap_and_preserve_accepted_work_after_wait_timeout runs 64 update round trips without a worker heartbeat. On a loaded host the loop outlasts the worker's staleness window, and the server then correctly refuses the stale lease owner's completion (409 stale_worker_registration) or hands out a different task. Eight concurrent runs all failed before; with a heartbeat per iteration all eight pass. Co-Authored-By: Claude Opus 5.5 --- rust/tests/execution.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 2319e19b0..769a92b15 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -2466,6 +2466,10 @@ async fn updates_have_no_lifetime_cap_and_preserve_accepted_work_after_wait_time open_update_wait(&app, "update-worker").await; let path = "/api/workflows/bounded-updates/update/set"; for index in 0..64 { + // Keep the worker active: on a loaded host the loop can outlast + // its staleness window, and a stale lease owner's task is refused. + let (status, beat) = request(&app, "POST", "/api/worker/heartbeat", json!({"worker_id":"update-worker"})).await; + assert_eq!(status, StatusCode::OK, "{beat}"); // PHP's shortest completion wait is one second; one update waits it // out and the rest return at acceptance. let mut body = json!({"input":envelope(Payload::Array(vec![Payload::Long(index)])), From e1cfaf78be866e6a1b21509a0939354a0cdf0b36 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 03:49:42 +0000 Subject: [PATCH 3/8] Rust #325: keep selection metadata within PHP's integer range From #851 review. The parallel preflight's selection member checks added base + size and member base + member size without overflow checks, so a member size of i64::MAX overflowed. It also accepted a numeric member key above i64::MAX, which PHP decodes as a float and refuses. Published PHP refuses both with the group path message; use checked sums and the signed 64-bit range. Co-Authored-By: Claude Opus 5.5 --- .../2026-10-29-large-command-batches.md | 9 ++++- rust/src/runtime/completion_rules.rs | 12 ++++-- rust/tests/execution/large_command_batches.rs | 39 +++++++++++++++++++ 3 files changed, 55 insertions(+), 5 deletions(-) diff --git a/docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md b/docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md index 390eab79b..fd0529a11 100644 --- a/docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md +++ b/docs/rust-server-port-checkpoints/2026-10-29-large-command-batches.md @@ -54,7 +54,12 @@ cancellation and cancellation scope checks. A valid group that native does not run (for example a child group) is still refused later with `development_parallel_group_unqualified`. + - **Integer range (from review).** PHP's member-range sums past the + integer range become floats that fail its bounds, and a numeric member + key above it decodes to a float. Native uses checked sums and the signed + 64-bit range for numeric keys; the earlier native sum overflowed. - **Tests.** `large_command_batches.rs` (150 timers and 150 activities accepted, the run described as waiting on its wait; a 150-member group - accepted and a 1,001-member group refused on both commands). Both fail on - the previous source. + accepted and a 1,001-member group refused on both commands; a select + member past the integer range and a key above it refused by the + preflight). All fail on the previous source. diff --git a/rust/src/runtime/completion_rules.rs b/rust/src/runtime/completion_rules.rs index 2e91bec62..567ca3483 100644 --- a/rust/src/runtime/completion_rules.rs +++ b/rust/src/runtime/completion_rules.rs @@ -713,11 +713,17 @@ fn parallel_entry(value: &Value, leaf: &str) -> Option= 0))?; let member_index = int("selection_member_index").filter(|index| *index >= 0)?; - let member_base = int("selection_member_base_sequence").filter(|member| (base..base + size).contains(member))?; - let member_size = int("selection_member_size").filter(|member| *member >= 1 && member_base + member <= base + size)?; + // PHP's sums past the integer range become floats that fail these + // bounds; checked sums refuse them the same way. + let end = base.checked_add(size)?; + let member_base = int("selection_member_base_sequence").filter(|member| (base..end).contains(member))?; + let member_size = int("selection_member_size") + .filter(|member| *member >= 1 && member_base.checked_add(*member).is_some_and(|last| last <= end))?; let member_kind = text("selection_member_kind") .filter(|kind| matches!(*kind, "activity" | "child" | "timer" | "signal" | "condition" | "group"))?; entry.extend([("parallel_group_mode", json!("select")), ("parallel_group_base_sequence", json!(base)), diff --git a/rust/tests/execution/large_command_batches.rs b/rust/tests/execution/large_command_batches.rs index ffdfbffa7..31dcd9fbb 100644 --- a/rust/tests/execution/large_command_batches.rs +++ b/rust/tests/execution/large_command_batches.rs @@ -80,3 +80,42 @@ async fn a_parallel_group_may_hold_up_to_the_command_batch_size() { runtime.close().await; database.remove().await; } + +fn select_member(base: i64, member_size: i64, key: Value) -> Value { + let entry = json!({"parallel_group_id":format!("select-calls:{base}:2"),"parallel_group_kind":"activity", + "parallel_group_mode":"select","parallel_group_base_sequence":base,"parallel_group_size":2,"parallel_group_index":0, + "selection_member_key":key,"selection_member_index":0,"selection_member_base_sequence":base, + "selection_member_size":member_size,"selection_member_kind":"activity"}); + let mut command = json!({"type":"schedule_activity","activity_type":"activity-echo","arguments":envelope(Payload::Array(vec![]))}); + command.as_object_mut().unwrap().extend(entry.as_object().unwrap().clone()); + command["parallel_group_path"] = json!([entry]); + command +} + +// PHP's preflight refuses a selection member whose range passes the integer +// range (its sums become floats) and a numeric member key above it (a float +// after JSON decoding), with the group path message. +#[tokio::test] +async fn select_metadata_past_the_integer_range_is_refused_by_the_preflight() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "worker", json!(["echo"]), json!(["activity-echo"])).await; + start(&app, "select-extremes").await; + let task = poll(&app, "worker", "workflow").await; + let message = "Parallel workflow commands require complete top-level metadata and a non-empty group path."; + for command in [ + select_member(1, i64::MAX, json!("a")), + select_member(i64::MAX, 1, json!("a")), + select_member(1, 1, json!(9_223_372_036_854_775_808_u64)), + ] { + let (status, body) = finish_task(&app, &task, json!([command])).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::UNPROCESSABLE_ENTITY, json!("validation_failed")), "{body}"); + assert_eq!(body["errors"], json!({"commands.0.parallel_group_path":[message]}), "{body}"); + } + // The same member with an ordinary range and key passes the preflight. + let (_, body) = finish_task(&app, &task, json!([select_member(1, 1, json!(9_223_372_036_854_775_807_i64))])).await; + assert_ne!(body["reason"], "validation_failed", "{body}"); + runtime.close().await; + database.remove().await; +} From bfd3ee878981726e9abd38c0ecb627f30ebea1c5 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 03:56:11 +0000 Subject: [PATCH 4/8] Rust #325: read the 150-member group through the history API The group test counted members with payload LIKE, which PostgreSQL refuses on its json payload column (operator does not exist: json ~~ unknown), failing both PostgreSQL native-test lanes on #851. Read the run's history through the API instead, the same on every database. Co-Authored-By: Claude Opus 5.5 --- rust/tests/execution/large_command_batches.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/rust/tests/execution/large_command_batches.rs b/rust/tests/execution/large_command_batches.rs index 31dcd9fbb..7a51fd7a1 100644 --- a/rust/tests/execution/large_command_batches.rs +++ b/rust/tests/execution/large_command_batches.rs @@ -74,9 +74,12 @@ async fn a_parallel_group_may_hold_up_to_the_command_batch_size() { assert_eq!(body["created_task_ids"].as_array().map(Vec::len), Some(150)); let (_, described) = request(&app, "GET", "/api/workflows/group-of-150", Value::Null).await; assert_eq!((described["status"].clone(), described["wait_kind"].clone()), (json!("waiting"), json!("activity")), "{described}"); + // Every member records the group, read through the history API so the + // check is the same on each database. let run = started["run_id"].as_str().unwrap(); - assert_eq!(task_retry::rows(&database, Box::leak(format!( - "SELECT COUNT(*) FROM workflow_history_events WHERE workflow_run_id='{run}' AND event_type='ActivityScheduled' AND payload LIKE '%parallel-activities:1:150%'").into_boxed_str())).await, 150); + let (_, history) = request(&app, "GET", &format!("/api/workflows/group-of-150/runs/{run}/history?page_size=1000"), Value::Null).await; + assert_eq!(history["events"].as_array().unwrap().iter().filter(|event| event["event_type"] == "ActivityScheduled" + && event["payload"]["parallel_group_id"] == "parallel-activities:1:150").count(), 150, "{history}"); runtime.close().await; database.remove().await; } From b4c2c677acf001dc752df37e79cb6f4a20d6233e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:14:44 +0000 Subject: [PATCH 5/8] Rust #325: record selection metadata past the integer range From #851 review. The large-command-batches fixture gains three raw-HTTP cases that published 2.11.5 refuses in its parallel preflight with the group path message and no created work: a selection member size of i64::MAX, a group base of i64::MAX, and a numeric member key of 2^63. Each case's commands are sent as written JSON text, so the 2^63 key reaches the server as a literal integer rather than a PHP float. Co-Authored-By: Claude Opus 5.5 --- .../large-command-batches-probe.php | 18 ++- .../ServerParity/large-command-batches.json | 104 +++++++++++++++++- ...rParityLargeCommandBatchesContractTest.mjs | 11 +- 3 files changed, 126 insertions(+), 7 deletions(-) diff --git a/scripts/conformance/server-parity/large-command-batches-probe.php b/scripts/conformance/server-parity/large-command-batches-probe.php index 9b7d5db2a..f3ae49ce4 100644 --- a/scripts/conformance/server-parity/large-command-batches-probe.php +++ b/scripts/conformance/server-parity/large-command-batches-probe.php @@ -29,9 +29,15 @@ function finishHttpLargeCommandBatches(Client $admin, array $fixture, string $wo '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 { + // A string body is sent as written, so a JSON number past PHP's integer + // range reaches the server as the literal token. + $raw = static function (string $method, string $path, array|string|null $body = null) use ($http, $headers): array { $response = $http->request($method, $path, ['headers' => $headers, - ...($body === null ? [] : ['json' => $body === [] ? (object) [] : $body])]); + ...(match (true) { + $body === null => [], + is_string($body) => ['body' => $body, 'headers' => [...$headers, 'Content-Type' => 'application/json']], + default => ['json' => $body === [] ? (object) [] : $body], + })]); return ['status' => $response->getStatusCode(), 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; @@ -76,8 +82,12 @@ function finishHttpLargeCommandBatches(Client $admin, array $fixture, string $wo $commands[$index] += [...$entry, 'parallel_group_path' => [$entry]]; } } - $completed = $raw('POST', 'worker/workflow-tasks/'.$task['task_id'].'/complete', ['lease_owner' => $task['lease_owner'], - 'workflow_task_attempt' => $task['workflow_task_attempt'], 'commands' => $commands]); + // A case's `raw_commands` is the commands array as JSON text, spliced + // into the body unchanged. + $fence = ['lease_owner' => $task['lease_owner'], 'workflow_task_attempt' => $task['workflow_task_attempt']]; + $completed = $raw('POST', 'worker/workflow-tasks/'.$task['task_id'].'/complete', isset($shape['raw_commands']) + ? substr(json_encode($fence, JSON_THROW_ON_ERROR), 0, -1).',"commands":'.$shape['raw_commands'].'}' + : [...$fence, 'commands' => $commands]); $observed = ['completed' => ['status' => $completed['status'], 'outcome' => $completed['response']['outcome'] ?? null, 'reason' => $completed['response']['reason'] ?? null, 'run_status' => $completed['response']['run_status'] ?? null, 'message' => $completed['response']['message'] ?? null, diff --git a/tests/Fixtures/ServerParity/large-command-batches.json b/tests/Fixtures/ServerParity/large-command-batches.json index b410a72fd..8eac0b12b 100644 --- a/tests/Fixtures/ServerParity/large-command-batches.json +++ b/tests/Fixtures/ServerParity/large-command-batches.json @@ -2,7 +2,7 @@ "schema": "durable-workflow.server-parity-fixture/v1", "id": "large-command-batches", "review_state": "reviewed", - "contract": "Workflow task completions with more than 100 commands (published Server 2.11.5 WorkerController::completeWorkflowTask with Workflow 2.15.1 WorkflowCommandNormalizer), over raw worker HTTP on protocol 1.20 in a fresh namespace. An external worker's completion has no command-count limit of its own; each command meets the run's structural limits instead. Completing a run's first workflow task with 150 start_timer commands, or 150 schedule_activity commands, is accepted (200 completed, run waiting) with a created task for each, records 150 TimerScheduled or ActivityScheduled events, and describes the run as waiting on its first timer or activity. An all group of 150 activities, each declaring the group with a one-entry parallel_group_path, is accepted the same way. A group declaring 1,001 members (above the default 1,000-command batch size that bounds a declared group) is refused 422 validation_failed before any guard, with an error on each command's parallel_group_path, and records nothing. Embedded applications have no worker protocol.", + "contract": "Workflow task completions with more than 100 commands (published Server 2.11.5 WorkerController::completeWorkflowTask with Workflow 2.15.1 WorkflowCommandNormalizer), over raw worker HTTP on protocol 1.20 in a fresh namespace. An external worker's completion has no command-count limit of its own; each command meets the run's structural limits instead. Completing a run's first workflow task with 150 start_timer commands, or 150 schedule_activity commands, is accepted (200 completed, run waiting) with a created task for each, records 150 TimerScheduled or ActivityScheduled events, and describes the run as waiting on its first timer or activity. An all group of 150 activities, each declaring the group with a one-entry parallel_group_path, is accepted the same way. A group declaring 1,001 members (above the default 1,000-command batch size that bounds a declared group) is refused 422 validation_failed before any guard, with an error on each command's parallel_group_path, and records nothing. A selection member whose range passes the 64-bit integer range (member size or group base 9223372036854775807), or whose numeric key is 9223372036854775808 (sent as a literal JSON integer), is refused by the same preflight with the group path message and records nothing. Embedded applications have no worker protocol.", "workflow_type": "parity.v1.echo", "input": { "message": "large command batches λ", @@ -54,6 +54,21 @@ "kind": "activities", "commands": 2, "group_size": 1001 + }, + "select_member_overflow": { + "kind": "activities", + "commands": 0, + "raw_commands": "[{\"type\":\"schedule_activity\",\"activity_type\":\"parity.v1.large_command_batches_activity\",\"arguments\":{\"codec\":\"avro\",\"blob\":\"wwHioz3/VYAiNwwA\"},\"parallel_group_id\":\"select-calls:1:2\",\"parallel_group_kind\":\"activity\",\"parallel_group_mode\":\"select\",\"parallel_group_base_sequence\":1,\"parallel_group_size\":2,\"parallel_group_index\":0,\"selection_member_key\":\"a\",\"selection_member_index\":0,\"selection_member_base_sequence\":1,\"selection_member_size\":9223372036854775807,\"selection_member_kind\":\"activity\",\"parallel_group_path\":[{\"parallel_group_id\":\"select-calls:1:2\",\"parallel_group_kind\":\"activity\",\"parallel_group_mode\":\"select\",\"parallel_group_base_sequence\":1,\"parallel_group_size\":2,\"parallel_group_index\":0,\"selection_member_key\":\"a\",\"selection_member_index\":0,\"selection_member_base_sequence\":1,\"selection_member_size\":9223372036854775807,\"selection_member_kind\":\"activity\"}]}]" + }, + "select_base_overflow": { + "kind": "activities", + "commands": 0, + "raw_commands": "[{\"type\":\"schedule_activity\",\"activity_type\":\"parity.v1.large_command_batches_activity\",\"arguments\":{\"codec\":\"avro\",\"blob\":\"wwHioz3/VYAiNwwA\"},\"parallel_group_id\":\"select-calls:9223372036854775807:2\",\"parallel_group_kind\":\"activity\",\"parallel_group_mode\":\"select\",\"parallel_group_base_sequence\":9223372036854775807,\"parallel_group_size\":2,\"parallel_group_index\":0,\"selection_member_key\":\"a\",\"selection_member_index\":0,\"selection_member_base_sequence\":9223372036854775807,\"selection_member_size\":1,\"selection_member_kind\":\"activity\",\"parallel_group_path\":[{\"parallel_group_id\":\"select-calls:9223372036854775807:2\",\"parallel_group_kind\":\"activity\",\"parallel_group_mode\":\"select\",\"parallel_group_base_sequence\":9223372036854775807,\"parallel_group_size\":2,\"parallel_group_index\":0,\"selection_member_key\":\"a\",\"selection_member_index\":0,\"selection_member_base_sequence\":9223372036854775807,\"selection_member_size\":1,\"selection_member_kind\":\"activity\"}]}]" + }, + "select_key_overflow": { + "kind": "activities", + "commands": 0, + "raw_commands": "[{\"type\":\"schedule_activity\",\"activity_type\":\"parity.v1.large_command_batches_activity\",\"arguments\":{\"codec\":\"avro\",\"blob\":\"wwHioz3/VYAiNwwA\"},\"parallel_group_id\":\"select-calls:1:2\",\"parallel_group_kind\":\"activity\",\"parallel_group_mode\":\"select\",\"parallel_group_base_sequence\":1,\"parallel_group_size\":2,\"parallel_group_index\":0,\"selection_member_key\":9223372036854775808,\"selection_member_index\":0,\"selection_member_base_sequence\":1,\"selection_member_size\":1,\"selection_member_kind\":\"activity\",\"parallel_group_path\":[{\"parallel_group_id\":\"select-calls:1:2\",\"parallel_group_kind\":\"activity\",\"parallel_group_mode\":\"select\",\"parallel_group_base_sequence\":1,\"parallel_group_size\":2,\"parallel_group_index\":0,\"selection_member_key\":9223372036854775808,\"selection_member_index\":0,\"selection_member_base_sequence\":1,\"selection_member_size\":1,\"selection_member_kind\":\"activity\"}]}]" } }, "expected": { @@ -168,6 +183,93 @@ "WorkflowStarted": 1 }, "parallel_groups": [] + }, + "select_member_overflow": { + "completed": { + "status": 422, + "outcome": null, + "reason": "validation_failed", + "run_status": null, + "message": "Parallel workflow commands require complete top-level metadata and a non-empty group path.", + "errors": { + "commands.0.parallel_group_path": [ + "Parallel workflow commands require complete top-level metadata and a non-empty group path." + ] + }, + "created_tasks": 0 + }, + "described": { + "status": "pending", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "workflow-task", + "wait_reason": "Workflow task leased to worker" + }, + "events": { + "StartAccepted": 1, + "WorkflowStarted": 1 + }, + "parallel_groups": [] + }, + "select_base_overflow": { + "completed": { + "status": 422, + "outcome": null, + "reason": "validation_failed", + "run_status": null, + "message": "Parallel workflow commands require complete top-level metadata and a non-empty group path.", + "errors": { + "commands.0.parallel_group_path": [ + "Parallel workflow commands require complete top-level metadata and a non-empty group path." + ] + }, + "created_tasks": 0 + }, + "described": { + "status": "pending", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "workflow-task", + "wait_reason": "Workflow task leased to worker" + }, + "events": { + "StartAccepted": 1, + "WorkflowStarted": 1 + }, + "parallel_groups": [] + }, + "select_key_overflow": { + "completed": { + "status": 422, + "outcome": null, + "reason": "validation_failed", + "run_status": null, + "message": "Parallel workflow commands require complete top-level metadata and a non-empty group path.", + "errors": { + "commands.0.parallel_group_path": [ + "Parallel workflow commands require complete top-level metadata and a non-empty group path." + ] + }, + "created_tasks": 0 + }, + "described": { + "status": "pending", + "status_bucket": "running", + "is_terminal": false, + "resume_step_sequence": null, + "recovery_kind": null, + "wait_kind": "workflow-task", + "wait_reason": "Workflow task leased to worker" + }, + "events": { + "StartAccepted": 1, + "WorkflowStarted": 1 + }, + "parallel_groups": [] } }, "described_fields": [ diff --git a/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs b/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs index 7eff343d7..ba2267f54 100644 --- a/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs +++ b/tests/Unit/ServerParityLargeCommandBatchesContractTest.mjs @@ -8,13 +8,13 @@ const expected = fixture.large_command_batches.expected; const observation = () => structuredClone({mode: 'http', large_command_batches: {applicable: true, ...expected}}); test('the published answers pass', () => { - assert.deepStrictEqual(checkLargeCommandBatches(fixture, observation()), {scope: 'large_command_batches', runs: 4}); + assert.deepStrictEqual(checkLargeCommandBatches(fixture, observation()), {scope: 'large_command_batches', runs: 7}); }); test('embedded applications have no worker protocol', () => { assert.deepStrictEqual(checkLargeCommandBatches(fixture, {mode: 'embedded', large_command_batches: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), - {scope: 'large_command_batches', runs: 4}); + {scope: 'large_command_batches', runs: 7}); }); test('150 commands are accepted in one completion', () => { @@ -46,6 +46,13 @@ for (const [name, mutate] of [ state.activity_group.completed = {status: 422, outcome: null, reason: 'development_parallel_group_unqualified', run_status: null, message: 'development_parallel_group_unqualified', errors: null, created_tasks: 0}; }], + ['an overflowing selection key accepted', state => { + state.select_key_overflow.completed = {status: 200, outcome: 'completed', reason: null, run_status: 'waiting', + message: null, errors: null, created_tasks: 1}; + }], + ['an overflowing member refused for another reason', state => { + state.select_member_overflow.completed.reason = 'development_parallel_group_unqualified'; + }], ['an over-limit group refused on its first command only', state => { delete state.group_over_limit.completed.errors['commands.1.parallel_group_path']; }], From 9eb0b23310b8022cf135ea50ad12323251e2fe19 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:18:13 +0000 Subject: [PATCH 6/8] Rust #325: give the large-command-batches probe its own budget The probe now runs seven cases, three of them with 150 commands each. Against published PHP it took 18 seconds on an idle host, but timed out at the runner's 45-second default on a loaded one (spawnSync php ETIMEDOUT), the likely cause of #851's corpus failures. Give it a 150-second budget, as other long probes declare. Co-Authored-By: Claude Opus 5.5 --- scripts/conformance/server-parity.mjs | 2 +- tests/Fixtures/ServerParity/large-command-batches.json | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/scripts/conformance/server-parity.mjs b/scripts/conformance/server-parity.mjs index b2e7a6b28..5e5fc68f6 100644 --- a/scripts/conformance/server-parity.mjs +++ b/scripts/conformance/server-parity.mjs @@ -105,7 +105,7 @@ try { for (const flag of ['url', 'application-root']) { if (values[flag]) arguments_.push(`--${flag}`, values[flag]); } - const probe = captureProbe('php', arguments_, {directory: dirname(resolve(values.output)), timeout: fixture.memo_replay ? fixture.memo_replay.overall_budget_seconds * 1000 + 5000 : fixture.corrected_markers ? fixture.corrected_markers.overall_budget_seconds * 1000 + 5000 : fixture.namespace_signal_pair || fixture.selected_run_signal_pair || fixture.cancellation_scopes ? 75000 : fixture.update_validation ? 90000 : fixture.worker_poll_fence ? fixture.worker_poll_fence.overall_budget_seconds * 1000 + 5000 : fixture.worker_sessions ? fixture.worker_sessions.overall_budget_seconds * 1000 + 5000 : fixture.worker_session_activities ? fixture.worker_session_activities.overall_budget_seconds * 1000 + 5000 : 45000, + const probe = captureProbe('php', arguments_, {directory: dirname(resolve(values.output)), timeout: fixture.memo_replay ? fixture.memo_replay.overall_budget_seconds * 1000 + 5000 : fixture.corrected_markers ? fixture.corrected_markers.overall_budget_seconds * 1000 + 5000 : fixture.namespace_signal_pair || fixture.selected_run_signal_pair || fixture.cancellation_scopes ? 75000 : fixture.update_validation ? 90000 : fixture.worker_poll_fence ? fixture.worker_poll_fence.overall_budget_seconds * 1000 + 5000 : fixture.worker_sessions ? fixture.worker_sessions.overall_budget_seconds * 1000 + 5000 : fixture.worker_session_activities ? fixture.worker_session_activities.overall_budget_seconds * 1000 + 5000 : fixture.large_command_batches ? fixture.large_command_batches.overall_budget_seconds * 1000 + 5000 : 45000, env: {...process.env, DW_PARITY_FAILURE_EVIDENCE_DIR: dirname(resolve(values.output))}, preserveStdout: join(sourceDirectory, name)}); const item = {fixture_id: fixture.id, fixture, outcome: 'runner-blocked'}; diff --git a/tests/Fixtures/ServerParity/large-command-batches.json b/tests/Fixtures/ServerParity/large-command-batches.json index 8eac0b12b..cb9807ab1 100644 --- a/tests/Fixtures/ServerParity/large-command-batches.json +++ b/tests/Fixtures/ServerParity/large-command-batches.json @@ -280,6 +280,7 @@ "is_terminal", "resume_step_sequence", "recovery_kind" - ] + ], + "overall_budget_seconds": 150 } } From fd2a4b5a20b872279a119b27b6adc0a523fda6fb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:55:11 +0000 Subject: [PATCH 7/8] Rust #325: keep a blocked probe's fixture in its failure excerpt From #851 review. When a probe fails before it observes anything, the runner writes a runner-blocked record and stops, but only the product-fail path wrote the failure excerpt that the CI artifact summary keeps. A timed-out probe therefore left no fixture name in the retained artifacts, as in #851's corpus failures. Write the bounded excerpt for a blocked record too; the new runner test drives the recorder with a failing fake php and checks the excerpt names the fixture. Also rename the MySQL and PostgreSQL bootstrap steps to say they run the native tests, since a failed native test was reported under "Bootstrap". Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 7 +++-- scripts/conformance/server-parity.mjs | 4 +++ tests/Unit/ServerParityPartsTest.mjs | 8 ++--- .../ServerParityRunnerBlockedExcerptTest.mjs | 31 +++++++++++++++++++ 4 files changed, 43 insertions(+), 7 deletions(-) create mode 100644 tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index ec1c39f85..b992a9218 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -30,6 +30,7 @@ on: - tests/Fixtures/ServerParityBaselines/** - tests/Fixtures/ServerParityNormalization/** - tests/Unit/ServerParityRunnerTest.mjs + - tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs - tests/Unit/ServerParitySdkDirectionLanesTest.mjs - tests/Unit/ServerParityCodecDeadlineCellsTest.mjs - tests/Unit/ServerParityArtifactSummaryTest.mjs @@ -525,7 +526,7 @@ jobs: -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target \ -v "$PWD:/source:ro" -v "$PWD/cargo-home:/cargo" -v "$PWD/rust-target:/target" -w /source/rust \ mirror.gcr.io/library/rust@sha256:ba81bc3eaa4422af576c0262515d96b0111a628a6ccc2c86557cf55c9a4bbee0 cargo clean --locked --offline --package durable-workflow-server - - name: Bootstrap independent published PHP and native MySQL databases + - name: Bootstrap published PHP and native MySQL databases and run the native tests run: | set -euo pipefail # Failed Rust tests are printed from their evidence files into this step's log. @@ -1206,7 +1207,7 @@ jobs: -e CARGO_HOME=/cargo -e CARGO_TARGET_DIR=/target \ -v "$PWD:/source:ro" -v "$PWD/cargo-home:/cargo" -v "$PWD/rust-target:/target" -w /source/rust \ mirror.gcr.io/library/rust@sha256:ba81bc3eaa4422af576c0262515d96b0111a628a6ccc2c86557cf55c9a4bbee0 cargo clean --locked --offline --package durable-workflow-server - - name: Bootstrap independent published PHP and native PostgreSQL databases + - name: Bootstrap published PHP and native PostgreSQL databases and run the native tests env: POSTGRES_IMAGE: ${{ matrix.image }} run: | @@ -1986,7 +1987,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/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/ServerParityLargeCommandBatchesContractTest.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/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/ServerParityLargeCommandBatchesContractTest.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/scripts/conformance/server-parity.mjs b/scripts/conformance/server-parity.mjs index 5e5fc68f6..baa25fbd8 100644 --- a/scripts/conformance/server-parity.mjs +++ b/scripts/conformance/server-parity.mjs @@ -148,6 +148,10 @@ try { record.outcome = item.outcome; record.finished_at = new Date().toISOString(); writeRecord(values.output, record, originals); + // The artifact summary keeps this outline, not the record, so a + // blocked probe still names its fixture and error. + writeFileSync(values.output.replace(/\.json$/, '') + '.failure-excerpt.txt', + failureExcerpt(record.cases.filter(entry => entry.outcome !== 'pass'))); throw error; } record.cases.push(item); diff --git a/tests/Unit/ServerParityPartsTest.mjs b/tests/Unit/ServerParityPartsTest.mjs index b6a5ff46b..4ffcdbf94 100644 --- a/tests/Unit/ServerParityPartsTest.mjs +++ b/tests/Unit/ServerParityPartsTest.mjs @@ -13,7 +13,7 @@ const PARTS = ['native-tests', 'corpus', 'restarts', 'cohorts', 'sdk-directions' const SHARED_NODES = 'Start the published PHP and native nodes the corpus parts share'; const EVERY_PART = [/^actions\/checkout@/, /^Pull the pinned images, retrying registry errors$/, /^Fetch /, /^Restore native Cargo dependencies from trusted main$/, /^Discard restored native workspace outputs before rebuilding$/, - /^Create TLS fixture certificates$/, /^Bootstrap independent /, /^Collect /, /^Summarize /, /^Retain /]; + /^Create TLS fixture certificates$/, /^Bootstrap published /, /^Collect /, /^Summarize /, /^Retain /]; function job(name) { const start = workflow.indexOf(`\n ${name}:\n`); @@ -119,10 +119,10 @@ for (const [name, key] of [['mysql-execution', 'Compare three runtimes'], ['post } test('the database bootstraps run the native suites only in the native-tests parts', () => { - const postgres = steps(job('postgres-storage')).find(step => step.name.startsWith('Bootstrap independent ')); + const postgres = steps(job('postgres-storage')).find(step => step.name.startsWith('Bootstrap published ')); assert.match(postgres.text, /-e DW_PARITY_PART \\\n/); assert.match(postgres.text, /'if \[ "\$DW_PARITY_PART" = native-tests \]; then cargo test [^']* --test execution [^']*; fi && cargo build --locked --offline'/); - const mysql = steps(job('mysql-execution')).find(step => step.name.startsWith('Bootstrap independent ')); + const mysql = steps(job('mysql-execution')).find(step => step.name.startsWith('Bootstrap published ')); assert.match(mysql.text, /-e DW_PARITY_PART -e DW_EXECUTION_SHARDS \\\n/); // native-tests runs the database suite and the first half of the execution // tests; native-tests-b, where a label has it, runs the second half. @@ -148,7 +148,7 @@ test('the MySQL 8 labels split their execution tests into two parts, every test } assert.ok(entries.filter(entry => !sharded.includes(entry.label)).every(entry => !entry.execution_shards && !entry.part)); // The selection the bootstrap runs: list, sort bytewise, deal round-robin. - const bootstrap = steps(block).find(step => step.name.startsWith('Bootstrap independent ')).text; + const bootstrap = steps(block).find(step => step.name.startsWith('Bootstrap published ')).text; const selection = /cargo test --locked --offline --test execution -- --list --format terse \| sed -n "s\/: test\\\$\/\/p" \| LC_ALL=C sort \| awk -v count="\$DW_EXECUTION_SHARDS" -v shard="\$shard" "([^"]+)" > \/evidence\/execution-shard\$suffix\.txt && test -s \/evidence\/execution-shard\$suffix\.txt && /.exec(bootstrap); assert.ok(selection, 'the execution tests are selected by the listed round-robin'); const names = Array.from({length: 489}, (_, index) => `suite::test_${String(index).padStart(3, '0')}`); diff --git a/tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs b/tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs new file mode 100644 index 000000000..82f83a269 --- /dev/null +++ b/tests/Unit/ServerParityRunnerBlockedExcerptTest.mjs @@ -0,0 +1,31 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {spawnSync} from 'node:child_process'; +import {chmodSync, mkdtempSync, readFileSync, rmSync, writeFileSync} from 'node:fs'; +import {tmpdir} from 'node:os'; +import {delimiter, join} from 'node:path'; +import {fileURLToPath} from 'node:url'; + +const runner = fileURLToPath(new URL('../../scripts/conformance/server-parity.mjs', import.meta.url)); +const fixture = fileURLToPath(new URL('../Fixtures/ServerParity/large-command-batches.json', import.meta.url)); + +// A probe that fails before it observes anything blocks the record. The +// artifact summary keeps the failure excerpt, not the record, so the +// excerpt must still name the blocked fixture and its error. +test('a blocked probe leaves a failure excerpt naming its fixture', () => { + const directory = mkdtempSync(join(tmpdir(), 'dw-runner-blocked-')); + try { + writeFileSync(join(directory, 'php'), '#!/bin/sh\necho "fake php refused" >&2\nexit 3\n'); + chmodSync(join(directory, 'php'), 0o755); + const output = join(directory, 'php-record.json'); + const run = spawnSync(process.execPath, [runner, 'record', '--mode', 'http', '--url', 'http://127.0.0.1:9', + '--target', 'php', '--output', output, '--runner-revision', '0'.repeat(40), '--fixture', fixture], + {encoding: 'utf8', env: {...process.env, PATH: `${directory}${delimiter}${process.env.PATH}`}}); + assert.notEqual(run.status, 0, run.stdout); + assert.equal(JSON.parse(readFileSync(output, 'utf8')).outcome, 'runner-blocked'); + assert.match(readFileSync(join(directory, 'php-record.failure-excerpt.txt'), 'utf8'), + /^== large-command-batches \(runner-blocked\): fake php refused/); + } finally { + rmSync(directory, {recursive: true, force: true}); + } +}); From 0cbdefb0d2a24af03168b5534dd0f60ada380e20 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:40:07 +0000 Subject: [PATCH 8/8] Rust #325: end the large-command-batches runs once they are recorded #851's merge-queue corpus failed twice on schedule-fixed-rate-one- occurrence: published PHP's installed scheduler did not fire within the probe's ten seconds. The large-command-batches fixture, which the corpus runs before it, left several hundred open timers and never-polled activity tasks behind. PHP's maintenance loop then repaired 25 of them per pass; one pass took 19.4 seconds locally and held up the scheduler evaluation that follows it. The probe now terminates each of its runs after recording them. Locally the passes after the fixture return to zero repair candidates and about 1.7 seconds. Co-Authored-By: Claude Opus 5.5 --- .../server-parity/large-command-batches-probe.php | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/scripts/conformance/server-parity/large-command-batches-probe.php b/scripts/conformance/server-parity/large-command-batches-probe.php index f3ae49ce4..ba5dfe6d9 100644 --- a/scripts/conformance/server-parity/large-command-batches-probe.php +++ b/scripts/conformance/server-parity/large-command-batches-probe.php @@ -115,6 +115,16 @@ function finishHttpLargeCommandBatches(Client $admin, array $fixture, string $wo $observed['parallel_groups'] = $groups; $state[$case] = $observed; } + // The runs' open timers and never-polled activity tasks would otherwise + // stay for the rest of the corpus, where every maintenance pass repairs + // them and delays the installed scheduler the schedule fixtures wait on. + foreach (array_keys($definition['cases']) as $case) { + $terminated = $raw('POST', 'workflows/'.rawurlencode($workflowId.'-lcb-'.$case).'/terminate', + ['reason' => 'large command batches recorded']); + if (! in_array($terminated['status'], [200, 202], true)) { + throw new RuntimeException('Large command batches cleanup: '.json_encode($terminated)); + } + } return $state; }