Rust #325: encode plain input in document order, record key order

.github/workflows/server-parity.yml

@@ -73,6 +73,7 @@ on:
       - tests/Unit/ServerParityChildVisibilityContractTest.mjs
       - tests/Unit/ServerParityClusterCapabilitiesContractTest.mjs
       - tests/Unit/ServerParityDeepJsonBodiesContractTest.mjs
+      - tests/Unit/ServerParityKeyOrderContractTest.mjs
       - tests/Unit/ServerParityVisibilityCollationContractTest.mjs
       - tests/Unit/ServerParitySearchAttributeDatetimesContractTest.mjs
       - tests/Unit/ServerParityRetriedChildCancellationContractTest.mjs
@@ -2229,6 +2230,7 @@ jobs:
               tests/Unit/ServerParityChildVisibilityContractTest.mjs \
               tests/Unit/ServerParityClusterCapabilitiesContractTest.mjs \
               tests/Unit/ServerParityDeepJsonBodiesContractTest.mjs \
+              tests/Unit/ServerParityKeyOrderContractTest.mjs \
               tests/Unit/ServerParityVisibilityCollationContractTest.mjs \
               tests/Unit/ServerParitySearchAttributeDatetimesContractTest.mjs \
               tests/Unit/ServerParityRetriedChildCancellationContractTest.mjs \

docs/rust-server-port-checkpoints/2026-11-01-key-order.md

@@ -0,0 +1,66 @@
+# November 1, key order in plain input and search attribute errors
+
+- **Before.** Native's plain JSON input reached the Avro writer through
+  `memos::from_json`, whose maps are `BTreeMap`s:
+  - a nested object was encoded with sorted keys, so its blob differed from
+    PHP's while the decoded values were equal;
+  - a nested object with keys 0 to n-1 in any order read as a list, so
+    `{"1": "a", "0": "b"}` became `["b", "a"]`;
+  - before #906 (`serde_json` without `preserve_order`), an object input's
+    values and several search attribute errors also came out in sorted key
+    order.
+- **Contract** (published 2.11.7, Workflow 2.15.7).
+  `PayloadEnvelopeResolver::resolve` takes a plain input's values in
+  document order (`array_values`) and serializes them. `Avro::toDatum`
+  encodes a PHP list (`array_is_list`) as an Avro array and any other array
+  as a map in document order. Its keys must be strings, and `json_decode`
+  turns a canonical integer string into an integer key, so an integer-keyed
+  object that is not a list fails with `invalid_map_key`:
+  - on a start, Core's exception reaches the controller, which answers
+    `422 validation_failed` with the message under `workflow_type`, and no
+    run is created;
+  - on a signal, it answers `500 control_plane_internal_error`
+    (`operation_failed`, `indeterminate`).
+
+  `SearchAttributeValueValidator` loops over the attributes in request
+  order, so several errors on a start (`search_attributes`) or on one
+  upsert command (`commands.N.attributes`) are reported in that order.
+- **Native.** `plain_input::ordered` converts plain input to an
+  `OrderedValue` in document order with `Avro::toDatum`'s rules, and the
+  ordered Avro writer encodes it (`queries::encode_ordered_arguments`). Every
+  plain input path uses it: start, bridge start, standalone activity
+  start, signal, update, query and message stream append. A refusal on a
+  non-bridge start is PHP's validation failure under `workflow_type`; the
+  other paths keep `422 invalid_workflow_arguments` rather than PHP's 500.
+  Search attribute errors already followed request order after #906.
+- **Ordered writer.** `ValueCodec::encode_ordered_value` used Apache's datum
+  writer over the binding schema and then read its own blob back. Plain
+  input can nest 511 levels (`deep-json-bodies`), and those two steps
+  overflowed even a 64 MiB stack there. The writer now emits the protocol
+  Value's Avro binary layout directly, one small frame per level, and does
+  not read the blob back. A 511-level value encodes in a 2 MiB stack. A unit
+  test checks it byte for byte against Apache's binding writer, and the
+  existing test against PHP's published bytes still passes. Query routing,
+  the other caller, writes the same bytes as before.
+- **Fixture.** `key-order`, recorded against published 2.11.7 on SQLite,
+  in a fresh namespace without registered attributes. The probe sends each
+  body as the fixture's JSON text, so key order and integer keys survive:
+  - a nested map, whose stored input blob is PHP's exact bytes (`zeta`
+    before `alpha`, `yak` before `bee`);
+  - `{"1": "a", "0": "b"}` and `{"second": "b", "first": "a"}` as the whole
+    input, stored as `["a", "b"]` and `["b", "a"]`; native before #906 gave
+    the second as `["a", "b"]`;
+  - `[{"1": "a", "0": "b"}]` and `[{"1": "a"}]`, refused with the 422;
+  - a signal's nested map, stored in document order;
+  - three invalid search attributes, `Zz`, `9bad`, `Aa`, on a start and on
+    an upsert command, reported in that order; native before #906 gave
+    `9bad`, `Aa`, `Zz`.
+
+  The fixture passes against published 2.11.7 (recorded twice) and against
+  native on SQLite. The native recording ran before the direct writer
+  replaced the binding writer; the two write identical bytes.
+- **Tests.** `plain_input.rs` unit test; `wire_order.rs`
+  `the_direct_writer_matches_apache_s_binding_writer`; `key_order.rs` execution tests
+  (PHP's exact blobs, the refusal, the signal blob and the error order);
+  `ServerParityKeyOrderContractTest.mjs`, which rejects sorted maps, sorted
+  object values, out-of-order integer keys read as a list and sorted errors.

docs/rust-server-port.md

@@ -429,9 +429,21 @@ contract fixture before its correction.
   and PostgreSQL. Native refuses such a name, and a name of only spaces,
   with `invalid_signal_name` or `invalid_update_name` on every database
   ([checkpoint](rust-server-port-checkpoints/2026-10-22-poll-waits.md)).
-- **Key order inside plain JSON input.** For input that is not an envelope,
-  PHP encodes nested objects as Avro maps in request order; native encodes
-  them with sorted keys. The decoded values are equal, but the blobs differ.
+- **Key order inside plain JSON input** (November 1, closed). For input
+  that is not an envelope, published PHP takes an object input's values in
+  document order (`array_values`) and encodes each nested object as an Avro
+  map in document order (`Avro::toDatum`). A nested object whose integer
+  keys are not 0 to n-1 in document order, such as `{"1": "a", "0": "b"}`,
+  is not a list, and its encoder fails with `invalid_map_key`. Native
+  encoded nested maps with sorted keys and read such an object as the list
+  `["b", "a"]`. It now encodes plain input in document order through the
+  ordered Avro writer, so its blobs are PHP's byte for byte, and refuses an
+  integer-keyed object that is not a list: on a start with PHP's 422 under
+  `workflow_type`. On a signal, update, query, message or bridge start it
+  keeps native's `422 invalid_workflow_arguments`, where PHP's signal answers
+  `500 control_plane_internal_error`. The `key-order` fixture records starts
+  and a signal against published 2.11.7
+  ([checkpoint](rust-server-port-checkpoints/2026-11-01-key-order.md)).
 - **Child completion while the parent has an open workflow task**
   (November 1; ready task closed, leased task decided;
   [checkpoint](rust-server-port-checkpoints/2026-11-01-child-open-parent-task.md)).
@@ -774,9 +786,14 @@ contract fixture before its correction.
   without trailing spaces). The `visibility-collation` fixture records both
   collations against published 2.11.7
   ([checkpoint](rust-server-port-checkpoints/2026-10-30-visibility-collation.md)).
-- **Visibility query key order.** When one command or query has several
-  errors, native reports them in sorted key order rather than request order.
-  The same holds for several invalid search attributes on one start.
+- **Search attribute validation order** (November 1, closed). Published
+  PHP reports several invalid search attributes in request order, on a
+  start and on one `upsert_search_attributes` command, as
+  `SearchAttributeValueValidator` loops over them. Native reported them in
+  sorted key order. Since `serde_json`'s `preserve_order` (#906) its
+  validator visits the request's keys in document order, and the `key-order`
+  fixture records both orders against published 2.11.7
+  ([checkpoint](rust-server-port-checkpoints/2026-11-01-key-order.md)).
 - **Inherited memos and search attributes on child runs** (October 29,
   closed). Published PHP copies the parent's memos and search attributes to
   a child run, and the failed run's to a retried child

rust/src/codec/wire_order.rs

@@ -8,7 +8,7 @@
 //! 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,
+    Schema, reader::datum::GenericDatumReader, types::Value as Datum,
 };
 use base64::{Engine, engine::general_purpose::STANDARD};
 use std::{collections::HashMap, sync::LazyLock};
@@ -62,45 +62,86 @@ impl ValueCodec {
     /// 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<String, CodecError> {
-        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());
+        // Written directly in the protocol Value's Avro binary layout: a deep
+        // value costs one small frame per level, where Apache's datum writer
+        // over the binding schema needs far more stack than a 511-level
+        // request leaves.
+        let mut bytes = Vec::with_capacity(64);
         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)
+        write_value(value, &mut bytes)?;
+        // The layout is checked against Apache's writer and PHP's published
+        // bytes in tests; reading it back here would cost a deep value more
+        // stack than writing it.
+        Ok(STANDARD.encode(bytes))
     }
 }
 
-fn record(field: &str, value: Datum) -> Datum {
-    Datum::Record(vec![(field.into(), value)])
+/// Avro's zig-zag variable-length long.
+fn write_long(number: i64, out: &mut Vec<u8>) {
+    let mut zigzag = ((number << 1) ^ (number >> 63)) as u64;
+    while zigzag >= 0x80 {
+        out.push((zigzag as u8 & 0x7f) | 0x80);
+        zigzag >>= 7;
+    }
+    out.push(zigzag as u8);
+}
+
+fn write_bytes(bytes: &[u8], out: &mut Vec<u8>) {
+    write_long(bytes.len() as i64, out);
+    out.extend_from_slice(bytes);
 }
 
-fn to_binding(value: &OrderedValue) -> Result<Datum, CodecError> {
-    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))),
+/// One protocol Value datum: the union branch index, then the branch
+/// record's one field. An array or map is one block of all its items (none
+/// when empty) and the end marker, as Apache's and PHP's writers emit.
+fn write_value(value: &OrderedValue, out: &mut Vec<u8>) -> Result<(), CodecError> {
+    match value {
+        OrderedValue::Null => write_long(0, out),
+        OrderedValue::Boolean(flag) => {
+            write_long(1, out);
+            out.push(u8::from(*flag));
+        }
+        OrderedValue::Long(number) => {
+            write_long(2, out);
+            write_long(*number, out);
+        }
         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::Double(number) => {
+            write_long(3, out);
+            out.extend_from_slice(&number.to_le_bytes());
+        }
+        OrderedValue::Bytes(bytes) => {
+            write_long(4, out);
+            write_bytes(bytes, out);
+        }
+        OrderedValue::String(text) => {
+            write_long(5, out);
+            write_bytes(text.as_bytes(), out);
+        }
         OrderedValue::Array(items) => {
-            (6, record("items", Datum::Array(items.iter().map(to_binding).collect::<Result<_, _>>()?)))
+            write_long(6, out);
+            if !items.is_empty() {
+                write_long(items.len() as i64, out);
+                for item in items {
+                    write_value(item, out)?;
+                }
+            }
+            write_long(0, out);
         }
         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::<Result<_, CodecError>>()?;
-            (7, record("entries", Datum::Array(entries)))
+            write_long(7, out);
+            if !entries.is_empty() {
+                write_long(entries.len() as i64, out);
+                for (key, item) in entries {
+                    write_bytes(key.as_bytes(), out);
+                    write_value(item, out)?;
+                }
+            }
+            write_long(0, out);
         }
-    };
-    Ok(record("value", Datum::Union(index, Box::new(branch))))
+    }
+    Ok(())
 }
 
 /// The one field of a record the binding schema read.
@@ -205,6 +246,59 @@ mod tests {
         assert!(codec.decode_ordered("AAAA").is_err());
     }
 
+    /// Apache's datum writer over the binding schema, the reference the
+    /// direct writer replaced.
+    fn apache_reference(value: &OrderedValue) -> String {
+        use apache_avro::writer::datum::GenericDatumWriter;
+        fn record(field: &str, value: Datum) -> Datum {
+            Datum::Record(vec![(field.into(), value)])
+        }
+        fn binding(value: &OrderedValue) -> Datum {
+            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) => (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(binding).collect()))),
+                OrderedValue::Map(entries) => (7, record("entries", Datum::Array(entries.iter()
+                    .map(|(key, value)| Datum::Record(vec![("key".into(), Datum::String(key.clone())), ("value".into(), binding(value))]))
+                    .collect()))),
+            };
+            record("value", Datum::Union(index, Box::new(branch)))
+        }
+        let datum = GenericDatumWriter::builder(&BINDING).build().unwrap().write_value_to_vec(binding(value)).unwrap();
+        STANDARD.encode([MARKER.as_slice(), VALUE_FINGERPRINT.as_slice(), datum.as_slice()].concat())
+    }
+
+    #[test]
+    fn the_direct_writer_matches_apache_s_binding_writer() {
+        let codec = ValueCodec::new().unwrap();
+        let many: Vec<OrderedValue> = (0..200).map(|index| OrderedValue::Long(index * 1_000_003 - 7)).collect();
+        let entries: Vec<(String, OrderedValue)> = (0..70).map(|index| (format!("k{}", 70 - index), OrderedValue::Bytes(vec![index as u8; index]))).collect();
+        for value in [
+            OrderedValue::Null,
+            OrderedValue::Boolean(true),
+            OrderedValue::Long(i64::MIN),
+            OrderedValue::Long(i64::MAX),
+            OrderedValue::Long(-1),
+            OrderedValue::Double(-0.0),
+            OrderedValue::Double(1.5e300),
+            OrderedValue::String("λ ключ 🙂".into()),
+            OrderedValue::Bytes(vec![]),
+            OrderedValue::Array(vec![]),
+            OrderedValue::Map(vec![]),
+            OrderedValue::Array(many),
+            OrderedValue::Map(entries),
+            OrderedValue::Array(vec![OrderedValue::Map(vec![("z".into(), OrderedValue::Array(vec![OrderedValue::Null])),
+                ("a".into(), OrderedValue::Map(vec![("".into(), OrderedValue::Boolean(false))]))])]),
+        ] {
+            assert_eq!(codec.encode_ordered_value(&value).unwrap(), apache_reference(&value), "{value:?}");
+        }
+        assert!(matches!(codec.encode_ordered_value(&OrderedValue::Double(f64::NAN)), Err(CodecError::NonFiniteDouble)));
+    }
+
     #[test]
     fn a_repeated_key_keeps_its_first_position_and_last_value() {
         let codec = ValueCodec::new().unwrap();

rust/src/runtime/execution.rs

@@ -469,17 +469,30 @@ impl Runtime {
     /// empty input is no arguments and any list or object other than an
     /// envelope is its values, both encoded here. An envelope is left as sent.
     async fn plain_arguments(&self, body: &Value) -> Result<Option<String>> {
+        self.plain_arguments_or(body, || refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_workflow_arguments")).await
+    }
+
+    /// The plain input in document order (`super::plain_input`), or
+    /// `refusal` where PHP's serializer fails with `invalid_map_key`.
+    async fn plain_arguments_or(
+        &self,
+        body: &Value,
+        refusal: impl FnOnce() -> super::RuntimeError,
+    ) -> Result<Option<String>> {
         super::input_envelope::check(body)?;
         let arguments = match super::start_request::input(body) {
             super::start_request::Input::Envelope => return Ok(None),
             super::start_request::Input::Empty => Vec::new(),
-            super::start_request::Input::Values(values) => values
-                .into_iter()
-                .map(|value| super::memos::from_json(value.clone()))
-                .collect::<Result<_>>()
-                .map_err(|_| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_workflow_arguments"))?,
+            super::start_request::Input::Values(values) => match values.into_iter().map(super::plain_input::ordered).collect() {
+                Some(arguments) => arguments,
+                None => return Err(refusal()),
+            },
         };
-        let blob = super::queries::encode_arguments(self.signal_codec.clone(), crate::codec::Value::Array(arguments)).await?;
+        let blob = super::queries::encode_ordered_arguments(
+            self.signal_codec.clone(),
+            crate::codec::OrderedValue::Array(arguments),
+        )
+        .await?;
         Ok(Some(blob))
     }
 
@@ -566,7 +579,21 @@ impl Runtime {
         let mut body = super::start_request::normalized(body);
         super::input_envelope::transport(&body)?;
         super::start_request::validate(&body)?;
-        match self.plain_arguments(&body).await? {
+        // Core's start reports the serializer's `invalid_map_key` under
+        // `workflow_type`, as the controller reports Core's LogicException.
+        let plain = if bridge {
+            self.plain_arguments(&body).await?
+        } else {
+            self.plain_arguments_or(&body, || {
+                super::laravel_validation::failure_messages(vec![(
+                    "workflow_type".to_owned(),
+                    vec![super::plain_input::INVALID_MAP_KEY.to_owned()],
+                )])
+                .expect("one message")
+            })
+            .await?
+        };
+        match plain {
             Some(blob) => body["input"] = json!({"codec": "avro", "blob": blob}),
             // PHP 2.5.14 answers an undecodable start blob with a 500
             // (server#601); native refuses it as PHP's signal does.

rust/src/runtime/mod.rs

@@ -100,6 +100,7 @@ mod php_value;
 #[cfg(debug_assertions)]
 #[doc(hidden)]
 pub mod poll_pauses;
+mod plain_input;
 mod poll_readiness;
 pub mod postgres;
 mod previews;