diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 168a88a56..4497d00d3 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, external payload storage diagnostic: - **Source.** Native serves `POST /api/storage/test` for operators, in the diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 5be3b8bf8..5a6a78653 100644 --- a/rust/src/runtime/http.rs +++ b/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, 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) @@ -697,7 +709,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( @@ -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, ) -> 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))) } @@ -843,7 +855,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))) } @@ -854,7 +866,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) @@ -871,7 +883,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))) } @@ -882,7 +894,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))) } @@ -893,7 +905,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 cccf73e5d..d7d9d49ab 100644 --- a/rust/tests/execution.rs +++ b/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 { 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; +}