Rust #325 train: query plain results, body admission

docs/rust-server-port-checkpoints/2026-10-22-payload-limits.md

@@ -0,0 +1,32 @@
+# October 22, request body admission
+
+- **Gap.** Published PHP prepends `EnforcePayloadLimits` to every request,
+  ahead of authentication and routing. Native checked malformed JSON only on
+  the two system passes. Elsewhere it answered a plain-text 400 or 415 from
+  its JSON extractor, after authentication. Its body cap was a fixed 3 MB.
+- **Source.** A new outermost layer applies PHP's checks in PHP's order:
+  - **Size.** A declared `Content-Length` or a body over
+    `server.limits.max_payload_bytes` (`DW_MAX_PAYLOAD_BYTES` or
+    `WORKFLOW_MAX_PAYLOAD_BYTES`, default 2 MiB) is refused with
+    `413 payload_too_large` and the limit.
+  - **Media type.** A POST, PUT, PATCH or DELETE with a body must use
+    `application/json` or a `+json` type, or it is refused with
+    `415 unsupported_media_type` and the accepted types.
+  - **JSON.** A body PHP's `json_decode` would refuse is refused with
+    `400 malformed_json` and PHP's `json_last_error_msg()` text: syntax,
+    malformed UTF-8, control character, unpaired surrogate, or nesting deeper
+    than 512 levels. A whitespace-only body is malformed, as in PHP.
+  - **Envelope.** Worker-plane routes answer in the worker protocol envelope.
+    Others use the control-plane envelope.
+  - **Exception.** The external payload upload keeps its own transport rules.
+  The layer reads each body once and bounds it, so the fixed 3 MB cap is
+  removed.
+- **Remaining gap.** A body nested 129 to 512 levels deep passes this check,
+  as in PHP, but native's handlers then refuse it (the port's gap list).
+- **Verification.**
+  - Unit tests map serde_json's errors to PHP's messages, including both
+    unpaired surrogate forms, a valid surrogate pair and the 512-level depth.
+  - A new integration test checks unauthenticated malformed bodies, worker
+    and control-plane envelopes, a missing or non-JSON media type, an
+    oversized body and a `+json` type passing on to authentication.
+  - The full SQLite suite passes.

docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md

@@ -0,0 +1,26 @@
+# October 22, query worker results
+
+- **Found by.** The Shared Server fixtures workflow on main. After #626, its
+  `mysql-execution` and `postgres-storage` jobs still failed in the SDK query
+  directions profile (main run 39704552600). The PHP-worker cases differed in
+  `control_plane.result`:
+  - native's wire response had `null`;
+  - the published Python client's hydrated response had the query state.
+- **Cause.** Published PHP SDK workers complete a query task with both the
+  plain `result` and its `result_envelope`. Published PHP Server's
+  `completeQueryTask` accepts both and stores both. Its answer returns the
+  plain `result` beside the envelope, and the control-plane envelope copies
+  it. Native required the envelope, ignored the plain `result`, and always
+  answered `result: null`.
+- **Source.**
+  - **Storage.** A completed query task keeps the worker's plain `result`.
+  - **Answer.** The answer returns it, and its control-plane envelope copies
+    it, as PHP does.
+  - **Plain-only completions.** A completion with a plain `result` and no
+    envelope is accepted, as PHP's rules allow. A completion with neither is
+    still refused.
+  - **Budget.** The plain result counts toward the query result budget.
+- **Verification.** A new integration test completes one query with a plain
+  result and an envelope, and one with only a plain result. Both answers
+  carry the plain result at the top level and in `control_plane`. The query
+  tests pass on SQLite, PostgreSQL and MySQL.

docs/rust-server-port.md

@@ -167,10 +167,10 @@ contract fixture before its correction.
   its message stream items. Native does not clean either up yet. It skips
   such a run with `external_payload_cleanup_unavailable` or
   `message_stream_cleanup_unavailable` and keeps its details.
-- **Malformed JSON bodies.** Published PHP answers a malformed JSON body
-  with `400 malformed_json` and a JSON body, and adds the control-plane
-  envelope on control-plane operations. Native does so only on the two
-  system passes. Elsewhere, native answers a plain-text 400.
+- **Deeply nested JSON bodies.** Published PHP decodes JSON up to 512 levels
+  deep. Native checks every body as PHP does before any route, but its
+  handlers parse with serde_json's 128-level limit. A body nested 129 to 512
+  levels deep passes the check and is then refused with a plain-text 400.
 - **Registration without a capability manifest.** At protocol 1.19,
   published PHP 2.5.14 refuses a registration that has no
   `capability_manifest` (422). Native accepts it.

rust/src/runtime/execution.rs

@@ -750,20 +750,25 @@ impl Runtime {
         if let Some(receipt) = receipt {
             return Ok(receipt);
         }
-        let result = super::envelope(&body, "result_envelope")?;
-        if result.len() > super::queries::RESULT_BYTES {
-            return Err(refuse(
-                StatusCode::PAYLOAD_TOO_LARGE,
-                "query_result_exceeds_budget",
-            ));
+        // A plain `result` without an envelope needs no decoding.
+        let plain_only = body.get("result").is_some_and(|value| !value.is_null())
+            && body.get("result_envelope").is_none_or(Value::is_null);
+        if !plain_only {
+            let result = super::envelope(&body, "result_envelope")?;
+            if result.len() > super::queries::RESULT_BYTES {
+                return Err(refuse(
+                    StatusCode::PAYLOAD_TOO_LARGE,
+                    "query_result_exceeds_budget",
+                ));
+            }
+            super::queries::decode(
+                self.signal_codec.clone(),
+                result,
+                false,
+                "invalid_query_result",
+            )
+            .await?;
         }
-        super::queries::decode(
-            self.signal_codec.clone(),
-            result,
-            false,
-            "invalid_query_result",
-        )
-        .await?;
         self.finish_query(task_id, body, false).await
     }
     pub(crate) async fn finish_query(

rust/src/runtime/http.rs

@@ -271,8 +271,10 @@ pub fn router(runtime: Runtime) -> Router {
         .route("/api/worker/query-tasks/{task_id}/fail", post(fail_query))
         .fallback(unavailable)
         .layer(middleware::from_fn(empty_list_body))
-        .layer(DefaultBodyLimit::max(3 * 1024 * 1024))
+        // `payload_limits` reads and bounds every body first.
+        .layer(DefaultBodyLimit::disable())
         .layer(middleware::from_fn_with_state(runtime.clone(), guard))
+        .layer(middleware::from_fn(super::payload_limits::enforce))
         .with_state(runtime)
 }
 
@@ -758,7 +760,7 @@ async fn delete_search_attribute(
     Ok(Json(runtime.delete_search_attribute(&name).await?))
 }
 
-async fn admission_error(
+pub(super) async fn admission_error(
     request: Request,
     worker: bool,
     status: StatusCode,

rust/src/runtime/mod.rs

@@ -42,6 +42,7 @@ pub mod mysql;
 mod namespaces;
 mod operator_dashboard;
 mod operator_metrics;
+mod payload_limits;
 mod poll_readiness;
 pub mod postgres;
 mod previews;

rust/src/runtime/payload_limits.rs

@@ -0,0 +1,152 @@
+//! Published PHP Server's `EnforcePayloadLimits`, which runs before every
+//! other middleware, authentication included: a body over the payload limit,
+//! a body without a JSON media type and a body that is not valid JSON are each
+//! refused with PHP's JSON error, in the worker or control-plane envelope.
+//! The external payload upload keeps its own transport rules.
+use axum::{
+    body::{Body, to_bytes},
+    extract::Request,
+    http::{Method, StatusCode, header},
+    middleware::Next,
+    response::Response,
+};
+use serde_json::{Value, json};
+
+/// `server.limits.max_payload_bytes`.
+fn max_payload_bytes() -> usize {
+    ["DW_MAX_PAYLOAD_BYTES", "WORKFLOW_MAX_PAYLOAD_BYTES"]
+        .iter()
+        .find_map(|key| std::env::var(key).ok())
+        .and_then(|value| value.trim().parse::<usize>().ok())
+        .unwrap_or(2 * 1024 * 1024)
+}
+
+pub(super) async fn enforce(request: Request, next: Next) -> Response {
+    let path = request.uri().path().to_owned();
+    if request.method() == Method::POST && path == "/api/external-payloads/v1" {
+        return next.run(request).await;
+    }
+    let max = max_payload_bytes();
+    let worker = path == "/api/worker" || path.starts_with("/api/worker/");
+    let declared = request
+        .headers()
+        .get(header::CONTENT_LENGTH)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.trim().parse::<u64>().ok());
+    let (parts, body) = request.into_parts();
+    let refuse = |status: StatusCode, body: Value| {
+        let parts = parts.clone();
+        async move { super::http::admission_error(Request::from_parts(parts, Body::empty()), worker, status, body).await }
+    };
+    let too_large = || json!({"message": format!("Request payload exceeds the maximum allowed size of {max} bytes."),
+        "reason": "payload_too_large", "limit": max});
+    if declared.is_some_and(|declared| declared > max as u64) {
+        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+    }
+    let Ok(bytes) = to_bytes(body, max.saturating_add(1)).await else {
+        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+    };
+    if bytes.len() > max {
+        return refuse(StatusCode::PAYLOAD_TOO_LARGE, too_large()).await;
+    }
+    let has_body = !bytes.is_empty() || declared.is_some_and(|declared| declared > 0);
+    if matches!(parts.method, Method::POST | Method::PUT | Method::PATCH | Method::DELETE) && has_body {
+        if !json_media_type(parts.headers.get(header::CONTENT_TYPE).and_then(|value| value.to_str().ok())) {
+            return refuse(StatusCode::UNSUPPORTED_MEDIA_TYPE, json!({"message": "Request bodies must use a JSON media type.",
+                "reason": "unsupported_media_type", "accepted_content_types": ["application/json", "application/*+json"]})).await;
+        }
+        if let Some(error) = json_error(&bytes) {
+            return refuse(StatusCode::BAD_REQUEST, json!({"message": "Request bodies must contain valid JSON.",
+                "reason": "malformed_json", "json_error": error})).await;
+        }
+    }
+    next.run(Request::from_parts(parts, Body::from(bytes))).await
+}
+
+/// `usesJsonMediaType`: `application/json` or any `+json` type.
+fn json_media_type(content_type: Option<&str>) -> bool {
+    let Some(content_type) = content_type.filter(|value| !value.trim().is_empty()) else {
+        return false;
+    };
+    let media = content_type.split(';').next().unwrap_or("").trim().to_ascii_lowercase();
+    media == "application/json" || media.ends_with("+json")
+}
+
+/// PHP's `json_last_error_msg()` after `json_decode` of the body, or `None`
+/// for valid JSON. PHP nests up to 512 levels.
+fn json_error(bytes: &[u8]) -> Option<&'static str> {
+    // Skipping a value leaves escapes and UTF-8 unchecked, so decode it.
+    let error = serde_json::from_slice::<Value>(bytes).err()?;
+    let message = error.to_string();
+    Some(if message.contains("recursion limit exceeded") {
+        if depth(bytes) <= 512 {
+            return None;
+        }
+        "Maximum stack depth exceeded"
+    } else if message.contains("control character") {
+        "Control character error, possibly incorrectly encoded"
+    } else if message.contains("surrogate") || message.contains("unexpected end of hex escape") {
+        "Single unpaired UTF-16 surrogate in unicode escape"
+    } else if message.contains("invalid unicode code point") {
+        "Malformed UTF-8 characters, possibly incorrectly encoded"
+    } else {
+        "Syntax error"
+    })
+}
+
+/// The deepest array or object nesting, outside strings.
+fn depth(bytes: &[u8]) -> usize {
+    let (mut current, mut deepest, mut in_string, mut escaped) = (0usize, 0usize, false, false);
+    for &byte in bytes {
+        if in_string {
+            match (escaped, byte) {
+                (true, _) => escaped = false,
+                (false, b'\\') => escaped = true,
+                (false, b'"') => in_string = false,
+                _ => {}
+            }
+            continue;
+        }
+        match byte {
+            b'"' => in_string = true,
+            b'[' | b'{' => {
+                current += 1;
+                deepest = deepest.max(current);
+            }
+            b']' | b'}' => current = current.saturating_sub(1),
+            _ => {}
+        }
+    }
+    deepest
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn errors_name_php_s_json_failures() {
+        assert_eq!(json_error(br#"{"a":1}"#), None);
+        assert_eq!(json_error(b"\"text\""), None);
+        assert_eq!(json_error(br#"{"a":"#), Some("Syntax error"));
+        assert_eq!(json_error(b"   "), Some("Syntax error"));
+        assert_eq!(json_error(b"{\"a\":\"\x01\"}"), Some("Control character error, possibly incorrectly encoded"));
+        assert_eq!(json_error(br#"{"a":"\ud800"}"#), Some("Single unpaired UTF-16 surrogate in unicode escape"));
+        assert_eq!(json_error(br#"{"a":"\udc00"}"#), Some("Single unpaired UTF-16 surrogate in unicode escape"));
+        assert_eq!(json_error(br#"{"a":"\ud83d\ude00"}"#), None);
+        assert_eq!(json_error(b"{\"a\":\"\xff\"}"), Some("Malformed UTF-8 characters, possibly incorrectly encoded"));
+        // PHP accepts nesting serde_json's default limit refuses.
+        let nested = |levels: usize| format!("{}{}", "[".repeat(levels), "]".repeat(levels)).into_bytes();
+        assert_eq!(json_error(&nested(200)), None);
+        assert_eq!(json_error(&nested(600)), Some("Maximum stack depth exceeded"));
+    }
+
+    #[test]
+    fn media_types_follow_php() {
+        assert!(json_media_type(Some("application/json")));
+        assert!(json_media_type(Some("Application/JSON; charset=utf-8")));
+        assert!(json_media_type(Some("application/vnd.api+json")));
+        assert!(!json_media_type(Some("text/plain")));
+        assert!(!json_media_type(None));
+    }
+}

rust/src/runtime/queries.rs

@@ -458,7 +458,7 @@ where
             match task["status"].as_str() {
                 Some("completed") => {
                     let mut answer = json!({"workflow_id":workflow_id,"run_id":run_id,"target_scope":"instance",
-                        "query_name":name,"result":null,"result_envelope":task["result_envelope"],"reason":null});
+                        "query_name":name,"result":task["result"],"result_envelope":task["result_envelope"],"reason":null});
                     if task["principal"].is_object() {
                         answer["principal"] = task["principal"].clone();
                     }
@@ -1153,6 +1153,7 @@ where
         let reason: Option<String>;
         let failure;
         let result;
+        let mut plain = None;
         if failed {
             let supplied = &body["failure"];
             if !supplied.is_object() || serde_json::to_vec(supplied)?.len() > 32 * 1024 {
@@ -1171,14 +1172,22 @@ where
             failure = Some(supplied.clone());
             result = None;
         } else {
-            let supplied = envelope(body, "result_envelope")?;
-            if supplied.len() > RESULT_BYTES {
+            // `completeQueryTask`: a worker may also send the plain `result`,
+            // which published PHP returns to the caller beside the envelope.
+            plain = body.get("result").filter(|value| !value.is_null()).cloned();
+            let supplied = if plain.is_none() || body.get("result_envelope").is_some_and(|value| !value.is_null()) {
+                Some(envelope(body, "result_envelope")?)
+            } else {
+                None
+            };
+            let plain_bytes = plain.as_ref().map(serde_json::to_vec).transpose()?.map_or(0, |bytes| bytes.len());
+            if supplied.as_ref().map_or(0, String::len) + plain_bytes > RESULT_BYTES {
                 return Err(refuse(
                     StatusCode::PAYLOAD_TOO_LARGE,
                     "query_result_exceeds_budget",
                 ));
             }
-            result = Some(wire(&supplied));
+            result = supplied.map(|supplied| wire(&supplied));
             failure = None;
             reason = None;
         }
@@ -1196,6 +1205,9 @@ where
         if let Some(result) = result {
             task["result_envelope"] = result;
         }
+        if let Some(plain) = plain {
+            task["result"] = plain;
+        }
         task["terminal_receipt"] = json!({"lease_owner":owner,"attempt":attempt,
             "fingerprint":fingerprint,"decision":decision,"decided_at":authority::wire(decided_at)});
         // Results reserve their slot at admission, and discard the large replay

rust/tests/execution.rs

@@ -79,6 +79,10 @@ mod worker_deregistration;
 mod worker_listing;
 #[path = "execution/activity_failure_fields.rs"]
 mod activity_failure_fields;
+#[path = "execution/query_worker_results.rs"]
+mod query_worker_results;
+#[path = "execution/payload_limits.rs"]
+mod payload_limits;
 #[path = "execution/query_open_tasks.rs"]
 mod query_open_tasks;
 #[path = "execution/retry_message_tasks.rs"]