Rust #325: route query arguments the server cannot decode

.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

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 [<name>] 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`).

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

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

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<crate::codec::Value>),
+    Undecodable,
+    NotAList,
+}
+
+/// Decodes a query's arguments without refusing them.
+pub(super) async fn read_arguments(semaphore: Arc<Semaphore>, blob: String) -> Result<QueryArguments> {
+    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<crate::codec::Value>,
+        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<QueryAttempt> {

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;