From bbbad63091273e45cbca1bce017dc29a25de912d Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 03:10:09 +0000 Subject: [PATCH] Ignore a caller's trace context where PHP Server admits it PHP Server's optional tracing admits a top-level observability_context on start, signal, update, cancel and request-cancellation requests. Published PHP 2.5.14 and PHP with tracing disabled ignore it; native refused it with unsupported_request_field, so a caller that sent it without checking discovery would have its business request refused. Those nine native handlers now drop exactly that field unread. Native never advertises the capability, and other routes, nested fields and unknown fields keep their handling. Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 25 ++++ rust/src/runtime/http.rs | 30 ++-- rust/tests/execution.rs | 3 + rust/tests/execution/trace_carriers.rs | 184 +++++++++++++++++++++++++ 4 files changed, 233 insertions(+), 9 deletions(-) create mode 100644 rust/tests/execution/trace_carriers.rs diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 0600015c9..7340d6062 100644 --- a/docs/rust-server-port.md +++ b/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, lost workflow task claims: - **Found by.** The [multi-node runs](../benchmarks/server-port/multinode/results-2026-10-20.md). diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 3f54d55d8..942a62c6e 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -611,11 +611,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, Json(body): Json, ) -> Result<(StatusCode, Json)> { - 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) @@ -680,7 +692,7 @@ async fn cancel_current( Path(workflow_id): Path, Json(body): Json, ) -> Result<(StatusCode, Json)> { - 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( @@ -789,7 +801,7 @@ async fn cancel_run( TerminalCommand::Cancel, &workflow_id, Some(&run_id), - body, + without_trace_carrier(body), ) .await } @@ -815,7 +827,7 @@ async fn request_cancellation_current( Json(body): Json, ) -> Result<(StatusCode, Json)> { let (status, response) = runtime - .request_cancellation(&workflow_id, None, body) + .request_cancellation(&workflow_id, None, without_trace_carrier(body)) .await?; Ok((status, Json(response))) } @@ -826,7 +838,7 @@ async fn request_cancellation_run( Json(body): Json, ) -> Result<(StatusCode, Json)> { 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))) } @@ -837,7 +849,7 @@ async fn signal_current( Json(body): Json, ) -> Result<(StatusCode, Json)> { 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) @@ -854,7 +866,7 @@ async fn update_current( Json(body): Json, ) -> Result<(StatusCode, Json)> { 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))) } @@ -865,7 +877,7 @@ async fn update_run( Json(body): Json, ) -> Result<(StatusCode, Json)> { 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))) } @@ -876,7 +888,7 @@ async fn signal_run( Json(body): Json, ) -> Result<(StatusCode, Json)> { 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( diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 0c82502d0..128361d8b 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -154,6 +154,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 { diff --git a/rust/tests/execution/trace_carriers.rs b/rust/tests/execution/trace_carriers.rs new file mode 100644 index 000000000..feed40e28 --- /dev/null +++ b/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::>() + }; + assert_eq!(events(&plain), events(&carried)); + runtime.close().await; + database.remove().await; +}