diff --git a/docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md b/docs/rust-server-port-checkpoints/2026-10-22-query-worker-result.md new file mode 100644 index 000000000..f202466fa --- /dev/null +++ b/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. diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 8427fe452..4137d1196 100644 --- a/rust/src/runtime/execution.rs +++ b/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( diff --git a/rust/src/runtime/queries.rs b/rust/src/runtime/queries.rs index 52a9b5bf3..b2245e017 100644 --- a/rust/src/runtime/queries.rs +++ b/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; 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 diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 46752e475..a041ba80e 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -79,6 +79,8 @@ 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/query_open_tasks.rs"] mod query_open_tasks; #[path = "execution/retry_message_tasks.rs"] diff --git a/rust/tests/execution/query_worker_results.rs b/rust/tests/execution/query_worker_results.rs new file mode 100644 index 000000000..14ae7ec86 --- /dev/null +++ b/rust/tests/execution/query_worker_results.rs @@ -0,0 +1,40 @@ +//! A query worker's plain `result`, which published PHP stores beside the +//! result envelope and returns to the caller (`completeQueryTask`). +use super::*; + +async fn ask(app: &Router, workflow_id: &str) -> tokio::task::JoinHandle<(StatusCode, Value)> { + let app = app.clone(); + let path = format!("/api/workflows/{workflow_id}/query/state"); + tokio::spawn(async move { + request(&app, "POST", &path, json!({"input":envelope(Payload::Array(vec![Payload::Long(1)]))})).await + }) +} + +#[tokio::test] +async fn query_answers_carry_the_worker_s_plain_result() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + query_worker(&app, "plain-query-worker").await; + let plain = json!({"received":0,"last":null}); + for (workflow_id, with_envelope) in [("plain-and-envelope", true), ("plain-only", false)] { + start(&app, workflow_id).await; + let asked = ask(&app, workflow_id).await; + let claim = await_query(&app, "plain-query-worker").await; + let mut completion = json!({"lease_owner":"plain-query-worker","query_task_attempt":claim["query_task_attempt"], + "result":plain}); + if with_envelope { + completion["result_envelope"] = envelope(Payload::Long(7)); + } + let (status, body) = request(&app, "POST", + &format!("/api/worker/query-tasks/{}/complete", claim["query_task_id"].as_str().unwrap()), completion).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let (status, answer) = asked.await.unwrap(); + assert_eq!(status, StatusCode::OK, "{answer}"); + assert_eq!(answer["result"], plain, "{answer}"); + assert_eq!(answer["control_plane"]["result"], plain, "{answer}"); + assert_eq!(answer["result_envelope"].is_object(), with_envelope, "{answer}"); + } + runtime.close().await; + database.remove().await; +}