From fc08735efd891983bc79dcaedd898e53897d92da Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:54:59 -0600 Subject: [PATCH] Echo the worker protocol version in every worker-plane body (#1001) The Server reference says worker-plane responses echo the advertised version in the header and in a protocol_version body member, and PHP's WorkerProtocol::json adds protocol_version and server_capabilities to every worker answer that lacks them. Native added only the header, so answers whose handler did not write the members itself (successful polls, query tasks, registrations, history pages and many refusals) went without, and the published CLI's `dw workflow-task:poll` refused them. The worker middleware now appends each missing member to a JSON object answer of known size, keeping the body's bytes and never re-serializing numbers; a member a handler already wrote is left as it is. Activity heartbeats wrote the request's version where PHP sends the server's, so they now send the server's. The execution suite checks every worker answer it sees: body protocol_version equal to the header, and server_capabilities present. A recorder compares the full suite before and after: only missing members are added, apart from the heartbeat version correction. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/activity_heartbeats.rs | 8 +- rust/src/runtime/http.rs | 110 ++++++++++++++++++ rust/tests/execution.rs | 48 +++++++- rust/tests/execution/activity_deadlines.rs | 6 +- rust/tests/execution/pending_limits.rs | 6 + .../execution/worker_protocol_envelope.rs | 52 +++++++++ 6 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 rust/tests/execution/worker_protocol_envelope.rs diff --git a/rust/src/runtime/activity_heartbeats.rs b/rust/src/runtime/activity_heartbeats.rs index c7f25397c..de7bb5f21 100644 --- a/rust/src/runtime/activity_heartbeats.rs +++ b/rust/src/runtime/activity_heartbeats.rs @@ -154,7 +154,7 @@ where "last_heartbeat_at":DB::optional_instant(&attempt,"last_heartbeat_at")?.or(DB::optional_instant(&activity,"last_heartbeat_at")?).map(super::cooperative::timestamp), "run_status":run_status,"run_closed_reason":DB::optional_string(&run,"closed_reason")?,"run_closed_at":DB::optional_instant(&run,"closed_at")?.map(super::cooperative::timestamp), "activity_status":activity_status,"attempt_status":attempt_status,"task_status":task_status, - "deadlines":super::activity_deadlines::deadlines_view::(&activity)?,"worker_session":worker_session,"protocol_version":protocol,"server_capabilities":super::store::capabilities()}); + "deadlines":super::activity_deadlines::deadlines_view::(&activity)?,"worker_session":worker_session,"protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}); // A cooperative request that reached this attempt shows its receipt. if cancel_requested && let Some(receipt) = Self::activity_cancellation_receipt(tx, &run, &DB::string(&activity, "id")?, attempt_id).await? @@ -233,7 +233,7 @@ where "run_closed_reason":observed["run_closed_reason"],"run_closed_at":observed["run_closed_at"], "heartbeat_recorded":false,"lease_expires_at":observed["lease_expires_at"], "last_heartbeat_at":observed["last_heartbeat_at"],"worker_session":observed["worker_session"], - "protocol_version":protocol,"server_capabilities":super::store::capabilities()}); + "protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}); tx.commit().await?; return Ok(response); } @@ -288,7 +288,7 @@ where "run_closed_reason":observed["run_closed_reason"],"run_closed_at":observed["run_closed_at"], "heartbeat_recorded":false,"lease_expires_at":observed["lease_expires_at"], "last_heartbeat_at":observed["last_heartbeat_at"],"error":renewal["error"],"worker_session":snapshot, - "worker_session_renewal":renewal,"protocol_version":protocol,"server_capabilities":super::store::capabilities()}) }); + "worker_session_renewal":renewal,"protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}) }); } } } @@ -353,7 +353,7 @@ where // PHP formats each time in UTC with microseconds, as status reads do. "heartbeat_recorded":true,"lease_expires_at":super::cooperative::timestamp(expires), "last_heartbeat_at":super::cooperative::timestamp(at),"worker_session":worker_session, - "protocol_version":protocol,"server_capabilities":super::store::capabilities()}), + "protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}), ) } } diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 90dabf957..f8631cd27 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -692,12 +692,75 @@ async fn guard(State(runtime): State, mut request: Request, next: Next) if response.extensions().get::().is_some() && (worker || route.is_some()) { response = typed_refusal_envelope(response, route, worker).await; } + if worker { + response = worker_protocol_envelope(response).await; + } response .headers_mut() .insert(header, HeaderValue::from_static(version)); response } +/// The largest worker answer whose envelope is completed here. Bigger or +/// unsized bodies pass through unchanged rather than being buffered. +const WORKER_ENVELOPE_MAX_BYTES: u64 = 64 * 1024 * 1024; + +/// PHP's `WorkerProtocol::json`: every worker-plane JSON object answer +/// carries `protocol_version` and `server_capabilities`, each added only when +/// the handler left it out (`??=`), so a value already present is never +/// changed (server#1001). The body's bytes are kept and the missing members +/// are appended before its closing brace, as PHP appends them; numbers are +/// never re-serialized. +async fn worker_protocol_envelope(response: Response) -> Response { + use axum::body::HttpBody as _; + let json = response + .headers() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.starts_with("application/json")); + let size = response.body().size_hint().exact(); + if !json || !size.is_some_and(|size| size <= WORKER_ENVELOPE_MAX_BYTES) { + return response; + } + let (mut parts, body) = response.into_parts(); + let Ok(bytes) = axum::body::to_bytes(body, WORKER_ENVELOPE_MAX_BYTES as usize).await else { + // An exact-sized in-memory body cannot fail to collect. + return (StatusCode::INTERNAL_SERVER_ERROR, Json(json!({"reason":"worker_response_unavailable"}))).into_response(); + }; + match complete_worker_envelope(&bytes) { + Some(completed) => { + parts.headers.remove(header::CONTENT_LENGTH); + Response::from_parts(parts, Body::from(completed)) + } + None => Response::from_parts(parts, Body::from(bytes)), + } +} + +/// The body with any missing envelope member appended, or `None` when it is +/// not a JSON object or already has both members. +fn complete_worker_envelope(bytes: &[u8]) -> Option> { + // Only the member names are read; values are skipped unparsed. + let members: std::collections::HashMap = serde_json::from_slice(bytes).ok()?; + let mut missing = Vec::new(); + if !members.contains_key("protocol_version") { + missing.push(format!("\"protocol_version\":{}", json!(super::WORKER_PROTOCOL_VERSION))); + } + if !members.contains_key("server_capabilities") { + missing.push(format!("\"server_capabilities\":{}", capabilities())); + } + if missing.is_empty() { + return None; + } + let end = bytes.iter().rposition(|byte| *byte == b'}')?; + let mut completed = bytes[..end].to_vec(); + if !members.is_empty() { + completed.push(b','); + } + completed.extend_from_slice(missing.join(",").as_bytes()); + completed.extend_from_slice(&bytes[end..]); + Some(completed) +} + /// PHP renders a typed external payload error through `WorkerProtocol` or /// `ControlPlaneProtocol`, which add the route's envelope fields. async fn typed_refusal_envelope(response: Response, route: Option, worker: bool) -> Response { @@ -2894,3 +2957,50 @@ async fn unavailable() -> Response { ) .into_response() } + +#[cfg(test)] +mod worker_envelope_tests { + use super::*; + + fn complete(text: &str) -> Option { + complete_worker_envelope(text.as_bytes()).map(|bytes| String::from_utf8(bytes).unwrap()) + } + + #[test] + fn missing_members_are_appended_and_present_ones_untouched() { + let completed = complete(r#"{"task":null,"poll_status":"empty"}"#).unwrap(); + let body: Value = serde_json::from_str(&completed).unwrap(); + assert_eq!(body["protocol_version"], super::super::WORKER_PROTOCOL_VERSION); + assert_eq!(body["server_capabilities"], capabilities()); + // The original members keep their exact bytes and order. + assert!(completed.starts_with(r#"{"task":null,"poll_status":"empty","protocol_version":"#)); + // A handler's own server_capabilities is kept; only the version is added. + let completed = complete(r#"{"task":null,"server_capabilities":{"own":true}}"#).unwrap(); + let body: Value = serde_json::from_str(&completed).unwrap(); + assert_eq!(body["server_capabilities"], json!({"own": true})); + assert_eq!(body["protocol_version"], super::super::WORKER_PROTOCOL_VERSION); + // A handler's own protocol_version is never replaced. + let completed = complete(r#"{"protocol_version":"1.7","x":1}"#).unwrap(); + let body: Value = serde_json::from_str(&completed).unwrap(); + assert_eq!(body["protocol_version"], "1.7"); + } + + #[test] + fn complete_bodies_and_non_objects_are_left_alone() { + assert_eq!(complete(r#"{"protocol_version":"1.20","server_capabilities":{}}"#), None); + for body in [r#"[1,2]"#, r#""text""#, "42", "null", "{not json", ""] { + assert_eq!(complete(body), None, "{body}"); + } + } + + #[test] + fn numbers_and_whitespace_keep_their_bytes() { + // Re-serializing would turn these into 1.1 and a rounded float. + let completed = complete("{\"a\":1.10,\"big\":123456789012345678901234567890} \n").unwrap(); + assert!(completed.starts_with("{\"a\":1.10,\"big\":123456789012345678901234567890,\"protocol_version\":")); + assert!(completed.ends_with("} \n")); + let completed = complete("{}").unwrap(); + assert!(completed.starts_with("{\"protocol_version\":")); + serde_json::from_str::(&completed).unwrap(); + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c7b547bd4..4585a3cbf 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -370,6 +370,8 @@ mod key_order; mod activity_result_repeats; #[path = "execution/task_fairness.rs"] mod task_fairness; +#[path = "execution/worker_protocol_envelope.rs"] +mod worker_protocol_envelope; enum TestDatabase { Sqlite(tempfile::TempDir), @@ -1123,10 +1125,54 @@ async fn request_protocol( .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); let status = response.status(); + let header = response + .headers() + .get("x-durable-workflow-protocol-version") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); let body = to_bytes(response.into_body(), 4 * 1024 * 1024) .await .unwrap(); - (status, serde_json::from_slice(&body).unwrap()) + let body: Value = serde_json::from_slice(&body).unwrap(); + if path.starts_with("/api/worker/") { + worker_envelope_sweep(method, path, status, header.as_deref(), &body); + } + (status, body) +} + +/// Every worker-plane JSON object answer this suite sees carries the +/// `protocol_version` body member its header advertises, plus +/// `server_capabilities`, as PHP's `WorkerProtocol::json` guarantees +/// (server#1001). With `DW_WORKER_ENVELOPE_SWEEP` set to a file, each +/// answer's shape is also recorded there, for a before/after comparison. +fn worker_envelope_sweep(method: &str, path: &str, status: StatusCode, header: Option<&str>, body: &Value) { + let Some(members) = body.as_object() else { + return; + }; + if let Ok(file) = std::env::var("DW_WORKER_ENVELOPE_SWEEP") { + use std::io::Write; + let route = path + .split('/') + .map(|segment| if segment.len() >= 20 && segment.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') { "{id}" } else { segment }) + .collect::>() + .join("/"); + let route = route.split('?').next().unwrap_or_default(); + let mut keys: Vec<&String> = members.keys().collect(); + keys.sort(); + let record = json!({"method": method, "route": route, "status": status.as_u16(), "keys": keys, + "protocol_version": members.get("protocol_version"), + "server_capabilities": members.get("server_capabilities").map(|value| value.to_string())}); + // Tests run in parallel: one locked write per record keeps lines whole. + static SWEEP_FILE: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _guard = SWEEP_FILE.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let mut out = std::fs::OpenOptions::new().create(true).append(true).open(file).unwrap(); + out.write_all(format!("{record}\n").as_bytes()).unwrap(); + } + assert_eq!(header, Some("1.20"), "{method} {path}: worker protocol header"); + assert_eq!(members.get("protocol_version").and_then(Value::as_str), header, + "{method} {path} {status}: body protocol_version must echo the header: {body}"); + assert!(members.get("server_capabilities").is_some_and(Value::is_object), + "{method} {path} {status}: body server_capabilities: {body}"); } async fn register(app: &Router, worker: &str, workflows: Value, activities: Value) { diff --git a/rust/tests/execution/activity_deadlines.rs b/rust/tests/execution/activity_deadlines.rs index ff93c43c1..974fd63b8 100644 --- a/rust/tests/execution/activity_deadlines.rs +++ b/rust/tests/execution/activity_deadlines.rs @@ -219,7 +219,11 @@ async fn every_authored_deadline_fences_outcomes_and_resolves_once() { if outcome == "heartbeat" { assert_eq!(status.1["can_continue"], false); assert_eq!(status.1["heartbeat_recorded"], false); - assert_eq!(status.1["protocol_version"], "1.19"); + // The server's advertised version, not the 1.19 request's: + // PHP 2.11.8's heartbeat answers (ActivityTaskController + // lines 865 and 903) set no protocol_version, so + // WorkerProtocol::json supplies the server's (server#1001). + assert_eq!(status.1["protocol_version"], "1.20"); } } } diff --git a/rust/tests/execution/pending_limits.rs b/rust/tests/execution/pending_limits.rs index befb278eb..6b2481b74 100644 --- a/rust/tests/execution/pending_limits.rs +++ b/rust/tests/execution/pending_limits.rs @@ -34,6 +34,12 @@ async fn a_pending_count_past_its_limit_refuses_the_whole_completion() { // The first command meets a limit of one; the second passes it. let (status, body) = finish_task(&app, &task, json!([command(kind), command(kind)])).await; assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{kind}: {body}"); + // The worker envelope PHP's WorkerProtocol::json adds (server#1001), + // then the refusal's own members exactly. + let mut body = body; + let members = body.as_object_mut().unwrap(); + assert_eq!(members.remove("protocol_version"), Some(json!("1.20")), "{kind}"); + assert!(members.remove("server_capabilities").is_some_and(|capabilities| capabilities.is_object()), "{kind}"); assert_eq!(body, json!({"task_id":task["task_id"],"workflow_task_attempt":task["workflow_task_attempt"], "outcome":"rejected","recorded":false,"error":format!("Structural limit exceeded: 1 {noun} (limit 1)."), "reason":"structural_limit_exceeded","limit_kind":limit_kind,"current_value":1,"configured_limit":1}), "{kind}"); diff --git a/rust/tests/execution/worker_protocol_envelope.rs b/rust/tests/execution/worker_protocol_envelope.rs new file mode 100644 index 000000000..59d5fe907 --- /dev/null +++ b/rust/tests/execution/worker_protocol_envelope.rs @@ -0,0 +1,52 @@ +use super::*; + +// Server #1001: like PHP's `WorkerProtocol::json`, every worker-plane JSON +// answer carries the `protocol_version` body member its header advertises, +// plus `server_capabilities`. Native 3.0.5 sent only the header on successful +// polls, so the published CLI's `dw workflow-task:poll` refused the answer. + +/// The raw answer to a worker poll, with its protocol header. +async fn raw_poll(app: &Router, kind: &str, worker: &str) -> (StatusCode, Option, Value) { + let request = Request::builder() + .method("POST") + .uri(format!("/api/worker/{kind}-tasks/poll")) + .header("authorization", "Bearer test-token") + .header("content-type", "application/json") + .header("x-namespace", "default") + .header("x-durable-workflow-protocol-version", "1.0") + .body(Body::from(json!({"worker_id":worker,"task_queue":"test","timeout_seconds":0}).to_string())) + .unwrap(); + let response = app.clone().oneshot(request).await.unwrap(); + let status = response.status(); + let header = response + .headers() + .get("x-durable-workflow-protocol-version") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let body = to_bytes(response.into_body(), 4 * 1024 * 1024).await.unwrap(); + (status, header, serde_json::from_slice(&body).unwrap()) +} + +#[tokio::test] +async fn leased_and_empty_polls_echo_the_protocol_version_in_the_body() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "envelope-worker", json!(["echo"]), json!(["activity"])).await; + start(&app, "envelope-poll").await; + for (kind, leased) in [("workflow", true), ("workflow", false), ("activity", false)] { + // A 1.0 worker, as the published CLI 2.3.3 sends, gets the server's + // advertised version back in both places. + let (status, header, body) = raw_poll(&app, kind, "envelope-worker").await; + assert_eq!(status, StatusCode::OK, "{kind}: {body}"); + assert_eq!(body["task"].is_object(), leased, "{kind}: {body}"); + assert_eq!(header.as_deref(), Some("1.20"), "{kind}"); + assert_eq!(body["protocol_version"], "1.20", "{kind}: {body}"); + assert!(body["server_capabilities"].is_object(), "{kind}: {body}"); + if kind == "workflow" { + assert_eq!(body["poll_status"], if leased { "leased" } else { "empty" }, "{kind}: {body}"); + } + } + runtime.close().await; + database.remove().await; +}