diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 22dc04bbb..1b77ab365 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -61,6 +61,7 @@ on: - tests/Unit/ServerParityContinueTransfersContractTest.mjs - tests/Unit/ServerParityStartSearchAttributesContractTest.mjs - tests/Unit/ServerParityQueryArgumentsContractTest.mjs + - tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs - tests/Unit/ServerParityParentCloseCancellationContractTest.mjs - tests/Unit/ServerParityBridgeAdaptersContractTest.mjs - tests/Unit/ServerParityNexusOperationsContractTest.mjs @@ -1984,7 +1985,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/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/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-28-query-arguments.md b/docs/rust-server-port-checkpoints/2026-10-28-query-arguments.md index 6898dc2d9..e12edde1a 100644 --- a/docs/rust-server-port-checkpoints/2026-10-28-query-arguments.md +++ b/docs/rust-server-port-checkpoints/2026-10-28-query-arguments.md @@ -19,7 +19,7 @@ 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). + review); the query-contract-arguments slice closes it. - **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. diff --git a/docs/rust-server-port-checkpoints/2026-10-28-query-contract-arguments.md b/docs/rust-server-port-checkpoints/2026-10-28-query-contract-arguments.md new file mode 100644 index 000000000..d50d5af69 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-28-query-contract-arguments.md @@ -0,0 +1,95 @@ +# October 28, named and validated query arguments + +- **Before.** For a declared query with a parameter contract, native + differed from published PHP in four ways: + - it checked positional arguments only; + - it refused a named-argument map as "must decode to an array"; + - it answered every violation with a bare `invalid_query_arguments`; + - it refused some contracts as `unsupported_query_contract` (defaults, + and declared types other than a few scalars). + + The worker then received the arguments exactly as the client sent them. +- **Contract** (published 2.11.5, Core 2.15.1; + `WorkflowQueryTaskBroker::validateContractedQueryArguments`, + `WorkflowQueryContract::validatedArgumentsForRun`, + `normalizePositionalArguments`, `normalizeNamedArguments`, + `valueMatchesDeclaredType`): + - **Arguments.** An Avro array is a positional list. A non-empty map + with no key PHP casts to an integer is a PHP array of named arguments. + `Avro::fromDatum` returns any other map (empty, or with a canonical + decimal integer key) as an `AvroMapValue` object, which is not an array. + At the top level it is refused before normalization with "must decode to + an array" (found in review). + - **Positional.** Values fill the parameters in order. A variadic + parameter takes the rest. A missing argument takes its default or is + reported if required. Extra values are reported under `arguments`. + - **Named.** Each parameter takes its name's value. A variadic parameter + spreads a list or a map's values. Defaults and required arguments are + handled as for positional ones. Each unknown name is reported under + itself. + - **Values.** A null is refused unless the parameter allows it. A value + must match the declared type: nullable, union, intersection, `int`, + `float` (an int too), `string`, `bool`, `array`, `iterable`, `scalar`, + `true`, `false`, `null`, `mixed`. An adapter map is not an `array`. A + class or `object` type matches only an object: decoded bytes or an + adapter map. Neither object is `callable`. A named variadic spreads a + PHP array's values in wire order and keeps any other value, an adapter + map included, as one argument (found in review). + - **Refusal.** Any violation answers `422 invalid_query_arguments`, with + the message "Workflow query [] argument validation failed." and + `validation_errors` by argument. + - **Worker.** Otherwise the worker receives the normalized positional + list, encoded again with the request's codec. + - **No parameter contract.** A declared query without one takes a list as + given and refuses named arguments: "Named arguments require a durable + or loadable workflow query contract." (a native server never loads a + workflow class). +- **Fixture.** `query-contract-arguments`, raw HTTP on protocol 1.20 in a + fresh namespace. The worker declares `state(string $prefix)`, + `greet(string $prefix = 'dflt')`, `plain`, `arr(array $value)`, + `obj(object $value)` and `rest(mixed ...$value)`, and three `mixed` + queries whose registered defaults are raw JSON objects. There are + twenty-two cases: + - named, positional, default and named-default queries are routed, and + the worker's blob is checked; + - unknown, missing, wrong-type, null, too-many and named-without-contract + queries are refused, with their messages checked; + - an empty map, a `{"0": "ok"}` map and a mixed-key map are refused as not + arrays; + - a nested empty map is refused by `array`, but routed whole by `object` + and by the variadic `mixed`; + - `{value: {z: 20, a: 10}}` reaches the variadic `rest` as `[20, 10]`; + - a registered default `{"0": "x", "1": "y"}`, sent as a JSON object + (the probe records the bytes it sends), reaches the worker as the list + `[["x", "y"]]`. On the previous source native routed it as a map; + - a registered default `{"0": "x", "k": "y"}` or `{"1": "x"}` fails + published PHP's Avro encoder (`invalid_map_key`) with an uncaught 500. + The fixture pins only that status. +- **Native.** `query_contract.rs` ports the normalization and type rules + over `codec::OrderedValue`. Apache's generic reader keeps maps unordered, + so the new `codec/wire_order.rs` reads the bytes the public reader has + already validated again through a private binding schema. That schema + spells each map as an array of key/value records, as PHP's + `ValueDatumWriter` does, so the binary layout is the same and Apache still + reads every primitive and block. Writing goes through the same binding, + under the public header, and the public reader checks the result. For every + fixture blob the bytes match PHP's. + A registered default takes PHP's array shape: canonical integer keys + 0 to n-1 make a list, string keys a map, and any other integer key is + answered 500 `query_default_unencodable`, naming the parameter. + Violations answer first, as PHP encodes only valid arguments. + `check_query_contract` returns the normalized list for a declared query, + or the refusal. `query_attempt` enqueues the re-encoded blob. +- **Kept.** + - A registered default map keeps its keys in sorted order, not the + registered order: native's JSON objects do not preserve order + (`serde_json` without `preserve_order`). The same limit makes + `{"1": "a", "0": "b"}` a list on native, where PHP's array is not a list + and fails to encode. + - `callable` matches nothing here: PHP's `is_callable` also accepts + strings naming functions, which a native server cannot resolve. + - A class name that a published PHP server can load (a real class) would + refuse other objects. Native, loading no classes, accepts any object. +- **Tests.** `query_contract_arguments.rs` (routed normalization; refusals + with messages) fails on the previous source. `query_contract.rs` unit + tests cover the type rules and integer-keyed maps. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index b40ae76cc..2e574aa10 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -275,20 +275,6 @@ 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. -- **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 @@ -552,6 +538,29 @@ contract fixture before its correction. 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. +- **Named and validated query arguments** (October 28, closed). For a + declared query, published PHP matches the arguments to the query's durable + parameter contract (`WorkflowQueryContract::validatedArgumentsForRun`): + - a string-keyed map is taken as named arguments; + - positional arguments are matched in order; + - defaults are filled in and variadics collected; + - unknown names, missing required arguments, nulls, wrong types and extra + arguments are each reported. Declared types may be nullable, unions, + intersections, scalars, `array`, `object` or class names. + + A violation answers `422 invalid_query_arguments`, "argument validation + failed", with per-argument messages in `validation_errors`. The worker + receives the normalized positional list, encoded again. A declared query + without a parameter contract takes a list as given and refuses named + arguments. Native now does the same (`query_contract.rs`). It used to + check positional arguments only, refuse a map, report a violation as a + bare `invalid_query_arguments`, and refuse some contracts as unsupported. + `Avro::fromDatum` returns an empty or integer-keyed map as an object, not + an array, and a named variadic map spreads in wire order. A registered + JSON default takes PHP's array shape. The `query-contract-arguments` + fixture records twenty-two cases against + published 2.11.5 + ([checkpoint](rust-server-port-checkpoints/2026-10-28-query-contract-arguments.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 @@ -647,6 +656,18 @@ contract fixture before its correction. - **Command route names.** Published PHP records each request's route name in its command context, and its unnamed workflow routes get a name generated per deployment. Native records names only for the named schedule routes. +- **Registered default key order.** A worker registers query parameter + defaults as JSON. PHP keeps each object's keys in the registered order, + and a default map reaches the worker in that order. Native's JSON objects + keep keys sorted (`serde_json` without `preserve_order`), so: + - a named default map such as `{"z": 20, "a": 10}` reaches the worker in + sorted order; + - `{"1": "a", "0": "b"}` becomes a list on native, where PHP's array is + not a list and its encoder fails with a 500. + + Both need registered-order JSON in native contracts and a fixture case + each; this remains open #325 parity work + ([checkpoint](rust-server-port-checkpoints/2026-10-28-query-contract-arguments.md)). ## Contract and verification inventory diff --git a/rust/src/codec.rs b/rust/src/codec.rs index c32bcc913..cbab8d3d2 100644 --- a/rust/src/codec.rs +++ b/rust/src/codec.rs @@ -6,6 +6,8 @@ use std::collections::BTreeMap; use std::io::{self, Read}; mod ordered; mod validation; +mod wire_order; +pub use wire_order::OrderedValue; use apache_avro::{ GenericSingleObjectReader, GenericSingleObjectWriter, Schema, types::Value as Datum, diff --git a/rust/src/codec/wire_order.rs b/rust/src/codec/wire_order.rs new file mode 100644 index 000000000..2d6265b95 --- /dev/null +++ b/rust/src/codec/wire_order.rs @@ -0,0 +1,216 @@ +//! Protocol values whose map entries keep the order the wire carried them. +//! PHP decodes an Avro map into an array in wire order, so a decoded +//! argument list can depend on it (a variadic named argument spreads a map's +//! values in that order). Apache's generic reader keeps maps in a hash map, +//! so this path reads and writes through a private binding schema that spells +//! each map as an array of key/value records, as PHP's `ValueDatumWriter` +//! does. The binary layout is identical; Apache still owns every primitive, +//! block and union, and the public schema still validates and frames. +use super::{CodecError, CompleteInput, VALUE_FINGERPRINT, VALUE_SCHEMA, ValueCodec, php_base64}; +use apache_avro::{ + Schema, reader::datum::GenericDatumReader, types::Value as Datum, writer::datum::GenericDatumWriter, +}; +use base64::{Engine, engine::general_purpose::STANDARD}; +use std::{collections::HashMap, sync::LazyLock}; + +/// The single-object marker; the public fingerprint follows it. +const MARKER: [u8; 2] = [0xc3, 0x01]; +const PUBLIC_MAP: &str = r#"{"type":"map","values":"Value"}"#; +const ENTRY_ARRAY: &str = r#"{"type":"array","items":{"type":"record","name":"MapEntry","fields":[{"name":"key","type":"string"},{"name":"value","type":"Value"}]}}"#; + +/// The public Value schema with its map spelled as an array of entries. +static BINDING: LazyLock = LazyLock::new(|| { + assert_eq!(VALUE_SCHEMA.matches(PUBLIC_MAP).count(), 1, "one map in the protocol Value schema"); + Schema::parse_str(&VALUE_SCHEMA.replace(PUBLIC_MAP, ENTRY_ARRAY)).expect("Value binding schema") +}); + +/// A protocol Value with map entries in wire order. A repeated key keeps its +/// first position and its last value, as a PHP array does. +#[derive(Clone, Debug, PartialEq)] +pub enum OrderedValue { + Null, + Boolean(bool), + Long(i64), + Double(f64), + Bytes(Vec), + String(String), + Array(Vec), + Map(Vec<(String, OrderedValue)>), +} + +impl ValueCodec { + /// Decodes a blob as `decode` does, keeping map entries in wire order. + pub fn decode_ordered(&self, blob: &str) -> Result { + // The public reader checks framing, the fingerprint, the schema, + // truncation and trailing bytes first; the binding pass then reads + // the same datum again only to recover the order. + self.decode(blob)?; + let bytes = php_base64(blob)?; + let datum = bytes.get(MARKER.len() + VALUE_FINGERPRINT.len()..).ok_or(CodecError::Truncated)?; + let mut input = CompleteInput { remaining: datum, truncated: false }; + let value = GenericDatumReader::builder(&BINDING).build()?.read_value(&mut input); + if input.truncated { + return Err(CodecError::Truncated); + } + let value = value?; + if !input.remaining.is_empty() { + return Err(CodecError::TrailingBytes); + } + from_binding(value) + } + + /// Encodes a value with its map entries in the order given, under the + /// public single-object header. + pub fn encode_ordered_value(&self, value: &OrderedValue) -> Result { + let datum = GenericDatumWriter::builder(&BINDING).build()?.write_value_to_vec(to_binding(value)?)?; + let mut bytes = Vec::with_capacity(MARKER.len() + VALUE_FINGERPRINT.len() + datum.len()); + bytes.extend_from_slice(&MARKER); + bytes.extend_from_slice(&VALUE_FINGERPRINT); + bytes.extend_from_slice(&datum); + let blob = STANDARD.encode(bytes); + // The public reader must accept what the binding wrote. + self.decode(&blob)?; + Ok(blob) + } +} + +fn record(field: &str, value: Datum) -> Datum { + Datum::Record(vec![(field.into(), value)]) +} + +fn to_binding(value: &OrderedValue) -> Result { + let (index, branch) = match value { + OrderedValue::Null => (0, Datum::Null), + OrderedValue::Boolean(flag) => (1, record("boolean", Datum::Boolean(*flag))), + OrderedValue::Long(number) => (2, record("long", Datum::Long(*number))), + OrderedValue::Double(number) if !number.is_finite() => return Err(CodecError::NonFiniteDouble), + OrderedValue::Double(number) => (3, record("double", Datum::Double(*number))), + OrderedValue::Bytes(bytes) => (4, record("bytes", Datum::Bytes(bytes.clone()))), + OrderedValue::String(text) => (5, record("string", Datum::String(text.clone()))), + OrderedValue::Array(items) => { + (6, record("items", Datum::Array(items.iter().map(to_binding).collect::>()?))) + } + OrderedValue::Map(entries) => { + let entries = entries + .iter() + .map(|(key, value)| { + Ok(Datum::Record(vec![("key".into(), Datum::String(key.clone())), ("value".into(), to_binding(value)?)])) + }) + .collect::>()?; + (7, record("entries", Datum::Array(entries))) + } + }; + Ok(record("value", Datum::Union(index, Box::new(branch)))) +} + +/// The one field of a record the binding schema read. +fn field(datum: Datum, name: &str) -> Result { + match datum { + Datum::Record(fields) => fields.into_iter().find(|(field, _)| field == name).map(|(_, value)| value), + _ => None, + } + .ok_or(CodecError::InvalidValue) +} + +fn from_binding(datum: Datum) -> Result { + let Datum::Union(index, branch) = field(datum, "value")? else { + return Err(CodecError::InvalidValue); + }; + Ok(match (index, *branch) { + (0, Datum::Null) => OrderedValue::Null, + (1, branch) => match field(branch, "boolean")? { + Datum::Boolean(flag) => OrderedValue::Boolean(flag), + _ => return Err(CodecError::InvalidValue), + }, + (2, branch) => match field(branch, "long")? { + Datum::Long(number) => OrderedValue::Long(number), + _ => return Err(CodecError::InvalidValue), + }, + (3, branch) => match field(branch, "double")? { + Datum::Double(number) => OrderedValue::Double(number), + _ => return Err(CodecError::InvalidValue), + }, + (4, branch) => match field(branch, "bytes")? { + Datum::Bytes(bytes) => OrderedValue::Bytes(bytes), + _ => return Err(CodecError::InvalidValue), + }, + (5, branch) => match field(branch, "string")? { + Datum::String(text) => OrderedValue::String(text), + _ => return Err(CodecError::InvalidValue), + }, + (6, branch) => match field(branch, "items")? { + Datum::Array(items) => OrderedValue::Array(items.into_iter().map(from_binding).collect::>()?), + _ => return Err(CodecError::InvalidValue), + }, + (7, branch) => match field(branch, "entries")? { + Datum::Array(records) => { + let mut entries: Vec<(String, OrderedValue)> = Vec::with_capacity(records.len()); + let mut slots: HashMap = HashMap::with_capacity(records.len()); + for entry in records { + let Datum::Record(fields) = entry else { + return Err(CodecError::InvalidValue); + }; + let mut fields = fields.into_iter(); + let (Some((_, Datum::String(key))), Some((_, value)), None) = (fields.next(), fields.next(), fields.next()) else { + return Err(CodecError::InvalidValue); + }; + let value = from_binding(value)?; + match slots.get(&key) { + Some(&slot) => entries[slot].1 = value, + None => { + slots.insert(key.clone(), entries.len()); + entries.push((key, value)); + } + } + } + OrderedValue::Map(entries) + } + _ => return Err(CodecError::InvalidValue), + }, + _ => return Err(CodecError::InvalidValue), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn map_entries_keep_their_wire_order_both_ways() { + let codec = ValueCodec::new().unwrap(); + // PHP's encoding of ['rest' => ['z' => 20, 'a' => 10]]. + let value = OrderedValue::Map(vec![("rest".into(), + OrderedValue::Map(vec![("z".into(), OrderedValue::Long(20)), ("a".into(), OrderedValue::Long(10))]))]); + let blob = codec.encode_ordered_value(&value).unwrap(); + assert_eq!(codec.decode_ordered(&blob).unwrap(), value); + // The schema-aware reader accepts the same bytes. + assert!(codec.decode(&blob).is_ok()); + } + + #[test] + fn the_writer_matches_the_published_bytes_for_every_branch() { + let codec = ValueCodec::new().unwrap(); + for blob in [ + "wwHioz3/VYAiNw4CDHByZWZpeAoEb2sA", + "wwHioz3/VYAiNwwCCgRvawA=", + "wwHioz3/VYAiNwwA", + "wwHioz3/VYAiNw4A", + "wwHioz3/VYAiNw4EAjAKAmEMcHJlZml4CgJiAA==", + "wwHioz3/VYAiNw4CCnZhbHVlDgAA", + "wwHioz3/VYAiNw4MDGJpbmFyeQgIc2FtZQxkb3VibGUGAAAAAAAAHEAcaW52YWxpZF9iaW5hcnkIBP8ACGxvbmcEDgxuZXN0ZWQOBAphbHBoYQQCCGJldGEEBAAIdGV4dAoIc2FtZQA=", + ] { + let value = codec.decode_ordered(blob).unwrap(); + assert_eq!(codec.encode_ordered_value(&value).unwrap(), blob, "{blob}"); + } + assert!(codec.decode_ordered("AAAA").is_err()); + } + + #[test] + fn a_repeated_key_keeps_its_first_position_and_last_value() { + let codec = ValueCodec::new().unwrap(); + let long = |number| OrderedValue::Long(number); + let written = OrderedValue::Map(vec![("a".into(), long(1)), ("b".into(), long(2)), ("a".into(), long(3))]); + let blob = codec.encode_ordered_value(&written).unwrap(); + assert_eq!(codec.decode_ordered(&blob).unwrap(), OrderedValue::Map(vec![("a".into(), long(3)), ("b".into(), long(2))])); + } +} diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 916db144d..1af7651f5 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -99,6 +99,7 @@ mod previews; mod prometheus_metrics; mod queries; mod query_authority; +mod query_contract; mod repair_candidates; mod repair_pass; mod retention; diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 4db17d405..279a681f9 100644 --- a/rust/src/runtime/queries.rs +++ b/rust/src/runtime/queries.rs @@ -7,7 +7,7 @@ use super::{ envelope, query_authority::{self as authority, QueryTaskConfig}, refuse, - store::{Store, validate_arguments}, + store::Store, text, wire, }; use axum::http::StatusCode; @@ -131,12 +131,11 @@ pub(super) async fn decode( /// 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. +/// arguments as given. A string-keyed map decodes to a PHP array too, of +/// named arguments (`WorkflowQueryContract::normalizeNamedArguments`); only a +/// scalar is not an array. pub(super) enum QueryArguments { - Decoded(Vec), + Decoded(super::query_contract::Arguments), Undecodable, NotAList, } @@ -152,9 +151,9 @@ pub(super) async fn read_arguments(semaphore: Arc, blob: String) -> R 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, + match codec.decode_ordered(&blob) { + Ok(value) => super::query_contract::Arguments::from_value(value) + .map_or(QueryArguments::NotAList, QueryArguments::Decoded), Err(_) => QueryArguments::Undecodable, } }) @@ -213,54 +212,63 @@ fn admits_queries(status: &str) -> bool { } /// `validateContractedQueryArguments`: a run that declares no queries leaves -/// 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. +/// the name and arguments to the worker (`None`). A declared query's +/// arguments must decode to a PHP array; `WorkflowQueryContract` then matches +/// them to the query's parameter contract, and the worker receives the +/// normalized positional list. fn check_query_contract( contract: &Value, workflow_id: &str, run_id: &str, name: &str, arguments: &QueryArguments, -) -> Result<()> { +) -> Result>> { let declared = |field: &str| contract[field].as_array().is_some_and(|items| !items.is_empty()); if !declared("declared_queries") && !declared("declared_query_contracts") { - return Ok(()); + return Ok(None); } if !contract["declared_queries"].as_array().is_some_and(|names| names.iter().any(|v| v == name)) { 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 refused = |message: String, errors: Value| { + let mut refusal = query_refusal(StatusCode::UNPROCESSABLE_ENTITY, workflow_id, run_id, name, "invalid_query_arguments", &message); + if let super::RuntimeError::Protocol { response, .. } = &mut refusal { + response["validation_errors"] = errors; + } + refusal + }; + let arguments = match arguments { + QueryArguments::Decoded(arguments) => arguments, + QueryArguments::Undecodable => return Err(refused(format!("Workflow query [{name}] arguments could not be decoded."), + json!({"arguments": ["Query arguments could not be decoded."]}))), + QueryArguments::NotAList => return Err(refused(format!("Workflow query [{name}] arguments must decode to an array."), + json!({"arguments": ["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]}); + let parameters = contract["declared_query_contracts"] + .as_array() + .and_then(|contracts| contracts.iter().find(|entry| entry["name"] == name)); + match super::query_contract::validate(parameters, name, arguments.clone()) { + Ok(normalized) => Ok(Some(normalized)), + Err(super::query_contract::Refusal::Invalid(errors)) => Err(refused( + format!("Workflow query [{name}] argument validation failed."), super::query_contract::errors_json(&errors))), + // Published PHP fails here with an uncaught Avro `invalid_map_key` + // (500). Native keeps the status and says which default failed. + Err(super::query_contract::Refusal::UnencodableDefault(parameter)) => Err(query_refusal( + StatusCode::INTERNAL_SERVER_ERROR, workflow_id, run_id, name, "query_default_unencodable", + &format!("Workflow query [{name}] parameter [{parameter}] has a registered default with an integer key outside a list, which the Avro Value codec cannot encode."))), } - 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)) - { - validate_arguments( - contract, - "declared_query_contracts", - name, - values, - "unsupported_query_contract", - "invalid_query_arguments", - )?; +/// The arguments a routed query's task carries: a declared query's +/// normalized list encoded again, or the request's arguments as given. +fn routed_arguments(normalized: Option>, arguments: &str) -> Result { + match normalized { + Some(values) => crate::codec::ValueCodec::new() + .and_then(|codec| codec.encode_ordered_value(&crate::codec::OrderedValue::Array(values))) + .map_err(|_| refuse(StatusCode::SERVICE_UNAVAILABLE, "query_codec_unavailable")), + None => Ok(arguments.to_owned()), } - Ok(()) } impl Store @@ -403,7 +411,10 @@ where } let contract = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='WorkflowStarted' ORDER BY sequence LIMIT 1") .bind(&run_id).fetch_one(&self.pool).await?; - check_query_contract(&DB::document_row(&contract, "payload")?, workflow_id, &run_id, name, values)?; + // The worker receives a declared query's normalized arguments, encoded + // again (`Serializer::serializeWithCodec`). + let normalized = check_query_contract(&DB::document_row(&contract, "payload")?, workflow_id, &run_id, name, values)?; + let arguments = routed_arguments(normalized, arguments)?; let route = self.query_route(&run, config.worker_stale_after_seconds).await?; if !route.servable { return Ok(QueryAttempt::Unroutable { run_id, route }); diff --git a/rust/src/runtime/query_contract.rs b/rust/src/runtime/query_contract.rs new file mode 100644 index 000000000..d7b3deae6 --- /dev/null +++ b/rust/src/runtime/query_contract.rs @@ -0,0 +1,473 @@ +//! Workflow 2.15.1 `WorkflowQueryContract::validatedArgumentsForRun`: a +//! declared query's arguments, positional or named, matched to its durable +//! parameter contract. The broker sends the worker the normalized positional +//! list and refuses the query with every violation it finds. +use crate::codec::OrderedValue as Value; +use serde_json::{Map, Value as Json}; + +/// A query's decoded arguments, as PHP holds them after `Avro::fromDatum`: +/// a list, or a string-keyed map of named arguments. +#[derive(Clone)] +pub(super) enum Arguments { + Positional(Vec), + Named(Vec<(String, Value)>), +} + +impl Arguments { + /// An Avro array is a list. A map is a PHP array of named arguments only + /// when `Avro::fromDatum` makes it one: non-empty, with no key PHP casts + /// to an integer. Any other map arrives as an `AvroMapValue` object, + /// which is not an array at all. + pub(super) fn from_value(value: Value) -> Option { + match value { + Value::Array(values) => Some(Self::Positional(values)), + Value::Map(entries) if !is_adapter(&entries) => Some(Self::Named(entries)), + _ => None, + } + } +} + +/// Whether `Avro::fromDatum` returns this map as an `AvroMapValue` object: +/// it is empty, or one of its keys is a canonical decimal integer, which a +/// PHP array would turn into an integer key. +fn is_adapter(entries: &[(String, Value)]) -> bool { + entries.is_empty() || entries.iter().any(|(key, _)| integer_key(key)) +} + +/// PHP's integer array keys: `0`, or an optional minus sign and a digit +/// string without a leading zero that fits in a 64-bit integer. +fn integer_key(key: &str) -> bool { + let digits = key.strip_prefix('-').unwrap_or(key); + let canonical = key == "0" + || (!digits.is_empty() && !digits.starts_with('0') && digits.bytes().all(|byte| byte.is_ascii_digit())); + canonical && key.parse::().is_ok() +} + +/// Whether PHP holds this decoded value as an array. +fn is_array(value: &Value) -> bool { + match value { + Value::Array(_) => true, + Value::Map(entries) => !is_adapter(entries), + _ => false, + } +} + +/// Whether PHP holds this decoded value as an object: bytes arrive as an +/// `AvroBinaryValue`, and an adapter map as an `AvroMapValue`. +fn is_object(value: &Value) -> bool { + match value { + Value::Bytes(_) => true, + Value::Map(entries) => is_adapter(entries), + _ => false, + } +} + +/// The validated arguments, or the violations by argument name, in the +/// order PHP reports them. +pub(super) type Validated = Result, Refusal>; + +/// Why a declared query's arguments are not routed. +#[derive(Debug, PartialEq)] +pub(super) enum Refusal { + /// The violations by argument, as PHP's `validation_errors`. + Invalid(Vec<(String, Vec)>), + /// A missing argument's registered default that PHP's Avro encoder + /// cannot write: an array with an integer key that is not a list. PHP + /// fails with an uncaught `invalid_map_key` (500). + UnencodableDefault(String), +} + +/// `validatedArgumentsForRun` for a run whose durable contract may hold this +/// query's parameter contract. A native server never loads a workflow class, +/// so a query without one takes a list as given and refuses named arguments. +pub(super) fn validate(contract: Option<&Json>, query: &str, arguments: Arguments) -> Validated { + let Some(contract) = contract else { + return match arguments { + Arguments::Positional(values) => Ok(values), + Arguments::Named(_) => Err(Refusal::Invalid(vec![("arguments".to_owned(), + vec!["Named arguments require a durable or loadable workflow query contract.".to_owned()])])), + }; + }; + let parameters = contract["parameters"].as_array().cloned().unwrap_or_default(); + let mut errors = Errors::default(); + let normalized = match arguments { + Arguments::Positional(values) => positional(¶meters, query, &values, &mut errors), + Arguments::Named(entries) => named(¶meters, &entries, &mut errors), + }; + errors.into_result(normalized) +} + +/// `normalizePositionalArguments`. +fn positional(parameters: &[Json], query: &str, values: &[Value], errors: &mut Errors) -> Vec { + let mut normalized = Vec::new(); + let mut consumed = 0; + for parameter in parameters { + let take = if parameter["variadic"] == true { values.len() - consumed } else { usize::from(consumed < values.len()) }; + if take > 0 { + for value in &values[consumed..consumed + take] { + errors.check(parameter, value); + normalized.push(value.clone()); + } + consumed += take; + } else if parameter["variadic"] != true { + missing(parameter, &mut normalized, errors); + } + } + if consumed < values.len() { + errors.push("arguments", format!("Too many arguments were provided for query [{query}].")); + } + normalized +} + +/// `normalizeNamedArguments`. +fn named(parameters: &[Json], entries: &[(String, Value)], errors: &mut Errors) -> Vec { + let mut normalized = Vec::new(); + let value_of = |name: &str| entries.iter().find(|(key, _)| key == name).map(|(_, value)| value); + for parameter in parameters { + let name = parameter["name"].as_str().unwrap_or_default(); + match (value_of(name), parameter["variadic"] == true) { + (Some(value), true) => { + for value in spread(value) { + errors.check(parameter, value); + normalized.push(value.clone()); + } + } + (Some(value), false) => { + errors.check(parameter, value); + normalized.push(value.clone()); + } + (None, true) => {} + (None, false) => missing(parameter, &mut normalized, errors), + } + } + for (key, _) in entries { + if !parameters.iter().any(|parameter| parameter["name"] == key.as_str()) { + errors.push(key, format!("Unknown argument [{key}].")); + } + } + normalized +} + +/// A variadic named value spreads a PHP array's values; any other value, +/// an object included, is one argument. +fn spread(value: &Value) -> Vec<&Value> { + match value { + Value::Array(values) => values.iter().collect(), + Value::Map(values) if !is_adapter(values) => values.iter().map(|(_, value)| value).collect(), + value => vec![value], + } +} + +/// A missing argument takes its default, or is reported if required. +fn missing(parameter: &Json, normalized: &mut Vec, errors: &mut Errors) { + if parameter["default_available"] == true { + match from_json(¶meter["default"]) { + Some(value) => normalized.push(value), + None => { + errors.unencodable.get_or_insert_with(|| parameter["name"].as_str().unwrap_or_default().to_owned()); + } + } + } else if parameter["required"] == true { + errors.required(parameter); + } +} + +/// The violations by argument name, keeping PHP's first-seen key order. +#[derive(Default)] +struct Errors { + found: Vec<(String, Vec)>, + /// The first parameter whose default PHP cannot encode. + unencodable: Option, +} + +impl Errors { + fn push(&mut self, key: &str, message: String) { + match self.found.iter_mut().find(|(name, _)| name == key) { + Some((_, messages)) => messages.push(message), + None => self.found.push((key.to_owned(), vec![message])), + } + } + + fn required(&mut self, parameter: &Json) { + let name = parameter["name"].as_str().unwrap_or_default(); + self.push(name, format!("The {name} argument is required.")); + } + + /// `appendParameterValidationErrors`. + fn check(&mut self, parameter: &Json, value: &Value) { + let name = parameter["name"].as_str().unwrap_or("argument"); + if let Some(message) = parameter_error(parameter, name, value) { + self.push(name, message); + } + } + + /// Violations answer first; PHP encodes the arguments only after they + /// validate. + fn into_result(self, normalized: Vec) -> Validated { + match (self.found.is_empty(), self.unencodable) { + (false, _) => Err(Refusal::Invalid(self.found)), + (true, Some(parameter)) => Err(Refusal::UnencodableDefault(parameter)), + (true, None) => Ok(normalized), + } + } +} + +/// `validationErrorsForParameterValue`. +fn parameter_error(parameter: &Json, name: &str, value: &Value) -> Option { + let declared = parameter["type"].as_str().map(str::trim); + if matches!(value, Value::Null) { + let allows = match parameter["allows_null"].as_bool() { + Some(allows) => allows, + None => declared.is_none_or(|kind| { + kind.is_empty() || kind.starts_with('?') || split(kind, '|').iter().any(|part| *part == "null") + }), + }; + return (!allows).then(|| format!("The {name} argument cannot be null.")); + } + let kind = declared.filter(|kind| !kind.is_empty() && *kind != "mixed")?; + (!matches_type(value, kind)).then(|| format!("The {name} argument must be of type {kind}.")) +} + +/// `splitDeclaredType`: the parts of a union or intersection, outside +/// parentheses. +fn split(kind: &str, delimiter: char) -> Vec<&str> { + let mut parts = Vec::new(); + let mut depth = 0i32; + let mut start = 0; + for (index, character) in kind.char_indices() { + match character { + '(' => depth += 1, + ')' => depth -= 1, + _ if character == delimiter && depth == 0 => { + parts.push(kind[start..index].trim()); + start = index + 1; + } + _ => {} + } + } + parts.push(kind[start..].trim()); + parts.into_iter().filter(|part| !part.is_empty()).collect() +} + +/// `valueMatchesDeclaredType` over the decoded Avro values PHP sees: bytes +/// and adapter maps arrive as objects, every other value as a PHP scalar or +/// array. +fn matches_type(value: &Value, kind: &str) -> bool { + let kind = kind.trim(); + if kind.is_empty() || kind == "mixed" { + return true; + } + if let Some(inner) = kind.strip_prefix('?') { + return matches!(value, Value::Null) || matches_type(value, inner); + } + let union = split(kind, '|'); + if union.len() > 1 { + return union.iter().any(|part| matches_type(value, part)); + } + let intersection = split(kind, '&'); + if intersection.len() > 1 { + return intersection.iter().all(|part| matches_type(value, part)); + } + let kind = kind.trim_matches(|character: char| character == '(' || character == ')' || character.is_whitespace() || character == '\0'); + let object = is_object(value); + match kind { + "int" => matches!(value, Value::Long(_)), + "float" => matches!(value, Value::Double(_) | Value::Long(_)), + "string" => matches!(value, Value::String(_)), + "bool" => matches!(value, Value::Boolean(_)), + "array" | "iterable" => is_array(value), + "scalar" => matches!(value, Value::Long(_) | Value::Double(_) | Value::String(_) | Value::Boolean(_)), + "true" => matches!(value, Value::Boolean(true)), + "false" => matches!(value, Value::Boolean(false)), + "null" => matches!(value, Value::Null), + "mixed" => true, + "never" | "void" => false, + // Neither decoded object (`AvroBinaryValue`, `AvroMapValue`) is + // invokable. A string naming a PHP function would be callable, but a + // native server has no PHP function table. + "callable" => false, + // A named class: the server loads no class, so any object passes + // (`! class_exists($type)`), as do `object` and the relative types. + _ => object, + } +} + +/// A registered JSON default as the Avro value PHP's serializer encodes, or +/// `None` where it cannot. PHP decodes the registration into arrays whose +/// canonical integer keys are integers. A list (keys 0 to n-1, including an +/// empty object) is an Avro array; an array with only string keys is a map; +/// an integer key in any other array fails PHP's encoder. +/// +/// Native's JSON objects keep their keys sorted, not in the registered +/// order, so integer keys are read in numeric order and a map's entries in +/// key order. +fn from_json(value: &Json) -> Option { + Some(match value { + Json::Null => Value::Null, + Json::Bool(flag) => Value::Boolean(*flag), + Json::Number(number) => number.as_i64().map(Value::Long).unwrap_or_else(|| Value::Double(number.as_f64().unwrap_or_default())), + Json::String(text) => Value::String(text.clone()), + Json::Array(items) => Value::Array(items.iter().map(from_json).collect::>()?), + Json::Object(entries) => { + let integers: Option> = entries.iter() + .map(|(key, value)| integer_key(key).then(|| (key.parse::().ok().map(|key| (key, value)))).flatten()) + .collect(); + match integers { + Some(mut items) => { + items.sort_by_key(|(key, _)| *key); + if !items.iter().enumerate().all(|(position, (key, _))| i64::try_from(position) == Ok(*key)) { + return None; + } + Value::Array(items.into_iter().map(|(_, value)| from_json(value)).collect::>()?) + } + None if entries.keys().any(|key| integer_key(key)) => return None, + None => Value::Map(entries.iter().map(|(key, value)| Some((key.clone(), from_json(value)?))).collect::>()?), + } + } + }) +} + +/// The violations as PHP's `validation_errors` object. +pub(super) fn errors_json(errors: &[(String, Vec)]) -> Json { + let mut object = Map::new(); + for (key, messages) in errors { + object.insert(key.clone(), Json::from(messages.clone())); + } + Json::Object(object) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn invalid(validated: Validated) -> Option)>> { + match validated { + Err(Refusal::Invalid(errors)) => Some(errors), + _ => None, + } + } + + #[test] + fn a_registered_default_takes_phps_array_shape() { + let with_default = |default: Json| json!({"name":"q","parameters":[{"name":"value","type":"mixed","required":false, + "variadic":false,"default_available":true,"allows_null":false,"default":default}]}); + let filled = |default: Json| validate(Some(&with_default(default)), "q", Arguments::Positional(vec![])); + let strings = |items: &[&str]| Value::Array(items.iter().map(|item| Value::String((*item).into())).collect()); + // A list, in numeric key order even past ten keys; an empty object. + assert_eq!(filled(json!({"0":"x","1":"y"})), Ok(vec![strings(&["x", "y"])])); + let eleven: serde_json::Map = (0..11).map(|key| (key.to_string(), json!(key.to_string()))).collect(); + let expected: Vec = (0..11).map(|key| key.to_string()).collect(); + assert_eq!(filled(Json::Object(eleven)), Ok(vec![strings(&expected.iter().map(String::as_str).collect::>())])); + assert_eq!(filled(json!({})), Ok(vec![Value::Array(vec![])])); + assert_eq!(filled(json!({"a":{"0":1}})), Ok(vec![Value::Map(vec![("a".into(), Value::Array(vec![Value::Long(1)]))])])); + // An integer key outside a list fails PHP's encoder, nested or not. + for default in [json!({"0":"x","k":"y"}), json!({"1":"x"}), json!({"a":{"1":"x"}}), json!([{"0":"x","2":"y"}])] { + assert_eq!(filled(default.clone()), Err(Refusal::UnencodableDefault("value".into())), "{default}"); + } + // Violations answer before the encoder runs. + let mut both = with_default(json!({"1":"x"})); + both["parameters"].as_array_mut().unwrap().push(json!({"name":"other","type":"int","required":true,"variadic":false, + "default_available":false,"allows_null":false})); + assert_eq!(invalid(validate(Some(&both), "q", Arguments::Positional(vec![]))), + Some(vec![("other".to_owned(), vec!["The other argument is required.".to_owned()])])); + } + + fn state() -> Json { + json!({"name":"state","parameters":[{"name":"prefix","type":"string","required":true,"variadic":false, + "default_available":false,"allows_null":false}]}) + } + + fn named(pairs: &[(&str, Value)]) -> Arguments { + Arguments::Named(pairs.iter().map(|(key, value)| ((*key).to_owned(), value.clone())).collect()) + } + + #[test] + fn named_and_positional_arguments_normalize_to_the_contract_order() { + assert_eq!(validate(Some(&state()), "state", named(&[("prefix", Value::String("ok".into()))])).ok(), + Some(vec![Value::String("ok".into())])); + assert_eq!(validate(Some(&state()), "state", Arguments::Positional(vec![Value::String("ok".into())])).ok(), + Some(vec![Value::String("ok".into())])); + let greet = json!({"name":"greet","parameters":[{"name":"prefix","type":"string","required":false,"variadic":false, + "default_available":true,"default":"dflt","allows_null":false}]}); + assert_eq!(validate(Some(&greet), "greet", Arguments::Positional(vec![])).ok(), Some(vec![Value::String("dflt".into())])); + } + + #[test] + fn violations_carry_phps_messages() { + let refused = |arguments| invalid(validate(Some(&state()), "state", arguments)).unwrap(); + assert_eq!(refused(named(&[("other", Value::String("x".into()))])), vec![ + ("prefix".to_owned(), vec!["The prefix argument is required.".to_owned()]), + ("other".to_owned(), vec!["Unknown argument [other].".to_owned()])]); + assert_eq!(refused(Arguments::Positional(vec![Value::Long(5)])), + vec![("prefix".to_owned(), vec!["The prefix argument must be of type string.".to_owned()])]); + assert_eq!(refused(Arguments::Positional(vec![Value::Null])), + vec![("prefix".to_owned(), vec!["The prefix argument cannot be null.".to_owned()])]); + assert_eq!(refused(Arguments::Positional(vec![Value::String("a".into()), Value::String("b".into())])), + vec![("arguments".to_owned(), vec!["Too many arguments were provided for query [state].".to_owned()])]); + assert_eq!(invalid(validate(None, "plain", named(&[("prefix", Value::Null)]))).unwrap(), + vec![("arguments".to_owned(), vec!["Named arguments require a durable or loadable workflow query contract.".to_owned()])]); + } + + #[test] + fn declared_types_follow_php() { + for (kind, value, expected) in [ + ("?string", Value::Null, true), ("int|string", Value::String("x".into()), true), ("int|string", Value::Boolean(true), false), + ("float", Value::Long(1), true), ("array", Value::Map(vec![]), false), ("scalar", Value::Array(vec![]), false), + ("object", Value::Bytes(vec![1]), true), ("App\\Thing", Value::String("x".into()), false), ("true", Value::Boolean(false), false), + ] { + assert_eq!(matches_type(&value, kind), expected, "{kind}"); + } + } + + #[test] + fn variadics_take_the_rest_or_a_named_list() { + let items = json!({"name":"items","parameters":[ + {"name":"first","type":"int","required":true,"variadic":false,"default_available":false,"allows_null":false}, + {"name":"rest","type":"int","required":false,"variadic":true,"default_available":false,"allows_null":false}]}); + let longs = |values: &[i64]| values.iter().map(|value| Value::Long(*value)).collect::>(); + assert_eq!(validate(Some(&items), "items", Arguments::Positional(longs(&[1, 2, 3]))).ok(), Some(longs(&[1, 2, 3]))); + assert_eq!(validate(Some(&items), "items", Arguments::Positional(longs(&[1]))).ok(), Some(longs(&[1]))); + assert_eq!(validate(Some(&items), "items", named(&[("first", Value::Long(1)), ("rest", Value::Array(longs(&[2, 3])))])).ok(), + Some(longs(&[1, 2, 3]))); + assert_eq!(invalid(validate(Some(&items), "items", Arguments::Positional(vec![Value::Long(1), Value::String("x".into())]))), + Some(vec![("rest".to_owned(), vec!["The rest argument must be of type int.".to_owned()])])); + } + + #[test] + fn empty_and_integer_keyed_maps_are_objects_not_arrays() { + let map = |pairs: &[(&str, Value)]| Value::Map(pairs.iter().map(|(key, value)| ((*key).to_owned(), value.clone())).collect()); + for value in [map(&[]), map(&[("0", Value::String("ok".into()))]), map(&[("0", Value::Long(1)), ("prefix", Value::Long(2))]), + map(&[("-7", Value::Null)])] { + assert!(Arguments::from_value(value.clone()).is_none(), "{value:?}"); + assert!(is_object(&value) && !is_array(&value), "{value:?}"); + } + for key in ["00", "-0", "01", "1.0", " 1", "9223372036854775808", "prefix"] { + let value = map(&[(key, Value::Long(1))]); + assert!(matches!(Arguments::from_value(value.clone()), Some(Arguments::Named(_))), "{key}"); + assert!(is_array(&value), "{key}"); + } + assert!(Arguments::from_value(Value::String("x".into())).is_none()); + } + + #[test] + fn a_named_variadic_map_spreads_in_wire_order() { + let items = json!({"name":"items","parameters":[{"name":"rest","type":"int","required":false,"variadic":true, + "default_available":false,"allows_null":false}]}); + let rest = named(&[("rest", Value::Map(vec![("z".into(), Value::Long(20)), ("a".into(), Value::Long(10))]))]); + assert_eq!(validate(Some(&items), "items", rest).ok(), Some(vec![Value::Long(20), Value::Long(10)])); + assert!(!matches_type(&Value::Bytes(vec![1]), "callable")); + assert!(!matches_type(&Value::Map(vec![]), "callable")); + } + + #[test] + fn a_nested_adapter_map_is_an_object_and_one_variadic_argument() { + let nested = named(&[("value", Value::Map(vec![]))]); + let contract = |kind: &str, variadic: bool| json!({"name":"q","parameters":[{"name":"value","type":kind,"required":!variadic, + "variadic":variadic,"default_available":false,"allows_null":false}]}); + assert_eq!(invalid(validate(Some(&contract("array", false)), "q", nested.clone())), + Some(vec![("value".to_owned(), vec!["The value argument must be of type array.".to_owned()])])); + assert_eq!(validate(Some(&contract("object", false)), "q", nested.clone()).ok(), Some(vec![Value::Map(vec![])])); + assert_eq!(validate(Some(&contract("mixed", true)), "q", nested).ok(), Some(vec![Value::Map(vec![])])); + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 7dfcac7c2..14e67cc6d 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -337,6 +337,9 @@ mod workflow_import_fences; #[path = "execution/query_argument_routing.rs"] mod query_argument_routing; +#[path = "execution/query_contract_arguments.rs"] +mod query_contract_arguments; + #[path = "execution/close_inventory.rs"] mod close_inventory; diff --git a/rust/tests/execution/query_contract_arguments.rs b/rust/tests/execution/query_contract_arguments.rs new file mode 100644 index 000000000..5f4967632 --- /dev/null +++ b/rust/tests/execution/query_contract_arguments.rs @@ -0,0 +1,121 @@ +use super::*; + +// Published Server 2.11.5 with Workflow 2.15.1 +// (`WorkflowQueryContract::validatedArgumentsForRun`): a declared query's +// arguments, positional or named, are matched to its durable parameter +// contract. The worker receives the normalized positional list, encoded +// again; a violation is refused with every message. The +// query-contract-arguments fixture records the same cases. + +async fn contract_worker(app: &Router) { + let parameter = |required: bool, default: Option<&str>| { + let mut parameter = json!({"name":"prefix","position":0,"type":"string","required":required,"variadic":false, + "default_available":default.is_some(),"allows_null":false}); + if let Some(default) = default { + parameter["default"] = json!(default); + } + parameter + }; + let (status, body) = request(app, "POST", "/api/worker/register", json!({"worker_id":"contract-worker","task_queue":"test", + "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":[],"capabilities":["query_tasks"], + "workflow_command_contracts":{"echo":{"signals":[],"updates":[],"queries":["state","greet","plain","listdef","gapdef"],"query_contracts":[ + {"name":"state","parameters":[parameter(true, None)]},{"name":"greet","parameters":[parameter(false, Some("dflt"))]}, + {"name":"listdef","parameters":[json_default(json!({"0":"x","1":"y"}))]}, + {"name":"gapdef","parameters":[json_default(json!({"1":"x"}))]}]}}})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +/// An optional `mixed` parameter whose registered default is raw JSON. +fn json_default(default: Value) -> Value { + json!({"name":"value","position":0,"type":"mixed","required":false,"variadic":false,"default_available":true, + "allows_null":false,"default":default}) +} + +fn named(entries: &[(&str, Payload)]) -> Value { + envelope(Payload::Map(entries.iter().map(|(key, value)| ((*key).to_owned(), value.clone())).collect())) +} + +/// Sends a query the contract accepts and returns the arguments its task carries. +async fn routed(app: &Router, query: &str, input: Value) -> Value { + let waiting = { + let app = app.clone(); + let path = format!("/api/workflows/contract-query/query/{query}"); + tokio::spawn(async move { request(&app, "POST", &path, json!({"input": input})).await }) + }; + let task = await_query(app, "contract-worker").await; + let (status, body) = request(app, "POST", &format!("/api/worker/query-tasks/{}/complete", task["query_task_id"].as_str().unwrap()), + json!({"lease_owner":"contract-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}"); + task["query_arguments"].clone() +} + +#[tokio::test] +async fn named_and_defaulted_query_arguments_reach_the_worker_as_positional_lists() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + contract_worker(&app).await; + start(&app, "contract-query").await; + let ok = envelope(Payload::Array(vec![Payload::String("ok".into())])); + assert_eq!(routed(&app, "state", named(&[("prefix", Payload::String("ok".into()))])).await, ok); + assert_eq!(routed(&app, "state", ok.clone()).await, ok); + assert_eq!(routed(&app, "greet", envelope(Payload::Array(vec![]))).await, + envelope(Payload::Array(vec![Payload::String("dflt".into())]))); + // A declared query without a parameter contract takes a list as given. + let two = envelope(Payload::Array(vec![Payload::String("a".into()), Payload::String("b".into())])); + assert_eq!(routed(&app, "plain", two.clone()).await, two); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn query_contract_violations_are_refused_with_phps_messages() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + contract_worker(&app).await; + start(&app, "contract-query").await; + for (query, input, errors) in [ + ("state", named(&[("prefix", Payload::String("ok".into())), ("other", Payload::String("x".into()))]), + json!({"other":["Unknown argument [other]."]})), + ("state", named(&[("other", Payload::String("x".into()))]), + json!({"prefix":["The prefix argument is required."],"other":["Unknown argument [other]."]})), + ("state", envelope(Payload::Array(vec![Payload::Long(5)])), json!({"prefix":["The prefix argument must be of type string."]})), + ("state", envelope(Payload::Array(vec![Payload::String("a".into()), Payload::String("b".into())])), + json!({"arguments":["Too many arguments were provided for query [state]."]})), + ("state", envelope(Payload::Array(vec![Payload::Null])), json!({"prefix":["The prefix argument cannot be null."]})), + ("state", envelope(Payload::Array(vec![])), json!({"prefix":["The prefix argument is required."]})), + ("plain", named(&[("prefix", Payload::String("ok".into()))]), + json!({"arguments":["Named arguments require a durable or loadable workflow query contract."]})), + ] { + let (status, body) = request(&app, "POST", &format!("/api/workflows/contract-query/query/{query}"), 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!(format!("Workflow query [{query}] argument validation failed.")), errors), "{body}"); + } + assert!(poll(&app, "contract-worker", "query").await.is_null(), "no query task was created"); + runtime.close().await; + database.remove().await; +} + +// A registered JSON default takes PHP's array shape: a list is an Avro array. +// An integer key outside a list fails PHP's encoder with an uncaught 500; +// native keeps the status and names the default. +#[tokio::test] +async fn a_registered_json_default_takes_phps_array_shape() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + contract_worker(&app).await; + start(&app, "contract-query").await; + let list = Payload::Array(vec![Payload::String("x".into()), Payload::String("y".into())]); + assert_eq!(routed(&app, "listdef", envelope(Payload::Array(vec![]))).await, envelope(Payload::Array(vec![list]))); + let (status, body) = request(&app, "POST", "/api/workflows/contract-query/query/gapdef", + json!({"input": envelope(Payload::Array(vec![]))})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::INTERNAL_SERVER_ERROR, json!("query_default_unencodable")), "{body}"); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 10120304e..14da87169 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -17,6 +17,7 @@ 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 {checkQueryContractArguments} from './query-contract-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'; @@ -176,6 +177,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { 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 queryContractArguments = fixture.query_contract_arguments ? checkQueryContractArguments(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; @@ -753,6 +755,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(continueTransfers ? {continue_transfers: continueTransfers} : {}), ...(startSearchAttributes ? {start_search_attributes: startSearchAttributes} : {}), ...(queryArguments ? {query_argument_routing: queryArguments} : {}), + ...(queryContractArguments ? {query_contract_arguments: queryContractArguments} : {}), ...(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 94104423a..a1b65b9ea 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -435,6 +435,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $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; + $queryContractArguments = isset($fixture['query_contract_arguments']) ? finishHttpQueryContractArguments($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; @@ -565,6 +566,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'continue_transfers' => $continueTransfers, 'start_search_attributes' => $startSearchAttributes, 'query_argument_routing' => $queryArgumentRouting, + 'query_contract_arguments' => $queryContractArguments, 'parent_close_cancellation' => $parentCloseCancellation, 'bridge_adapters' => $bridgeAdapters, 'nexus_operations' => $nexusOperations, @@ -898,6 +900,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/continue-transfers-probe.php'; require __DIR__.'/start-search-attributes-probe.php'; require __DIR__.'/query-arguments-probe.php'; + require __DIR__.'/query-contract-arguments-probe.php'; require __DIR__.'/parent-close-cancellation-probe.php'; require __DIR__.'/bridge-adapters-probe.php'; require __DIR__.'/nexus-operations-probe.php'; @@ -1093,6 +1096,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa 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['query_contract_arguments'])) { + $observation['query_contract_arguments'] = ['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-contract-arguments-contract.mjs b/scripts/conformance/server-parity/query-contract-arguments-contract.mjs new file mode 100644 index 000000000..0f3cb713d --- /dev/null +++ b/scripts/conformance/server-parity/query-contract-arguments-contract.mjs @@ -0,0 +1,40 @@ +import assert from 'node:assert/strict'; + +// Reviewed query arguments matched to a durable query contract +// (query-contract-arguments-probe.php), recorded against published Server +// 2.11.5 with Workflow 2.15.1: named and positional arguments normalize to +// the contract's positional order, a default fills a missing argument, the +// worker receives the normalized list encoded again, and each violation is +// refused with PHP's per-argument messages. A registered JSON default takes +// PHP's array shape; `sent_defaults` pins that those defaults were sent as +// JSON objects. The fixture's `expected` holds the whole observation. +export function checkQueryContractArguments(fixture, observation) { + const definition = fixture.query_contract_arguments; + const state = observation.query_contract_arguments; + const projection = {scope: 'query_contract_arguments', cases: Object.keys(definition.cases).length}; + assert.ok(state && typeof state === 'object', 'actual query contract arguments observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_worker_protocol'}, + 'embedded query contract arguments explicitly inapplicable'); + return projection; + } + assert.equal(state.applicable, true, 'HTTP targets record query contract arguments'); + for (const [run, steps] of Object.entries(definition.expected)) { + assert.ok(state[run] && typeof state[run] === 'object', `query contract arguments ${run} recorded`); + // Published PHP answers an unencodable registered default with an + // uncaught 500 and a generic body; only the status is pinned, so a + // server may say what failed. + if (steps.status === 500) { + assert.equal(state[run].status, 500, `query contract arguments ${run} status`); + assert.equal(state[run].validation_errors, null, `query contract arguments ${run} validation errors`); + continue; + } + for (const [step, expected] of Object.entries(steps)) { + assert.deepStrictEqual(state[run][step], expected, `query contract arguments ${run} ${step}`); + } + assert.deepStrictEqual(Object.keys(state[run]).sort(), Object.keys(steps).sort(), `query contract arguments ${run} steps`); + } + assert.deepStrictEqual(Object.keys(state).sort(), ['applicable', ...Object.keys(definition.expected)].sort(), + 'query contract arguments runs'); + return projection; +} diff --git a/scripts/conformance/server-parity/query-contract-arguments-probe.php b/scripts/conformance/server-parity/query-contract-arguments-probe.php new file mode 100644 index 000000000..409719eb9 --- /dev/null +++ b/scripts/conformance/server-parity/query-contract-arguments-probe.php @@ -0,0 +1,108 @@ +createNamespace($namespace, 'query contract arguments λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new QueryContractArgumentsHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $headers = array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', 'X-Durable-Workflow-Protocol-Version' => '1.20', + ]); + $raw = static function (string $method, string $path, ?array $body = null) use ($http, $headers): array { + $response = $http->request($method, $path, ['headers' => $headers, + ...($body === null ? [] : ['json' => $body === [] ? (object) [] : $body])]); + + return ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + }; + $type = $definition['workflow_type']; + $id = $workflowId.'-qca'; + $queue = $id.'-queue'; + $worker = $id.'-worker'; + // The fixture is loaded into PHP arrays, so a default written as a JSON + // object with keys 0..n-1 would be sent as a list. These queries' + // defaults are sent as JSON objects, as a worker registering raw JSON + // would; the exact bytes sent are recorded. + $contracts = array_map(static fn (array $contract): array => in_array($contract['name'], $definition['object_defaults'], true) + ? [...$contract, 'parameters' => array_map(static fn (array $parameter): array => array_key_exists('default', $parameter) + ? [...$parameter, 'default' => (object) $parameter['default']] : $parameter, $contract['parameters'])] + : $contract, $definition['query_contracts']); + $sentDefaults = []; + foreach ($contracts as $contract) { + if (in_array($contract['name'], $definition['object_defaults'], true)) { + $sentDefaults[$contract['name']] = json_encode($contract['parameters'][0]['default'], JSON_THROW_ON_ERROR); + } + } + $registered = $raw('POST', 'worker/register', ['worker_id' => $worker, 'task_queue' => $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']), + 'workflow_command_contracts' => [$type => ['signals' => [], 'updates' => [], + 'queries' => $definition['queries'], 'query_contracts' => $contracts]]]); + if ($registered['status'] !== 201) { + throw new RuntimeException('Query contract arguments registration: '.json_encode($registered)); + } + $started = $raw('POST', 'workflows', ['workflow_id' => $id, 'workflow_type' => $type, 'task_queue' => $queue, + 'input' => ['codec' => 'avro', 'blob' => $definition['blobs']['empty']]]); + if ($started['status'] !== 201) { + throw new RuntimeException('Query contract arguments start: '.json_encode($started)); + } + + $state = ['applicable' => true, 'sent_defaults' => $sentDefaults]; + foreach ($definition['cases'] as $case => [$query, $blob, $routed]) { + $path = '/api/workflows/'.rawurlencode($id).'/query/'.$query; + $body = ['input' => ['codec' => 'avro', 'blob' => $definition['blobs'][$blob]]]; + if (! $routed) { + $answer = $raw('POST', ltrim(substr($path, strlen('/api/')), '/'), $body); + $state[$case] = ['status' => $answer['status'], 'reason' => $answer['response']['reason'] ?? null, + 'message' => $answer['response']['message'] ?? null, + 'validation_errors' => $answer['response']['validation_errors'] ?? null]; + + continue; + } + $pending = startHeldRequest($url, $headers, $path, $body, 0.0); + $leased = null; + for ($attempt = 0; $attempt < 10 && $leased === null; $attempt++) { + $polled = $raw('POST', 'worker/query-tasks/poll', ['worker_id' => $worker, 'task_queue' => $queue, + 'timeout_seconds' => 2]); + $leased = $polled['response']['task'] ?? null; + } + $observed = ['leased' => is_array($leased), 'query_arguments' => is_array($leased) + ? ($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[$case] = $observed; + } + + return $state; +} diff --git a/tests/Fixtures/ServerParity/query-contract-arguments.json b/tests/Fixtures/ServerParity/query-contract-arguments.json new file mode 100644 index 000000000..b9f53e8b4 --- /dev/null +++ b/tests/Fixtures/ServerParity/query-contract-arguments.json @@ -0,0 +1,550 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "query-contract-arguments", + "review_state": "reviewed", + "contract": "Query arguments matched to a durable query contract (published Server 2.11.5 WorkflowQueryTaskBroker::validateContractedQueryArguments with Workflow 2.15.1 WorkflowQueryContract::validatedArgumentsForRun), over raw HTTP on protocol 1.20 in a fresh namespace. The worker that starts the run declares state(string $prefix), greet(string $prefix = 'dflt'), plain without a parameter contract, arr(array $value), obj(object $value) and rest(mixed ...$value). A query the contract accepts is routed and the worker receives the normalized positional list encoded again: a named map {prefix: ok} and the list [ok] both reach state as [ok], an empty list reaches greet as [dflt], a named map reaches greet as [ok], and plain takes a list as given. A violation is refused before routing with 422 invalid_query_arguments, 'Workflow query [] argument validation failed.' and validation_errors by argument: Unknown argument [other]., The prefix argument is required., The prefix argument must be of type string., The prefix argument cannot be null., Too many arguments were provided for query [state]., and for plain, Named arguments require a durable or loadable workflow query contract. Avro::fromDatum returns an empty map, or a map with a key PHP casts to an integer, as an AvroMapValue object: at the top level an empty map, {\"0\": \"ok\"} and a mixed-key map are refused before normalization with 'arguments must decode to an array.'; nested as {value: {}} it is refused by arr (The value argument must be of type array.) and routed whole as one argument by obj and by the variadic rest. A named variadic spreads a string-keyed map's values in wire order: {value: {z: 20, a: 10}} reaches rest as [20, 10]. A registered default the worker sends as JSON takes PHP's array shape: {\"0\": \"x\", \"1\": \"y\"} is a list and reaches the worker as [[\"x\", \"y\"]]. A default with an integer key outside a list ({\"0\": \"x\", \"k\": \"y\"} or {\"1\": \"x\"}) fails published PHP's Avro encoder with an uncaught 500; only that status is pinned, and native explains it as query_default_unencodable. Embedded applications have no worker protocol.", + "workflow_type": "parity.v1.echo", + "input": { + "message": "query contract arguments λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "query contract arguments λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "query_contract_arguments": { + "workflow_type": "parity.v1.query_contract_arguments", + "queries": [ + "state", + "greet", + "plain", + "arr", + "obj", + "rest", + "listdef", + "mixeddef", + "gapdef" + ], + "query_contracts": [ + { + "name": "state", + "parameters": [ + { + "name": "prefix", + "position": 0, + "type": "string", + "required": true, + "variadic": false, + "default_available": false, + "allows_null": false + } + ] + }, + { + "name": "greet", + "parameters": [ + { + "name": "prefix", + "position": 0, + "type": "string", + "required": false, + "variadic": false, + "default_available": true, + "allows_null": false, + "default": "dflt" + } + ] + }, + { + "name": "arr", + "parameters": [ + { + "name": "value", + "position": 0, + "type": "array", + "required": true, + "variadic": false, + "default_available": false, + "allows_null": false + } + ] + }, + { + "name": "obj", + "parameters": [ + { + "name": "value", + "position": 0, + "type": "object", + "required": true, + "variadic": false, + "default_available": false, + "allows_null": false + } + ] + }, + { + "name": "rest", + "parameters": [ + { + "name": "value", + "position": 0, + "type": "mixed", + "required": false, + "variadic": true, + "default_available": false, + "allows_null": false + } + ] + }, + { + "name": "listdef", + "parameters": [ + { + "name": "value", + "position": 0, + "type": "array", + "required": false, + "variadic": false, + "default_available": true, + "allows_null": false, + "default": { + "0": "x", + "1": "y" + } + } + ] + }, + { + "name": "mixeddef", + "parameters": [ + { + "name": "value", + "position": 0, + "type": "mixed", + "required": false, + "variadic": false, + "default_available": true, + "allows_null": false, + "default": { + "0": "x", + "k": "y" + } + } + ] + }, + { + "name": "gapdef", + "parameters": [ + { + "name": "value", + "position": 0, + "type": "mixed", + "required": false, + "variadic": false, + "default_available": true, + "allows_null": false, + "default": { + "1": "x" + } + } + ] + } + ], + "blobs": { + "named": "wwHioz3/VYAiNw4CDHByZWZpeAoEb2sA", + "positional": "wwHioz3/VYAiNwwCCgRvawA=", + "wrong_type": "wwHioz3/VYAiNwwCBAoA", + "too_many": "wwHioz3/VYAiNwwECgJhCgJiAA==", + "null": "wwHioz3/VYAiNwwCAAA=", + "empty": "wwHioz3/VYAiNwwA", + "unknown": "wwHioz3/VYAiNw4CCm90aGVyCgJ4AA==", + "default_filled": "wwHioz3/VYAiNwwCCghkZmx0AA==", + "named_extra": "wwHioz3/VYAiNw4EDHByZWZpeAoEb2sKb3RoZXIKAngA", + "empty_map": "wwHioz3/VYAiNw4A", + "numeric_map": "wwHioz3/VYAiNw4CAjAKBG9rAA==", + "mixed_map": "wwHioz3/VYAiNw4EAjAKAmEMcHJlZml4CgJiAA==", + "nested_empty": "wwHioz3/VYAiNw4CCnZhbHVlDgAA", + "one_empty_map": "wwHioz3/VYAiNwwCDgAA", + "unsorted_variadic": "wwHioz3/VYAiNw4CCnZhbHVlDgQCegQoAmEEFAAA", + "spread_in_order": "wwHioz3/VYAiNwwEBCgEFAA=" + }, + "cases": { + "named": [ + "state", + "named", + true + ], + "positional": [ + "state", + "positional", + true + ], + "named_extra": [ + "state", + "named_extra", + false + ], + "unknown": [ + "state", + "unknown", + false + ], + "wrong_type": [ + "state", + "wrong_type", + false + ], + "too_many": [ + "state", + "too_many", + false + ], + "null": [ + "state", + "null", + false + ], + "missing": [ + "state", + "empty", + false + ], + "default": [ + "greet", + "empty", + true + ], + "default_named": [ + "greet", + "named", + true + ], + "plain_named": [ + "plain", + "named", + false + ], + "plain_list": [ + "plain", + "too_many", + true + ], + "empty_map": [ + "greet", + "empty_map", + false + ], + "numeric_map": [ + "state", + "numeric_map", + false + ], + "mixed_map": [ + "state", + "mixed_map", + false + ], + "nested_array": [ + "arr", + "nested_empty", + false + ], + "nested_object": [ + "obj", + "nested_empty", + true + ], + "nested_variadic": [ + "rest", + "nested_empty", + true + ], + "unsorted_variadic": [ + "rest", + "unsorted_variadic", + true + ], + "list_default": [ + "listdef", + "empty", + true + ], + "mixed_default": [ + "mixeddef", + "empty", + false + ], + "gap_default": [ + "gapdef", + "empty", + false + ] + }, + "expected": { + "named": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCgRvawA=" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "positional": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCgRvawA=" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "named_extra": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] argument validation failed.", + "validation_errors": { + "other": [ + "Unknown argument [other]." + ] + } + }, + "unknown": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] argument validation failed.", + "validation_errors": { + "prefix": [ + "The prefix argument is required." + ], + "other": [ + "Unknown argument [other]." + ] + } + }, + "wrong_type": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] argument validation failed.", + "validation_errors": { + "prefix": [ + "The prefix argument must be of type string." + ] + } + }, + "too_many": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] argument validation failed.", + "validation_errors": { + "arguments": [ + "Too many arguments were provided for query [state]." + ] + } + }, + "null": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] argument validation failed.", + "validation_errors": { + "prefix": [ + "The prefix argument cannot be null." + ] + } + }, + "missing": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [state] argument validation failed.", + "validation_errors": { + "prefix": [ + "The prefix argument is required." + ] + } + }, + "default": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCghkZmx0AA==" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "default_named": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCCgRvawA=" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "plain_named": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [plain] argument validation failed.", + "validation_errors": { + "arguments": [ + "Named arguments require a durable or loadable workflow query contract." + ] + } + }, + "plain_list": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwECgJhCgJiAA==" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "empty_map": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [greet] arguments must decode to an array.", + "validation_errors": { + "arguments": [ + "Query arguments must decode to an array." + ] + } + }, + "numeric_map": { + "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." + ] + } + }, + "mixed_map": { + "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." + ] + } + }, + "nested_array": { + "status": 422, + "reason": "invalid_query_arguments", + "message": "Workflow query [arr] argument validation failed.", + "validation_errors": { + "value": [ + "The value argument must be of type array." + ] + } + }, + "nested_object": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCDgAA" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "nested_variadic": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCDgAA" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "unsorted_variadic": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwEBCgEFAA=" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "list_default": { + "leased": true, + "query_arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCDAQKAngKAnkAAA==" + }, + "answered": 200, + "client": { + "status": 200, + "reason": null + } + }, + "mixed_default": { + "status": 500, + "reason": null, + "message": "Server Error", + "validation_errors": null + }, + "gap_default": { + "status": 500, + "reason": null, + "message": "Server Error", + "validation_errors": null + }, + "sent_defaults": { + "listdef": "{\"0\":\"x\",\"1\":\"y\"}", + "mixeddef": "{\"0\":\"x\",\"k\":\"y\"}", + "gapdef": "{\"1\":\"x\"}" + } + }, + "object_defaults": [ + "listdef", + "mixeddef", + "gapdef" + ] + } +} diff --git a/tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs b/tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs new file mode 100644 index 000000000..df3f7e176 --- /dev/null +++ b/tests/Unit/ServerParityQueryContractArgumentsContractTest.mjs @@ -0,0 +1,85 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkQueryContractArguments} from '../../scripts/conformance/server-parity/query-contract-arguments-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/query-contract-arguments.json', import.meta.url), 'utf8')); +const definition = fixture.query_contract_arguments; +const expected = definition.expected; +const observation = () => structuredClone({mode: 'http', query_contract_arguments: {applicable: true, ...expected}}); +const cases = Object.keys(definition.cases).length; + +test('the published answers pass', () => { + assert.deepStrictEqual(checkQueryContractArguments(fixture, observation()), {scope: 'query_contract_arguments', cases}); +}); + +test('an unencodable default pins only the 500 status, so a server may explain it', () => { + const raw = observation(); + raw.query_contract_arguments.mixed_default = {status: 500, reason: 'query_default_unencodable', + message: 'Workflow query [mixeddef] parameter [value] has a registered default with an integer key outside a list, which the Avro Value codec cannot encode.', + validation_errors: null}; + assert.deepStrictEqual(checkQueryContractArguments(fixture, raw), {scope: 'query_contract_arguments', cases}); +}); + +test('embedded applications have no worker protocol', () => { + assert.deepStrictEqual(checkQueryContractArguments(fixture, + {mode: 'embedded', query_contract_arguments: {applicable: false, reason: 'embedded_has_no_worker_protocol'}}), + {scope: 'query_contract_arguments', cases}); + assert.throws(() => checkQueryContractArguments(fixture, {mode: 'embedded', query_contract_arguments: {applicable: true}})); +}); + +test('accepted queries reach the worker as normalized positional lists', () => { + const blob = name => definition.blobs[name]; + assert.deepStrictEqual(expected.named.query_arguments.blob, blob('positional')); + assert.deepStrictEqual(expected.positional.query_arguments.blob, blob('positional')); + assert.deepStrictEqual(expected.default.query_arguments.blob, blob('default_filled')); + assert.deepStrictEqual(expected.default_named.query_arguments.blob, blob('positional')); + assert.deepStrictEqual(expected.plain_list.query_arguments.blob, blob('too_many')); + assert.deepStrictEqual(expected.nested_object.query_arguments.blob, blob('one_empty_map')); + assert.deepStrictEqual(expected.nested_variadic.query_arguments.blob, blob('one_empty_map')); + assert.deepStrictEqual(expected.unsorted_variadic.query_arguments.blob, blob('spread_in_order')); + for (const name of ['empty_map', 'numeric_map', 'mixed_map']) { + assert.match(expected[name].message, /must decode to an array\.$/, name); + } + for (const name of ['named_extra', 'unknown', 'wrong_type', 'too_many', 'null', 'missing', 'plain_named', 'nested_array']) { + assert.deepStrictEqual([expected[name].status, expected[name].reason], [422, 'invalid_query_arguments'], name); + assert.match(expected[name].message, /argument validation failed\.$/, name); + } +}); + +for (const [name, mutate] of [ + ['a named map refused as not a list', state => { + state.named = {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.']}}; + }], + ['the client blob passed through un-normalized', state => { state.named.query_arguments.blob = definition.blobs.named; }], + ['a default left unfilled', state => { state.default.query_arguments.blob = definition.blobs.empty; }], + ['a bare refusal', state => { state.wrong_type.message = 'invalid_query_arguments'; state.wrong_type.validation_errors = null; }], + ['a missing unknown-argument message', state => { delete state.unknown.validation_errors.other; }], + ['an empty map taken as an empty list', state => { + state.empty_map = {leased: true, query_arguments: {codec: 'avro', blob: definition.blobs.default_filled}, answered: 200, + client: {status: 200, reason: null}}; + }], + ['a nested empty map accepted as an array', state => { + state.nested_array = {leased: true, query_arguments: {codec: 'avro', blob: definition.blobs.one_empty_map}, answered: 200, + client: {status: 200, reason: null}}; + }], + ['a nested empty map spread to nothing', state => { state.nested_variadic.query_arguments.blob = definition.blobs.empty; }], + ['a variadic map spread in sorted key order', state => { + state.unsorted_variadic.query_arguments.blob = 'wwHioz3/VYAiNwwEBBQEKAA='; + }], + ['a list default routed as a map', state => { + state.list_default.query_arguments.blob = 'wwHioz3/VYAiNwwCDgQCMAoCeAIxCgJ5AAA='; + }], + ['an unencodable default routed', state => { + state.mixed_default = {status: 200, reason: null, message: null, validation_errors: null}; + }], + ['an unencodable default refused as invalid arguments', state => { + state.gap_default = {status: 422, reason: 'invalid_query_arguments', message: 'x', validation_errors: {value: ['x']}}; + }], + ['a list default sent as a JSON array', state => { state.sent_defaults.listdef = '["x","y"]'; }], + ['an unexpected case', state => { state.extra = {}; }], +]) test(`the contract rejects ${name}`, () => { + const raw = observation(); mutate(raw.query_contract_arguments); + assert.throws(() => checkQueryContractArguments(fixture, raw)); +});