diff --git a/docs/server-reference.md b/docs/server-reference.md index 9b83d1eef..bf7a2912d 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -908,13 +908,13 @@ can still ask for the expired diagnostic set with `status=stale`, and stale workers stop being considered for query-task dispatch and routing-gate admission. -The current server advertises worker protocol `1.17` by default. Worker-plane +The current server advertises worker protocol `1.20` by default. Worker-plane requests should send the highest `X-Durable-Workflow-Protocol-Version` their SDK implements, and worker-plane responses always echo the server's advertised version in the same header plus the `protocol_version` body field. Worker-protocol compatibility is same-major with a worker minor less than or -equal to the server minor: a server advertising `1.17` accepts worker requests -for `1.0` through `1.17`, including older `1.16` and `1.15` workers. Requests +equal to the server minor: a server advertising `1.20` accepts worker requests +for `1.0` through `1.20`, including older `1.19` and `1.17` workers. Requests with a missing, malformed, different-major, or higher-minor protocol version are rejected with `missing_protocol_version` or `unsupported_protocol_version` and a `supported_version` value that tells the worker what the server advertises. diff --git a/rust/Cargo.toml b/rust/Cargo.toml index c539fbed3..9936e57d8 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -17,7 +17,7 @@ miniz_oxide = "0.9" thiserror = "2" axum = { version = "=0.8.9", default-features = false, features = ["http1", "json", "query", "tokio"] } serde = { version = "1", features = ["derive"] } -serde_json = { version = "1", features = ["preserve_order", "unbounded_depth"] } +serde_json = { version = "1", features = ["preserve_order", "raw_value", "unbounded_depth"] } sha2 = { version = "0.11", default-features = false } hmac = "0.13" form_urlencoded = "1.2" diff --git a/rust/src/runtime/activity_heartbeats.rs b/rust/src/runtime/activity_heartbeats.rs index de7bb5f21..0f02263a4 100644 --- a/rust/src/runtime/activity_heartbeats.rs +++ b/rust/src/runtime/activity_heartbeats.rs @@ -35,15 +35,15 @@ where response: json!({"reason":"activity_attempt_status_unavailable","minimum_protocol_version":"1.20"}), }); } - self.observe_activity(task_id, body, protocol).await + self.observe_activity(task_id, body).await } // Heartbeat can report lost authority on legacy protocols without granting // the separately capability-gated non-renewing status endpoint. - async fn observe_activity(&self, task_id: &str, body: Value, protocol: &str) -> Result { + async fn observe_activity(&self, task_id: &str, body: Value) -> Result { let mut tx = self.begin_read().await?; let (response, _) = self - .activity_observation(&mut tx, task_id, body, protocol, true) + .activity_observation(&mut tx, task_id, body, true) .await?; tx.commit().await?; Ok(response) @@ -54,7 +54,6 @@ where tx: &mut Transaction<'_, DB>, task_id: &str, body: Value, - protocol: &str, check_session: bool, ) -> Result<(Value, bool)> { reject_fields(&body, &["activity_attempt_id", "lease_owner"])?; @@ -164,12 +163,7 @@ where Ok((response, Self::cancellation_request(&run)?.is_none())) } - pub(crate) async fn heartbeat_activity( - &self, - task_id: &str, - body: Value, - protocol: &str, - ) -> Result { + pub(crate) async fn heartbeat_activity(&self, task_id: &str, body: Value) -> Result { reject_fields( &body, &[ @@ -209,7 +203,6 @@ where &mut tx, task_id, json!({"activity_attempt_id":attempt_id,"lease_owner":text(&body,"lease_owner")?}), - protocol, false, ) .await diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 7c18289ff..01ccecb1e 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -1786,7 +1786,7 @@ impl Runtime { Storage::MySql(store) => store.complete_activity(task_id, body, external).await, } } - delegate!(heartbeat_activity(task_id: &str, body: Value, protocol: &str) -> Result); + delegate!(heartbeat_activity(task_id: &str, body: Value) -> Result); delegate!(activity_status(task_id: &str, body: Value, protocol: &str) -> Result); delegate!(renew_activity_lease(task_id: &str, body: Value) -> Result); delegate!(workflow_in_namespace(workflow_id: &str) -> Result); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 65942f12d..628cc2bb2 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -652,7 +652,9 @@ async fn guard(State(runtime): State, mut request: Request, next: Next) }; let exists = match runtime.namespace_exists(&namespace).await { Ok(exists) => exists, - Err(error) => return error.into_response(), + // A storage failure here is still an answer on this plane: it gets + // the same envelope and version header as a handler's answer. + Err(error) => return finish_answer(error.into_response(), worker, header, version).await, }; if !exists { let body = json!({"reason":"namespace_not_found", "namespace":namespace, @@ -692,6 +694,12 @@ 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; } + finish_answer(response, worker, header, version).await +} + +/// Every answer on a plane carries its version header, and every worker +/// answer its body envelope. +async fn finish_answer(mut response: Response, worker: bool, header: &'static str, version: &'static str) -> Response { if worker { response = worker_protocol_envelope(response).await; } @@ -706,11 +714,11 @@ async fn guard(State(runtime): State, mut request: Request, next: Next) 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. +/// carries `protocol_version` and `server_capabilities`, each set only when +/// the handler left it out or null (`??=`), so a non-null value is never +/// changed (server#1001). A missing member is appended before the body's +/// closing brace, as PHP appends it, and the other bytes are kept; numbers +/// are never re-serialized. async fn worker_protocol_envelope(response: Response) -> Response { use axum::body::HttpBody as _; let json = response @@ -736,29 +744,80 @@ async fn worker_protocol_envelope(response: Response) -> Response { } } -/// The body with any missing envelope member appended, or `None` when it is -/// not a JSON object or already has both members. +/// The body with each missing or null envelope member set, or `None` when it +/// is not a UTF-8 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() { + // A body that is not UTF-8 is not JSON; it passes through unchanged. + let text = std::str::from_utf8(bytes).ok()?; + // Values are kept as their raw text and never parsed into numbers. + let ObjectMembers(members) = serde_json::from_str(text).ok()?; + let envelope = [ + ("protocol_version", json!(super::WORKER_PROTOCOL_VERSION)), + ("server_capabilities", capabilities()), + ]; + // As in PHP's decoded array, the last of duplicate names is the value. + let value = |name: &str| { + members.iter().rev().find(|(key, _)| key == name).map(|(_, raw)| raw.get()) + }; + let unset: Vec<&(&str, Value)> = envelope + .iter() + .filter(|(name, _)| value(name).is_none_or(|raw| raw == "null")) + .collect(); + if unset.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','); + let member = |name: &str, value: &Value| format!("{}:{value}", json!(name)); + if unset.iter().all(|(name, _)| value(name).is_none()) { + let end = text.rfind('}')?; + let mut completed = text[..end].to_owned(); + if !members.is_empty() { + completed.push(','); + } + let appended: Vec = unset.iter().map(|(name, value)| member(name, value)).collect(); + completed.push_str(&appended.join(",")); + completed.push_str(&text[end..]); + return Some(completed.into_bytes()); + } + // An explicit null is replaced where it stands, as `??=` assigns to the + // existing key. The members are rewritten from their raw text, so only + // the whitespace between them changes. + let mut rendered: Vec = members + .iter() + .map(|(key, raw)| match unset.iter().find(|(name, _)| name == key) { + Some((name, value)) => member(name, value), + None => format!("{}:{}", json!(key), raw.get()), + }) + .collect(); + rendered.extend( + unset + .iter() + .filter(|(name, _)| value(name).is_none()) + .map(|(name, value)| member(name, value)), + ); + Some(format!("{{{}}}", rendered.join(",")).into_bytes()) +} + +/// A JSON object's members in order, each value kept as its raw text. +struct ObjectMembers<'a>(Vec<(String, &'a serde_json::value::RawValue)>); + +impl<'de: 'a, 'a> serde::Deserialize<'de> for ObjectMembers<'a> { + fn deserialize>(deserializer: D) -> std::result::Result { + struct Members<'a>(std::marker::PhantomData<&'a ()>); + impl<'de: 'a, 'a> serde::de::Visitor<'de> for Members<'a> { + type Value = ObjectMembers<'a>; + fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + formatter.write_str("a JSON object") + } + fn visit_map>(self, mut map: M) -> std::result::Result { + let mut members = Vec::new(); + while let Some(member) = map.next_entry()? { + members.push(member); + } + Ok(ObjectMembers(members)) + } + } + deserializer.deserialize_map(Members(std::marker::PhantomData)) } - 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 @@ -2536,9 +2595,12 @@ async fn heartbeat_activity( headers: HeaderMap, Json(body): Json, ) -> Result> { + // The worker-plane middleware already requires the header; keep the + // handler's own refusal so the error order is unchanged. + request_protocol(&headers)?; let attempt = body["activity_attempt_id"].clone(); runtime - .heartbeat_activity(&task_id, body, request_protocol(&headers)?) + .heartbeat_activity(&task_id, body) .await .map_err(|error| activity_not_found(error, &task_id, attempt)) .map(Json) @@ -3007,4 +3069,35 @@ mod worker_envelope_tests { assert!(completed.starts_with("{\"protocol_version\":")); serde_json::from_str::(&completed).unwrap(); } + + #[test] + fn explicit_null_members_are_set_where_they_stand() { + // PHP's `??=` assigns to a null member and keeps its position. + let completed = complete(r#"{"a":1.10,"protocol_version":null,"b":"x"}"#).unwrap(); + assert!(completed.starts_with(r#"{"a":1.10,"protocol_version":"1.20","b":"x","server_capabilities":{"#), "{completed}"); + let body: Value = serde_json::from_str(&completed).unwrap(); + assert_eq!(body["server_capabilities"], capabilities()); + let completed = complete(r#"{"server_capabilities":null,"protocol_version":"1.7"}"#).unwrap(); + let body: Value = serde_json::from_str(&completed).unwrap(); + assert_eq!(body["server_capabilities"], capabilities()); + assert_eq!(body["protocol_version"], "1.7"); + // Raw values, including big numbers, keep their bytes on this path too. + let completed = complete(r#"{ "big" : 123456789012345678901234567890, "protocol_version" : null }"#).unwrap(); + assert!(completed.starts_with(r#"{"big":123456789012345678901234567890,"protocol_version":"1.20","#), "{completed}"); + // A non-null value, even a falsy one, is never replaced. + for value in [r#""""#, "false", "0", "{}", "[]"] { + let body = format!(r#"{{"protocol_version":{value},"server_capabilities":{value}}}"#); + assert_eq!(complete(&body), None, "{body}"); + } + // The last of duplicate names decides, as in PHP's decoded array. + assert_eq!(complete(r#"{"protocol_version":null,"protocol_version":"1.7","server_capabilities":{}}"#), None); + let completed = complete(r#"{"protocol_version":"1.7","protocol_version":null,"server_capabilities":{}}"#).unwrap(); + assert!(!completed.contains("1.7"), "{completed}"); + } + + #[test] + fn bodies_that_are_not_utf8_pass_through() { + assert_eq!(complete_worker_envelope(b"{\"a\":\"\xff\"}"), None); + assert_eq!(complete_worker_envelope(b"{\"\xc3\x28\":1}"), None); + } } diff --git a/rust/tests/execution/admission.rs b/rust/tests/execution/admission.rs index 46822bb97..db8c635b3 100644 --- a/rust/tests/execution/admission.rs +++ b/rust/tests/execution/admission.rs @@ -40,6 +40,14 @@ pub(super) async fn admission_http( let headers = response.headers().clone(); let body = serde_json::from_slice(&to_bytes(response.into_body(), 1024 * 1024).await.unwrap()) .unwrap(); + // Admission answers on the worker plane are worker answers too + // (server#1001), so the suite's envelope sweep covers them. + if uri.starts_with("/api/worker/") { + let header = headers + .get("x-durable-workflow-protocol-version") + .and_then(|value| value.to_str().ok()); + super::worker_envelope_sweep(method, uri, status, header, &body); + } (status, headers, body) } diff --git a/rust/tests/execution/worker_protocol_envelope.rs b/rust/tests/execution/worker_protocol_envelope.rs index 59d5fe907..98a265b50 100644 --- a/rust/tests/execution/worker_protocol_envelope.rs +++ b/rust/tests/execution/worker_protocol_envelope.rs @@ -50,3 +50,51 @@ async fn leased_and_empty_polls_echo_the_protocol_version_in_the_body() { runtime.close().await; database.remove().await; } + +// A route's own refusal is an answer like any other: prepare and deliver +// carry the envelope at each stage they can refuse. They answer for a +// claimed task, so they are reached with a real claim. +#[tokio::test] +async fn scope_prepare_and_deliver_refusals_carry_the_envelope() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let registered = request_protocol(&app, "POST", "/api/worker/register", + json!({"worker_id":"envelope-scope","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[],"capabilities":["cooperative_cancellation"]}), "1.20").await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); + start(&app, "envelope-scope").await; + let (status, polled) = request_protocol(&app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":"envelope-scope","task_queue":"test","timeout_seconds":2}), "1.20").await; + assert_eq!(status, StatusCode::OK, "{polled}"); + let task = &polled["task"]; + let delivery = json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "scope_id":"no-such-scope","request_id":"envelope-request","sequence":1,"call_kind":"timer"}); + for route in ["prepare", "deliver"] { + let path = format!("/api/worker/workflow-tasks/{}/cancellation-scopes/{route}", task["task_id"].as_str().unwrap()); + // request_protocol's sweep asserts the header and both members. + for (body, protocol) in [(delivery.clone(), "1.19"), (json!({}), "1.20"), (delivery.clone(), "1.20")] { + let (status, answer) = request_protocol(&app, "POST", &path, body, protocol).await; + assert!(status.is_client_error(), "{route} {protocol}: {status} {answer}"); + assert!(answer["reason"].is_string() || answer["errors"].is_object(), "{route} {protocol}: {answer}"); + } + } + runtime.close().await; + database.remove().await; +} + +// A storage failure while resolving the namespace is still a worker answer: +// it gets the version header and the body envelope (server#1001 follow-up). +#[tokio::test] +async fn a_namespace_storage_failure_carries_the_envelope() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + runtime.close().await; + let (status, header, body) = raw_poll(&app, "workflow", "envelope-closed").await; + assert!(status.is_server_error(), "{status}: {body}"); + assert_eq!(header.as_deref(), Some("1.20")); + assert_eq!(body["protocol_version"], "1.20", "{body}"); + assert!(body["server_capabilities"].is_object(), "{body}"); + database.remove().await; +}