From 1d44b1b435f35c6a62e0704a9e47ec375a95b982 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 06:14:58 +0000 Subject: [PATCH 1/6] Prepare 4.15.0 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 ++ Cargo.lock | 2 +- Cargo.toml | 4 ++-- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bc7e36..e8a8e78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +## 4.15.0 + - A workflow-task completion that native Server 3.6.0 refuses as too large (HTTP 413, `limit_kind: workflow_task_completion_bytes`, server#1152, durable-workflow/.github#228) is now a typed, non-retryable outcome. diff --git a/Cargo.lock b/Cargo.lock index 200ff6a..dbd0271 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -250,7 +250,7 @@ dependencies = [ [[package]] name = "durable-workflow" -version = "4.14.0" +version = "4.15.0" dependencies = [ "apache-avro", "base64 0.22.1", diff --git a/Cargo.toml b/Cargo.toml index 4cdc190..fa0c7ce 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "durable-workflow" -version = "4.14.0" +version = "4.15.0" edition = "2021" description = "Rust client and worker SDK for Durable Workflow Cloud and self-hosted Server" license = "MIT" @@ -14,7 +14,7 @@ categories = ["api-bindings", "asynchronous"] include = ["/src/**", "/schema/**", "/examples/**", "/Cargo.toml", "/README.md", "/CHANGELOG.md", "/LICENSE", "/tests/fixtures/replay-regressions/published-php-complete-explicit-backoff.json", "/tests/fixtures/sdk-metrics-contract.json", "/tests/fixtures/auth-configuration/static-credentials-v1.json"] [package.metadata.durable-workflow] -product-train = "4.14.0" +product-train = "4.15.0" compatibility-authority = "protocol-manifests" supported-server-versions = "2.0.0" supported-server-release-baselines = { "2" = "2.0.0", "3" = "3.0.2" } From 6d2650a6046b813a053542f27601ccdb483755cf Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 06:24:18 +0000 Subject: [PATCH 2/6] Read the cancellation context's timeout-cleanup cause The portable cancellation context gains an optional top-level `cause` (workflow#863, workflow#1129). `CancellationRequest` now carries a `#[non_exhaustive]` `CancellationCause`: `Requested` for an ordinary request (`cause` absent or "cancellation") or `Timeout { timeout_kind, work_deadline_at, cleanup_deadline_at }` for a timeout-cleanup request. The interruption still arrives as `Error::CooperativeCancellationRequested`, so existing handlers catch timeouts unchanged; there is no new `Error` variant. Unknown causes, timeout fields without the timeout cause, unknown timeout kinds and a work deadline that doesn't precede the cleanup deadline are refused. `to_value()` writes `cause` and the timeout fields back exactly, and ordinary contexts are byte-identical. Core's cause vectors are ported. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 28 +++ docs/cooperative-cancellation-design.md | 15 ++ src/cancellation_context.rs | 163 ++++++++++++++ src/cooperative_cancellation.rs | 12 + src/lib.rs | 3 +- src/tests/cooperative_cancellation.rs | 277 ++++++++++++++++++++++++ src/tests/nexus.rs | 1 + 7 files changed, 498 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bc7e36..b728458 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,34 @@ ## Unreleased +- Cooperative cancellation now reads the optional top-level `cause` of the + portable cancellation context (workflow#863, workflow#1129). A + timeout-cleanup request (`cause: "timeout"`: the run passed its work + deadline and is now in cleanup) arrives through the existing + `Error::CooperativeCancellationRequested`, so existing + `Error::CooperativeCancellationRequested(_)` arms, such as saga + compensation, run for it unchanged. There is no new `Error` variant. Tell + the two apart with the new `#[non_exhaustive]` `CancellationCause`: + `Requested` for an ordinary request (`cause` absent or `"cancellation"`), or + `Timeout { timeout_kind, work_deadline_at, cleanup_deadline_at }`, with the + new `#[non_exhaustive]` `TimeoutKind` (`Execution` or `Run`). + `timeout_kind` and `work_deadline_at` are `None` when the context doesn't + carry them. `CancellationContext::cause()` and `is_timeout()` read it from + the context. A context with any other `cause`, with a timeout field but no + timeout cause, with an unknown `timeout_kind`, or with a `work_deadline_at` + that doesn't precede its cleanup deadline is refused as + `Error::InvalidCooperativeCancellation`. A descendant cancelled under the + request carries no `cause`, so it is an ordinary cancellation. + `CancellationContext::to_value()` writes `cause` and the timeout fields back + exactly as read, and an ordinary context's output is unchanged. + + **Minor-release note:** `CancellationRequest` gains a public + `cause: CancellationCause` field. Code that builds a `CancellationRequest` + with a struct literal or destructures it exhaustively must add the field + (`cause: CancellationCause::Requested` for an ordinary request). Requests + the SDK produces already set it. `CancellationRequest` will become + `#[non_exhaustive]` in the next major release. + - A workflow-task completion that native Server 3.6.0 refuses as too large (HTTP 413, `limit_kind: workflow_task_completion_bytes`, server#1152, durable-workflow/.github#228) is now a typed, non-retryable outcome. diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md index cb37c21..4709fcc 100644 --- a/docs/cooperative-cancellation-design.md +++ b/docs/cooperative-cancellation-design.md @@ -39,6 +39,21 @@ time, original deadline and ordered lineage. Requester fields are limited to caller type, ID and label. Timestamp helpers return immutable UTC dates. `to_value()` produces a detached portable snapshot. +## Timeout cleanup cause + +A run that opted in to deadline-triggered cleanup receives its timeout-cleanup +interruption as an ordinary cooperative cancellation request whose context has +the top-level `cause: "timeout"`. It arrives as +`Error::CooperativeCancellationRequested`, so cancellation handlers such as saga +compensation run for it unchanged, and the run still ends timed out. +`request.cause` (and `CancellationContext::cause()`) is the discriminator: +`CancellationCause::Requested` for an ordinary request, or +`CancellationCause::Timeout { timeout_kind, work_deadline_at, cleanup_deadline_at }`. +`timeout_kind` and `work_deadline_at` are `None` when the context does not carry +them. A context with any other `cause` value, or with timeout fields but no +timeout cause, is refused. Descendants cancelled under the request carry no +`cause`, so they are ordinary cancellations. + A scoped child context preserves its complete immutable `scope_origin` alongside the local run lineage. `ScopedCancellationContext::root_context()` retains the original root deadline. Its scoped lineage records each run, instance, scope, diff --git a/src/cancellation_context.rs b/src/cancellation_context.rs index 1917bce..5a91615 100644 --- a/src/cancellation_context.rs +++ b/src/cancellation_context.rs @@ -32,6 +32,65 @@ impl CancellationLineage { } } +/// Which hard timeout passed its work deadline in a timeout-cleanup request. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +#[non_exhaustive] +pub enum TimeoutKind { + /// The execution timeout, which spans continue-as-new. + Execution, + /// The run timeout of the current run. + Run, +} + +impl TimeoutKind { + /// The portable wire value: `execution` or `run`. + pub fn as_str(self) -> &'static str { + match self { + Self::Execution => "execution", + Self::Run => "run", + } + } +} + +/// Why a cooperative cancellation request was delivered to this run. +/// +/// Read from the optional top-level `cause` of the portable cancellation +/// context. `Requested` is an ordinary cancellation request (`cause` absent or +/// `cancellation`). `Timeout` is a deadline-triggered cleanup request: the run +/// passed its work deadline and has until `cleanup_deadline_at` to clean up, +/// and it still ends timed out. A descendant cancelled under that request is an +/// ordinary cancellation; it does not inherit the cause. +/// +/// `timeout_kind` and `work_deadline_at` are `None` when the context does not +/// carry them; the embedded engine's first implementation records only +/// `cause`. `cleanup_deadline_at` is the context's own cleanup deadline. +#[derive(Clone, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum CancellationCause { + /// An ordinary cooperative cancellation request. + Requested, + /// A timeout-cleanup request after the run's work deadline passed. + Timeout { + timeout_kind: Option, + work_deadline_at: Option, + cleanup_deadline_at: String, + }, +} + +impl CancellationCause { + /// Whether this is a timeout-cleanup request. + pub fn is_timeout(&self) -> bool { + matches!(self, Self::Timeout { .. }) + } +} + +/// The context's `cause` exactly as recorded, so it round-trips unchanged. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RecordedCause { + Cancellation, + Timeout, +} + /// Original cancellation metadata restored from canonical request history. /// /// Fields and nested metadata have read-only accessors. Descendants retain @@ -49,6 +108,9 @@ pub struct CancellationContext { requested_at: DateTime, cleanup_deadline_at: DateTime, lineage: Vec, + cause: Option, + timeout_kind: Option, + work_deadline_at: Option>, scope_origin: Option>, replay: Option>>, } @@ -66,6 +128,9 @@ impl PartialEq for CancellationContext { && self.requested_at == other.requested_at && self.cleanup_deadline_at == other.cleanup_deadline_at && self.lineage == other.lineage + && self.cause == other.cause + && self.timeout_kind == other.timeout_kind + && self.work_deadline_at == other.work_deadline_at && self.scope_origin == other.scope_origin } } @@ -93,6 +158,68 @@ fn nullable_text(value: &Value, key: &str, allow_empty: bool) -> Result(value: &'a Value, key: &str) -> Option<&'a Value> { + value.get(key).filter(|field| !field.is_null()) +} + +/// A recorded cause with the timeout kind and work deadline only it may carry. +type RecordedCauseFields = ( + Option, + Option, + Option>, +); + +/// Read the optional `cause` and the timeout fields only it may carry. +fn read_cause(value: &Value, cleanup_deadline_at: DateTime) -> Result { + let cause = match present(value, "cause") { + None => None, + Some(Value::String(cause)) if cause == "cancellation" => Some(RecordedCause::Cancellation), + Some(Value::String(cause)) if cause == "timeout" => Some(RecordedCause::Timeout), + Some(_) => { + return Err(invalid_context( + "cancellation cause must be cancellation or timeout", + )) + } + }; + let kind = present(value, "timeout_kind"); + let work = present(value, "work_deadline_at"); + if cause != Some(RecordedCause::Timeout) { + if kind.is_some() || work.is_some() { + return Err(invalid_context( + "cancellation timeout fields require the timeout cause", + )); + } + return Ok((cause, None, None)); + } + let timeout_kind = match kind { + None => None, + Some(Value::String(kind)) if kind == "execution" => Some(TimeoutKind::Execution), + Some(Value::String(kind)) if kind == "run" => Some(TimeoutKind::Run), + Some(_) => { + return Err(invalid_context( + "cancellation timeout kind must be execution or run", + )) + } + }; + let work_deadline_at = match work { + None => None, + Some(work) => { + let work = work + .as_str() + .and_then(|text| DateTime::parse_from_rfc3339(text).ok()) + .ok_or_else(|| invalid_context("cancellation work deadline timestamp is invalid"))? + .with_timezone(&Utc); + if work >= cleanup_deadline_at { + return Err(invalid_context( + "cancellation work deadline must precede its cleanup deadline", + )); + } + Some(work) + } + }; + Ok((cause, timeout_kind, work_deadline_at)) +} + impl CancellationContext { /// Read legacy v1 and candidate scoped v2 without granting a new cleanup budget. pub fn from_value(value: &Value) -> Result { @@ -187,6 +314,7 @@ impl CancellationContext { "cancellation deadline must follow the original request", )); } + let (cause, timeout_kind, work_deadline_at) = read_cause(value, cleanup_deadline_at)?; let context = Self { request_id, root_request_id, @@ -199,6 +327,9 @@ impl CancellationContext { requested_at, cleanup_deadline_at, lineage: normalized, + cause, + timeout_kind, + work_deadline_at, scope_origin, replay: None, }; @@ -293,6 +424,27 @@ impl CancellationContext { self.scope_origin.as_deref() } + /// Why this request was delivered: an ordinary request or timeout cleanup. + pub fn cause(&self) -> CancellationCause { + match self.cause { + Some(RecordedCause::Timeout) => CancellationCause::Timeout { + timeout_kind: self.timeout_kind, + work_deadline_at: self + .work_deadline_at + .map(|work| work.to_rfc3339_opts(SecondsFormat::Micros, true)), + cleanup_deadline_at: self + .cleanup_deadline_at + .to_rfc3339_opts(SecondsFormat::Micros, true), + }, + _ => CancellationCause::Requested, + } + } + + /// Whether this is a timeout-cleanup request (`cause: "timeout"`). + pub fn is_timeout(&self) -> bool { + self.cause == Some(RecordedCause::Timeout) + } + /// Remaining cleanup budget at the last blocking result consumed by this replay. /// /// Never reads host time. Detached metadata, an ended replay, and a missing or @@ -337,6 +489,17 @@ impl CancellationContext { "cleanup_deadline_at": self.cleanup_deadline_at.to_rfc3339_opts(SecondsFormat::Micros, true), "lineage": self.lineage.iter().map(CancellationLineage::to_value).collect::>(), }); + match self.cause { + Some(RecordedCause::Cancellation) => value["cause"] = json!("cancellation"), + Some(RecordedCause::Timeout) => value["cause"] = json!("timeout"), + None => {} + } + if let Some(kind) = self.timeout_kind { + value["timeout_kind"] = json!(kind.as_str()); + } + if let Some(work) = self.work_deadline_at { + value["work_deadline_at"] = json!(work.to_rfc3339_opts(SecondsFormat::Micros, true)); + } if let Some(origin) = &self.scope_origin { value["scope_origin"] = origin.to_value(); value["scope_authority_deadline_at"] = value["cleanup_deadline_at"].clone(); diff --git a/src/cooperative_cancellation.rs b/src/cooperative_cancellation.rs index 81eab2a..92e86b0 100644 --- a/src/cooperative_cancellation.rs +++ b/src/cooperative_cancellation.rs @@ -131,6 +131,12 @@ pub struct CancellationRequest { pub cleanup_deadline_at: String, pub history_refresh_page_token: Option, pub context: Option, + /// Why the request was delivered, read from the canonical context's + /// `cause`. A timeout-cleanup request arrives through the same + /// [`Error::CooperativeCancellationRequested`] as an ordinary one, so + /// existing handlers run for both; match on this to tell them apart. + /// `Requested` when no canonical context is known yet. + pub cause: CancellationCause, } /// Acknowledgment of a request targeting the current durable run. @@ -220,6 +226,7 @@ impl CancellationRequest { cleanup_deadline_at: cleanup_deadline_at.to_owned(), history_refresh_page_token: Some(text(value, "history_refresh_page_token")?.to_owned()), context: None, + cause: CancellationCause::Requested, }) } } @@ -501,6 +508,7 @@ impl CancellationHistory { let mut state = Self { request: observation.cloned().map(|mut request| { request.context = None; + request.cause = CancellationCause::Requested; request }), delivery: None, @@ -589,6 +597,9 @@ impl CancellationHistory { } } if context.is_some() || observation.is_none() { + let cause = context + .as_ref() + .map_or(CancellationCause::Requested, CancellationContext::cause); state.request = Some(CancellationRequest { request_id: request_id.to_owned(), requested_at: context.as_ref().map_or_else( @@ -603,6 +614,7 @@ impl CancellationHistory { history_refresh_page_token: observation .and_then(|request| request.history_refresh_page_token.clone()), context, + cause, }); } state.request_index = index; diff --git a/src/lib.rs b/src/lib.rs index 05632ab..a7a9604 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -40,7 +40,8 @@ mod workflow_hydration; pub use activity_error::ActivityError; pub use activity_retry::{ActivityAttemptLimit, ActivityRetryInfo}; pub use cancellation_context::{ - CancellationContext, CancellationLineage, ScopedCancellationContext, ScopedCancellationLineage, + CancellationCause, CancellationContext, CancellationLineage, ScopedCancellationContext, + ScopedCancellationLineage, TimeoutKind, }; pub use cancellation_scope::CancellationScopeOpenReceipt; #[doc(hidden)] diff --git a/src/tests/cooperative_cancellation.rs b/src/tests/cooperative_cancellation.rs index 1566788..96f10ca 100644 --- a/src/tests/cooperative_cancellation.rs +++ b/src/tests/cooperative_cancellation.rs @@ -5598,3 +5598,280 @@ async fn cooperative_request_bounds_discovery_and_mutation_with_one_total_budget ); assert_eq!(server.requests.lock().unwrap().len(), 2); } + +/// Core's `CancellationContextCauseTest` snapshot (workflow#863). +fn core_cause_snapshot() -> Value { + json!({ + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "req-1", + "root_request_id": "req-1", + "root_workflow_instance_id": "instance-1", + "root_workflow_run_id": "run-1", + "parent_request_id": null, + "reason": "work deadline passed", + "requester": {"type": "system", "id": "timeout_cleanup"}, + "source": "system", + "requested_at": "2026-11-13T00:01:00.000000Z", + "cleanup_deadline_at": "2026-11-13T00:01:10.000000Z", + "lineage": [{"request_id": "req-1", "workflow_instance_id": "instance-1", "workflow_run_id": "run-1"}], + }) +} + +fn with_fields(mut snapshot: Value, fields: Value) -> Value { + for (key, value) in fields.as_object().unwrap() { + snapshot[key] = value.clone(); + } + snapshot +} + +#[test] +fn cancellation_cause_ordinary_context_is_unchanged_and_stays_a_cancellation() { + let snapshot = core_cause_snapshot(); + let context = CancellationContext::from_value(&snapshot).unwrap(); + assert!(!context.is_timeout()); + assert_eq!(context.cause(), CancellationCause::Requested); + assert_eq!( + serde_json::to_string(&context.to_value()).unwrap(), + serde_json::to_string(&snapshot).unwrap() + ); + assert!(context.to_value().get("cause").is_none()); + for cause in [json!("cancellation"), Value::Null] { + let explicit = with_fields(snapshot.clone(), json!({ "cause": cause })); + let context = CancellationContext::from_value(&explicit).unwrap(); + assert_eq!(context.cause(), CancellationCause::Requested); + assert!(!context.is_timeout()); + } + let explicit = with_fields(snapshot.clone(), json!({"cause": "cancellation"})); + let context = CancellationContext::from_value(&explicit).unwrap(); + assert_eq!(context.to_value(), explicit); + assert_ne!( + context, + CancellationContext::from_value(&snapshot).unwrap(), + "an explicit cause is preserved, not normalized away" + ); +} + +#[test] +fn cancellation_cause_timeout_round_trips_with_and_without_deadline_fields() { + let timeout = with_fields(core_cause_snapshot(), json!({"cause": "timeout"})); + let context = CancellationContext::from_value(&timeout).unwrap(); + assert!(context.is_timeout()); + assert_eq!( + context.cause(), + CancellationCause::Timeout { + timeout_kind: None, + work_deadline_at: None, + cleanup_deadline_at: "2026-11-13T00:01:10.000000Z".into(), + } + ); + assert_eq!(context.to_value()["cause"], "timeout"); + assert_eq!( + serde_json::to_string(&context.to_value()).unwrap(), + serde_json::to_string(&timeout).unwrap() + ); + + let full = with_fields( + timeout, + json!({"timeout_kind": "execution", "work_deadline_at": "2026-11-13T00:01:00.000000Z"}), + ); + let context = CancellationContext::from_value(&full).unwrap(); + assert_eq!( + context.cause(), + CancellationCause::Timeout { + timeout_kind: Some(TimeoutKind::Execution), + work_deadline_at: Some("2026-11-13T00:01:00.000000Z".into()), + cleanup_deadline_at: "2026-11-13T00:01:10.000000Z".into(), + } + ); + assert!(context.cause().is_timeout()); + assert_eq!( + serde_json::to_string(&context.to_value()).unwrap(), + serde_json::to_string(&full).unwrap() + ); + assert_eq!( + CancellationContext::from_value(&context.to_value()).unwrap(), + context + ); + let run = with_fields(full, json!({"timeout_kind": "run"})); + assert!(matches!( + CancellationContext::from_value(&run).unwrap().cause(), + CancellationCause::Timeout { + timeout_kind: Some(TimeoutKind::Run), + .. + } + )); +} + +#[test] +fn cancellation_cause_unknown_or_inconsistent_values_are_refused() { + let base = core_cause_snapshot(); + for fields in [ + json!({"cause": "deadline"}), + json!({"cause": "Timeout"}), + json!({"cause": ""}), + json!({"cause": true}), + json!({"cause": 1}), + json!({"cause": {"kind": "timeout"}}), + json!({"timeout_kind": "execution"}), + json!({"work_deadline_at": "2026-11-13T00:01:00.000000Z"}), + json!({"cause": "cancellation", "timeout_kind": "run"}), + json!({"cause": "timeout", "timeout_kind": "activity"}), + json!({"cause": "timeout", "timeout_kind": 1}), + json!({"cause": "timeout", "work_deadline_at": "not a time"}), + json!({"cause": "timeout", "work_deadline_at": "2026-11-13T00:01:10.000000Z"}), + json!({"cause": "timeout", "work_deadline_at": "2026-11-13T00:02:00.000000Z"}), + ] { + assert!( + matches!( + CancellationContext::from_value(&with_fields(base.clone(), fields.clone())), + Err(Error::InvalidCooperativeCancellation(_)) + ), + "{fields} must be refused" + ); + } +} + +#[test] +fn cancellation_cause_descendant_is_cancelled_not_timed_out() { + let parent = with_fields(core_cause_snapshot(), json!({"cause": "timeout"})); + assert!(CancellationContext::from_value(&parent) + .unwrap() + .is_timeout()); + // Core's forDescendant: the cause is dropped and the lineage extended. + let mut child = parent.clone(); + child.as_object_mut().unwrap().remove("cause"); + child["request_id"] = json!("req-child"); + child["parent_request_id"] = json!("req-1"); + child["lineage"].as_array_mut().unwrap().push(json!({ + "request_id": "req-child", "workflow_instance_id": "child-instance", "workflow_run_id": "child-run", + })); + let child = CancellationContext::from_value(&child).unwrap(); + assert!(!child.is_timeout()); + assert_eq!(child.cause(), CancellationCause::Requested); +} + +#[test] +fn cancellation_cause_scoped_root_from_a_timeout_run_keeps_its_cause() { + let timeout = with_fields(context_snapshot(), json!({"cause": "timeout"})); + let context = CancellationContext::from_value(&timeout).unwrap(); + let scoped = ScopedCancellationContext::from_run_context(&context).unwrap(); + assert!(scoped.root_context().is_timeout()); + assert_eq!( + ScopedCancellationContext::from_value(&scoped.to_value()).unwrap(), + scoped + ); +} + +fn timeout_request_event(snapshot: &Value) -> HistoryEvent { + serde_json::from_value(json!({ + "event_type": "CooperativeCancellationRequested", "recorded_at": "2026-10-01T00:00:05Z", + "payload": { + "workflow_command_id": "request-1", "workflow_run_id": "run-1", "workflow_instance_id": "child-instance", + "reason": "maintenance", "cleanup_deadline_at": "2026-10-01T00:00:30.123456Z", + "cancellation": snapshot, + } + })) + .unwrap() +} + +fn timeout_delivery_event(snapshot: &Value) -> HistoryEvent { + event( + "CooperativeCancellationDelivered", + json!({ + "workflow_command_id": "request-1", "workflow_run_id": "run-1", + "sequence": 1, "call_kind": "timer", "cancellation": snapshot, + }), + ) +} + +#[test] +fn cancellation_cause_timeout_reaches_an_existing_cooperative_cancellation_arm() { + let snapshot = with_fields( + context_snapshot(), + json!({"cause": "timeout", "timeout_kind": "run", "work_deadline_at": "2026-10-01T00:00:20.123456Z"}), + ); + let mut worker = cancellation_worker(); + worker.register_workflow("cancel", |ctx, _input| async move { + let Err(Error::CooperativeCancellationRequested(cancelled)) = + ctx.sleep(Duration::from_secs(10)).await + else { + panic!("a timeout cleanup request is a cooperative cancellation request"); + }; + let CancellationCause::Timeout { + timeout_kind, + work_deadline_at, + cleanup_deadline_at, + } = cancelled.request.cause.clone() + else { + panic!("expected the timeout cause"); + }; + assert_eq!(timeout_kind, Some(TimeoutKind::Run)); + assert_eq!( + work_deadline_at.as_deref(), + Some("2026-10-01T00:00:20.123456Z") + ); + assert_eq!(cleanup_deadline_at, "2026-10-01T00:00:30.123456Z"); + let context = cancelled.request.context.unwrap(); + assert_eq!(context.cause(), cancelled.request.cause); + assert!(ctx.cancellation_context()?.unwrap().is_timeout()); + Ok(context.to_value()) + }); + let result = worker + .execute_workflow_task(context_task(vec![ + timeout_request_event(&snapshot), + timeout_delivery_event(&snapshot), + ])) + .unwrap(); + assert_eq!(result[0]["type"], "complete_workflow"); + assert_eq!( + decode_wire_value(&result[0]["result"], DEFAULT_CODEC).unwrap(), + snapshot + ); +} + +#[test] +fn cancellation_cause_ordinary_delivery_and_observation_stay_requested() { + let mut worker = cancellation_worker(); + worker.register_workflow("cancel", |ctx, _input| async move { + let Err(Error::CooperativeCancellationRequested(cancelled)) = + ctx.sleep(Duration::from_secs(10)).await + else { + panic!("expected delivery"); + }; + assert_eq!(cancelled.request.cause, CancellationCause::Requested); + Ok(Value::Null) + }); + worker + .execute_workflow_task(context_task(vec![context_request(), context_delivery()])) + .unwrap(); + assert_eq!(context_observation().cause, CancellationCause::Requested); +} + +#[test] +fn cancellation_cause_delivery_cannot_add_or_drop_the_timeout_cause() { + let timeout = with_fields(context_snapshot(), json!({"cause": "timeout"})); + for (request, delivery) in [ + (context_snapshot(), timeout.clone()), + (timeout.clone(), context_snapshot()), + ] { + assert!(CancellationHistory::from_events( + &[ + timeout_request_event(&request), + timeout_delivery_event(&delivery) + ], + "run-1", + None + ) + .is_err()); + } + let history = CancellationHistory::from_events( + &[ + timeout_request_event(&timeout), + timeout_delivery_event(&timeout), + ], + "run-1", + None, + ) + .unwrap(); + assert!(history.request.unwrap().cause.is_timeout()); +} diff --git a/src/tests/nexus.rs b/src/tests/nexus.rs index de27ef1..a03c55a 100644 --- a/src/tests/nexus.rs +++ b/src/tests/nexus.rs @@ -157,6 +157,7 @@ fn root_request_selects_original_service_boundary_without_readmission() { cleanup_deadline_at: "2026-10-12T14:00:30Z".into(), history_refresh_page_token: Some("original-page".into()), context: None, + cause: CancellationCause::Requested, }; let decision = worker() .execute_workflow_task_decision_with_cancellation(task, Some(&observation)) From d586b1949cd66e005e3051275ac45f2a730b0b81 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 06:53:18 +0000 Subject: [PATCH 3/6] Carry only the cause in the cancellation context Core settled the context shape on workflow#1129 (14a4b944): the context carries only `cause` beside its existing cleanup deadline, and the timeout kind and work deadline stay on WorkflowTimeoutCleanupStarted. Drop `TimeoutKind` and the timeout fields; `CancellationCause::Timeout` is now a unit variant, and other keys are not interpreted, as in Core. The cause is never inherited. Port Core's scoped-descendant vector: a v2 child of a timeout root has no top-level cause, parses as `Requested` and round-trips without one, while its origin root context keeps the cause. Core now also preserves an explicit "cancellation" on write-back. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 22 ++- docs/cooperative-cancellation-design.md | 14 +- src/cancellation_context.rs | 135 +++-------------- src/lib.rs | 2 +- src/tests/cooperative_cancellation.rs | 192 +++++++++++------------- 5 files changed, 125 insertions(+), 240 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b728458..577b732 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,18 +10,16 @@ `Error::CooperativeCancellationRequested(_)` arms, such as saga compensation, run for it unchanged. There is no new `Error` variant. Tell the two apart with the new `#[non_exhaustive]` `CancellationCause`: - `Requested` for an ordinary request (`cause` absent or `"cancellation"`), or - `Timeout { timeout_kind, work_deadline_at, cleanup_deadline_at }`, with the - new `#[non_exhaustive]` `TimeoutKind` (`Execution` or `Run`). - `timeout_kind` and `work_deadline_at` are `None` when the context doesn't - carry them. `CancellationContext::cause()` and `is_timeout()` read it from - the context. A context with any other `cause`, with a timeout field but no - timeout cause, with an unknown `timeout_kind`, or with a `work_deadline_at` - that doesn't precede its cleanup deadline is refused as - `Error::InvalidCooperativeCancellation`. A descendant cancelled under the - request carries no `cause`, so it is an ordinary cancellation. - `CancellationContext::to_value()` writes `cause` and the timeout fields back - exactly as read, and an ordinary context's output is unchanged. + `Requested` for an ordinary request (`cause` absent or `"cancellation"`) or + `Timeout`. The request's cleanup deadline is its existing + `cleanup_deadline_at`. `CancellationContext::cause()` and `is_timeout()` + read it from the context. A context with any other `cause` is refused as + `Error::InvalidCooperativeCancellation`. The cause is never inherited: a + descendant cancelled under the request, scoped or not, carries no `cause` + and is an ordinary cancellation, even when its scope origin's root context + carries `cause: "timeout"`. `CancellationContext::to_value()` writes `cause` + back exactly as read, including an explicit `"cancellation"`, and an + ordinary context's output is unchanged. **Minor-release note:** `CancellationRequest` gains a public `cause: CancellationCause` field. Code that builds a `CancellationRequest` diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md index 4709fcc..47926af 100644 --- a/docs/cooperative-cancellation-design.md +++ b/docs/cooperative-cancellation-design.md @@ -47,12 +47,14 @@ the top-level `cause: "timeout"`. It arrives as `Error::CooperativeCancellationRequested`, so cancellation handlers such as saga compensation run for it unchanged, and the run still ends timed out. `request.cause` (and `CancellationContext::cause()`) is the discriminator: -`CancellationCause::Requested` for an ordinary request, or -`CancellationCause::Timeout { timeout_kind, work_deadline_at, cleanup_deadline_at }`. -`timeout_kind` and `work_deadline_at` are `None` when the context does not carry -them. A context with any other `cause` value, or with timeout fields but no -timeout cause, is refused. Descendants cancelled under the request carry no -`cause`, so they are ordinary cancellations. +`CancellationCause::Requested` for an ordinary request or +`CancellationCause::Timeout`. The cleanup deadline is the request's existing +`cleanup_deadline_at`; the timeout kind and work deadline are recorded on the +`WorkflowTimeoutCleanupStarted` event, not in the context. A context with any +other `cause` value is refused. The cause is never inherited: descendants +cancelled under the request, scoped or not, carry no `cause`, so they are +ordinary cancellations, even when a scoped child's origin root context carries +`cause: "timeout"`. A scoped child context preserves its complete immutable `scope_origin` alongside the local run lineage. `ScopedCancellationContext::root_context()` retains the diff --git a/src/cancellation_context.rs b/src/cancellation_context.rs index 5a91615..7c7624f 100644 --- a/src/cancellation_context.rs +++ b/src/cancellation_context.rs @@ -32,55 +32,28 @@ impl CancellationLineage { } } -/// Which hard timeout passed its work deadline in a timeout-cleanup request. -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -#[non_exhaustive] -pub enum TimeoutKind { - /// The execution timeout, which spans continue-as-new. - Execution, - /// The run timeout of the current run. - Run, -} - -impl TimeoutKind { - /// The portable wire value: `execution` or `run`. - pub fn as_str(self) -> &'static str { - match self { - Self::Execution => "execution", - Self::Run => "run", - } - } -} - /// Why a cooperative cancellation request was delivered to this run. /// /// Read from the optional top-level `cause` of the portable cancellation /// context. `Requested` is an ordinary cancellation request (`cause` absent or /// `cancellation`). `Timeout` is a deadline-triggered cleanup request: the run -/// passed its work deadline and has until `cleanup_deadline_at` to clean up, -/// and it still ends timed out. A descendant cancelled under that request is an -/// ordinary cancellation; it does not inherit the cause. -/// -/// `timeout_kind` and `work_deadline_at` are `None` when the context does not -/// carry them; the embedded engine's first implementation records only -/// `cause`. `cleanup_deadline_at` is the context's own cleanup deadline. -#[derive(Clone, Debug, PartialEq, Eq)] +/// passed its work deadline and has until the request's cleanup deadline to +/// clean up, and it still ends timed out. The cause is never inherited: a +/// descendant cancelled under the request, scoped or not, is an ordinary +/// cancellation. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] #[non_exhaustive] pub enum CancellationCause { /// An ordinary cooperative cancellation request. Requested, /// A timeout-cleanup request after the run's work deadline passed. - Timeout { - timeout_kind: Option, - work_deadline_at: Option, - cleanup_deadline_at: String, - }, + Timeout, } impl CancellationCause { /// Whether this is a timeout-cleanup request. - pub fn is_timeout(&self) -> bool { - matches!(self, Self::Timeout { .. }) + pub fn is_timeout(self) -> bool { + self == Self::Timeout } } @@ -109,8 +82,6 @@ pub struct CancellationContext { cleanup_deadline_at: DateTime, lineage: Vec, cause: Option, - timeout_kind: Option, - work_deadline_at: Option>, scope_origin: Option>, replay: Option>>, } @@ -129,8 +100,6 @@ impl PartialEq for CancellationContext { && self.cleanup_deadline_at == other.cleanup_deadline_at && self.lineage == other.lineage && self.cause == other.cause - && self.timeout_kind == other.timeout_kind - && self.work_deadline_at == other.work_deadline_at && self.scope_origin == other.scope_origin } } @@ -158,66 +127,18 @@ fn nullable_text(value: &Value, key: &str, allow_empty: bool) -> Result(value: &'a Value, key: &str) -> Option<&'a Value> { - value.get(key).filter(|field| !field.is_null()) -} - -/// A recorded cause with the timeout kind and work deadline only it may carry. -type RecordedCauseFields = ( - Option, - Option, - Option>, -); - -/// Read the optional `cause` and the timeout fields only it may carry. -fn read_cause(value: &Value, cleanup_deadline_at: DateTime) -> Result { - let cause = match present(value, "cause") { - None => None, - Some(Value::String(cause)) if cause == "cancellation" => Some(RecordedCause::Cancellation), - Some(Value::String(cause)) if cause == "timeout" => Some(RecordedCause::Timeout), - Some(_) => { - return Err(invalid_context( - "cancellation cause must be cancellation or timeout", - )) - } - }; - let kind = present(value, "timeout_kind"); - let work = present(value, "work_deadline_at"); - if cause != Some(RecordedCause::Timeout) { - if kind.is_some() || work.is_some() { - return Err(invalid_context( - "cancellation timeout fields require the timeout cause", - )); +/// Read the optional `cause`. Other keys, as in Core, are not interpreted. +fn read_cause(value: &Value) -> Result> { + match value.get("cause") { + None | Some(Value::Null) => Ok(None), + Some(Value::String(cause)) if cause == "cancellation" => { + Ok(Some(RecordedCause::Cancellation)) } - return Ok((cause, None, None)); + Some(Value::String(cause)) if cause == "timeout" => Ok(Some(RecordedCause::Timeout)), + Some(_) => Err(invalid_context( + "cancellation cause must be cancellation or timeout", + )), } - let timeout_kind = match kind { - None => None, - Some(Value::String(kind)) if kind == "execution" => Some(TimeoutKind::Execution), - Some(Value::String(kind)) if kind == "run" => Some(TimeoutKind::Run), - Some(_) => { - return Err(invalid_context( - "cancellation timeout kind must be execution or run", - )) - } - }; - let work_deadline_at = match work { - None => None, - Some(work) => { - let work = work - .as_str() - .and_then(|text| DateTime::parse_from_rfc3339(text).ok()) - .ok_or_else(|| invalid_context("cancellation work deadline timestamp is invalid"))? - .with_timezone(&Utc); - if work >= cleanup_deadline_at { - return Err(invalid_context( - "cancellation work deadline must precede its cleanup deadline", - )); - } - Some(work) - } - }; - Ok((cause, timeout_kind, work_deadline_at)) } impl CancellationContext { @@ -314,7 +235,7 @@ impl CancellationContext { "cancellation deadline must follow the original request", )); } - let (cause, timeout_kind, work_deadline_at) = read_cause(value, cleanup_deadline_at)?; + let cause = read_cause(value)?; let context = Self { request_id, root_request_id, @@ -328,8 +249,6 @@ impl CancellationContext { cleanup_deadline_at, lineage: normalized, cause, - timeout_kind, - work_deadline_at, scope_origin, replay: None, }; @@ -427,15 +346,7 @@ impl CancellationContext { /// Why this request was delivered: an ordinary request or timeout cleanup. pub fn cause(&self) -> CancellationCause { match self.cause { - Some(RecordedCause::Timeout) => CancellationCause::Timeout { - timeout_kind: self.timeout_kind, - work_deadline_at: self - .work_deadline_at - .map(|work| work.to_rfc3339_opts(SecondsFormat::Micros, true)), - cleanup_deadline_at: self - .cleanup_deadline_at - .to_rfc3339_opts(SecondsFormat::Micros, true), - }, + Some(RecordedCause::Timeout) => CancellationCause::Timeout, _ => CancellationCause::Requested, } } @@ -494,12 +405,6 @@ impl CancellationContext { Some(RecordedCause::Timeout) => value["cause"] = json!("timeout"), None => {} } - if let Some(kind) = self.timeout_kind { - value["timeout_kind"] = json!(kind.as_str()); - } - if let Some(work) = self.work_deadline_at { - value["work_deadline_at"] = json!(work.to_rfc3339_opts(SecondsFormat::Micros, true)); - } if let Some(origin) = &self.scope_origin { value["scope_origin"] = origin.to_value(); value["scope_authority_deadline_at"] = value["cleanup_deadline_at"].clone(); diff --git a/src/lib.rs b/src/lib.rs index a7a9604..bdca4d1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -41,7 +41,7 @@ pub use activity_error::ActivityError; pub use activity_retry::{ActivityAttemptLimit, ActivityRetryInfo}; pub use cancellation_context::{ CancellationCause, CancellationContext, CancellationLineage, ScopedCancellationContext, - ScopedCancellationLineage, TimeoutKind, + ScopedCancellationLineage, }; pub use cancellation_scope::CancellationScopeOpenReceipt; #[doc(hidden)] diff --git a/src/tests/cooperative_cancellation.rs b/src/tests/cooperative_cancellation.rs index 96f10ca..e40ba17 100644 --- a/src/tests/cooperative_cancellation.rs +++ b/src/tests/cooperative_cancellation.rs @@ -5617,123 +5617,80 @@ fn core_cause_snapshot() -> Value { }) } -fn with_fields(mut snapshot: Value, fields: Value) -> Value { - for (key, value) in fields.as_object().unwrap() { - snapshot[key] = value.clone(); - } +fn with_cause(mut snapshot: Value, cause: Value) -> Value { + snapshot["cause"] = cause; snapshot } +fn same_bytes(left: &Value, right: &Value) { + assert_eq!( + serde_json::to_string(left).unwrap(), + serde_json::to_string(right).unwrap() + ); +} + #[test] fn cancellation_cause_ordinary_context_is_unchanged_and_stays_a_cancellation() { let snapshot = core_cause_snapshot(); let context = CancellationContext::from_value(&snapshot).unwrap(); assert!(!context.is_timeout()); assert_eq!(context.cause(), CancellationCause::Requested); - assert_eq!( - serde_json::to_string(&context.to_value()).unwrap(), - serde_json::to_string(&snapshot).unwrap() - ); + same_bytes(&context.to_value(), &snapshot); assert!(context.to_value().get("cause").is_none()); - for cause in [json!("cancellation"), Value::Null] { - let explicit = with_fields(snapshot.clone(), json!({ "cause": cause })); - let context = CancellationContext::from_value(&explicit).unwrap(); - assert_eq!(context.cause(), CancellationCause::Requested); - assert!(!context.is_timeout()); - } - let explicit = with_fields(snapshot.clone(), json!({"cause": "cancellation"})); + // An explicit cancellation cause round-trips with exactly its own keys. + let explicit = with_cause(snapshot.clone(), json!("cancellation")); let context = CancellationContext::from_value(&explicit).unwrap(); - assert_eq!(context.to_value(), explicit); - assert_ne!( - context, - CancellationContext::from_value(&snapshot).unwrap(), - "an explicit cause is preserved, not normalized away" - ); + assert!(!context.is_timeout()); + assert_eq!(context.cause(), CancellationCause::Requested); + same_bytes(&context.to_value(), &explicit); + let null = CancellationContext::from_value(&with_cause(snapshot, Value::Null)).unwrap(); + assert_eq!(null.cause(), CancellationCause::Requested); } #[test] -fn cancellation_cause_timeout_round_trips_with_and_without_deadline_fields() { - let timeout = with_fields(core_cause_snapshot(), json!({"cause": "timeout"})); +fn cancellation_cause_timeout_round_trips() { + let timeout = with_cause(core_cause_snapshot(), json!("timeout")); let context = CancellationContext::from_value(&timeout).unwrap(); assert!(context.is_timeout()); - assert_eq!( - context.cause(), - CancellationCause::Timeout { - timeout_kind: None, - work_deadline_at: None, - cleanup_deadline_at: "2026-11-13T00:01:10.000000Z".into(), - } - ); - assert_eq!(context.to_value()["cause"], "timeout"); - assert_eq!( - serde_json::to_string(&context.to_value()).unwrap(), - serde_json::to_string(&timeout).unwrap() - ); - - let full = with_fields( - timeout, - json!({"timeout_kind": "execution", "work_deadline_at": "2026-11-13T00:01:00.000000Z"}), - ); - let context = CancellationContext::from_value(&full).unwrap(); - assert_eq!( - context.cause(), - CancellationCause::Timeout { - timeout_kind: Some(TimeoutKind::Execution), - work_deadline_at: Some("2026-11-13T00:01:00.000000Z".into()), - cleanup_deadline_at: "2026-11-13T00:01:10.000000Z".into(), - } - ); + assert_eq!(context.cause(), CancellationCause::Timeout); assert!(context.cause().is_timeout()); - assert_eq!( - serde_json::to_string(&context.to_value()).unwrap(), - serde_json::to_string(&full).unwrap() - ); + assert_eq!(context.to_value()["cause"], "timeout"); + same_bytes(&context.to_value(), &timeout); assert_eq!( CancellationContext::from_value(&context.to_value()).unwrap(), context ); - let run = with_fields(full, json!({"timeout_kind": "run"})); - assert!(matches!( - CancellationContext::from_value(&run).unwrap().cause(), - CancellationCause::Timeout { - timeout_kind: Some(TimeoutKind::Run), - .. - } - )); + // The timeout kind and work deadline live on WorkflowTimeoutCleanupStarted, + // not in the context; like any other unknown key, they are not interpreted. + let mut extra = timeout.clone(); + extra["timeout_kind"] = json!("bogus"); + extra["work_deadline_at"] = json!(1); + assert_eq!(CancellationContext::from_value(&extra).unwrap(), context); } #[test] -fn cancellation_cause_unknown_or_inconsistent_values_are_refused() { - let base = core_cause_snapshot(); - for fields in [ - json!({"cause": "deadline"}), - json!({"cause": "Timeout"}), - json!({"cause": ""}), - json!({"cause": true}), - json!({"cause": 1}), - json!({"cause": {"kind": "timeout"}}), - json!({"timeout_kind": "execution"}), - json!({"work_deadline_at": "2026-11-13T00:01:00.000000Z"}), - json!({"cause": "cancellation", "timeout_kind": "run"}), - json!({"cause": "timeout", "timeout_kind": "activity"}), - json!({"cause": "timeout", "timeout_kind": 1}), - json!({"cause": "timeout", "work_deadline_at": "not a time"}), - json!({"cause": "timeout", "work_deadline_at": "2026-11-13T00:01:10.000000Z"}), - json!({"cause": "timeout", "work_deadline_at": "2026-11-13T00:02:00.000000Z"}), +fn cancellation_cause_unknown_values_are_refused() { + for cause in [ + json!("deadline"), + json!("Timeout"), + json!(""), + json!(true), + json!(1), + json!({"kind": "timeout"}), ] { assert!( matches!( - CancellationContext::from_value(&with_fields(base.clone(), fields.clone())), + CancellationContext::from_value(&with_cause(core_cause_snapshot(), cause.clone())), Err(Error::InvalidCooperativeCancellation(_)) ), - "{fields} must be refused" + "{cause} must be refused" ); } } #[test] fn cancellation_cause_descendant_is_cancelled_not_timed_out() { - let parent = with_fields(core_cause_snapshot(), json!({"cause": "timeout"})); + let parent = with_cause(core_cause_snapshot(), json!("timeout")); assert!(CancellationContext::from_value(&parent) .unwrap() .is_timeout()); @@ -5750,16 +5707,51 @@ fn cancellation_cause_descendant_is_cancelled_not_timed_out() { assert_eq!(child.cause(), CancellationCause::Requested); } +/// Core's `scopedDescendantSnapshot` for a child run of a run-level root scope. +fn scoped_child_of(root: &CancellationContext) -> Value { + let origin = ScopedCancellationContext::from_run_context(root).unwrap(); + let mut child = origin.root_context().to_value(); + // A descendant is cancelled because its ancestor was (Core strips the cause). + child.as_object_mut().unwrap().remove("cause"); + let deadline = origin + .deadline() + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true); + child["schema"] = json!("durable-workflow.cancellation-context/v2"); + child["request_id"] = json!("req-child"); + child["parent_request_id"] = json!(origin.request_id()); + child["cleanup_deadline_at"] = json!(deadline); + child["lineage"] = json!([ + child["lineage"][0].clone(), + {"request_id": "req-child", "workflow_instance_id": "child-instance", "workflow_run_id": "child-run"}, + ]); + child["scope_origin"] = origin.to_value(); + child["scope_authority_deadline_at"] = json!(deadline); + child +} + #[test] -fn cancellation_cause_scoped_root_from_a_timeout_run_keeps_its_cause() { - let timeout = with_fields(context_snapshot(), json!({"cause": "timeout"})); - let context = CancellationContext::from_value(&timeout).unwrap(); - let scoped = ScopedCancellationContext::from_run_context(&context).unwrap(); - assert!(scoped.root_context().is_timeout()); +fn cancellation_cause_scoped_descendant_of_a_timeout_is_cancelled_not_timed_out() { + let root = + CancellationContext::from_value(&with_cause(core_cause_snapshot(), json!("timeout"))) + .unwrap(); + let origin = ScopedCancellationContext::from_run_context(&root).unwrap(); + // fromRunContext keeps the run's own cause on the scope root. + assert!(origin.root_context().is_timeout()); assert_eq!( - ScopedCancellationContext::from_value(&scoped.to_value()).unwrap(), - scoped + ScopedCancellationContext::from_value(&origin.to_value()).unwrap(), + origin ); + let snapshot = scoped_child_of(&root); + assert_eq!(snapshot["scope_origin"]["root_context"]["cause"], "timeout"); + let child = CancellationContext::from_value(&snapshot).unwrap(); + assert!(!child.is_timeout()); + assert_eq!(child.cause(), CancellationCause::Requested); + assert!(child.scope_origin().unwrap().root_context().is_timeout()); + assert!(child.to_value().get("cause").is_none()); + same_bytes(&child.to_value(), &snapshot); + assert!(!CancellationContext::from_value(&child.to_value()) + .unwrap() + .is_timeout()); } fn timeout_request_event(snapshot: &Value) -> HistoryEvent { @@ -5786,10 +5778,7 @@ fn timeout_delivery_event(snapshot: &Value) -> HistoryEvent { #[test] fn cancellation_cause_timeout_reaches_an_existing_cooperative_cancellation_arm() { - let snapshot = with_fields( - context_snapshot(), - json!({"cause": "timeout", "timeout_kind": "run", "work_deadline_at": "2026-10-01T00:00:20.123456Z"}), - ); + let snapshot = with_cause(context_snapshot(), json!("timeout")); let mut worker = cancellation_worker(); worker.register_workflow("cancel", |ctx, _input| async move { let Err(Error::CooperativeCancellationRequested(cancelled)) = @@ -5797,20 +5786,11 @@ fn cancellation_cause_timeout_reaches_an_existing_cooperative_cancellation_arm() else { panic!("a timeout cleanup request is a cooperative cancellation request"); }; - let CancellationCause::Timeout { - timeout_kind, - work_deadline_at, - cleanup_deadline_at, - } = cancelled.request.cause.clone() - else { - panic!("expected the timeout cause"); - }; - assert_eq!(timeout_kind, Some(TimeoutKind::Run)); + assert_eq!(cancelled.request.cause, CancellationCause::Timeout); assert_eq!( - work_deadline_at.as_deref(), - Some("2026-10-01T00:00:20.123456Z") + cancelled.request.cleanup_deadline_at, + "2026-10-01T00:00:30.123456Z" ); - assert_eq!(cleanup_deadline_at, "2026-10-01T00:00:30.123456Z"); let context = cancelled.request.context.unwrap(); assert_eq!(context.cause(), cancelled.request.cause); assert!(ctx.cancellation_context()?.unwrap().is_timeout()); @@ -5849,7 +5829,7 @@ fn cancellation_cause_ordinary_delivery_and_observation_stay_requested() { #[test] fn cancellation_cause_delivery_cannot_add_or_drop_the_timeout_cause() { - let timeout = with_fields(context_snapshot(), json!({"cause": "timeout"})); + let timeout = with_cause(context_snapshot(), json!("timeout")); for (request, delivery) in [ (context_snapshot(), timeout.clone()), (timeout.clone(), context_snapshot()), From fb5bb5ec7445da998dd392f188a71d876d58a966 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:04:05 +0000 Subject: [PATCH 4/6] fix: validate every workflow route path segment --- CHANGELOG.md | 5 + src/cooperative_cancellation.rs | 10 +- src/lib.rs | 92 +++++++++--- src/pagination.rs | 6 +- src/run_activities.rs | 6 +- src/terminal_failure_detail.rs | 8 +- src/testing/history.rs | 6 +- src/tests/path_segments.rs | 246 ++++++++++++++++++++++++++++++++ src/update_outcome.rs | 16 +-- src/update_outcome/tests.rs | 7 +- 10 files changed, 352 insertions(+), 50 deletions(-) create mode 100644 src/tests/path_segments.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bc7e36..f6113fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +- All workflow and run routes encode each identifier or name as one path + segment. Empty identifiers and `.` or `..` segments return + `Error::InvalidPathSegment` before any HTTP request, including capability + discovery for cooperative cancellation. + - A workflow-task completion that native Server 3.6.0 refuses as too large (HTTP 413, `limit_kind: workflow_task_completion_bytes`, server#1152, durable-workflow/.github#228) is now a typed, non-retryable outcome. diff --git a/src/cooperative_cancellation.rs b/src/cooperative_cancellation.rs index 81eab2a..2d5ac05 100644 --- a/src/cooperative_cancellation.rs +++ b/src/cooperative_cancellation.rs @@ -1769,6 +1769,11 @@ impl Client { run_id: Option<&str>, options: CooperativeCancellationOptions, ) -> Result { + let mut path = format!("/workflows/{}", path_segment("workflow_id", workflow_id)?); + if let Some(run_id) = run_id { + path.push_str(&format!("/runs/{}", path_segment("run_id", run_id)?)); + } + path.push_str("/request-cancellation"); if workflow_id.trim().is_empty() || run_id.is_some_and(|id| id.trim().is_empty()) { return Err(invalid( "workflow and selected run identities must be non-empty", @@ -1798,11 +1803,6 @@ impl Client { ) .await?; require_discovery(&info)?; - let mut path = format!("/workflows/{}", percent_encode_path_segment(workflow_id)); - if let Some(run_id) = run_id { - path.push_str(&format!("/runs/{}", percent_encode_path_segment(run_id))); - } - path.push_str("/request-cancellation"); let response: Value = self .request_json( reqwest::Method::POST, diff --git a/src/lib.rs b/src/lib.rs index 05632ab..5ab4a39 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3426,7 +3426,11 @@ impl Client { }); self.request_json( reqwest::Method::POST, - &format!("/workflows/{workflow_id}/message-streams/{stream_name}/messages"), + &format!( + "/workflows/{}/message-streams/{}/messages", + path_segment("workflow_id", workflow_id)?, + path_segment("stream_name", stream_name)?, + ), RequestProtocol::ControlPlane, Some(&body), ) @@ -3482,9 +3486,18 @@ impl Client { })); let path = match run_id { Some(run_id) => { - format!("/workflows/{workflow_id}/runs/{run_id}/signal/{signal_name}") + format!( + "/workflows/{}/runs/{}/signal/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, + path_segment("signal_name", signal_name)?, + ) } - None => format!("/workflows/{workflow_id}/signal/{signal_name}"), + None => format!( + "/workflows/{}/signal/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("signal_name", signal_name)?, + ), }; let receipt = self .request_json( @@ -3567,7 +3580,11 @@ impl Client { let data: Value = self .request_json( reqwest::Method::POST, - &format!("/workflows/{workflow_id}/runs/{failed_run_id}/redrive"), + &format!( + "/workflows/{}/runs/{}/redrive", + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", failed_run_id)?, + ), RequestProtocol::ControlPlane, Some(&body), ) @@ -3626,10 +3643,16 @@ impl Client { let body = &options; let path = match run_id { Some(run_id) => format!( - "/workflows/{workflow_id}/runs/{run_id}/{}", + "/workflows/{}/runs/{}/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, command.as_str() ), - None => format!("/workflows/{workflow_id}/{}", command.as_str()), + None => format!( + "/workflows/{}/{}", + path_segment("workflow_id", workflow_id)?, + command.as_str(), + ), }; let data = match self .request_json( @@ -3717,9 +3740,18 @@ impl Client { let body = json!({"input": encode_typed_envelope(&input, DEFAULT_CODEC)?}); let path = match run_id { Some(run_id) => { - format!("/workflows/{workflow_id}/runs/{run_id}/query/{query_name}") + format!( + "/workflows/{}/runs/{}/query/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, + path_segment("query_name", query_name)?, + ) } - None => format!("/workflows/{workflow_id}/query/{query_name}"), + None => format!( + "/workflows/{}/query/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("query_name", query_name)?, + ), }; let response: Value = match self .request_json( @@ -3763,9 +3795,18 @@ impl Client { }); let path = match run_id { Some(run_id) => { - format!("/workflows/{workflow_id}/runs/{run_id}/query/{query_name}") + format!( + "/workflows/{}/runs/{}/query/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, + path_segment("query_name", query_name)?, + ) } - None => format!("/workflows/{workflow_id}/query/{query_name}"), + None => format!( + "/workflows/{}/query/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("query_name", query_name)?, + ), }; let response: Value = match self .request_json( @@ -3891,7 +3932,11 @@ impl Client { workflow_id: &str, run_id: &str, ) -> Result { - let path = format!("/workflows/{workflow_id}/runs/{run_id}"); + let path = format!( + "/workflows/{}/runs/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, + ); let mut data: WorkflowDescription = self .request_json( reqwest::Method::GET, @@ -3904,17 +3949,21 @@ impl Client { Ok(data) } - fn workflow_stream_path(workflow_id: &str, run_id: &str, stream_name: Option<&str>) -> String { + fn workflow_stream_path( + workflow_id: &str, + run_id: &str, + stream_name: Option<&str>, + ) -> Result { let mut path = format!( "/workflows/{}/runs/{}/streams", - percent_encode_path_segment(workflow_id), - percent_encode_path_segment(run_id), + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, ); if let Some(stream_name) = stream_name { path.push('/'); - path.push_str(&percent_encode_path_segment(stream_name)); + path.push_str(&path_segment("stream_name", stream_name)?); } - path + Ok(path) } /// List the run-scoped output streams already opened by a workflow. @@ -3926,7 +3975,7 @@ impl Client { let response: WorkflowStreamListResponse = self .request_json( reqwest::Method::GET, - &Self::workflow_stream_path(workflow_id, run_id, None), + &Self::workflow_stream_path(workflow_id, run_id, None)?, RequestProtocol::ControlPlane, Option::<&Value>::None, ) @@ -3944,7 +3993,7 @@ impl Client { let response: WorkflowStreamDescriptionResponse = self .request_json( reqwest::Method::GET, - &Self::workflow_stream_path(workflow_id, run_id, Some(stream_name)), + &Self::workflow_stream_path(workflow_id, run_id, Some(stream_name))?, RequestProtocol::ControlPlane, Option::<&Value>::None, ) @@ -3971,7 +4020,7 @@ impl Client { let wait_seconds = wait.as_secs().min(MAX_LONG_POLL_TIMEOUT_SECONDS); let path = format!( "{}/items?from={from_offset}&max_items={max_items}&wait_seconds={wait_seconds}", - Self::workflow_stream_path(workflow_id, run_id, Some(stream_name)), + Self::workflow_stream_path(workflow_id, run_id, Some(stream_name))?, ); let response: WorkflowStreamPageResponse = self .request_json_with_timeout( @@ -4078,7 +4127,7 @@ impl Client { reqwest::Method::POST, &format!( "{}/items", - Self::workflow_stream_path(workflow_id, run_id, Some(stream_name)), + Self::workflow_stream_path(workflow_id, run_id, Some(stream_name))?, ), RequestProtocol::ControlPlane, Some(&body), @@ -4119,7 +4168,7 @@ impl Client { reqwest::Method::POST, &format!( "{}/close", - Self::workflow_stream_path(workflow_id, run_id, Some(stream_name)), + Self::workflow_stream_path(workflow_id, run_id, Some(stream_name))?, ), RequestProtocol::ControlPlane, Some(&body), @@ -19719,6 +19768,7 @@ mod tests { mod cooperative_cancellation; mod local_activity; mod nexus; + mod path_segments; mod run_closed_settlement; mod runtime_payloads; mod runtime_uploads; diff --git a/src/pagination.rs b/src/pagination.rs index 5af0b33..c9d48f5 100644 --- a/src/pagination.rs +++ b/src/pagination.rs @@ -6,7 +6,7 @@ //! the Server's ordering and consistency contract; it is not a snapshot across //! concurrent changes. No new Server query fields or filter grammar are added. -use crate::{percent_encode_path_segment, Client, HistoryEvent, RequestProtocol, Result, Value}; +use crate::{path_segment, Client, HistoryEvent, RequestProtocol, Result, Value}; use serde::{de::DeserializeOwned, Deserialize, Deserializer, Serialize}; use std::collections::{BTreeMap, HashMap, VecDeque}; @@ -191,8 +191,8 @@ impl Client { let path = query_path( &format!( "/workflows/{}/runs/{}/history", - percent_encode_path_segment(workflow_id), - percent_encode_path_segment(run_id) + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)? ), [ ("page_size", options.page_size.map(|n| n.to_string())), diff --git a/src/run_activities.rs b/src/run_activities.rs index 51c4603..dceba0d 100644 --- a/src/run_activities.rs +++ b/src/run_activities.rs @@ -1,6 +1,6 @@ //! Read-only activity visibility for one explicit workflow run. -use crate::{percent_encode_path_segment, Client, RequestProtocol, Result, Value, WorkflowHandle}; +use crate::{path_segment, Client, RequestProtocol, Result, Value, WorkflowHandle}; use serde::{Deserialize, Deserializer, Serialize}; use std::collections::BTreeMap; @@ -136,8 +136,8 @@ impl Client { ) -> Result { let path = format!( "/workflows/{}/runs/{}/activities", - percent_encode_path_segment(workflow_id), - percent_encode_path_segment(run_id), + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)?, ); let document: Value = self .request_json( diff --git a/src/terminal_failure_detail.rs b/src/terminal_failure_detail.rs index 896731b..3564123 100644 --- a/src/terminal_failure_detail.rs +++ b/src/terminal_failure_detail.rs @@ -6,7 +6,9 @@ use serde_json::Value; -use crate::{percent_encode_path_segment, Client, RequestProtocol, WorkflowDescription}; +use crate::{ + path_segment, percent_encode_path_segment, Client, RequestProtocol, WorkflowDescription, +}; /// History pages read to find a terminal event when a description lacks detail. const TERMINAL_HISTORY_PAGE_LIMIT: usize = 20; @@ -101,8 +103,8 @@ impl Client { ) -> Option { let base = format!( "/workflows/{}/runs/{}/history", - percent_encode_path_segment(workflow_id), - percent_encode_path_segment(run_id), + path_segment("workflow_id", workflow_id).ok()?, + path_segment("run_id", run_id).ok()?, ); let mut next_page_token: Option = None; let mut payload = None; diff --git a/src/testing/history.rs b/src/testing/history.rs index 44cb4e9..9d3d103 100644 --- a/src/testing/history.rs +++ b/src/testing/history.rs @@ -1,7 +1,7 @@ use super::TestingError; use crate::{ decode_wire_avro_value, encode_typed_envelope, enrich_query_history_from_export, - hydrate_query_history_from_export, json, normalize_avro_arguments, percent_encode_path_segment, + hydrate_query_history_from_export, json, normalize_avro_arguments, path_segment, validate_query_task_payloads, AvroValue, Client, HistoryEvent, QueryTask, RequestProtocol, Result, Value, WorkflowTask, DEFAULT_CODEC, }; @@ -198,8 +198,8 @@ impl Client { reqwest::Method::GET, &format!( "/workflows/{}/runs/{}/history/export", - percent_encode_path_segment(workflow_id), - percent_encode_path_segment(run_id) + path_segment("workflow_id", workflow_id)?, + path_segment("run_id", run_id)? ), RequestProtocol::ControlPlane, Option::<&Value>::None, diff --git a/src/tests/path_segments.rs b/src/tests/path_segments.rs new file mode 100644 index 0000000..64cec33 --- /dev/null +++ b/src/tests/path_segments.rs @@ -0,0 +1,246 @@ +use super::*; + +// Exercise public entry points so validation cannot accidentally happen after +// capability discovery or be skipped by a typed/selected-run variant. +const ROUTES: &[(&str, &str, bool, Option<&str>)] = &[ + ("describe", "", false, None), + ( + "outcome", + "/updates/{n}?wait_for=completed", + false, + Some("update_id"), + ), + ( + "outcome-avro", + "/updates/{n}?wait_for=completed", + false, + Some("update_id"), + ), + ("outcome-receipt", "/updates/{n}", false, Some("update_id")), + ("describe-run", "/runs/{r}", true, None), + ("signal", "/signal/{n}", false, Some("signal_name")), + ( + "signal-run", + "/runs/{r}/signal/{n}", + true, + Some("signal_name"), + ), + ("query", "/query/{n}", false, Some("query_name")), + ("query-run", "/runs/{r}/query/{n}", true, Some("query_name")), + ("query-avro", "/query/{n}", false, Some("query_name")), + ( + "query-run-avro", + "/runs/{r}/query/{n}", + true, + Some("query_name"), + ), + ("update", "/update/{n}", false, Some("update_name")), + ("update-avro", "/update/{n}", false, Some("update_name")), + ("cancel", "/cancel", false, None), + ("cancel-run", "/runs/{r}/cancel", true, None), + ("terminate", "/terminate", false, None), + ("terminate-run", "/runs/{r}/terminate", true, None), + ("redrive", "/runs/{r}/redrive", true, None), + ( + "message", + "/message-streams/{n}/messages", + false, + Some("stream_name"), + ), + ("streams", "/runs/{r}/streams", true, None), + ("stream", "/runs/{r}/streams/{n}", true, Some("stream_name")), + ( + "subscribe", + "/runs/{r}/streams/{n}/items?from=0&max_items=1&wait_seconds=0", + true, + Some("stream_name"), + ), + ( + "append", + "/runs/{r}/streams/{n}/items", + true, + Some("stream_name"), + ), + ( + "close", + "/runs/{r}/streams/{n}/close", + true, + Some("stream_name"), + ), + ("activities", "/runs/{r}/activities", true, None), + ("history", "/runs/{r}/history", true, None), + ("export", "/runs/{r}/history/export", true, None), + ("cooperative", "/request-cancellation", false, None), + ( + "cooperative-run", + "/runs/{r}/request-cancellation", + true, + None, + ), +]; + +async fn exercise(client: &Client, route: &str, w: &str, r: &str, n: &str) -> Result<()> { + match route { + "outcome" => client.update_outcome(w, n, None).await.map(|_| ()), + "outcome-avro" => client + .update_outcome_avro_value(w, n, None) + .await + .map(|_| ()), + "outcome-receipt" => client + .update_outcome_receipt(w, n, UpdateWaitStage::Accepted, None) + .await + .map(|_| ()), + "describe" => client.describe_workflow(w).await.map(|_| ()), + "describe-run" => client.describe_workflow_run(w, r).await.map(|_| ()), + "signal" => client.signal_workflow(w, n, json!([])).await.map(|_| ()), + "signal-run" => client + .signal_workflow_run(w, r, n, json!([])) + .await + .map(|_| ()), + "query" => client.query_workflow(w, n, json!([])).await.map(|_| ()), + "query-run" => client + .query_workflow_run(w, r, n, json!([])) + .await + .map(|_| ()), + "query-avro" => client + .query_workflow_avro_value(w, n, json!([])) + .await + .map(|_| ()), + "query-run-avro" => client + .query_workflow_run_avro_value(w, r, n, json!([])) + .await + .map(|_| ()), + "update" => client + .update_workflow(w, n, json!([]), None) + .await + .map(|_| ()), + "update-avro" => client + .update_workflow_avro_value(w, n, json!([]), None) + .await + .map(|_| ()), + "cancel" => client + .cancel_workflow(w, Default::default()) + .await + .map(|_| ()), + "cancel-run" => client + .cancel_workflow_run(w, r, Default::default()) + .await + .map(|_| ()), + "terminate" => client + .terminate_workflow(w, Default::default()) + .await + .map(|_| ()), + "terminate-run" => client + .terminate_workflow_run(w, r, Default::default()) + .await + .map(|_| ()), + "redrive" => client.redrive_workflow_run(w, r, None).await.map(|_| ()), + "message" => client + .append_message_stream(w, n, "message-1", json!([])) + .await + .map(|_| ()), + "streams" => client.list_workflow_streams(w, r).await.map(|_| ()), + "stream" => client.describe_workflow_stream(w, r, n).await.map(|_| ()), + "subscribe" => client + .subscribe_workflow_stream(w, r, n, 0, 1, Duration::ZERO) + .await + .map(|_| ()), + "append" => client + .append_workflow_stream(w, r, n, &[WorkflowStreamAppendItem::new(json!(1))?], None) + .await + .map(|_| ()), + "close" => client + .close_workflow_stream(w, r, n, None, None) + .await + .map(|_| ()), + "activities" => client.get_run_activities(w, r).await.map(|_| ()), + "history" => client + .get_workflow_history(w, r, Default::default()) + .await + .map(|_| ()), + "export" => client.export_workflow_history(w, r).await.map(|_| ()), + "cooperative" => client + .request_workflow_cancellation(w, Default::default()) + .await + .map(|_| ()), + "cooperative-run" => client + .request_workflow_run_cancellation(w, r, Default::default()) + .await + .map(|_| ()), + _ => unreachable!("unknown route"), + } +} + +fn refuse_request(path: &str, _: &str, _: usize) -> Option<(&'static str, String)> { + if path == "/api/cluster/info" { + return Some(( + "200 OK", + json!({"capabilities": {"update_outcome_reads": true}}).to_string(), + )); + } + path.starts_with("/api/workflows/") + .then(|| ("400 Bad Request", "{}".into())) +} + +#[tokio::test] +async fn all_routes_keep_reserved_and_unicode_identifiers_in_one_segment() { + let server = MockWorkerServer::start_with_behavior(MockWorkerBehavior { + request_override: Some(refuse_request), + ..Default::default() + }); + let client = keyed_client(&server); + let raw = "a/b?c#d%2F\\é"; + let encoded = "a%2Fb%3Fc%23d%252F%5C%C3%A9"; + for &(route, suffix, _, _) in ROUTES { + // Cooperative encoding has its own successful discovery/ack test. + if route.starts_with("cooperative") { + continue; + } + let before = server.captured_paths().len(); + exercise(&client, route, raw, raw, raw) + .await + .expect_err("server refuses the encoded request"); + let expected = format!( + "/api/workflows/{encoded}{}", + suffix.replace("{r}", encoded).replace("{n}", encoded) + ); + let paths = server.captured_paths(); + let requests: Vec<_> = paths[before..] + .iter() + .filter(|path| path.as_str() != "/api/cluster/info") + .cloned() + .collect(); + assert_eq!(requests, vec![expected], "{route}"); + } +} + +#[tokio::test] +async fn all_routes_refuse_empty_and_dot_segments_before_any_http_request() { + let server = MockWorkerServer::start(); + let client = keyed_client(&server); + for &(route, _, has_run, name_field) in ROUTES { + for invalid in ["", ".", ".."] { + let mut cases = vec![(invalid, "run-1", "name", "workflow_id")]; + if has_run { + cases.push(("workflow-1", invalid, "name", "run_id")); + } + if let Some(field) = name_field { + cases.push(("workflow-1", "run-1", invalid, field)); + } + for (w, r, n, field) in cases { + let error = exercise(&client, route, w, r, n) + .await + .expect_err("invalid segment"); + assert!( + matches!(error, Error::InvalidPathSegment { field: actual } if actual == field), + "{route}, {field}={invalid:?}: {error:?}" + ); + } + } + } + assert!( + server.captured_paths().is_empty(), + "no discovery or mutation: {:?}", + server.captured_paths() + ); +} diff --git a/src/update_outcome.rs b/src/update_outcome.rs index 295bbc5..519f0c5 100644 --- a/src/update_outcome.rs +++ b/src/update_outcome.rs @@ -13,7 +13,7 @@ use std::{fmt, sync::atomic::Ordering, time::Duration}; use serde_json::Value; use crate::{ - decode_wire_avro_value, decode_wire_value, AvroValue, Client, Error, RawRequest, + decode_wire_avro_value, decode_wire_value, path_segment, AvroValue, Client, Error, RequestProtocol, Result, WorkflowHandle, DEFAULT_CODEC, }; @@ -261,20 +261,16 @@ impl Client { wait_for: UpdateWaitStage, wait_timeout_seconds: Option, ) -> Result { + let mut path = format!( + "/workflows/{}/updates/{}", + path_segment("workflow_id", workflow_id)?, + path_segment("update_id", update_id)?, + ); if update_id.trim().is_empty() { return Err(Error::Codec( "an update outcome read needs the update id".to_string(), )); } - // Each id must stay one path segment: "." or ".." would collapse the - // read onto another route, such as the workflow description. - let mut path = RawRequest::get(["workflows", workflow_id, "updates", update_id]) - .path() - .map_err(|_| { - Error::Codec( - "an update outcome read needs a workflow id and update id that are each one path segment: not empty, \".\" or \"..\"".to_string(), - ) - })?; self.require_update_outcome_reads().await?; let mut query = Vec::new(); diff --git a/src/update_outcome/tests.rs b/src/update_outcome/tests.rs index 6dd1b1a..e4d03b7 100644 --- a/src/update_outcome/tests.rs +++ b/src/update_outcome/tests.rs @@ -397,14 +397,17 @@ async fn dot_segment_and_empty_ids_are_refused_before_any_request() { .await .expect_err("a dot segment would read another route"); assert!( - matches!(error, Error::Codec(_)), + matches!(error, Error::InvalidPathSegment { .. }), "{workflow_id:?}/{update_id:?}: {error:?}" ); let error = client .update_outcome_receipt(workflow_id, update_id, UpdateWaitStage::Accepted, None) .await .expect_err("a dot segment would read another route"); - assert!(matches!(error, Error::Codec(_)), "{error:?}"); + assert!( + matches!(error, Error::InvalidPathSegment { .. }), + "{error:?}" + ); } assert!( requests.await.unwrap().is_empty(), From b3b553be181baf327df803b3b73ab3ff3e4105a5 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:07:46 +0000 Subject: [PATCH 5/6] Preserve CancellationRequest source compatibility with a cause accessor --- CHANGELOG.md | 13 ++++++------- docs/cooperative-cancellation-design.md | 6 +++++- src/cooperative_cancellation.rs | 22 ++++++++++------------ src/tests/cooperative_cancellation.rs | 11 ++++++----- src/tests/nexus.rs | 1 - tests/public_task_model_migration.rs | 24 ++++++++++++++++++++++++ 6 files changed, 51 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 577b732..09b285c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ the two apart with the new `#[non_exhaustive]` `CancellationCause`: `Requested` for an ordinary request (`cause` absent or `"cancellation"`) or `Timeout`. The request's cleanup deadline is its existing - `cleanup_deadline_at`. `CancellationContext::cause()` and `is_timeout()` + `cleanup_deadline_at`. `CancellationRequest::cause()`, `CancellationContext::cause()` and `is_timeout()` read it from the context. A context with any other `cause` is refused as `Error::InvalidCooperativeCancellation`. The cause is never inherited: a descendant cancelled under the request, scoped or not, carries no `cause` @@ -21,12 +21,11 @@ back exactly as read, including an explicit `"cancellation"`, and an ordinary context's output is unchanged. - **Minor-release note:** `CancellationRequest` gains a public - `cause: CancellationCause` field. Code that builds a `CancellationRequest` - with a struct literal or destructures it exhaustively must add the field - (`cause: CancellationCause::Requested` for an ordinary request). Requests - the SDK produces already set it. `CancellationRequest` will become - `#[non_exhaustive]` in the next major release. + `CancellationRequest::cause()` derives the discriminator from the existing + context, preserving stable 4.x struct literals and exhaustive destructuring. + This parser support is experimental. Service-mode timeout-cleanup admission + and execution require the upstream Core/native implementation and separate + published-artifact qualification. - A workflow-task completion that native Server 3.6.0 refuses as too large (HTTP 413, `limit_kind: workflow_task_completion_bytes`, server#1152, diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md index 47926af..f2085b6 100644 --- a/docs/cooperative-cancellation-design.md +++ b/docs/cooperative-cancellation-design.md @@ -41,12 +41,16 @@ caller type, ID and label. Timestamp helpers return immutable UTC dates. ## Timeout cleanup cause +This parser support is experimental. Service-mode timeout-cleanup admission +and execution require the upstream Core/native implementation and separate +published-artifact qualification. + A run that opted in to deadline-triggered cleanup receives its timeout-cleanup interruption as an ordinary cooperative cancellation request whose context has the top-level `cause: "timeout"`. It arrives as `Error::CooperativeCancellationRequested`, so cancellation handlers such as saga compensation run for it unchanged, and the run still ends timed out. -`request.cause` (and `CancellationContext::cause()`) is the discriminator: +`request.cause()` (and `CancellationContext::cause()`) is the discriminator: `CancellationCause::Requested` for an ordinary request or `CancellationCause::Timeout`. The cleanup deadline is the request's existing `cleanup_deadline_at`; the timeout kind and work deadline are recorded on the diff --git a/src/cooperative_cancellation.rs b/src/cooperative_cancellation.rs index 92e86b0..9114bff 100644 --- a/src/cooperative_cancellation.rs +++ b/src/cooperative_cancellation.rs @@ -131,12 +131,6 @@ pub struct CancellationRequest { pub cleanup_deadline_at: String, pub history_refresh_page_token: Option, pub context: Option, - /// Why the request was delivered, read from the canonical context's - /// `cause`. A timeout-cleanup request arrives through the same - /// [`Error::CooperativeCancellationRequested`] as an ordinary one, so - /// existing handlers run for both; match on this to tell them apart. - /// `Requested` when no canonical context is known yet. - pub cause: CancellationCause, } /// Acknowledgment of a request targeting the current durable run. @@ -185,6 +179,16 @@ fn text<'a>(value: &'a Value, field: &str) -> Result<&'a str> { } impl CancellationRequest { + /// Why this request was delivered, derived from its canonical context. + /// + /// An observation without a canonical context is an ordinary request. + /// Both causes arrive through [`Error::CooperativeCancellationRequested`]. + pub fn cause(&self) -> CancellationCause { + self.context + .as_ref() + .map_or(CancellationCause::Requested, CancellationContext::cause) + } + pub(super) fn preserve_observation(&self, current: &Self) -> Result { self.validate_observation()?; current.validate_observation()?; @@ -226,7 +230,6 @@ impl CancellationRequest { cleanup_deadline_at: cleanup_deadline_at.to_owned(), history_refresh_page_token: Some(text(value, "history_refresh_page_token")?.to_owned()), context: None, - cause: CancellationCause::Requested, }) } } @@ -508,7 +511,6 @@ impl CancellationHistory { let mut state = Self { request: observation.cloned().map(|mut request| { request.context = None; - request.cause = CancellationCause::Requested; request }), delivery: None, @@ -597,9 +599,6 @@ impl CancellationHistory { } } if context.is_some() || observation.is_none() { - let cause = context - .as_ref() - .map_or(CancellationCause::Requested, CancellationContext::cause); state.request = Some(CancellationRequest { request_id: request_id.to_owned(), requested_at: context.as_ref().map_or_else( @@ -614,7 +613,6 @@ impl CancellationHistory { history_refresh_page_token: observation .and_then(|request| request.history_refresh_page_token.clone()), context, - cause, }); } state.request_index = index; diff --git a/src/tests/cooperative_cancellation.rs b/src/tests/cooperative_cancellation.rs index e40ba17..7239068 100644 --- a/src/tests/cooperative_cancellation.rs +++ b/src/tests/cooperative_cancellation.rs @@ -5786,13 +5786,14 @@ fn cancellation_cause_timeout_reaches_an_existing_cooperative_cancellation_arm() else { panic!("a timeout cleanup request is a cooperative cancellation request"); }; - assert_eq!(cancelled.request.cause, CancellationCause::Timeout); + assert_eq!(cancelled.request.cause(), CancellationCause::Timeout); assert_eq!( cancelled.request.cleanup_deadline_at, "2026-10-01T00:00:30.123456Z" ); + let cause = cancelled.request.cause(); let context = cancelled.request.context.unwrap(); - assert_eq!(context.cause(), cancelled.request.cause); + assert_eq!(context.cause(), cause); assert!(ctx.cancellation_context()?.unwrap().is_timeout()); Ok(context.to_value()) }); @@ -5818,13 +5819,13 @@ fn cancellation_cause_ordinary_delivery_and_observation_stay_requested() { else { panic!("expected delivery"); }; - assert_eq!(cancelled.request.cause, CancellationCause::Requested); + assert_eq!(cancelled.request.cause(), CancellationCause::Requested); Ok(Value::Null) }); worker .execute_workflow_task(context_task(vec![context_request(), context_delivery()])) .unwrap(); - assert_eq!(context_observation().cause, CancellationCause::Requested); + assert_eq!(context_observation().cause(), CancellationCause::Requested); } #[test] @@ -5853,5 +5854,5 @@ fn cancellation_cause_delivery_cannot_add_or_drop_the_timeout_cause() { None, ) .unwrap(); - assert!(history.request.unwrap().cause.is_timeout()); + assert!(history.request.unwrap().cause().is_timeout()); } diff --git a/src/tests/nexus.rs b/src/tests/nexus.rs index a03c55a..de27ef1 100644 --- a/src/tests/nexus.rs +++ b/src/tests/nexus.rs @@ -157,7 +157,6 @@ fn root_request_selects_original_service_boundary_without_readmission() { cleanup_deadline_at: "2026-10-12T14:00:30Z".into(), history_refresh_page_token: Some("original-page".into()), context: None, - cause: CancellationCause::Requested, }; let decision = worker() .execute_workflow_task_decision_with_cancellation(task, Some(&observation)) diff --git a/tests/public_task_model_migration.rs b/tests/public_task_model_migration.rs index 7e6cd97..78b1966 100644 --- a/tests/public_task_model_migration.rs +++ b/tests/public_task_model_migration.rs @@ -68,3 +68,27 @@ fn tasks_remain_constructible_with_explicit_absent_optional_metadata() { assert_eq!(workflow.payload_codec, activity.payload_codec); assert_eq!(activity.payload_codec, query.payload_codec); } + +#[test] +fn cancellation_request_keeps_its_existing_literal_and_exhaustive_pattern() { + use durable_workflow::{CancellationCause, CancellationRequest}; + let request = CancellationRequest { + request_id: "request-1".into(), + requested_at: "2026-10-01T00:00:00Z".into(), + cleanup_deadline_at: "2026-10-01T00:00:30Z".into(), + history_refresh_page_token: None, + context: None, + }; + assert_eq!(request.cause(), CancellationCause::Requested); + let CancellationRequest { + request_id, + requested_at, + cleanup_deadline_at, + history_refresh_page_token, + context, + } = request; + assert_eq!(request_id, "request-1"); + assert_ne!(requested_at, cleanup_deadline_at); + assert!(history_refresh_page_token.is_none()); + assert!(context.is_none()); +} From 5633fc8a03dec87f7ff988178ac24cbc34d3ed3f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:09:28 +0000 Subject: [PATCH 6/6] Share query route validation across result formats --- src/lib.rs | 72 ++++++++++++++++++++---------------------------------- 1 file changed, 26 insertions(+), 46 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 5ab4a39..160f2f7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3736,39 +3736,9 @@ impl Client { query_name: &str, input: T, ) -> Result { - let input = normalize_avro_arguments(AvroValue::from_serialize(&input)?); - let body = json!({"input": encode_typed_envelope(&input, DEFAULT_CODEC)?}); - let path = match run_id { - Some(run_id) => { - format!( - "/workflows/{}/runs/{}/query/{}", - path_segment("workflow_id", workflow_id)?, - path_segment("run_id", run_id)?, - path_segment("query_name", query_name)?, - ) - } - None => format!( - "/workflows/{}/query/{}", - path_segment("workflow_id", workflow_id)?, - path_segment("query_name", query_name)?, - ), - }; - let response: Value = match self - .request_json( - reqwest::Method::POST, - &path, - RequestProtocol::ControlPlane, - Some(&body), - ) - .await - { - Ok(response) => response, - Err(Error::Http { status, body }) => { - return Err(Error::QueryFailed(query_failure(status, body))); - } - Err(error) => return Err(error), - }; - + let response = self + .query_workflow_response(workflow_id, run_id, query_name, input) + .await?; let envelope = response .get("result_envelope") .filter(|envelope| !envelope.is_null()) @@ -3787,12 +3757,30 @@ impl Client { run_id: Option<&str>, query_name: &str, input: T, + ) -> Result { + let response = self + .query_workflow_response(workflow_id, run_id, query_name, input) + .await?; + if let Some(envelope) = response + .get("result_envelope") + .filter(|envelope| !envelope.is_null()) + { + return decode_wire_value(envelope, DEFAULT_CODEC); + } + + Ok(response.get("result").cloned().unwrap_or(Value::Null)) + } + + // Typed and JSON readers share route validation and HTTP refusal handling. + async fn query_workflow_response( + &self, + workflow_id: &str, + run_id: Option<&str>, + query_name: &str, + input: T, ) -> Result { let input = normalize_avro_arguments(AvroValue::from_serialize(&input)?); - let input_envelope = encode_typed_envelope(&input, DEFAULT_CODEC)?; - let body = json!({ - "input": input_envelope - }); + let body = json!({"input": encode_typed_envelope(&input, DEFAULT_CODEC)?}); let path = match run_id { Some(run_id) => { format!( @@ -3823,15 +3811,7 @@ impl Client { } Err(error) => return Err(error), }; - - if let Some(envelope) = response - .get("result_envelope") - .filter(|envelope| !envelope.is_null()) - { - return decode_wire_value(envelope, DEFAULT_CODEC); - } - - Ok(response.get("result").cloned().unwrap_or(Value::Null)) + Ok(response) } /// Send a synchronous update using fixed Avro Value arguments.