diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bc7e36..09b285c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,31 @@ ## 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`. The request's cleanup deadline is its existing + `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` + 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. + + `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, 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..f2085b6 100644 --- a/docs/cooperative-cancellation-design.md +++ b/docs/cooperative-cancellation-design.md @@ -39,6 +39,27 @@ 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 + +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: +`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 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..7c7624f 100644 --- a/src/cancellation_context.rs +++ b/src/cancellation_context.rs @@ -32,6 +32,38 @@ impl CancellationLineage { } } +/// 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 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, +} + +impl CancellationCause { + /// Whether this is a timeout-cleanup request. + pub fn is_timeout(self) -> bool { + 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 +81,7 @@ pub struct CancellationContext { requested_at: DateTime, cleanup_deadline_at: DateTime, lineage: Vec, + cause: Option, scope_origin: Option>, replay: Option>>, } @@ -66,6 +99,7 @@ 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.scope_origin == other.scope_origin } } @@ -93,6 +127,20 @@ fn nullable_text(value: &Value, key: &str, allow_empty: bool) -> Result Result> { + match value.get("cause") { + None | Some(Value::Null) => Ok(None), + Some(Value::String(cause)) if cause == "cancellation" => { + Ok(Some(RecordedCause::Cancellation)) + } + Some(Value::String(cause)) if cause == "timeout" => Ok(Some(RecordedCause::Timeout)), + Some(_) => Err(invalid_context( + "cancellation cause must be cancellation or timeout", + )), + } +} + impl CancellationContext { /// Read legacy v1 and candidate scoped v2 without granting a new cleanup budget. pub fn from_value(value: &Value) -> Result { @@ -187,6 +235,7 @@ impl CancellationContext { "cancellation deadline must follow the original request", )); } + let cause = read_cause(value)?; let context = Self { request_id, root_request_id, @@ -199,6 +248,7 @@ impl CancellationContext { requested_at, cleanup_deadline_at, lineage: normalized, + cause, scope_origin, replay: None, }; @@ -293,6 +343,19 @@ 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, + _ => 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 +400,11 @@ 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(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..9114bff 100644 --- a/src/cooperative_cancellation.rs +++ b/src/cooperative_cancellation.rs @@ -179,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()?; diff --git a/src/lib.rs b/src/lib.rs index 05632ab..bdca4d1 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, }; pub use cancellation_scope::CancellationScopeOpenReceipt; #[doc(hidden)] diff --git a/src/tests/cooperative_cancellation.rs b/src/tests/cooperative_cancellation.rs index 1566788..7239068 100644 --- a/src/tests/cooperative_cancellation.rs +++ b/src/tests/cooperative_cancellation.rs @@ -5598,3 +5598,261 @@ 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_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); + same_bytes(&context.to_value(), &snapshot); + assert!(context.to_value().get("cause").is_none()); + // 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!(!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() { + 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); + assert!(context.cause().is_timeout()); + assert_eq!(context.to_value()["cause"], "timeout"); + same_bytes(&context.to_value(), &timeout); + assert_eq!( + CancellationContext::from_value(&context.to_value()).unwrap(), + context + ); + // 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_values_are_refused() { + for cause in [ + json!("deadline"), + json!("Timeout"), + json!(""), + json!(true), + json!(1), + json!({"kind": "timeout"}), + ] { + assert!( + matches!( + CancellationContext::from_value(&with_cause(core_cause_snapshot(), cause.clone())), + Err(Error::InvalidCooperativeCancellation(_)) + ), + "{cause} must be refused" + ); + } +} + +#[test] +fn cancellation_cause_descendant_is_cancelled_not_timed_out() { + let parent = with_cause(core_cause_snapshot(), json!("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); +} + +/// 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_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(&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 { + 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_cause(context_snapshot(), json!("timeout")); + 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"); + }; + 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(), 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_cause(context_snapshot(), json!("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/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()); +}