Close the worker-envelope follow-ups from the #1003 review

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.

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"

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<Value> {
+    async fn observe_activity(&self, task_id: &str, body: Value) -> Result<Value> {
         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<Value> {
+    pub(crate) async fn heartbeat_activity(&self, task_id: &str, body: Value) -> Result<Value> {
         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

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<Value>);
+    delegate!(heartbeat_activity(task_id: &str, body: Value) -> Result<Value>);
     delegate!(activity_status(task_id: &str, body: Value, protocol: &str) -> Result<Value>);
     delegate!(renew_activity_lease(task_id: &str, body: Value) -> Result<Value>);
     delegate!(workflow_in_namespace(workflow_id: &str) -> Result<bool>);

rust/src/runtime/http.rs

@@ -652,7 +652,9 @@ async fn guard(State(runtime): State<Runtime>, 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<Runtime>, mut request: Request, next: Next)
     if response.extensions().get::<super::external_transport::TypedRefusal>().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<Runtime>, 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<Vec<u8>> {
-    // Only the member names are read; values are skipped unparsed.
-    let members: std::collections::HashMap<String, serde::de::IgnoredAny> = 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<String> = 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<String> = 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<D: serde::Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
+        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<M: serde::de::MapAccess<'de>>(self, mut map: M) -> std::result::Result<Self::Value, M::Error> {
+                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<Value>,
 ) -> Result<Json<Value>> {
+    // 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::<Value>(&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);
+    }
 }

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)
 }
 

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;
+}