From 0d854ac16d281e587931af29781c543a5317006d Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:14:53 +0000 Subject: [PATCH] Rust #325: route query arguments the server cannot decode Published Server decodes a query's arguments only to check them against a query contract the run's start recorded (validateContractedQueryArguments). A run that declares no queries routes the query with its arguments as given, so its worker receives even a blob that does not decode; a declared query refuses such arguments before routing, with PHP's message and validation_errors. Native refused any undecodable blob at admission with a bare invalid_query_arguments. The query-arguments fixture records both against published 2.11.5. The port log's stale "Start build-id pin" gap is corrected: native has pinned starts since #696. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 3 +- .../2026-10-28-query-arguments.md | 42 ++++++ docs/rust-server-port.md | 38 ++++-- rust/src/runtime/execution.rs | 10 +- rust/src/runtime/queries.rs | 61 ++++++++- rust/tests/execution.rs | 3 + .../tests/execution/query_argument_routing.rs | 70 ++++++++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../query-arguments-contract.mjs | 30 +++++ .../server-parity/query-arguments-probe.php | 122 ++++++++++++++++++ .../ServerParity/query-arguments.json | 94 ++++++++++++++ ...ServerParityQueryArgumentsContractTest.mjs | 45 +++++++ 13 files changed, 506 insertions(+), 21 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-28-query-arguments.md create mode 100644 rust/tests/execution/query_argument_routing.rs create mode 100644 scripts/conformance/server-parity/query-arguments-contract.mjs create mode 100644 scripts/conformance/server-parity/query-arguments-probe.php create mode 100644 tests/Fixtures/ServerParity/query-arguments.json create mode 100644 tests/Unit/ServerParityQueryArgumentsContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 9cbd00669..22dc04bbb 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -60,6 +60,7 @@ on: - tests/Unit/ServerParityConcurrentWaitsContractTest.mjs - tests/Unit/ServerParityContinueTransfersContractTest.mjs - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs + - tests/Unit/ServerParityQueryArgumentsContractTest.mjs - tests/Unit/ServerParityParentCloseCancellationContractTest.mjs - tests/Unit/ServerParityBridgeAdaptersContractTest.mjs - tests/Unit/ServerParityNexusOperationsContractTest.mjs @@ -1983,7 +1984,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/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/ServerParityStartSearchAttributesContractTest.mjs tests/Unit/ServerParityQueryArgumentsContractTest.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-28-query-arguments.md b/docs/rust-server-port-checkpoints/2026-10-28-query-arguments.md new file mode 100644 index 000000000..6898dc2d9 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-28-query-arguments.md @@ -0,0 +1,42 @@ +# October 28, query arguments the server cannot decode + +- **Before.** Native decoded every query's arguments at admission. It + refused an undecodable blob, or one that did not decode to a list, with a + bare `422 invalid_query_arguments`, whatever the run declared. +- **Contract** (published 2.11.5; `WorkflowController::query`, + `WorkflowQueryTaskBroker::query`, `validateContractedQueryArguments`): + - The server decodes the arguments only when the run's start recorded a + query contract (declared queries). + - **Declared query.** After the unknown-query check, arguments that do + not decode are refused before routing: `422 invalid_query_arguments`, + with the message `Workflow query [] arguments could not be + decoded.` and `validation_errors.arguments` `["Query arguments could + not be decoded."]`. Arguments that decode to a scalar rather than a PHP + array are refused the same way with "must decode to an array". No query + task is created. + - **Named arguments (not ported here).** A string-keyed Avro map decodes + to a PHP array too. With a durable query contract, Core normalizes it as + named arguments (`WorkflowQueryContract::normalizeNamedArguments`), and + the worker receives the positional result. Native still refuses a map as + "must decode to an array". The port log records the gap (found in + review). + - **No declared queries.** The query is routed with its arguments as + given. A query-capable worker leases a task carrying the blob unchanged, + and the client receives the worker's answer. +- **Fixture.** `query-arguments` (fixture key `query_argument_routing`; a + `query_arguments` field already belongs to `state-queries`). Raw HTTP on + protocol 1.20 in a fresh namespace: + - a run whose worker declares no queries: the client query is held while + the probe leases and answers the task; + - a run whose worker declared the query. + + Each run receives an undecodable blob (`AAAA`) and the avro encoding of + a single string. +- **Native.** Admission reads the arguments without refusing them + (`queries::read_arguments`). `check_query_contract` refuses + undecodable or non-list arguments only for a declared query, with PHP's + message and `validation_errors`. +- **Tests.** `query_argument_routing.rs` (routed to the worker; refused + before routing). Both fail on the previous source. +- **Also corrected.** The port log's "Start build-id pin" gap was stale: + native has pinned starts since #696 (`build_pins.rs`). diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 68d01e64f..b40ae76cc 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -275,10 +275,20 @@ contract fixture before its correction. each priority tier across fairness keys by their weights (`TaskFairnessScheduler`); native stores the keys and weights but does not rebalance yet. -- **Undecodable query input before routing.** Native decodes query - arguments at admission and refuses an undecodable blob with - `invalid_query_arguments`. Published PHP first looks for a query worker, - and answers `409 query_worker_unavailable` when there is none. +- **Named and validated query arguments.** For a declared query with a + durable parameter contract, published PHP normalizes the arguments through + `WorkflowQueryContract::validatedArgumentsForRun`, then gives the worker the + re-encoded positional result: + - a string-keyed map is taken as named arguments; + - positional arguments are matched to the parameters; + - defaults are filled in, variadics collected, and unknown names + reported; + - a violation answers `422 invalid_query_arguments`, "argument validation + failed", with per-parameter messages in `validation_errors`. + + Native checks positional arguments only. It refuses a map as "must decode + to an array", reports a violation as a bare `invalid_query_arguments`, and + gives the worker the arguments as sent. - **When an external payload reference is claimed.** Published PHP retains a referenced object once it fetches the bytes, before the command is recorded, so a refused command still retains it. Native claims it in the @@ -449,10 +459,6 @@ contract fixture before its correction. then did not stop within the 5-second shutdown budget. Forty local repetitions of that case alone passed. Find what holds the query thread after the fence before changing either side. -- **Start build-id pin.** Published PHP pins a new run to its start cohort's - build id: a promoted build-id rollout, else the single build id among the - live workers for the type, else among all the queue's live workers. Native - starts every run unpinned, and has no build-id rollouts. - **Named-namespace routes native does not serve.** Native now serves every workflow command and operation it implements in any namespace, as PHP does ([checkpoint](rust-server-port-checkpoints/2026-10-23-named-namespace-commands.md)). @@ -530,6 +536,22 @@ contract fixture before its correction. `start-search-attributes` fixture records typed, inferred, null, empty and refused starts against published 2.11.5 ([checkpoint](rust-server-port-checkpoints/2026-10-28-start-search-attributes.md)). +- **Query arguments the server cannot decode** (October 28, closed). + Published PHP decodes a query's arguments only to check them against a + query contract the run's start recorded + (`validateContractedQueryArguments`). A run that declares no queries + routes the query with its arguments as given, so its worker receives even + a blob that does not decode. A declared query refuses such arguments + before routing, with PHP's message and `validation_errors`. Native used to + refuse any undecodable blob at admission with a bare + `invalid_query_arguments`. The `query-arguments` fixture records both + against published 2.11.5 + ([checkpoint](rust-server-port-checkpoints/2026-10-28-query-arguments.md)). +- **Start build-id pin** (closed; entry corrected October 28). Native pins + a new run to its start cohort's build id as published PHP does: a promoted + build-id rollout, else the single build id among the live workers for the + type, else among all the queue's live workers (`start_in`, + `deployments.rs`, `build_pins.rs`). The open-gap entry predated #696. - **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/execution.rs b/rust/src/runtime/execution.rs index 605bc44c4..f6c73df8a 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -1124,13 +1124,9 @@ impl Runtime { "query_arguments_exceed_budget", )); } - let values = super::queries::decode( - self.signal_codec.clone(), - arguments.clone(), - true, - "invalid_query_arguments", - ) - .await?; + // Read, not refused: only a declared query contract needs them + // (`validateContractedQueryArguments`). + let values = super::queries::read_arguments(self.signal_codec.clone(), arguments.clone()).await?; match &*self.storage { Storage::Sqlite(store) => { store diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index a09d7dd68..4db17d405 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -128,6 +128,40 @@ pub(super) async fn decode( .map_err(|_| refuse(StatusCode::SERVICE_UNAVAILABLE, "query_codec_unavailable"))? } +/// A query's arguments as the server reads them: a list of values, or why +/// they are not one. Only a declared query contract reads them +/// (`validateContractedQueryArguments`); otherwise the worker receives the +/// arguments as given. A map is refused as not a list, although PHP decodes +/// a string-keyed map to an array and normalizes it as named arguments +/// (`WorkflowQueryContract::normalizeNamedArguments`); the port's gap list +/// records that. +pub(super) enum QueryArguments { + Decoded(Vec), + Undecodable, + NotAList, +} + +/// Decodes a query's arguments without refusing them. +pub(super) async fn read_arguments(semaphore: Arc, blob: String) -> Result { + let permit = semaphore + .acquire_owned() + .await + .map_err(|_| refuse(StatusCode::SERVICE_UNAVAILABLE, "query_codec_unavailable"))?; + tokio::task::spawn_blocking(move || { + let _permit = permit; + let Ok(codec) = crate::codec::ValueCodec::new() else { + return QueryArguments::Undecodable; + }; + match codec.decode(&blob) { + Ok(crate::codec::Value::Array(values)) => QueryArguments::Decoded(values), + Ok(_) => QueryArguments::NotAList, + Err(_) => QueryArguments::Undecodable, + } + }) + .await + .map_err(|_| refuse(StatusCode::SERVICE_UNAVAILABLE, "query_codec_unavailable")) +} + /// One admission attempt: an answer, or no worker to route the query to. pub(super) enum QueryAttempt { Answered((StatusCode, Value)), @@ -179,14 +213,15 @@ fn admits_queries(status: &str) -> bool { } /// `validateContractedQueryArguments`: a run that declares no queries leaves -/// the name and arguments to the worker. A declared query without a -/// parameter contract takes its arguments as given. +/// the name and arguments to the worker. A declared query's arguments must +/// decode to a list, and its parameter contract, when it has one, checks +/// them. fn check_query_contract( contract: &Value, workflow_id: &str, run_id: &str, name: &str, - values: &[crate::codec::Value], + arguments: &QueryArguments, ) -> Result<()> { let declared = |field: &str| contract[field].as_array().is_some_and(|items| !items.is_empty()); if !declared("declared_queries") && !declared("declared_query_contracts") { @@ -196,6 +231,22 @@ fn check_query_contract( return Err(query_refusal(StatusCode::NOT_FOUND, workflow_id, run_id, name, "rejected_unknown_query", &format!("Workflow query [{name}] is not declared on workflow [{workflow_id}]."))); } + // A declared query's arguments must decode to a list before routing. + let (message, error) = match arguments { + QueryArguments::Decoded(values) => return check_declared_arguments(contract, name, values), + QueryArguments::Undecodable => ("could not be decoded", "Query arguments could not be decoded."), + QueryArguments::NotAList => ("must decode to an array", "Query arguments must decode to an array."), + }; + let mut refusal = query_refusal(StatusCode::UNPROCESSABLE_ENTITY, workflow_id, run_id, name, "invalid_query_arguments", + &format!("Workflow query [{name}] arguments {message}.")); + if let super::RuntimeError::Protocol { response, .. } = &mut refusal { + response["validation_errors"] = json!({"arguments": [error]}); + } + Err(refusal) +} + +/// A declared query's parameter contract, when it has one. +fn check_declared_arguments(contract: &Value, name: &str, values: &[crate::codec::Value]) -> Result<()> { if contract["declared_query_contracts"] .as_array() .is_some_and(|contracts| contracts.iter().any(|entry| entry["name"] == name)) @@ -298,7 +349,7 @@ where selected_run: Option<&str>, name: &str, arguments: String, - values: Vec, + values: QueryArguments, config: &QueryTaskConfig, command: Option<&Value>, ) -> Result<(StatusCode, Value)> { @@ -328,7 +379,7 @@ where selected_run: Option<&str>, name: &str, arguments: &str, - values: &[crate::codec::Value], + values: &QueryArguments, config: &QueryTaskConfig, command: Option<&Value>, ) -> Result { diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c984906a9..7dfcac7c2 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/query_argument_routing.rs"] +mod query_argument_routing; + #[path = "execution/close_inventory.rs"] mod close_inventory; diff --git a/rust/tests/execution/query_argument_routing.rs b/rust/tests/execution/query_argument_routing.rs new file mode 100644 index 000000000..213b65853 --- /dev/null +++ b/rust/tests/execution/query_argument_routing.rs @@ -0,0 +1,70 @@ +use super::*; + +// Published Server 2.11.5 (`WorkflowQueryTaskBroker::query`, +// `validateContractedQueryArguments`) decodes a query's arguments only to +// check them against a declared query contract. A run that declares no +// queries routes the query with its arguments as given, so the worker +// receives even a blob that does not decode. A run that declares the query +// refuses such arguments before routing. The query-arguments fixture +// records the same cases. + +const UNDECODABLE: &str = "AAAA"; + +/// The avro encoding of a single string, which is not a list. +fn not_a_list() -> Value { + envelope(Payload::String("not a list".into())) +} + +async fn uncontracted_worker(app: &Router, worker: &str) { + let (status, body) = request(app, "POST", "/api/worker/register", json!({"worker_id":worker,"task_queue":"test", + "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":[],"capabilities":["query_tasks"]})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +#[tokio::test] +async fn an_uncontracted_run_routes_arguments_it_cannot_decode_to_the_worker() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + uncontracted_worker(&app, "plain-worker").await; + start(&app, "plain-query").await; + for input in [json!({"codec":"avro","blob":UNDECODABLE}), not_a_list()] { + let waiting = { + let app = app.clone(); + let input = input.clone(); + tokio::spawn(async move { request(&app, "POST", "/api/workflows/plain-query/query/state", json!({"input": input})).await }) + }; + let task = await_query(&app, "plain-worker").await; + assert_eq!(task["query_arguments"], input, "the worker receives the arguments as given: {task}"); + let (status, body) = request(&app, "POST", &format!("/api/worker/query-tasks/{}/complete", task["query_task_id"].as_str().unwrap()), + json!({"lease_owner":"plain-worker","query_task_attempt":task["query_task_attempt"], + "result_envelope":envelope(Payload::Array(vec![]))})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, body) = waiting.await.unwrap(); + assert_eq!(status, StatusCode::OK, "{body}"); + } + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn a_declared_query_refuses_arguments_that_do_not_decode_to_a_list() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + query_worker(&app, "contract-worker").await; + let started = start(&app, "declared-query").await; + for (input, message, error) in [ + (json!({"codec":"avro","blob":UNDECODABLE}), "Workflow query [state] arguments could not be decoded.", "Query arguments could not be decoded."), + (not_a_list(), "Workflow query [state] arguments must decode to an array.", "Query arguments must decode to an array."), + ] { + let (status, body) = request(&app, "POST", "/api/workflows/declared-query/query/state", json!({"input": input})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!((body["reason"].clone(), body["message"].clone(), body["validation_errors"].clone()), + (json!("invalid_query_arguments"), json!(message), json!({"arguments": [error]})), "{body}"); + assert_eq!(body["run_id"], started["run_id"], "{body}"); + } + assert!(poll(&app, "contract-worker", "query").await.is_null(), "no query task was created"); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 588dc7a8a..10120304e 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -16,6 +16,7 @@ import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; import {checkConcurrentWaits} from './concurrent-waits-contract.mjs'; import {checkContinueTransfers} from './continue-transfers-contract.mjs'; import {checkStartSearchAttributes} from './start-search-attributes-contract.mjs'; +import {checkQueryArgumentRouting} from './query-arguments-contract.mjs'; import {checkParentCloseCancellation} from './parent-close-cancellation-contract.mjs'; import {checkBridgeAdapters} from './bridge-adapters-contract.mjs'; import {checkNexusOperations} from './nexus-operations-contract.mjs'; @@ -174,6 +175,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const concurrentWaits = fixture.concurrent_waits ? checkConcurrentWaits(fixture, observation) : null; const continueTransfers = fixture.continue_transfers ? checkContinueTransfers(fixture, observation) : null; const startSearchAttributes = fixture.start_search_attributes ? checkStartSearchAttributes(fixture, observation) : null; + const queryArguments = fixture.query_argument_routing ? checkQueryArgumentRouting(fixture, observation) : null; const parentCloseCancellation = fixture.parent_close_cancellation ? checkParentCloseCancellation(fixture, observation) : null; const bridgeAdapters = fixture.bridge_adapters ? checkBridgeAdapters(fixture, observation, workflowId) : null; const nexusOperations = fixture.nexus_operations ? checkNexusOperations(fixture, observation, workflowId) : null; @@ -750,6 +752,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(concurrentWaits ? {concurrent_waits: concurrentWaits} : {}), ...(continueTransfers ? {continue_transfers: continueTransfers} : {}), ...(startSearchAttributes ? {start_search_attributes: startSearchAttributes} : {}), + ...(queryArguments ? {query_argument_routing: queryArguments} : {}), ...(parentCloseCancellation ? {parent_close_cancellation: parentCloseCancellation} : {}), ...(bridgeAdapters ? {bridge_adapters: bridgeAdapters} : {}), ...(nexusOperations ? {nexus_operations: nexusOperations} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 93a1e6921..94104423a 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -434,6 +434,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $concurrentWaits = isset($fixture['concurrent_waits']) ? finishHttpConcurrentWaits($client, $fixture, $workflowId, $url) : null; $continueTransfers = isset($fixture['continue_transfers']) ? finishHttpContinueTransfers($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; $parentCloseCancellation = isset($fixture['parent_close_cancellation']) ? finishHttpParentCloseCancellation($client, $fixture, $workflowId, $url) : null; $bridgeAdapters = isset($fixture['bridge_adapters']) ? finishHttpBridgeAdapters($client, $fixture, $workflowId, $url) : null; $nexusOperations = isset($fixture['nexus_operations']) ? finishHttpNexusOperations($client, $fixture, $workflowId, $url) : null; @@ -563,6 +564,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'concurrent_waits' => $concurrentWaits, 'continue_transfers' => $continueTransfers, 'start_search_attributes' => $startSearchAttributes, + 'query_argument_routing' => $queryArgumentRouting, 'parent_close_cancellation' => $parentCloseCancellation, 'bridge_adapters' => $bridgeAdapters, 'nexus_operations' => $nexusOperations, @@ -895,6 +897,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/concurrent-waits-probe.php'; require __DIR__.'/continue-transfers-probe.php'; require __DIR__.'/start-search-attributes-probe.php'; + require __DIR__.'/query-arguments-probe.php'; require __DIR__.'/parent-close-cancellation-probe.php'; require __DIR__.'/bridge-adapters-probe.php'; require __DIR__.'/nexus-operations-probe.php'; @@ -1087,6 +1090,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['start_search_attributes'])) { $observation['start_search_attributes'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_search_attribute_registry']; } + if ($mode === 'embedded' && isset($fixture['query_argument_routing'])) { + $observation['query_argument_routing'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; + } if ($mode === 'embedded' && isset($fixture['parent_close_cancellation'])) { $observation['parent_close_cancellation'] = ['applicable' => false, 'reason' => 'embedded_has_no_worker_protocol']; } diff --git a/scripts/conformance/server-parity/query-arguments-contract.mjs b/scripts/conformance/server-parity/query-arguments-contract.mjs new file mode 100644 index 000000000..9f8457a4e --- /dev/null +++ b/scripts/conformance/server-parity/query-arguments-contract.mjs @@ -0,0 +1,30 @@ +import assert from 'node:assert/strict'; + +// Reviewed query arguments the server cannot decode +// (query-arguments-probe.php), recorded against published Server 2.11.5: +// a run that declares no queries routes the query with its arguments as +// given, so its worker leases the undecodable or non-list blob and answers; +// a run that declares the query refuses those arguments before routing. +// The fixture's `expected` holds the whole observation. +export function checkQueryArgumentRouting(fixture, observation) { + const definition = fixture.query_argument_routing; + const state = observation.query_argument_routing; + const projection = {scope: 'query_argument_routing', runs: Object.keys(definition.expected).length}; + assert.ok(state && typeof state === 'object', 'actual query arguments observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}, + 'embedded query arguments explicitly inapplicable'); + return projection; + } + assert.equal(state.applicable, true, 'HTTP targets record query arguments'); + for (const [run, steps] of Object.entries(definition.expected)) { + assert.ok(state[run] && typeof state[run] === 'object', `query arguments ${run} recorded`); + for (const [step, expected] of Object.entries(steps)) { + assert.deepStrictEqual(state[run][step], expected, `query arguments ${run} ${step}`); + } + assert.deepStrictEqual(Object.keys(state[run]).sort(), Object.keys(steps).sort(), `query arguments ${run} steps`); + } + assert.deepStrictEqual(Object.keys(state).sort(), ['applicable', ...Object.keys(definition.expected)].sort(), + 'query arguments runs'); + return projection; +} diff --git a/scripts/conformance/server-parity/query-arguments-probe.php b/scripts/conformance/server-parity/query-arguments-probe.php new file mode 100644 index 000000000..1d2117d28 --- /dev/null +++ b/scripts/conformance/server-parity/query-arguments-probe.php @@ -0,0 +1,122 @@ +createNamespace($namespace, 'query arguments λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + // The SQLite reference shares its database with a one-second maintenance + // loop; retry its retryable lock-pressure refusals, as the other probes do. + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new QueryArgumentsHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $raw = static function (string $method, string $path, ?array $body = null) use ($http, $token, $namespace): array { + $response = $http->request($method, $path, ['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', + ]), ...($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']; + $query = $definition['query']; + $start = static function (string $suffix) use ($raw, $type, $workflowId, $definition): string { + $id = $workflowId.'-qa-'.$suffix; + $started = $raw('POST', 'workflows', ['workflow_id' => $id, 'workflow_type' => $type, 'task_queue' => $id.'-queue', + 'input' => ['codec' => 'avro', 'blob' => $definition['blobs']['empty']]]); + if ($started['status'] !== 201) { + throw new RuntimeException('Query arguments start: '.json_encode($started)); + } + + return $id; + }; + $ask = static function (string $id, string $blob) use ($raw, $query): array { + $answer = $raw('POST', 'workflows/'.rawurlencode($id).'/query/'.$query, ['input' => ['codec' => 'avro', 'blob' => $blob]]); + + return ['status' => $answer['status'], 'reason' => $answer['response']['reason'] ?? null, + 'message' => $answer['response']['message'] ?? null, + 'validation_errors' => $answer['response']['validation_errors'] ?? null]; + }; + + $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', + ]); + $register = static function (string $id, bool $contract) use ($raw, $type, $query): string { + $worker = $id.'-worker'; + $registration = ['worker_id' => $worker, 'task_queue' => $id.'-queue', 'runtime' => 'php', + 'sdk_version' => 'raw-http', 'supported_workflow_types' => [$type], 'supported_activity_types' => [], + 'capabilities' => ['query_tasks'], 'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', + 'sticky_execution'], ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http'])]; + if ($contract) { + $registration['workflow_command_contracts'] = [$type => ['signals' => [], 'updates' => [], 'queries' => [$query], + 'query_contracts' => [['name' => $query, 'parameters' => []]]]]; + } + $registered = $raw('POST', 'worker/register', $registration); + if ($registered['status'] !== 201) { + throw new RuntimeException('Query arguments registration: '.json_encode($registered)); + } + + return $worker; + }; + + $state = ['applicable' => true]; + // A query-capable worker that declares no queries: the start records no + // contract, so the server routes the query with its arguments as given. + // The client waits while the worker leases the query task and answers. + $uncontracted = $workflowId.'-qa-uncontracted'; + $worker = $register($uncontracted, false); + $start('uncontracted'); + foreach (['undecodable', 'not_a_list'] as $blob) { + $pending = startHeldRequest($url, $headers, '/api/workflows/'.rawurlencode($uncontracted).'/query/'.$query, + ['input' => ['codec' => 'avro', 'blob' => $definition['blobs'][$blob]]], 0.0); + $leased = null; + for ($attempt = 0; $attempt < 10 && $leased === null; $attempt++) { + $polled = $raw('POST', 'worker/query-tasks/poll', ['worker_id' => $worker, 'task_queue' => $uncontracted.'-queue', + 'timeout_seconds' => 2]); + $leased = $polled['response']['task'] ?? null; + } + $observed = ['leased' => is_array($leased), + 'arguments' => is_array($leased) ? ($leased['arguments'] ?? $leased['query_arguments'] ?? 'absent') : null]; + if (is_array($leased)) { + $observed['answered'] = $raw('POST', 'worker/query-tasks/'.$leased['query_task_id'].'/complete', [ + 'lease_owner' => $leased['lease_owner'], 'query_task_attempt' => $leased['query_task_attempt'], + 'result_envelope' => ['codec' => 'avro', 'blob' => $definition['blobs']['empty']]])['status']; + } + $answer = collectHeldRequest($pending); + $observed['client'] = ['status' => $answer['status'], 'reason' => $answer['response']['reason'] ?? null]; + $state['uncontracted'][$blob] = $observed; + } + // A worker that declares the query, so the start records its contract: + // the server decodes the arguments to check them, before any routing. + $contracted = $workflowId.'-qa-contracted'; + $register($contracted, true); + $start('contracted'); + foreach (['undecodable', 'not_a_list'] as $blob) { + $state['contracted'][$blob] = $ask($contracted, $definition['blobs'][$blob]); + } + + return $state; +} diff --git a/tests/Fixtures/ServerParity/query-arguments.json b/tests/Fixtures/ServerParity/query-arguments.json new file mode 100644 index 000000000..05d77e5b4 --- /dev/null +++ b/tests/Fixtures/ServerParity/query-arguments.json @@ -0,0 +1,94 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "query-arguments", + "review_state": "reviewed", + "contract": "Query arguments the server cannot decode (published Server 2.11.5 WorkflowController::query and WorkflowQueryTaskBroker::query with validateContractedQueryArguments), over raw HTTP on protocol 1.20 in a fresh namespace. The server decodes a query's arguments only to check them against a query contract the run's start recorded. A run whose query-capable worker declares no queries routes the query with its arguments as given: the worker leases a query task carrying the undecodable blob, or the blob of a single string, unchanged, answers it, and the waiting client receives 200. A run whose worker declared the query refuses the same arguments before routing with 422 invalid_query_arguments, the message 'Workflow query [state] arguments could not be decoded.' or 'Workflow query [state] arguments must decode to an array.', and validation_errors.arguments naming the cause; no query task is created. Embedded applications have no worker protocol.", + "workflow_type": "parity.v1.echo", + "input": { + "message": "query arguments λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "query arguments λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "query_argument_routing": { + "workflow_type": "parity.v1.query_arguments", + "query": "state", + "blobs": { + "empty": "wwHioz3/VYAiNwwA", + "undecodable": "AAAA", + "not_a_list": "wwHioz3/VYAiNwoUbm90IGEgbGlzdA==" + }, + "expected": { + "uncontracted": { + "undecodable": { + "leased": true, + "arguments": { + "codec": "avro", + "blob": "AAAA" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "not_a_list": { + "leased": true, + "arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoUbm90IGEgbGlzdA==" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + } + }, + "contracted": { + "undecodable": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] arguments could not be decoded.", + "validation_errors": { + "arguments": [ + "Query arguments could not be decoded." + ] + } + }, + "not_a_list": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] arguments must decode to an array.", + "validation_errors": { + "arguments": [ + "Query arguments must decode to an array." + ] + } + } + } + } + } +} diff --git a/tests/Unit/ServerParityQueryArgumentsContractTest.mjs b/tests/Unit/ServerParityQueryArgumentsContractTest.mjs new file mode 100644 index 000000000..2c0838ba5 --- /dev/null +++ b/tests/Unit/ServerParityQueryArgumentsContractTest.mjs @@ -0,0 +1,45 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkQueryArgumentRouting} from '../../scripts/conformance/server-parity/query-arguments-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/query-arguments.json', import.meta.url), 'utf8')); +const expected = fixture.query_argument_routing.expected; +const observation = () => structuredClone({mode: 'http', query_argument_routing: {applicable: true, ...expected}}); + +test('the published answers pass', () => { + assert.deepStrictEqual(checkQueryArgumentRouting(fixture, observation()), {scope: 'query_argument_routing', runs: 2}); +}); + +test('embedded applications have no worker protocol', () => { + assert.deepStrictEqual(checkQueryArgumentRouting(fixture, + {mode: 'embedded', query_argument_routing: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), + {scope: 'query_argument_routing', runs: 2}); + assert.throws(() => checkQueryArgumentRouting(fixture, {mode: 'embedded', query_argument_routing: {applicable: true}})); +}); + +test('an uncontracted run routes the blob; a declared query refuses it', () => { + for (const blob of ['undecodable', 'not_a_list']) { + const routed = expected.uncontracted[blob]; + assert.deepStrictEqual([routed.leased, routed.arguments.blob, routed.answered, routed.client.status], + [true, fixture.query_argument_routing.blobs[blob], 200, 200], blob); + const refused = expected.contracted[blob]; + assert.deepStrictEqual([refused.status, refused.reason], [422, 'invalid_query_arguments'], blob); + } +}); + +for (const [name, mutate] of [ + ['an undecodable blob refused before routing', state => { + state.uncontracted.undecodable = {leased: false, arguments: null, + client: {status: 422, reason: 'invalid_query_arguments'}}; + }], + ['a bare refusal without its message', state => { + state.contracted.undecodable.message = 'invalid_query_arguments'; + state.contracted.undecodable.validation_errors = null; + }], + ['a re-encoded blob', state => { state.uncontracted.not_a_list.arguments.blob = 'wwHioz3/VYAiNwwA'; }], + ['an unexpected run', state => { state.extra = {}; }], +]) test(`the contract rejects ${name}`, () => { + const raw = observation(); mutate(raw.query_argument_routing); + assert.throws(() => checkQueryArgumentRouting(fixture, raw)); +});