Native: ignore a caller's trace context where PHP Server admits it

docs/rust-server-port.md

@@ -229,6 +229,31 @@ support is not evidence of multi-node safety.
 Newest checkpoints first. Each entry is exact to the source, tree and runs it
 names, including unsuccessful observations.
 
+October 21, optional caller trace context:
+
+- **Why.** PHP Server's optional tracing ([#558](https://github.com/durable-workflow/server/pull/558),
+  Workflow #773) admits a top-level `observability_context` on start, signal,
+  update, cancel and request-cancellation requests when it is enabled. PHP
+  with tracing disabled, and published PHP 2.5.14, ignore the field. Native
+  refused it with 422 `unsupported_request_field`, so a caller that sent it
+  without checking discovery would have its business request refused.
+- **Source.** Those native handlers now drop exactly that top-level field
+  before reading the request. It is not validated, stored, logged or
+  compared, so a retried start with another trace context is the same
+  request. Native never advertises
+  `worker_protocol.server_capabilities.observability_context`. Terminate,
+  every other route, the same name anywhere else in a body, and any other
+  unknown field keep their existing handling.
+- **Test.** `trace_carriers` sends each of the nine affected routes with and
+  without a carrier to two identical workflows. It checks equal outcomes,
+  equal history event types, no stored trace identifier, the refused
+  terminate carrier and a refused unknown field beside the carrier. Removing
+  the change from any one of the nine handlers fails the test.
+- **Takeover.** Decided on [#325](https://github.com/durable-workflow/server/issues/325#issuecomment-6306545520):
+  a PHP 2.10 database with Core 2.9's nullable `observability_context`
+  columns is refused until its published layout is recorded. Its recipe will
+  count and then drop the three columns, like 2.9.7's `queue_delivery_reference`.
+
 October 20, external payload storage diagnostic:
 
 - **Source.** Native serves `POST /api/storage/test` for operators, in the

rust/src/runtime/http.rs

@@ -628,11 +628,23 @@ async fn activity_result(
     ))
 }
 
+/// PHP Server admits an optional top-level caller trace context on start,
+/// signal, update and cancellation requests when its tracing is enabled.
+/// Native has no carrier and never advertises one, so it drops that one field
+/// unread, as PHP does with tracing disabled. The same name anywhere else
+/// stays ordinary request data, and other unknown fields are still refused.
+fn without_trace_carrier(mut body: Value) -> Value {
+    if let Some(fields) = body.as_object_mut() {
+        fields.remove("observability_context");
+    }
+    body
+}
+
 async fn start(
     Extension(runtime): Extension<Runtime>,
     Json(body): Json<super::memo_json::StartBody>,
 ) -> Result<(StatusCode, Json<Value>)> {
-    let body = body.0;
+    let body = without_trace_carrier(body.0);
     let workflow_id = body["workflow_id"].as_str().unwrap_or("").to_owned();
     let result = runtime.start(body).await.map_err(|error| {
         super::control_plane::Operation::Start.refusal(error, &workflow_id, None)
@@ -697,7 +709,7 @@ async fn cancel_current(
     Path(workflow_id): Path<String>,
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
-    close(runtime, TerminalCommand::Cancel, &workflow_id, None, body).await
+    close(runtime, TerminalCommand::Cancel, &workflow_id, None, without_trace_carrier(body)).await
 }
 
 async fn terminate_current(
@@ -806,7 +818,7 @@ async fn cancel_run(
         TerminalCommand::Cancel,
         &workflow_id,
         Some(&run_id),
-        body,
+        without_trace_carrier(body),
     )
     .await
 }
@@ -832,7 +844,7 @@ async fn request_cancellation_current(
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
     let (status, response) = runtime
-        .request_cancellation(&workflow_id, None, body)
+        .request_cancellation(&workflow_id, None, without_trace_carrier(body))
         .await?;
     Ok((status, Json(response)))
 }
@@ -843,7 +855,7 @@ async fn request_cancellation_run(
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
     let (status, response) = runtime
-        .request_cancellation(&workflow_id, Some(&run_id), body)
+        .request_cancellation(&workflow_id, Some(&run_id), without_trace_carrier(body))
         .await?;
     Ok((status, Json(response)))
 }
@@ -854,7 +866,7 @@ async fn signal_current(
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
     let response = runtime
-        .signal(&workflow_id, None, &name, body)
+        .signal(&workflow_id, None, &name, without_trace_carrier(body))
         .await
         .map_err(|error| {
             super::control_plane::Operation::Signal(&name).refusal(error, &workflow_id, None)
@@ -871,7 +883,7 @@ async fn update_current(
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
     let (status, result) = runtime
-        .update_workflow(&workflow_id, None, &name, body)
+        .update_workflow(&workflow_id, None, &name, without_trace_carrier(body))
         .await?;
     Ok((status, Json(result)))
 }
@@ -882,7 +894,7 @@ async fn update_run(
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
     let (status, result) = runtime
-        .update_workflow(&workflow_id, Some(&run_id), &name, body)
+        .update_workflow(&workflow_id, Some(&run_id), &name, without_trace_carrier(body))
         .await?;
     Ok((status, Json(result)))
 }
@@ -893,7 +905,7 @@ async fn signal_run(
     Json(body): Json<Value>,
 ) -> Result<(StatusCode, Json<Value>)> {
     let response = runtime
-        .signal(&workflow_id, Some(&run_id), &name, body)
+        .signal(&workflow_id, Some(&run_id), &name, without_trace_carrier(body))
         .await
         .map_err(|error| {
             super::control_plane::Operation::Signal(&name).refusal(

rust/tests/execution.rs

@@ -157,6 +157,9 @@ mod cross_kind_poll;
 #[path = "execution/external_payloads.rs"]
 mod external_payloads;
 
+#[path = "execution/trace_carriers.rs"]
+mod trace_carriers;
+
 enum TestDatabase {
     Sqlite(tempfile::TempDir),
     Postgres {

rust/tests/execution/trace_carriers.rs

@@ -0,0 +1,184 @@
+//! A caller's optional trace context never changes a business outcome.
+use super::*;
+
+const TRACE_ID: &str = "4bf92f3577b34da6a3ce929d0e0e4736";
+
+fn traced(mut body: Value) -> Value {
+    body["observability_context"] = json!({"schema":"durable-workflow.trace-context","version":1,
+        "carrier":{"traceparent":format!("00-{TRACE_ID}-00f067aa0ba902b7-01"),"baggage":"tenant.id=t1"}});
+    body
+}
+
+fn number(value: i64) -> Value {
+    json!({"input":envelope(Payload::Array(vec![Payload::Long(value)]))})
+}
+
+/// What a caller can observe of an outcome, without generated identifiers.
+fn outcome(response: &(StatusCode, Value)) -> (StatusCode, Value, Value, Value) {
+    let body = &response.1;
+    (
+        response.0,
+        body["reason"].clone(),
+        body["outcome"].clone(),
+        body["update_status"].clone(),
+    )
+}
+
+async fn both(app: &Router, method: &str, path: &str, body: Value) {
+    let mut outcomes = Vec::new();
+    for (workflow, carried) in [("trace-plain", false), ("trace-carried", true)] {
+        let run = request(app, "GET", &format!("/api/workflows/{workflow}"), Value::Null)
+            .await
+            .1["run_id"]
+            .as_str()
+            .unwrap()
+            .to_owned();
+        let path = path.replace("{workflow}", workflow).replace("{run}", &run);
+        let body = if carried {
+            traced(body.clone())
+        } else {
+            body.clone()
+        };
+        let response = request(app, method, &path, body).await;
+        assert_ne!(
+            response.1["reason"], "unsupported_request_field",
+            "{path}: {}",
+            response.1
+        );
+        outcomes.push(outcome(&response));
+    }
+    assert_eq!(outcomes[0], outcomes[1], "{method} {path}");
+}
+
+async fn visible(app: &Router, workflow: &str) -> Value {
+    let described = request(app, "GET", &format!("/api/workflows/{workflow}"), Value::Null)
+        .await
+        .1;
+    let observed = request(
+        app,
+        "GET",
+        &format!("/api/workflows/{workflow}/observation"),
+        Value::Null,
+    )
+    .await
+    .1;
+    let history = run_history(app, &json!(workflow), &described["run_id"]).await;
+    json!({"described":described,"observed":observed,"history":history})
+}
+
+#[tokio::test]
+async fn caller_trace_context_is_ignored_where_php_admits_it_and_nowhere_else() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    update_worker(&app, "trace-worker").await;
+    for (workflow, carried) in [("trace-plain", false), ("trace-carried", true)] {
+        let body = json!({"workflow_id":workflow,"workflow_type":"echo","task_queue":"test",
+            "input":envelope(Payload::Array(vec![Payload::Long(9007199254740993)]))});
+        let started = request(
+            &app,
+            "POST",
+            "/api/workflows",
+            if carried { traced(body) } else { body },
+        )
+        .await;
+        assert_eq!(started.0, StatusCode::CREATED, "{}", started.1);
+        open_update_wait(&app, "trace-worker").await;
+    }
+    // A start retried with a different carrier is still the same request.
+    let mut retry = traced(json!({"workflow_id":"trace-carried","workflow_type":"echo",
+        "task_queue":"test","duplicate_policy":"use-existing",
+        "input":envelope(Payload::Array(vec![Payload::Long(9007199254740993)]))}));
+    retry["observability_context"]["carrier"]["traceparent"] =
+        json!("00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01");
+    let retried = request(&app, "POST", "/api/workflows", retry).await;
+    assert_eq!(retried.0, StatusCode::OK, "{}", retried.1);
+    assert_eq!(retried.1["outcome"], "used_existing_active");
+
+    // Terminate is not an operation that admits the carrier.
+    let refused = request(
+        &app,
+        "POST",
+        "/api/workflows/trace-carried/terminate",
+        traced(json!({"reason":"not admitted"})),
+    )
+    .await;
+    assert_eq!(refused.0, StatusCode::UNPROCESSABLE_ENTITY, "{}", refused.1);
+    assert_eq!(refused.1["reason"], "unsupported_request_field");
+    // The carrier never admits another unknown field beside it.
+    let mut unknown = traced(number(1));
+    unknown["business_key"] = json!("not supported");
+    let refused = request(
+        &app,
+        "POST",
+        "/api/workflows/trace-carried/signal/finish",
+        unknown,
+    )
+    .await;
+    assert_eq!(refused.0, StatusCode::UNPROCESSABLE_ENTITY, "{}", refused.1);
+    assert_eq!(refused.1["reason"], "unsupported_request_field");
+
+    let mut update = terminal_updates::input("trace-current");
+    both(&app, "POST", "/api/workflows/{workflow}/update/set", update).await;
+    update = terminal_updates::input("trace-run");
+    both(
+        &app,
+        "POST",
+        "/api/workflows/{workflow}/runs/{run}/update/set",
+        update,
+    )
+    .await;
+    both(
+        &app,
+        "POST",
+        "/api/workflows/{workflow}/signal/finish",
+        number(7),
+    )
+    .await;
+    both(
+        &app,
+        "POST",
+        "/api/workflows/{workflow}/runs/{run}/signal/finish",
+        number(8),
+    )
+    .await;
+    for path in [
+        "/api/workflows/{workflow}/request-cancellation",
+        "/api/workflows/{workflow}/runs/{run}/request-cancellation",
+    ] {
+        both(&app, "POST", path, json!({"reason":"trace λ"})).await;
+    }
+    both(
+        &app,
+        "POST",
+        "/api/workflows/{workflow}/runs/{run}/cancel",
+        json!({"reason":"trace λ"}),
+    )
+    .await;
+    both(
+        &app,
+        "POST",
+        "/api/workflows/{workflow}/cancel",
+        json!({"reason":"again"}),
+    )
+    .await;
+
+    let plain = visible(&app, "trace-plain").await;
+    let carried = visible(&app, "trace-carried").await;
+    for document in [&plain, &carried] {
+        let text = document.to_string();
+        assert!(!text.contains(TRACE_ID) && !text.contains("observability_context"));
+    }
+    assert_eq!(carried["described"]["status"], "cancelled");
+    let events = |document: &Value| {
+        document["history"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|event| event["event_type"].clone())
+            .collect::<Vec<_>>()
+    };
+    assert_eq!(events(&plain), events(&carried));
+    runtime.close().await;
+    database.remove().await;
+}