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 fed2ee077..5fe3195d4 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; #[path = "execution/child_start_turn.rs"] mod child_start_turn; @@ -1125,10 +1127,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; +}