diff --git a/docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md b/docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md new file mode 100644 index 000000000..bf70b8580 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md @@ -0,0 +1,130 @@ +# October 27, native serves cancellation scope prefix checkpoints + +- **The route.** `POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint` + was the last published PHP route native answered + `501 rust_capability_not_implemented`. Native now answers all 175 API + route methods in the published route table. +- **Why it blocked the release candidate** + ([#325 comment 6449500902](https://github.com/durable-workflow/server/issues/325#issuecomment-6449500902)). + - The published PHP SDK (3.6.1) calls it under cooperative cancellation, + which native advertises (`cooperative_cancellation: true`). It does so + when a turn recorded commands (a side effect, a version marker, a memo + or search-attribute upsert) before it opens or delivers a scope. + - The SDK turns any non-transient refusal into `WorkflowClaimAborted` + and fails the task. Replay is deterministic, so the workflow is stuck. + - A typed 409 refusal could not be made consistent. In Core, open, + prepare, deliver and checkpoint sit on one `CancellationScopeTaskBridge`, + and no capability flag tells an SDK to skip the checkpoint. + - The Python SDK (2.12.0), the Rust SDK (4.10.1) and the CLI (2.3.1) never + call it. +- **Contract.** Server's `checkpointCancellationScopePrefix` and + `mutateWorkflowTaskCommands`, over Core 2.15's + `DefaultWorkflowTaskBridge::checkpointCancellationScopePrefix` + (`checkpointCommands`). In order: + - protocol 1.20, or `409 cancellation_scope_checkpoint_requires_protocol_1_20`; + - `checkpoint_id` (string, at most 255), `start_sequence` (integer, at + least 1), and the prohibited `message_stream_cursors`, + `message_stream_waits` and `sticky_cache`, with Laravel's messages; + - a completion's own validation and guards: ownership, scope admission, + worker sessions, child cancellation policies, local activities; + - the claim must be the cooperative claim Server issued, or + `409 active_claim_cancellation_not_supported` (no refresh token); + - Core's refusals: the checkpoint arguments; a terminal command or a + turn-closing one (`record_local_activity`, `open_condition_wait`, + `open_signal_wait`, `cancel_selection_operation`); the claim; a closed + run or passed run deadline; the retained receipt; a root cancellation + request not yet delivered, or a prefix inside the delivered boundary, + or a passed cleanup deadline; the authored sequence; scope membership; + update commands; parallel sequences; the history meter. + - An accepted prefix applies its commands as a completion does, metered. + The task stays leased with its lease unchanged, and no history event of + its own is written. The task payload keeps one receipt, + `portable_scope_checkpoint`: checkpoint id, task, run, attempt, lease + owner and expiry, start and next sequence, fingerprint, created task + ids, and record time. + - An identical retry answers the receipt with `duplicate: true`. The same + id with anything else is `cancellation_scope_checkpoint_mismatch`. + A new receipt replaces the last one. + - Answers are 200, 404 for an unknown task, or 409, in the worker + envelope with `history_refresh_page_token`. +- **Native.** + - `scope_checkpoint.rs` holds the route's rules and Core's checks. + - `complete_workflow`'s non-terminal command arms move unchanged into + `apply_non_terminal`, which the completion and the checkpoint share. + - The completion's HTTP checks are shared too (`completion_checks`). +- **Divergence: the fingerprint value.** Core hashes its own normalized + command arrays (Server's and Core's normalizers). Native hashes the + validated request commands. The two servers therefore give different + values for the same prefix. The value identifies a retry on the server + that issued it; no client computes or compares it. The fixture names each + fingerprint and compares only its reuse. +- **Review corrections, against Core 2.15.2.** Three also change the + completion, where native had diverged from Core: + - *Completion and checkpoint:* a signal-resumed claim applies its signal + once (`recordAppliedSignalForSignalResume`). An already-applied signal + is left as it is, instead of refused with `signal_already_applied`, so + a claim that checkpointed still completes. + - *Completion, checkpoint and scope open:* the authored sequence counts + `SearchAttributesUpserted`, as `nextDurableCommandSequence` does. + - *Completion and checkpoint:* `complete_update` and `fail_update` take + the batch's next sequence and advance it (`applyCompleteUpdate`, + `applyFailUpdate`). A later command in the batch takes the following + sequence. Update events still do not count toward the next turn. + - *Checkpoint only:* + - A signal Core would not apply (its command unavailable, its wait not + open, another wait open) is skipped, as Core skips it. The completion + keeps native's refusals for those. + - A checkpoint changes no run status or summary; Core's + `markRunWaiting` and projection run at completion. + - Core's own refusals answer as a refused checkpoint. Server's + exception answers (structural limits, memo validation) keep their + shapes, as Server's catch block gives them for both routes. +- **Deferred: wait rows.** Native writes an activity's or timer's wait row + in the transaction that schedules it. Later events, such as the activity + completing while the turn is still open, update that row and require it + to exist. PHP's projector writes the row at completion. After a + checkpoint, native therefore lists the new wait while the claim is still + held, and PHP lists it only after completion. Both agree once the turn + completes. Deferring the row needs a rebuild-from-history path in + native's wait projection. +- **Not ported:** Server's `persistTypedSearchAttributeHistoryIdentity` + reads the receipt's `next_sequence` to check that Core wrote typed + search-attribute history. Native writes that history itself. +- **Shared fixture.** The `cancellation-scopes` probe adds run R5, under + `scope_checkpoints`. + - It covers the gates, validation and Core's refusals. + - It commits a prefix, retries it, sends a mismatch and a stale sequence, + then replaces the receipt with a second prefix (a timer and its task). + - The claim then opens its scope and completes. + - R3's plain claim and R4's protocol 1.19 claim are refused. + - Every new identity and fingerprint is named, and the key sorts last, so + no earlier label changes. + - The expectations are published 2.11.5's answers, where every existing + step still matched. +- **Checked.** + - Native `cargo test` on SQLite: the full library and execution suites + after the refactor (338 and 562 tests), and the four new execution + tests `scope_checkpoints::*`. These cover success and retry, mismatch, + sequence, refused commands, protocol, validation, owner and claim gates, + a timer with its task, a prefix after a delivered root cancellation, and + the claim's later scope opening and completion. Route coverage + expects no unserved route. + - The whole execution suite, the new tests included, passes on + PostgreSQL 17 and MySQL 8.0 (566 each; CI's pinned images, the test + binary run in CI's Rust image on the same network). On SQLite one + timing test, `updates_have_no_lifetime_cap_and_preserve_accepted_work_after_wait_timeout` + (a one-second update wait), failed once while a release image build + saturated the host, and passed 3 of 3 alone. + - The contract unit test (27 cases). + - On main's 2.11.5 reference (after #815), `native-image-corpus.sh` + ran with a native image built from this branch: CI's whole SQLite + selection, `update-validation` and `service-calls` included, against + published 2.11.5 (`sha256:bcc67285…`) and embedded 2.11.5. All three + pass 73 of 73 cases, `cancellation-scopes` among them, and the records + compare equal. + - After the review corrections, the whole native suite passes: library + 339 on SQLite, and execution 570 on SQLite, PostgreSQL 17 and MySQL 8.0. + `updates_have_no_lifetime_cap_and_preserve_accepted_work_after_wait_timeout`, + which waits one second for an update, failed on PostgreSQL and MySQL + while a release image build saturated the host. Run alone it passed 3 + of 3 on each. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 0d4aa3496..c4a9384b3 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -71,8 +71,11 @@ Updated October 27, 2026. for `update-validation` and `service-calls` (#723, #726, #796). No expectation is edited to make native pass. -- **Parity surface.** Native answers 174 of the 175 API route methods in - published 2.11.1's route table, which 2.11.4 and 2.11.5 leave unchanged. Eight of those are the prepared +- **Parity surface.** Native answers all 175 API route methods in + published 2.11.1's route table, which 2.11.4 and 2.11.5 leave unchanged. + The last, the cancellation scope prefix checkpoint, is served since + October 27 + ([checkpoint](rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md)). Eight of those are the prepared local-activity routes, which it answers with PHP's refusal for a bridge without the prepared role. Nexus workflow-bound handler kinds answer 501 until #794 settles their contract. An execution test sends each route @@ -141,8 +144,11 @@ any other unknown path or method answers PHP's own 404 or 405 pins the answers to `PreparedLocalActivityController::refused`. Serving them is planned in [#325 comment 6423707058](https://github.com/durable-workflow/server/issues/325#issuecomment-6423707058). - - Cancellation scope checkpoints. (Open, prepare and deliver are served, - with the `cancellation-scopes` fixture.) + - (Cancellation scopes are served: open, prepare, deliver and the prefix + checkpoint, with the `cancellation-scopes` fixture. The published PHP + SDK checkpoints a turn's earlier commands before it opens or delivers a + scope, and fails the task on any refusal + ([checkpoint](rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md)).) - (Worker sessions are served: create, heartbeat, close, list and detail, with the `worker-sessions` and `worker-session-activities` fixtures.) - (Activity cancellation acknowledgement is served: @@ -395,7 +401,8 @@ contract fixture before its correction. refuses a request whose lease is otherwise valid when its owner's registration is no longer fresh (`409 stale_worker_registration`). Native's workflow task guard does not check registration freshness yet, and the - local activity and cancellation scope checkpoints still use the older fence. + local activity checkpoints still use the older fence. The cancellation + scope checkpoint shares the completion's guard. Add a fixture for each before changing them. - **Retention of runs with external payloads.** Published PHP deletes a pruned run's external payload objects. Native does not delete them yet: it diff --git a/rust/src/runtime/cancellation_scopes.rs b/rust/src/runtime/cancellation_scopes.rs index 41b7dcf0d..a8ae67498 100644 --- a/rust/src/runtime/cancellation_scopes.rs +++ b/rust/src/runtime/cancellation_scopes.rs @@ -1114,7 +1114,7 @@ const DELIVERY_RULES: &laravel_rules::Rules = &[ ("operation_sequence_span", &[Rule::Sometimes, Rule::Required, Rule::Integer, Rule::Min(1)]), ]; -fn protocol_supported(protocol: &str) -> bool { +pub(super) fn protocol_supported(protocol: &str) -> bool { super::cooperative::supports(&json!(["cooperative_cancellation"]), protocol) } diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 56e20d1e4..a15c9d80e 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -1527,6 +1527,30 @@ impl Runtime { body: Value, worker: Option, protocol: Option<&str>, + ) -> Result { + self.complete_or_checkpoint(task_id, body, worker, protocol, None).await + } + + /// POST …/cancellation-scopes/checkpoint, after its route checks: the + /// completion's payload preparation, then the store's checkpoint. + pub(crate) async fn checkpoint_cancellation_scope( + &self, + task_id: &str, + body: Value, + worker: Option, + protocol: Option<&str>, + checkpoint: &super::scope_checkpoint::Checkpoint, + ) -> Result { + self.complete_or_checkpoint(task_id, body, worker, protocol, Some(checkpoint)).await + } + + async fn complete_or_checkpoint( + &self, + task_id: &str, + body: Value, + worker: Option, + protocol: Option<&str>, + checkpoint: Option<&super::scope_checkpoint::Checkpoint>, ) -> Result { self.check_command_payloads(&body).await?; let external = self.prepare_command_payloads(&body).await?; @@ -1572,17 +1596,17 @@ impl Runtime { match &*self.storage { Storage::Sqlite(store) => { store - .complete_workflow(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural) + .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint) .await } Storage::Postgres(store) => { store - .complete_workflow(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural) + .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint) .await } Storage::MySql(store) => { store - .complete_workflow(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural) + .complete_or_checkpoint(task_id, body, self.transition_codec.clone(), external, worker, protocol, self.structural, checkpoint) .await } } diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 736b94dff..6ab2f927d 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -311,6 +311,10 @@ pub fn router(runtime: Runtime) -> Router { "/api/worker/workflow-tasks/{task_id}/cancellation-scopes/deliver", post(deliver_cancellation_scope), ) + .route( + "/api/worker/workflow-tasks/{task_id}/cancellation-scopes/checkpoint", + post(checkpoint_cancellation_scope), + ) // Native lacks the prepared local-activity role: refused as PHP // refuses them without it (prepared_local.rs). .route("/api/worker/workflow-tasks/{task_id}/local-activities/prepare", post(prepared_local_refused)) @@ -863,7 +867,7 @@ fn worker_route(rest: &str) -> Option { | ["update-validation-tasks", "poll"] | ["update-validation-tasks", _, "approve" | "reject"] | ["workflow-tasks", _, "complete" | "fail" | "history" | "heartbeat" | "deliver-cancellation"] - | ["workflow-tasks", _, "cancellation-scopes", "open" | "prepare" | "deliver"] + | ["workflow-tasks", _, "cancellation-scopes", "open" | "prepare" | "deliver" | "checkpoint"] | ["workflow-tasks", _, "local-activities", "prepare" | "recover" | "checkpoint" | "checkpoint-group"] | ["workflow-tasks", _, "local-activities", _, "control" | "heartbeat" | "outcome" | "acknowledge-cancellation"] | ["activity-tasks", _, "complete" | "fail" | "heartbeat" | "status" | "renew-lease" | "acknowledge-cancellation"] @@ -2177,6 +2181,26 @@ async fn complete_workflow( headers: HeaderMap, Json(body): Json, ) -> Result> { + let body = completion_checks(&task_id, &headers, body)?; + let attempt = body["workflow_task_attempt"].clone(); + runtime + .complete_workflow(&task_id, body, Some(request.principal().clone()), headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok())) + .await + .map_err(|error| { + completion_refusal( + error, + &task_id, + "workflow_task_attempt", + attempt, + "Workflow task not found.", + ) + }) + .map(|body| Json(worker_response(body))) +} + +/// The completion request's checks before the runtime, shared by the +/// completion and the cancellation scope checkpoint. +fn completion_checks(task_id: &str, headers: &HeaderMap, body: Value) -> Result { // PHP validates the message-stream fields first, then its transport // rules, before every guard, ownership included. super::message_streams::check_completion_fields(&body).map_err(worker_refusal)?; @@ -2186,48 +2210,69 @@ async fn complete_workflow( .iter() .any(|command| command["type"] == "upsert_memo") }) { - memo_protocol(&headers)?; + memo_protocol(headers)?; } // `guardTypedSearchAttributeCommandsAvailable`. if body["commands"].as_array().is_some_and(|commands| { commands.iter().any(|command| { command["type"] == "upsert_search_attributes" && command.get("attribute_types").is_some() }) - }) && !protocol_at_least(&headers, 16)? + }) && !protocol_at_least(headers, 16)? { let minimum = super::search_attribute_upserts::MINIMUM_PROTOCOL_VERSION; return Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response: json!({"task_id":task_id,"workflow_task_attempt":body["workflow_task_attempt"], "outcome":"rejected","recorded":false,"reason":"typed_search_attributes_unavailable", - "requested_version":request_protocol(&headers)?,"minimum_protocol_version":minimum, + "requested_version":request_protocol(headers)?,"minimum_protocol_version":minimum, "remediation":format!("Complete typed search-attribute updates only through server nodes advertising worker protocol {minimum} or newer.")}), }); } - if !super::message_streams::Completion::of(&body).is_empty() && !protocol_at_least(&headers, 15)? { + if !super::message_streams::Completion::of(&body).is_empty() && !protocol_at_least(headers, 15)? { let minimum = super::message_streams::MINIMUM_PROTOCOL_VERSION; return Err(super::RuntimeError::Protocol { status: StatusCode::CONFLICT, response: json!({"task_id":task_id,"workflow_task_attempt":body["workflow_task_attempt"], "outcome":"rejected","reason":"message_streams_unavailable", - "requested_version":request_protocol(&headers)?,"minimum_protocol_version":minimum, + "requested_version":request_protocol(headers)?,"minimum_protocol_version":minimum, "remediation":format!("Send the X-Durable-Workflow-Protocol-Version header with worker protocol {minimum} or newer.")}), }); } + Ok(body) +} + +/// POST …/cancellation-scopes/checkpoint (`checkpointCancellationScopePrefix`): +/// protocol 1.20, the checkpoint fields, then a completion's checks and +/// guards, with Core's checkpoint in place of the completion. Every answer +/// is in the worker envelope; Core's answers carry the history refresh +/// token, at 200, 404 for an unknown task, or 409. +async fn checkpoint_cancellation_scope( + Extension(runtime): Extension, + request: CommandRequest, + Path(task_id): Path, + headers: HeaderMap, + Json(body): Json, +) -> Result<(StatusCode, Json)> { + let protocol = request_protocol(&headers)?; + if !super::cancellation_scopes::protocol_supported(protocol) { + return Ok((StatusCode::CONFLICT, Json(worker_response(json!({"checkpointed": false, + "reason": "cancellation_scope_checkpoint_requires_protocol_1_20"}))))); + } + let normalized = super::laravel_rules::normalize(&body); + super::laravel_rules::validate(&normalized, super::scope_checkpoint::RULES).map_err(worker_refusal)?; + let checkpoint = super::scope_checkpoint::Checkpoint::of(&normalized); + let body = completion_checks(&task_id, &headers, body)?; let attempt = body["workflow_task_attempt"].clone(); - runtime - .complete_workflow(&task_id, body, Some(request.principal().clone()), headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok())) + let mut response = runtime + .checkpoint_cancellation_scope(&task_id, body, Some(request.principal().clone()), Some(protocol), &checkpoint) .await - .map_err(|error| { - completion_refusal( - error, - &task_id, - "workflow_task_attempt", - attempt, - "Workflow task not found.", - ) - }) - .map(|body| Json(worker_response(body))) + .map_err(|error| completion_refusal(error, &task_id, "workflow_task_attempt", attempt, "Workflow task not found."))?; + let status = super::scope_checkpoint::status(&response); + // Server's claim check answers before Core, without the token. + if response["reason"] != "active_claim_cancellation_not_supported" { + response["history_refresh_page_token"] = json!("MA=="); + } + Ok((status, Json(worker_response(response)))) } /// Whether the request's worker protocol is `1.` or newer. diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 0b5bbe60f..d202ecd16 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -41,6 +41,7 @@ mod bundle_verifier; mod cancellation; mod cancellation_cascade; mod cancellation_scopes; +mod scope_checkpoint; mod child_summary; mod child_cancellation; mod children; diff --git a/rust/src/runtime/scope_checkpoint.rs b/rust/src/runtime/scope_checkpoint.rs new file mode 100644 index 000000000..bce017bee --- /dev/null +++ b/rust/src/runtime/scope_checkpoint.rs @@ -0,0 +1,351 @@ +//! `POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint`: +//! Server's `WorkerController::checkpointCancellationScopePrefix` and Core's +//! `DefaultWorkflowTaskBridge::checkpointCancellationScopePrefix`. +//! +//! A cooperative worker that recorded commands earlier in its turn (a side +//! effect, a version marker, a memo or search-attribute upsert, or an +//! activity, timer or child) checkpoints them before it opens or delivers a +//! cancellation scope. The commands are applied as a completion applies them, +//! under the same claim: the task stays leased with its lease unchanged, and +//! no history event of its own is written. The task payload keeps one +//! receipt, `portable_scope_checkpoint`, which an identical retry returns +//! with `duplicate: true`. +//! +//! The receipt's `fingerprint` is a SHA-256 over the checkpointed commands. +//! It identifies a retry of the same prefix on this server. Core hashes its +//! own normalized command arrays; native hashes the validated request +//! commands, so the two servers' values differ for the same commands. No +//! client computes or compares it (Server #325). +use super::laravel_rules::{self, Rule}; +use super::store::now; +use super::transaction::Transaction; +use super::{Result, RuntimeError}; +use axum::http::StatusCode; +use chrono::{DateTime, Utc}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; + +/// The checkpoint fields Server validates before the completion's own rules, +/// in its order (`mutateWorkflowTaskCommands`). +pub(super) const RULES: &laravel_rules::Rules = &[ + ("checkpoint_id", &[Rule::Required, Rule::String, Rule::Max(255)]), + ("start_sequence", &[Rule::Required, Rule::Integer, Rule::Min(1)]), + ("message_stream_cursors", &[Rule::Prohibited]), + ("message_stream_waits", &[Rule::Prohibited]), + ("sticky_cache", &[Rule::Prohibited]), +]; + +/// The validated checkpoint request. +pub(crate) struct Checkpoint { + pub(crate) id: String, + pub(crate) start_sequence: i64, +} + +impl Checkpoint { + /// The validated fields of a normalized body that passed [`RULES`]. + pub(super) fn of(body: &Value) -> Self { + let start_sequence = match &body["start_sequence"] { + Value::Number(number) => number.as_i64().or_else(|| number.as_f64().map(|float| float as i64)), + Value::String(text) => text.trim().parse().ok(), + _ => None, + }; + Self { + id: body["checkpoint_id"].as_str().unwrap_or_default().to_owned(), + start_sequence: start_sequence.unwrap_or_default(), + } + } +} + +/// The reason a checkpoint answer carries, and the status Server answers it +/// with (`workflowOutcomeStatus`). +pub(super) fn status(response: &Value) -> StatusCode { + match response["reason"].as_str() { + None => StatusCode::OK, + Some("task_not_found") => StatusCode::NOT_FOUND, + Some(_) => StatusCode::CONFLICT, + } +} + +/// Core's refused checkpoint. +fn refused(task_id: &str, reason: &str) -> Value { + json!({"checkpointed": false, "duplicate": false, "task_id": task_id, "reason": reason}) +} + +const INVALID_COMMANDS: &str = "invalid_cancellation_scope_checkpoint_commands"; + +/// Commands that cannot be part of a retained-claim prefix: a terminal +/// command, a wait (it closes a turn) and a local report (it reconstructs an +/// executed callback). Core also refuses `cancel_selection_operation` here; +/// native refuses that command type for any workflow task before this point +/// (`unsupported_workflow_command`). +fn excluded(command: &Value) -> bool { + matches!( + command["type"].as_str(), + Some("record_local_activity" | "open_condition_wait" | "open_signal_wait" | "complete_workflow" | "fail_workflow" + | "continue_as_new") + ) +} + +/// The prefix's identity for retries: SHA-256 over its commands' JSON. +fn fingerprint(commands: &[Value]) -> Result { + let encoded = serde_json::to_vec(commands).map_err(|_| super::refuse(StatusCode::CONFLICT, INVALID_COMMANDS))?; + Ok(super::bundle_verifier::hex(&Sha256::digest(&encoded))) +} + +/// A context's RFC 3339 instant. +fn instant(value: &Value) -> Option> { + value.as_str().and_then(|text| DateTime::parse_from_rfc3339(text).ok()).map(|at| at.with_timezone(&Utc)) +} + +store_methods! { + /// `PortableLocalActivityCleanup::snapshot` for the run's own request: a + /// prefix may follow a root cancellation request only after the request + /// was delivered, starting past the delivered boundary, with the + /// delivery's context the request's own. + async fn root_cleanup_proved(tx: &mut Transaction<'_, DB>, run: &DB::Row, start_sequence: i64) -> Result { + let Some(request_id) = DB::optional_string(run, "cancellation_request_command_id")? else { + return Ok(false); + }; + let run_id = DB::string(run, "id")?; + let event = |kind: &'static str| { + Self::query("SELECT sequence,payload,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type=$2 AND workflow_command_id=$3 ORDER BY sequence LIMIT 1") + .bind(run_id.clone()).bind(kind).bind(request_id.clone()) + }; + let request = event("CooperativeCancellationRequested").fetch_optional(&mut **tx).await?; + let delivery = event("CooperativeCancellationDelivered").fetch_optional(&mut **tx).await?; + let (Some(request), Some(delivery)) = (request, delivery) else { + return Ok(false); + }; + let payload = DB::document_row(&delivery, "payload")?; + let delivered_at = DB::optional_instant(run, "cancellation_delivered_at")?; + let Some(sequence) = payload["sequence"].as_i64() else { + return Ok(false); + }; + if DB::number(&delivery, "sequence")? <= DB::number(&request, "sequence")? + || payload["workflow_command_id"].as_str() != Some(request_id.as_str()) + || payload["workflow_run_id"].as_str() != Some(run_id.as_str()) + || DB::optional_number(run, "cancellation_delivery_sequence")? != Some(sequence) + || delivered_at.is_none_or(|delivered_at| DB::optional_instant(&delivery, "recorded_at").ok().flatten() + .is_none_or(|recorded_at| recorded_at < delivered_at)) + { + return Ok(false); + } + let span = match payload.get("sequence_span") { + None | Some(Value::Null) => 1, + Some(span) => match span.as_i64() { + Some(span) if span >= 1 && span <= i64::MAX - sequence => span, + _ => return Ok(false), + }, + }; + if start_sequence < sequence + span { + return Ok(false); + } + let requested = &DB::document_row(&request, "payload")?["cancellation"]; + let delivered = &payload["cancellation"]; + let deadline = DB::optional_instant(run, "cancellation_deadline_at")?; + Ok(requested.is_object() && delivered.is_object() + && requested["request_id"].as_str() == Some(request_id.as_str()) + && requested["root_request_id"] == delivered["root_request_id"] + && requested["request_id"] == delivered["request_id"] + && instant(&requested["requested_at"]).is_some_and(|at| Some(at) == instant(&delivered["requested_at"])) + && instant(&requested["cleanup_deadline_at"]).is_some_and(|at| { + Some(at) == instant(&delivered["cleanup_deadline_at"]) && Some(at) == deadline + })) + } + + /// Core's `checkpointCommands` for a cancellation scope prefix, after + /// Server's guards: refusals answer Core's refused checkpoint and change + /// nothing; an accepted prefix commits its commands and the receipt. + #[allow(clippy::too_many_arguments)] + pub(super) async fn checkpoint_prefix( + &self, + mut tx: Transaction<'_, DB>, + task: DB::Row, + body: &Value, + commands: &[Value], + checkpoint: &Checkpoint, + transition_codec: std::sync::Arc, + external: &super::external_resolution::PreparedCommands, + local: &std::collections::BTreeMap, + exceptions: &std::collections::BTreeMap, + limits: super::memo_limits::StructuralLimits, + ) -> Result { + let task_id = DB::string(&task, "id")?; + // Server: the claim must be the cooperative claim it issued. + if !Self::claim_supports_cancellation(&task)? { + tx.rollback().await?; + return Ok(json!({"checkpointed": false, "reason": "active_claim_cancellation_not_supported"})); + } + let refusal = |reason: &str| Ok(refused(&task_id, reason)); + let owner = body["lease_owner"].as_str().unwrap_or_default(); + let attempt = body["workflow_task_attempt"].as_i64().unwrap_or_default(); + // Core's argument checks. Its command-count bound (100) is already + // the completion's, refused before this point. + if attempt < 1 || owner.is_empty() || checkpoint.id.trim().is_empty() || checkpoint.id.len() > 255 { + tx.rollback().await?; + return refusal("invalid_cancellation_scope_checkpoint"); + } + if commands.iter().any(excluded) { + tx.rollback().await?; + return refusal(INVALID_COMMANDS); + } + let fingerprint = fingerprint(commands)?; + // The claim, read in this transaction. + let expires = DB::optional_instant(&task, "lease_expires_at")?; + if DB::string(&task, "task_type")? != "workflow" || DB::string(&task, "status")? != "leased" + || DB::optional_string(&task, "lease_owner")?.as_deref() != Some(owner) + || DB::number(&task, "attempt_count")? != attempt + { + tx.rollback().await?; + return refusal("workflow_claim_mismatch"); + } + let Some(expires) = expires.filter(|expires| *expires > now()) else { + tx.rollback().await?; + return refusal("workflow_claim_expired"); + }; + let run_id = DB::string(&task, "workflow_run_id")?; + let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1").bind(&run_id).fetch_one(&mut *tx).await?; + if super::run_observation_store::terminal(&DB::string(&run, "status")?) { + tx.rollback().await?; + return refusal("run_closed"); + } + if [DB::optional_instant(&run, "execution_deadline_at")?, DB::optional_instant(&run, "run_deadline_at")?] + .into_iter() + .flatten() + .any(|deadline| deadline <= now()) + { + tx.rollback().await?; + return refusal("run_deadline_expired"); + } + // A retry of the retained receipt. + let payload = DB::document_row(&task, "payload")?; + let receipt = &payload[RECEIPT]; + if receipt.is_object() && receipt["checkpoint_id"].as_str() == Some(checkpoint.id.as_str()) { + tx.rollback().await?; + if receipt["workflow_task_attempt"].as_i64() != Some(attempt) + || receipt["lease_owner"].as_str() != Some(owner) + || receipt["start_sequence"].as_i64() != Some(checkpoint.start_sequence) + || receipt["fingerprint"].as_str() != Some(fingerprint.as_str()) + { + return refusal("cancellation_scope_checkpoint_mismatch"); + } + let mut response = receipt.clone(); + response["checkpointed"] = json!(true); + response["duplicate"] = json!(true); + response["reason"] = Value::Null; + return Ok(response); + } + if DB::optional_string(&run, "cancellation_request_command_id")?.is_some() { + if !Self::root_cleanup_proved(&mut tx, &run, checkpoint.start_sequence).await? { + tx.rollback().await?; + return refusal("cancellation_requested"); + } + if DB::optional_instant(&run, "cancellation_deadline_at")?.is_none_or(|deadline| deadline <= now()) { + tx.rollback().await?; + return refusal("cancellation_deadline_expired"); + } + } + let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?; + if sequence + 1 != checkpoint.start_sequence { + tx.rollback().await?; + return refusal("cancellation_scope_checkpoint_sequence_mismatch"); + } + let members: Vec<&Value> = commands.iter().collect(); + if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &members).await? { + tx.rollback().await?; + return refusal(reason); + } + match Self::validate_update_commands(&mut tx, &run_id, &task, commands).await { + Err(RuntimeError::Refused { reason, .. }) => { + tx.rollback().await?; + return refusal(reason); + } + other => other?, + } + // Metered like a completion (Workflow #930). + tx.meter = Some(super::history_meter::Meter::new(&limits, &task_id, &body["workflow_task_attempt"])); + // Core answers its own refusals as a refused checkpoint. + match Self::apply_task_signal(&mut tx, &run_id, &task_id, &task, transition_codec, true).await { + Ok(()) => {} + Err(RuntimeError::Refused { reason, .. }) => { + tx.rollback().await?; + return refusal(reason); + } + Err(error) => return Err(error), + } + let apply = super::store::Apply { + run: &run, + task: &task, + task_id: &task_id, + run_id: &run_id, + attempt: &body["workflow_task_attempt"], + limits: &limits, + external, + local, + exceptions, + }; + let mut created = Vec::new(); + for (index, command) in commands.iter().enumerate() { + match self.apply_non_terminal(&mut tx, &apply, index, command, &mut sequence, &mut created).await { + Ok(()) => {} + // `parallelCommandsMatchSequences`, and the meter's refusal. + Err(RuntimeError::Refused { reason: "parallel_sequence_mismatch", .. }) => { + tx.rollback().await?; + return refusal(INVALID_COMMANDS); + } + Err(RuntimeError::Protocol { response, .. }) + if response["reason"] == "history_transaction_limit_exceeded" => + { + tx.rollback().await?; + return refusal("history_transaction_limit_exceeded"); + } + Err(error) => return Err(error), + } + } + // Command application can change the task payload (a consumed + // signal): keep that, and add the receipt. + let current = Self::query("SELECT payload FROM workflow_tasks WHERE id=$1").bind(&task_id).fetch_one(&mut *tx).await?; + let mut payload = DB::document_row(¤t, "payload")?; + if !payload.is_object() { + payload = json!({}); + } + let receipt = json!({ + "checkpoint_id": checkpoint.id, + "task_id": task_id, + "workflow_run_id": run_id, + "workflow_task_attempt": attempt, + "lease_owner": owner, + "lease_expires_at": super::cancellation_scopes::iso(expires), + "start_sequence": checkpoint.start_sequence, + "next_sequence": sequence + 1, + "fingerprint": fingerprint, + "created_task_ids": created, + "recorded_at": super::cancellation_scopes::iso(now()), + }); + payload[RECEIPT] = receipt.clone(); + Self::query("UPDATE workflow_tasks SET payload=$1 WHERE id=$2") + .bind(DB::document(&payload)) + .bind(&task_id) + .execute(&mut *tx) + .await?; + if let Some(error) = tx.meter.as_ref().and_then(|meter| meter.tripped()) { + tx.rollback().await?; + return match error { + RuntimeError::Protocol { .. } => refusal("history_transaction_limit_exceeded"), + error => Err(error), + }; + } + // Core projects the run at completion, not at a checkpoint: the run's + // status and summary stay as the claim found them. + tx.commit().await?; + self.wake.notify_waiters(); + let mut response = receipt; + response["checkpointed"] = json!(true); + response["duplicate"] = json!(false); + response["reason"] = Value::Null; + Ok(response) + } +} + +/// The task payload key of the retained receipt. +const RECEIPT: &str = "portable_scope_checkpoint"; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 98781bd4a..f001098ab 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1192,7 +1192,7 @@ where .map(|row| DB::number(&row, "sequence")) .transpose()? .unwrap_or(0); - let wait = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('SignalWaitOpened','ConditionWaitOpened','VersionMarkerRecorded','MemoUpserted','SideEffectRecorded') ORDER BY sequence DESC LIMIT 1") + let wait = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('SignalWaitOpened','ConditionWaitOpened','VersionMarkerRecorded','MemoUpserted','SideEffectRecorded','SearchAttributesUpserted') ORDER BY sequence DESC LIMIT 1") .bind(run_id).fetch_optional(&mut **tx).await?; let wait_sequence = wait .map(|row| DB::document_row(&row, "payload")) @@ -1442,12 +1442,19 @@ where Ok(None) } - async fn apply_task_signal( + /// `recordAppliedSignalForSignalResume`: a signal-resumed task applies + /// its signal once. A signal already applied (by an earlier prefix + /// checkpoint on the same claim) is left as it is, as Core leaves it. + /// For a `checkpoint`, a signal Core would not apply (its command + /// unavailable, its wait not open or another wait open) is skipped as + /// Core skips it; a completion keeps native's refusals for those. + pub(super) async fn apply_task_signal( tx: &mut Transaction<'_, DB>, run_id: &str, task_id: &str, task: &DB::Row, transition_codec: Arc, + checkpoint: bool, ) -> Result<()> { let payload = DB::document_row(task, "payload")?; if payload["resume_source_kind"] != "workflow_signal" { @@ -1455,11 +1462,15 @@ where } let signal_id = text(&payload, "workflow_signal_id")?; // The signal's own command, never one of another workflow (Server #765). - let signal = Self::query("SELECT s.*,c.message_sequence FROM workflow_signal_records s JOIN workflow_commands c ON c.id=s.workflow_command_id AND c.workflow_instance_id=s.workflow_instance_id WHERE s.id=$1 AND s.workflow_run_id=$2") + let Some(signal) = Self::query("SELECT s.*,c.message_sequence FROM workflow_signal_records s JOIN workflow_commands c ON c.id=s.workflow_command_id AND c.workflow_instance_id=s.workflow_instance_id WHERE s.id=$1 AND s.workflow_run_id=$2") .bind(signal_id).bind(run_id).fetch_optional(&mut **tx).await? - .ok_or_else(|| refuse(StatusCode::CONFLICT, "signal_command_unavailable"))?; - if DB::string(&signal, "status")? != "received" { - return Err(refuse(StatusCode::CONFLICT, "signal_already_applied")); + else { + return if checkpoint { Ok(()) } else { Err(refuse(StatusCode::CONFLICT, "signal_command_unavailable")) }; + }; + match DB::string(&signal, "status")?.as_str() { + "received" => {} + "applied" => return Ok(()), + _ => return Err(refuse(StatusCode::CONFLICT, "signal_already_applied")), } let name = DB::string(&signal, "signal_name")?; let wait_id = DB::string(&signal, "signal_wait_id")?; @@ -1474,6 +1485,9 @@ where .await? .is_some_and(|names| names.contains(&name)); if wait.is_none() && !declared { + if checkpoint { + return Ok(()); + } let reason = if open.is_some() { "signal_wait_mismatch" } else { "signal_wait_not_open" }; return Err(refuse(StatusCode::CONFLICT, reason)); } @@ -2607,7 +2621,7 @@ where } // `acknowledgeWorkflowTaskWaitingForHistory` completes a signal task // through the bridge, which records its signal as applied. - Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec).await?; + Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false).await?; payload["waiting_for_history_acknowledged"] = json!(true); tx.summary_changed(&run_id); payload["waiting_for_history_message"] = json!(message); @@ -2637,7 +2651,11 @@ where ) } - pub(crate) async fn complete_workflow( + /// A workflow task completion, or with `checkpoint` a cancellation scope + /// prefix checkpoint: the same validation and Server guards, then Core's + /// checkpoint instead of the completion (`mutateWorkflowTaskCommands`). + #[allow(clippy::too_many_arguments)] + pub(crate) async fn complete_or_checkpoint( &self, task_id: &str, body: Value, @@ -2646,6 +2664,7 @@ where worker: Option, protocol: Option<&str>, limits: super::memo_limits::StructuralLimits, + checkpoint: Option<&super::scope_checkpoint::Checkpoint>, ) -> Result { let transition_codec_for_streams = transition_codec.clone(); let commands = body["commands"] @@ -2805,7 +2824,8 @@ where // A completion that differs from the recorded one, or follows a // waiting acknowledgement, comes from a lost claim: the guard refuses // it with the fields a worker uses to recognize that. - let replay = match Self::duplicate_receipt(&task, &body) { + // A checkpoint has no completion receipt to replay. + let replay = checkpoint.is_none() && match Self::duplicate_receipt(&task, &body) { Err(super::RuntimeError::Refused { reason: "conflicting_completion" | "task_not_leased", .. @@ -2881,6 +2901,11 @@ where local = super::local_activities::normalize(commands, protocol)?; exceptions = super::local_activities::exceptions(&local, transition_codec.clone()).await?; } + if let Some(checkpoint) = checkpoint { + return self + .checkpoint_prefix(tx, task, &body, commands, checkpoint, transition_codec, &external, &local, &exceptions, limits) + .await; + } // `parseCommands`: at most one terminal command, anywhere in the // batch. It applies after every other command. if commands.iter().filter(|command| is_terminal(command)).count() > 1 { @@ -2966,9 +2991,20 @@ where let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?; // The tasks this batch's commands create, in order, as PHP reports them. let mut created = Vec::new(); - Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec).await?; + Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false).await?; // Published parseCommands applies every other command first, whatever // the terminal command's position in the batch. + let apply = Apply { + run: &run, + task: &task, + task_id, + run_id: &run_id, + attempt: &body["workflow_task_attempt"], + limits: &limits, + external: &external, + local: &local, + exceptions: &exceptions, + }; let order = (0..commands.len()) .filter(|index| !is_terminal(&commands[*index])) .chain((0..commands.len()).filter(|index| is_terminal(&commands[*index]))); @@ -2977,11 +3013,6 @@ where if command["type"] == "fail_workflow" { // Applied after the loop. continue; - } else if matches!( - command["type"].as_str(), - Some("complete_update" | "fail_update") - ) { - Self::apply_update(&mut tx, &run, &task, command, sequence + 1).await?; } else if command["type"] == "continue_as_new" { sequence += 1; if Self::cancellation_request(&run)?.is_none() @@ -2991,275 +3022,20 @@ where } else { created.push(self.continue_run(&mut tx, &run, &task, command, sequence).await?); } - } else if command["type"] == "start_child_workflow" { - Self::guard_pending(&mut tx, &run_id, super::memo_limits::Pending::Children, &limits, task_id, &body["workflow_task_attempt"]).await?; - sequence += 1; - created.push(Self::start_child(&mut tx, &run, task_id, command, sequence).await?); - } else if command["type"] == super::local_activities::COMMAND { - // `applyRecordLocalActivity`: one authored sequence. - sequence += 1; - let record = local.get(&index).expect("a normalized local activity record"); - Self::record_local_activity(&mut tx, &run, &task, record, exceptions.get(&index).map(String::as_str), sequence).await?; - } else if command["type"] == "record_side_effect" { - // `applyRecordSideEffect`: one authored sequence and its result. - sequence += 1; - let result = side_effect_result(command)?; - let payload = json!({"sequence": sequence, "result": wire(&result), "task": Self::task_snapshot(&task)?}); - Self::append(&mut tx, &run_id, "SideEffectRecorded", payload, Some(task_id), None).await?; - } else if command["type"] == "record_version_marker" { - sequence += 1; - let mut payload = version_marker_payload(command)?; - payload["sequence"] = json!(sequence); - payload["task"] = Self::task_snapshot(&task)?; - Self::append( - &mut tx, - &run_id, - "VersionMarkerRecorded", - payload, - Some(task_id), - None, - ) - .await?; - } else if command["type"] == "upsert_memo" { - sequence += 1; - self.upsert_memo(&mut tx, &run, &task, command, sequence) - .await?; - } else if command["type"] == "upsert_search_attributes" { - sequence += 1; - self.upsert_search_attributes(&mut tx, &run, &task, command, index, sequence) - .await?; - } else if command["type"] == "schedule_activity" { - Self::guard_pending(&mut tx, &run_id, super::memo_limits::Pending::Activities, &limits, task_id, &body["workflow_task_attempt"]).await?; - sequence += 1; - let metadata = super::selections::metadata(command, "activity")?; - super::selections::check_sequence(&metadata, sequence)?; - let activity_id = id(); - // A session-bound activity without its own queue or - // connection takes the session's (`applyWorkerSessionRouting`). - let session = super::worker_sessions::command_session(command); - let queue = match command.get("queue").filter(|queue| !queue.is_null()) { - Some(_) => text(command, "queue")?.to_owned(), - None => session.as_ref().and_then(|session| session.queue.clone()).map_or_else(|| DB::string(&run, "queue"), Ok)?, - }; - let connection = command["connection"].as_str().map(str::trim).filter(|connection| !connection.is_empty()).map(str::to_owned) - .or_else(|| session.as_ref().and_then(|session| session.connection.clone())); - let activity_options = session.as_ref().map(|session| json!({"worker_session": session.snapshot()})); - let activity_type = text(command, "activity_type")?; - let arguments = match external.get(&index) { - Some(prepared) => { - self.claim_external_in(&mut tx, &prepared.reference, &prepared.uri) - .await?; - prepared.stored() - } - None => envelope(command, "arguments")?, - }; - let policy = super::activity_failures::retry_policy(command)?; - let scheduled_at = now(); - let schedule_deadline = super::activity_deadlines::deadline( - policy.as_ref().unwrap_or(&Value::Null), - "schedule_to_start_timeout", - scheduled_at, - )?; - let lifecycle_deadline = super::activity_deadlines::deadline( - policy.as_ref().unwrap_or(&Value::Null), - "schedule_to_close_timeout", - scheduled_at, - )?; - // PHP keeps an activity's scope in its options, after the - // worker session; its snapshot and the scope's members read it - // from there. - let mut activity_options = activity_options; - if let Some(scope) = command.get("cancellation_scope_id") { - activity_options.get_or_insert_with(|| json!({}))["cancellation_scope_id"] = scope.clone(); - } - // The preserved SQL schema defaults this counter to one. - // Published authored scheduling explicitly starts at zero; - // only the original worker claim consumes its first attempt. - Self::query("INSERT INTO activity_executions(id,workflow_run_id,sequence,activity_type,activity_class,status,attempt_count,payload_codec,arguments,queue,retry_policy,parallel_group_path,schedule_deadline_at,schedule_to_close_deadline_at,connection,activity_options) VALUES ($1,$2,$3,$4,$5,'pending',0,'avro',$6,$7,$8,$9,$10,$11,$12,$13)") - .bind(&activity_id).bind(&run_id).bind(sequence).bind(activity_type).bind(activity_type) - .bind(&arguments).bind(&queue).bind(policy.as_ref().map(DB::document)) - .bind(DB::optional_document(&metadata["parallel_group_path"])) - .bind(DB::optional_time(schedule_deadline)).bind(DB::optional_time(lifecycle_deadline)) - .bind(connection.as_deref()).bind(activity_options.as_ref().map(DB::document)).execute(&mut *tx).await?; - let activity = Self::query("SELECT * FROM activity_executions WHERE id=$1") - .bind(&activity_id) - .fetch_one(&mut *tx) - .await?; - let mut payload = Self::activity_payload(&activity)?; - payload["task"] = Self::task_snapshot(&task)?; - let recorded_at = now(); - Self::append_at( - &mut tx, - &run_id, - "ActivityScheduled", - payload, - Some(task_id), - None, - Some(recorded_at), - ) - .await?; - // PHP publishes only timer tasks a completion creates, so this - // activity task waits for a poller undispatched. - let activity_task = Self::create_task_dispatched( - &mut tx, - &run_id, - &queue, - "activity", - json!({"activity_execution_id": activity_id}), - false, - ) - .await?; - created.push(activity_task.clone()); - Self::create_activity_wait( - &mut tx, - &run_id, - &activity_id, - &activity_task, - recorded_at, - ) - .await?; - Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") - .bind(&run_id) - .execute(&mut *tx) - .await?; - } else if command["type"] == "start_timer" { - Self::guard_pending(&mut tx, &run_id, super::memo_limits::Pending::Timers, &limits, task_id, &body["workflow_task_attempt"]).await?; - sequence += 1; - let metadata = super::selections::metadata(command, "timer")?; - super::selections::check_sequence(&metadata, sequence)?; - let timer_id = id(); - let delay = command["delay_seconds"].as_i64().unwrap(); - let fire_at = timer_deadline(command)?; - let scheduled_at = now(); - Self::query("INSERT INTO workflow_run_timers(id,workflow_run_id,sequence,status,delay_seconds,fire_at) VALUES ($1,$2,$3,'pending',$4,$5)") - .bind(&timer_id).bind(&run_id).bind(sequence).bind(delay).bind(DB::bind_time(fire_at)).execute(&mut *tx).await?; - // PHP records the fire time with `toJSON` (microseconds); a - // scoped timer's descriptor reads it back exactly. - let mut timer_payload = json!({"timer_id":timer_id,"sequence":sequence,"delay_seconds":delay, - "fire_at":super::cancellation_scopes::iso(fire_at),"task":Self::task_snapshot(&task)?}); - if let Some(scope) = command.get("cancellation_scope_id") { - timer_payload["cancellation_scope_id"] = scope.clone(); - } - super::selections::merge(&mut timer_payload, &metadata); - Self::append_at( - &mut tx, - &run_id, - "TimerScheduled", - timer_payload, - Some(task_id), - None, - Some(scheduled_at), - ) - .await?; - let mut timer_task_payload = json!({"timer_id":timer_id}); - if let Some(scope) = command.get("cancellation_scope_id") { - timer_task_payload["cancellation_scope_id"] = scope.clone(); - } - super::selections::merge(&mut timer_task_payload, &metadata); - let timer_task = Self::create_task( - &mut tx, - &run_id, - &DB::string(&run, "queue")?, - "timer", - timer_task_payload, - ) - .await?; - created.push(timer_task.clone()); - Self::query("UPDATE workflow_tasks SET available_at=$1 WHERE id=$2") - .bind(DB::bind_time(fire_at)) - .bind(&timer_task) - .execute(&mut *tx) - .await?; - Self::create_timer_wait( - &mut tx, - &run_id, - &timer_id, - &timer_task, - sequence, - scheduled_at, - fire_at, - ) - .await?; - Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") - .bind(&run_id) - .execute(&mut *tx) - .await?; - } else if command["type"] == "open_condition_wait" - || command["type"] == "open_signal_wait" - { - sequence += 1; - if let Some(wait) = Self::open_wait(&mut tx, &run_id).await? - && !Self::condition_timeout_fired(&mut tx, &run_id, &wait).await? - { - return Err(refuse(StatusCode::CONFLICT, "pending_durable_operations")); - } - let direct = command["type"] == "open_signal_wait"; - let mut payload = command.clone(); - payload.as_object_mut().unwrap().remove("type"); - // Match PHP's optional occurrence payload: retain authored - // nonempty strings and omit null/empty legacy values. - if !direct - && payload - .get("condition_wait_occurrence_id") - .is_some_and(|value| value.is_null() || value.as_str() == Some("")) - { - payload - .as_object_mut() - .unwrap() - .remove("condition_wait_occurrence_id"); - } - // `normalizeOptionalString`: PHP trims the condition's key and - // definition fingerprint and drops them when empty. - if !direct { - for field in ["condition_key", "condition_definition_fingerprint"] { - match payload[field].as_str().map(str::trim).filter(|value| !value.is_empty()) { - Some(value) => payload[field] = json!(value), - None => { - payload.as_object_mut().unwrap().remove(field); - } - } - } - } - let pending_wait_id = if direct { - let pending = Self::query("SELECT signal_wait_id FROM workflow_signal_records WHERE workflow_run_id=$1 AND signal_name=$2 AND status='received' ORDER BY command_sequence,id LIMIT 1") - .bind(&run_id).bind(text(command, "signal_name")?).fetch_optional(&mut *tx).await?; - pending - .as_ref() - .map(|row| DB::string(row, "signal_wait_id")) - .transpose()? - } else { - None - }; - let wait_id = pending_wait_id.clone().unwrap_or_else(id); - if payload["timeout_seconds"].is_null() { - payload.as_object_mut().unwrap().remove("timeout_seconds"); - } - payload[if direct { - "signal_wait_id" - } else { - "condition_wait_id" - }] = json!(wait_id); - payload["sequence"] = json!(sequence); - Self::append( - &mut tx, - &run_id, - if direct { - "SignalWaitOpened" - } else { - "ConditionWaitOpened" - }, - payload.clone(), - Some(task_id), - None, - ) - .await?; - if !direct || pending_wait_id.is_none() { - Self::schedule_wait_timeout(&mut tx, &run, &payload, task_id).await?; + } else if command["type"] != "complete_workflow" { + self.apply_non_terminal(&mut tx, &apply, index, command, &mut sequence, &mut created).await?; + // The completion leaves the run waiting on what it scheduled + // or opened. A prefix checkpoint changes no run status: + // Core's markRunWaiting runs only at completion. + if matches!( + command["type"].as_str(), + Some("schedule_activity" | "start_timer" | "open_condition_wait" | "open_signal_wait") + ) { + Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") + .bind(&run_id) + .execute(&mut *tx) + .await?; } - Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") - .bind(&run_id) - .execute(&mut *tx) - .await?; } else { if Self::cancellation_request(&run)?.is_some() { Self::finish_root_cancellation(&mut tx, &run, Some(&task)).await?; @@ -3356,6 +3132,295 @@ where ) } + /// One non-terminal command of a workflow task completion or of a + /// cancellation scope prefix checkpoint (`applyNonTerminalCommand`), at + /// `sequence`, adding the tasks it creates to `created`. + #[allow(clippy::too_many_arguments)] + pub(super) async fn apply_non_terminal( + &self, + tx: &mut Transaction<'_, DB>, + apply: &Apply<'_, DB::Row>, + index: usize, + command: &Value, + sequence_ref: &mut i64, + created: &mut Vec, + ) -> Result<()> { + let Apply { run, task, task_id, run_id, attempt, limits, external, local, exceptions } = *apply; + let mut sequence = *sequence_ref; + if matches!( + command["type"].as_str(), + Some("complete_update" | "fail_update") + ) { + // `applyCompleteUpdate` and `applyFailUpdate` record the update at + // the next sequence and return the one after it, so a later + // command in the batch takes the following sequence. History + // does not count update events toward the next turn's sequence. + sequence += 1; + Self::apply_update(tx, &run, &task, command, sequence).await?; + } else if command["type"] == "start_child_workflow" { + Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Children, &limits, task_id, attempt).await?; + sequence += 1; + created.push(Self::start_child(tx, &run, task_id, command, sequence).await?); + } else if command["type"] == super::local_activities::COMMAND { + // `applyRecordLocalActivity`: one authored sequence. + sequence += 1; + let record = local.get(&index).expect("a normalized local activity record"); + Self::record_local_activity(tx, &run, &task, record, exceptions.get(&index).map(String::as_str), sequence).await?; + } else if command["type"] == "record_side_effect" { + // `applyRecordSideEffect`: one authored sequence and its result. + sequence += 1; + let result = side_effect_result(command)?; + let payload = json!({"sequence": sequence, "result": wire(&result), "task": Self::task_snapshot(&task)?}); + Self::append(tx, &run_id, "SideEffectRecorded", payload, Some(task_id), None).await?; + } else if command["type"] == "record_version_marker" { + sequence += 1; + let mut payload = version_marker_payload(command)?; + payload["sequence"] = json!(sequence); + payload["task"] = Self::task_snapshot(&task)?; + Self::append( + tx, + &run_id, + "VersionMarkerRecorded", + payload, + Some(task_id), + None, + ) + .await?; + } else if command["type"] == "upsert_memo" { + sequence += 1; + self.upsert_memo(tx, &run, &task, command, sequence) + .await?; + } else if command["type"] == "upsert_search_attributes" { + sequence += 1; + self.upsert_search_attributes(tx, &run, &task, command, index, sequence) + .await?; + } else if command["type"] == "schedule_activity" { + Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Activities, &limits, task_id, attempt).await?; + sequence += 1; + let metadata = super::selections::metadata(command, "activity")?; + super::selections::check_sequence(&metadata, sequence)?; + let activity_id = id(); + // A session-bound activity without its own queue or + // connection takes the session's (`applyWorkerSessionRouting`). + let session = super::worker_sessions::command_session(command); + let queue = match command.get("queue").filter(|queue| !queue.is_null()) { + Some(_) => text(command, "queue")?.to_owned(), + None => session.as_ref().and_then(|session| session.queue.clone()).map_or_else(|| DB::string(&run, "queue"), Ok)?, + }; + let connection = command["connection"].as_str().map(str::trim).filter(|connection| !connection.is_empty()).map(str::to_owned) + .or_else(|| session.as_ref().and_then(|session| session.connection.clone())); + let activity_options = session.as_ref().map(|session| json!({"worker_session": session.snapshot()})); + let activity_type = text(command, "activity_type")?; + let arguments = match external.get(&index) { + Some(prepared) => { + self.claim_external_in(tx, &prepared.reference, &prepared.uri) + .await?; + prepared.stored() + } + None => envelope(command, "arguments")?, + }; + let policy = super::activity_failures::retry_policy(command)?; + let scheduled_at = now(); + let schedule_deadline = super::activity_deadlines::deadline( + policy.as_ref().unwrap_or(&Value::Null), + "schedule_to_start_timeout", + scheduled_at, + )?; + let lifecycle_deadline = super::activity_deadlines::deadline( + policy.as_ref().unwrap_or(&Value::Null), + "schedule_to_close_timeout", + scheduled_at, + )?; + // PHP keeps an activity's scope in its options, after the + // worker session; its snapshot and the scope's members read it + // from there. + let mut activity_options = activity_options; + if let Some(scope) = command.get("cancellation_scope_id") { + activity_options.get_or_insert_with(|| json!({}))["cancellation_scope_id"] = scope.clone(); + } + // The preserved SQL schema defaults this counter to one. + // Published authored scheduling explicitly starts at zero; + // only the original worker claim consumes its first attempt. + Self::query("INSERT INTO activity_executions(id,workflow_run_id,sequence,activity_type,activity_class,status,attempt_count,payload_codec,arguments,queue,retry_policy,parallel_group_path,schedule_deadline_at,schedule_to_close_deadline_at,connection,activity_options) VALUES ($1,$2,$3,$4,$5,'pending',0,'avro',$6,$7,$8,$9,$10,$11,$12,$13)") + .bind(&activity_id).bind(&run_id).bind(sequence).bind(activity_type).bind(activity_type) + .bind(&arguments).bind(&queue).bind(policy.as_ref().map(DB::document)) + .bind(DB::optional_document(&metadata["parallel_group_path"])) + .bind(DB::optional_time(schedule_deadline)).bind(DB::optional_time(lifecycle_deadline)) + .bind(connection.as_deref()).bind(activity_options.as_ref().map(DB::document)).execute(&mut **tx).await?; + let activity = Self::query("SELECT * FROM activity_executions WHERE id=$1") + .bind(&activity_id) + .fetch_one(&mut **tx) + .await?; + let mut payload = Self::activity_payload(&activity)?; + payload["task"] = Self::task_snapshot(&task)?; + let recorded_at = now(); + Self::append_at( + tx, + &run_id, + "ActivityScheduled", + payload, + Some(task_id), + None, + Some(recorded_at), + ) + .await?; + // PHP publishes only timer tasks a completion creates, so this + // activity task waits for a poller undispatched. + let activity_task = Self::create_task_dispatched( + tx, + &run_id, + &queue, + "activity", + json!({"activity_execution_id": activity_id}), + false, + ) + .await?; + created.push(activity_task.clone()); + Self::create_activity_wait( + tx, + &run_id, + &activity_id, + &activity_task, + recorded_at, + ) + .await?; + } else if command["type"] == "start_timer" { + Self::guard_pending(tx, &run_id, super::memo_limits::Pending::Timers, &limits, task_id, attempt).await?; + sequence += 1; + let metadata = super::selections::metadata(command, "timer")?; + super::selections::check_sequence(&metadata, sequence)?; + let timer_id = id(); + let delay = command["delay_seconds"].as_i64().unwrap(); + let fire_at = timer_deadline(command)?; + let scheduled_at = now(); + Self::query("INSERT INTO workflow_run_timers(id,workflow_run_id,sequence,status,delay_seconds,fire_at) VALUES ($1,$2,$3,'pending',$4,$5)") + .bind(&timer_id).bind(&run_id).bind(sequence).bind(delay).bind(DB::bind_time(fire_at)).execute(&mut **tx).await?; + // PHP records the fire time with `toJSON` (microseconds); a + // scoped timer's descriptor reads it back exactly. + let mut timer_payload = json!({"timer_id":timer_id,"sequence":sequence,"delay_seconds":delay, + "fire_at":super::cancellation_scopes::iso(fire_at),"task":Self::task_snapshot(&task)?}); + if let Some(scope) = command.get("cancellation_scope_id") { + timer_payload["cancellation_scope_id"] = scope.clone(); + } + super::selections::merge(&mut timer_payload, &metadata); + Self::append_at( + tx, + &run_id, + "TimerScheduled", + timer_payload, + Some(task_id), + None, + Some(scheduled_at), + ) + .await?; + let mut timer_task_payload = json!({"timer_id":timer_id}); + if let Some(scope) = command.get("cancellation_scope_id") { + timer_task_payload["cancellation_scope_id"] = scope.clone(); + } + super::selections::merge(&mut timer_task_payload, &metadata); + let timer_task = Self::create_task( + tx, + &run_id, + &DB::string(&run, "queue")?, + "timer", + timer_task_payload, + ) + .await?; + created.push(timer_task.clone()); + Self::query("UPDATE workflow_tasks SET available_at=$1 WHERE id=$2") + .bind(DB::bind_time(fire_at)) + .bind(&timer_task) + .execute(&mut **tx) + .await?; + Self::create_timer_wait( + tx, + &run_id, + &timer_id, + &timer_task, + sequence, + scheduled_at, + fire_at, + ) + .await?; + } else if command["type"] == "open_condition_wait" + || command["type"] == "open_signal_wait" + { + sequence += 1; + if let Some(wait) = Self::open_wait(tx, &run_id).await? + && !Self::condition_timeout_fired(tx, &run_id, &wait).await? + { + return Err(refuse(StatusCode::CONFLICT, "pending_durable_operations")); + } + let direct = command["type"] == "open_signal_wait"; + let mut payload = command.clone(); + payload.as_object_mut().unwrap().remove("type"); + // Match PHP's optional occurrence payload: retain authored + // nonempty strings and omit null/empty legacy values. + if !direct + && payload + .get("condition_wait_occurrence_id") + .is_some_and(|value| value.is_null() || value.as_str() == Some("")) + { + payload + .as_object_mut() + .unwrap() + .remove("condition_wait_occurrence_id"); + } + // `normalizeOptionalString`: PHP trims the condition's key and + // definition fingerprint and drops them when empty. + if !direct { + for field in ["condition_key", "condition_definition_fingerprint"] { + match payload[field].as_str().map(str::trim).filter(|value| !value.is_empty()) { + Some(value) => payload[field] = json!(value), + None => { + payload.as_object_mut().unwrap().remove(field); + } + } + } + } + let pending_wait_id = if direct { + let pending = Self::query("SELECT signal_wait_id FROM workflow_signal_records WHERE workflow_run_id=$1 AND signal_name=$2 AND status='received' ORDER BY command_sequence,id LIMIT 1") + .bind(&run_id).bind(text(command, "signal_name")?).fetch_optional(&mut **tx).await?; + pending + .as_ref() + .map(|row| DB::string(row, "signal_wait_id")) + .transpose()? + } else { + None + }; + let wait_id = pending_wait_id.clone().unwrap_or_else(id); + if payload["timeout_seconds"].is_null() { + payload.as_object_mut().unwrap().remove("timeout_seconds"); + } + payload[if direct { + "signal_wait_id" + } else { + "condition_wait_id" + }] = json!(wait_id); + payload["sequence"] = json!(sequence); + Self::append( + tx, + &run_id, + if direct { + "SignalWaitOpened" + } else { + "ConditionWaitOpened" + }, + payload.clone(), + Some(task_id), + None, + ) + .await?; + if !direct || pending_wait_id.is_none() { + Self::schedule_wait_timeout(tx, &run, &payload, task_id).await?; + } + } else { + return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "unsupported_workflow_command")); + } + *sequence_ref = sequence; + Ok(()) + } + pub(super) async fn upsert_memo( &self, tx: &mut Transaction<'_, DB>, @@ -4630,6 +4695,28 @@ pub(super) fn validate_arguments( Ok(()) } +/// What a workflow task's non-terminal commands are applied with: the run +/// and task they belong to, the request's attempt, the structural limits, +/// and the batch's prepared external payloads and local activity records. +pub(super) struct Apply<'a, R> { + pub(super) run: &'a R, + pub(super) task: &'a R, + pub(super) task_id: &'a str, + pub(super) run_id: &'a str, + pub(super) attempt: &'a Value, + pub(super) limits: &'a super::memo_limits::StructuralLimits, + pub(super) external: &'a super::external_resolution::PreparedCommands, + pub(super) local: &'a std::collections::BTreeMap, + pub(super) exceptions: &'a std::collections::BTreeMap, +} + +impl Clone for Apply<'_, R> { + fn clone(&self) -> Self { + *self + } +} +impl Copy for Apply<'_, R> {} + /// A workflow task completion's stored receipt: its request and the tasks it /// created, so a repeat answers as the original did. Older receipts, and /// activity receipts, are the bare request. diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 2b97f2278..05d85c375 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -101,6 +101,8 @@ mod payload_discovery; mod signal_rejection; #[path = "execution/cooperative_waits.rs"] mod cooperative_waits; +#[path = "execution/scope_checkpoints.rs"] +mod scope_checkpoints; #[path = "execution/child_policies.rs"] mod child_policies; #[path = "execution/definition_routing.rs"] diff --git a/rust/tests/execution/scope_checkpoints.rs b/rust/tests/execution/scope_checkpoints.rs new file mode 100644 index 000000000..b126a28a9 --- /dev/null +++ b/rust/tests/execution/scope_checkpoints.rs @@ -0,0 +1,307 @@ +use super::*; + +// Server's POST …/cancellation-scopes/checkpoint over Core's +// checkpointCancellationScopePrefix: a cooperative worker commits the +// commands it recorded earlier in its turn, under the same claim, before +// opening a cancellation scope. + +async fn register(app: &Router, worker: &str, capabilities: Value) { + let registered = request_protocol(app, "POST", "/api/worker/register", + json!({"worker_id":worker,"task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[],"capabilities":capabilities}), "1.20").await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); +} + +async fn claim(app: &Router, worker: &str) -> Value { + let response = request_protocol(app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":worker,"task_queue":"test","timeout_seconds":2}), "1.20").await; + assert_eq!(response.0, StatusCode::OK, "{}", response.1); + response.1["task"].clone() +} + +async fn checkpoint(app: &Router, task: &Value, fields: Value, protocol: &str) -> (StatusCode, Value) { + let mut body = json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"]}); + for (key, value) in fields.as_object().unwrap() { + body[key] = value.clone(); + } + request_protocol(app, "POST", &format!("/api/worker/workflow-tasks/{}/cancellation-scopes/checkpoint", + task["task_id"].as_str().unwrap()), body, protocol).await +} + +fn marker(change: &str) -> Value { + json!({"type":"record_version_marker","change_id":change,"version":1,"min_supported":1,"max_supported":1}) +} + +fn prefix() -> Value { + json!([{"type":"record_side_effect","result":envelope(Payload::Long(7))}, marker("before-scope")]) +} + +fn history_types(history: &Value) -> Vec { + history.as_array().unwrap().iter().map(|event| event["event_type"].as_str().unwrap().to_owned()).collect() +} + +#[tokio::test] +async fn a_scope_prefix_commits_under_the_claim_and_retries_from_its_receipt() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "scope-worker", json!(["cooperative_cancellation"])).await; + let run = start(&app, "scope-prefix").await; + let task = claim(&app, "scope-worker").await; + + let (status, receipt) = checkpoint(&app, &task, json!({"checkpoint_id":"prefix-1","start_sequence":1,"commands":prefix()}), "1.20").await; + assert_eq!(status, StatusCode::OK, "{receipt}"); + assert_eq!((receipt["checkpointed"].clone(), receipt["duplicate"].clone(), receipt["reason"].clone()), + (json!(true), json!(false), Value::Null), "{receipt}"); + assert_eq!((receipt["checkpoint_id"].clone(), receipt["task_id"].clone(), receipt["workflow_run_id"].clone(), + receipt["workflow_task_attempt"].clone(), receipt["lease_owner"].clone(), receipt["start_sequence"].clone(), + receipt["next_sequence"].clone(), receipt["created_task_ids"].clone()), + (json!("prefix-1"), task["task_id"].clone(), run["run_id"].clone(), task["workflow_task_attempt"].clone(), + task["lease_owner"].clone(), json!(1), json!(3), json!([]))); + let fingerprint = receipt["fingerprint"].as_str().unwrap(); + assert!(fingerprint.len() == 64 && fingerprint.bytes().all(|byte| byte.is_ascii_hexdigit()), "{receipt}"); + assert!(receipt["lease_expires_at"].is_string() && receipt["recorded_at"].is_string(), "{receipt}"); + assert_eq!(receipt["history_refresh_page_token"], "MA=="); + assert!(receipt["protocol_version"].is_string() && receipt["server_capabilities"].is_object(), "{receipt}"); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let recorded = history_types(&history); + assert!(recorded.ends_with(&["SideEffectRecorded".to_owned(), "VersionMarkerRecorded".to_owned()]), "{recorded:?}"); + let events = recorded.len(); + + // An identical retry answers the retained receipt and writes nothing. + let (status, again) = checkpoint(&app, &task, json!({"checkpoint_id":"prefix-1","start_sequence":1,"commands":prefix()}), "1.20").await; + assert_eq!((status, again["duplicate"].clone()), (StatusCode::OK, json!(true)), "{again}"); + for field in ["fingerprint", "next_sequence", "recorded_at", "lease_expires_at", "created_task_ids"] { + assert_eq!(again[field], receipt[field], "{field}"); + } + // The same id with other commands, or a new id at a used sequence. + let (status, mismatch) = checkpoint(&app, &task, + json!({"checkpoint_id":"prefix-1","start_sequence":1,"commands":[marker("other")]}), "1.20").await; + assert_eq!((status, mismatch["reason"].clone()), (StatusCode::CONFLICT, json!("cancellation_scope_checkpoint_mismatch"))); + assert_eq!((mismatch["checkpointed"].clone(), mismatch["duplicate"].clone(), mismatch["task_id"].clone()), + (json!(false), json!(false), task["task_id"].clone()), "{mismatch}"); + let (status, stale) = checkpoint(&app, &task, + json!({"checkpoint_id":"prefix-2","start_sequence":1,"commands":[marker("late")]}), "1.20").await; + assert_eq!((status, stale["reason"].clone()), (StatusCode::CONFLICT, json!("cancellation_scope_checkpoint_sequence_mismatch"))); + assert_eq!(history_types(&run_history(&app, &run["workflow_id"], &run["run_id"]).await).len(), events); + + // The claim is unchanged: the worker opens its scope at the next + // sequence and completes the turn. + let (status, opened) = request_protocol(&app, "POST", + &format!("/api/worker/workflow-tasks/{}/cancellation-scopes/open", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"],"sequence":3}), "1.20").await; + assert_eq!((status, opened["opened"].clone()), (StatusCode::OK, json!(true)), "{opened}"); + let finished = finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await; + assert_eq!(finished.0, StatusCode::OK, "{}", finished.1); + assert_eq!(finished.1["run_status"], "completed", "{}", finished.1); +} + +#[tokio::test] +async fn a_prefix_schedules_operations_and_refuses_what_closes_a_turn() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "scope-worker", json!(["cooperative_cancellation"])).await; + let run = start(&app, "scope-prefix-ops").await; + let task = claim(&app, "scope-worker").await; + for (commands, reason) in [ + (json!([{"type":"open_signal_wait","signal_name":"go"}]), "invalid_cancellation_scope_checkpoint_commands"), + (json!([{"type":"open_condition_wait","condition_key":"ready"}]), "invalid_cancellation_scope_checkpoint_commands"), + (json!([marker("a"), {"type":"complete_workflow","result":envelope(Payload::Null)}]), "invalid_cancellation_scope_checkpoint_commands"), + ] { + let (status, refused) = checkpoint(&app, &task, json!({"checkpoint_id":"ops","start_sequence":1,"commands":commands}), "1.20").await; + assert_eq!((status, refused["reason"].clone()), (StatusCode::CONFLICT, json!(reason)), "{refused}"); + assert_eq!(refused["history_refresh_page_token"], "MA==", "{refused}"); + } + assert!(!history_types(&run_history(&app, &run["workflow_id"], &run["run_id"]).await).contains(&"VersionMarkerRecorded".to_owned())); + + // A timer in a prefix creates its task, as a completion would. + let (status, receipt) = checkpoint(&app, &task, + json!({"checkpoint_id":"ops","start_sequence":1,"commands":[{"type":"start_timer","delay_seconds":3600}]}), "1.20").await; + assert_eq!(status, StatusCode::OK, "{receipt}"); + assert_eq!((receipt["next_sequence"].clone(), receipt["created_task_ids"].as_array().map(Vec::len)), (json!(2), Some(1)), "{receipt}"); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + assert!(history_types(&history).contains(&"TimerScheduled".to_owned())); +} + +#[tokio::test] +async fn the_route_checks_protocol_fields_and_claim_before_core() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "scope-worker", json!(["cooperative_cancellation"])).await; + register(&app, "plain-worker", json!([])).await; + start(&app, "scope-gates").await; + let task = claim(&app, "scope-worker").await; + + let (status, old) = checkpoint(&app, &task, json!({"checkpoint_id":"g","start_sequence":1,"commands":prefix()}), "1.19").await; + assert_eq!((status, old["checkpointed"].clone(), old["reason"].clone()), + (StatusCode::CONFLICT, json!(false), json!("cancellation_scope_checkpoint_requires_protocol_1_20")), "{old}"); + assert!(old["server_capabilities"].is_object(), "{old}"); + + let (status, invalid) = checkpoint(&app, &task, json!({"start_sequence":0,"sticky_cache":{"ttl_seconds":1},"commands":prefix()}), "1.20").await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{invalid}"); + assert_eq!(invalid["errors"], json!({"checkpoint_id":["The checkpoint id field is required."], + "start_sequence":["The start sequence field must be at least 1."], + "sticky_cache":["The sticky cache field is prohibited."]}), "{invalid}"); + + let mut other = task.clone(); + other["lease_owner"] = json!("someone-else"); + let (status, foreign) = checkpoint(&app, &other, json!({"checkpoint_id":"g","start_sequence":1,"commands":prefix()}), "1.20").await; + assert_eq!((status, foreign["reason"].clone()), (StatusCode::CONFLICT, json!("lease_owner_mismatch")), "{foreign}"); + + // A claim issued to a worker without cooperative cancellation. + let finished = finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await; + assert_eq!(finished.0, StatusCode::OK, "{}", finished.1); + start(&app, "scope-gates-plain").await; + let plain = claim(&app, "plain-worker").await; + let (status, refused) = checkpoint(&app, &plain, json!({"checkpoint_id":"g","start_sequence":1,"commands":prefix()}), "1.20").await; + assert_eq!((status, refused.get("task_id").cloned(), refused["reason"].clone()), + (StatusCode::CONFLICT, None, json!("active_claim_cancellation_not_supported")), "{refused}"); + assert!(refused.get("history_refresh_page_token").is_none(), "{refused}"); +} + +#[tokio::test] +async fn a_prefix_waits_for_a_root_cancellation_to_be_delivered() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "scope-worker", json!(["cooperative_cancellation"])).await; + start(&app, "scope-cancelled").await; + let response = request(&app, "POST", "/api/workflows/scope-cancelled/request-cancellation", + json!({"reason":"stop","cleanup_timeout_seconds":30})).await; + assert_eq!(response.0, StatusCode::ACCEPTED, "{}", response.1); + let pending = response.1["cancellation_request"].clone(); + let task = claim(&app, "scope-worker").await; + let (status, refused) = checkpoint(&app, &task, json!({"checkpoint_id":"c","start_sequence":1,"commands":prefix()}), "1.20").await; + assert_eq!((status, refused["reason"].clone()), (StatusCode::CONFLICT, json!("cancellation_requested")), "{refused}"); + + // Delivered at sequence 1, a prefix after the boundary is cleanup. + let delivered = super::cooperative_children::deliver(&app, &task, &pending, 1, "timer").await; + assert_eq!(delivered.1["delivered"], true, "{}", delivered.1); + let (status, early) = checkpoint(&app, &task, json!({"checkpoint_id":"c","start_sequence":1,"commands":prefix()}), "1.20").await; + assert_eq!((status, early["reason"].clone()), (StatusCode::CONFLICT, json!("cancellation_requested")), "{early}"); + let (status, receipt) = checkpoint(&app, &task, json!({"checkpoint_id":"c","start_sequence":2,"commands":prefix()}), "1.20").await; + assert_eq!((status, receipt["next_sequence"].clone()), (StatusCode::OK, json!(4)), "{receipt}"); +} + +/// A cooperative worker that declares the `finish` signal and the `set` +/// update, as `update_worker` does. +async fn contract_worker(app: &Router, worker: &str) { + let registered = request_protocol(app, "POST", "/api/worker/register", json!({ + "worker_id":worker,"task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[],"capabilities":["cooperative_cancellation","workflow_updates"],"workflow_command_contracts":{"echo":{ + "signals":["finish"],"signal_contracts":[{"name":"finish","parameters":[{"name":"value","type":"int","required":true,"variadic":false,"default_available":false,"allows_null":false}]}], + "updates":["set"],"update_validators":[],"update_contracts":[{"name":"set","parameters":[{"name":"value","type":"int","required":true,"variadic":false,"default_available":false,"allows_null":false}]}] + }}}), "1.20").await; + assert_eq!(registered.0, StatusCode::CREATED, "{}", registered.1); +} + +fn events_of<'a>(history: &'a Value, kind: &str) -> Vec<&'a Value> { + history.as_array().unwrap().iter().filter(|event| event["event_type"] == kind).collect() +} + +// Core's recordAppliedSignalForSignalResume applies a resumed task's signal +// once: a prefix checkpoint applies it, and the second checkpoint and the +// completion on the same claim leave it applied. +#[tokio::test] +async fn a_signal_resumed_claim_checkpoints_then_completes() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + contract_worker(&app, "signal-worker").await; + let run = start(&app, "scope-signal").await; + let first = claim(&app, "signal-worker").await; + let parked = finish_task(&app, &first, json!([{"type":"open_signal_wait","signal_name":"finish"}])).await; + assert_eq!(parked.0, StatusCode::OK, "{}", parked.1); + let signalled = request(&app, "POST", "/api/workflows/scope-signal/signal/finish", + json!({"input":envelope(Payload::Array(vec![Payload::Long(1)]))})).await; + assert_eq!(signalled.0, StatusCode::ACCEPTED, "{}", signalled.1); + let task = claim(&app, "signal-worker").await; + assert_eq!(task["resume_source_kind"], "workflow_signal", "{task}"); + + let (status, receipt) = checkpoint(&app, &task, + json!({"checkpoint_id":"after-signal","start_sequence":2,"commands":[marker("after-signal")]}), "1.20").await; + assert_eq!((status, receipt["next_sequence"].clone()), (StatusCode::OK, json!(3)), "{receipt}"); + let (status, second) = checkpoint(&app, &task, + json!({"checkpoint_id":"second","start_sequence":3,"commands":[marker("second")]}), "1.20").await; + assert_eq!((status, second["next_sequence"].clone()), (StatusCode::OK, json!(4)), "{second}"); + let finished = finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await; + assert_eq!((finished.0, finished.1["run_status"].clone()), (StatusCode::OK, json!("completed")), "{}", finished.1); + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + assert_eq!(events_of(&history, "SignalApplied").len(), 1, "{history}"); +} + +// Core's nextDurableCommandSequence counts SearchAttributesUpserted, so a +// prefix that ends in an upsert is followed at its receipt's next sequence. +#[tokio::test] +async fn a_prefix_ending_in_a_search_attribute_upsert_is_followed_at_its_next_sequence() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let defined = request(&app, "POST", "/api/search-attributes", json!({"name":"ScopeCount","type":"int"})).await; + assert_eq!(defined.0, StatusCode::CREATED, "{}", defined.1); + register(&app, "scope-worker", json!(["cooperative_cancellation", "typed_search_attributes"])).await; + start(&app, "scope-upsert").await; + let task = claim(&app, "scope-worker").await; + let (status, receipt) = checkpoint(&app, &task, json!({"checkpoint_id":"upsert","start_sequence":1, + "commands":[{"type":"upsert_search_attributes","attributes":{"ScopeCount":1}}]}), "1.20").await; + assert_eq!((status, receipt["next_sequence"].clone()), (StatusCode::OK, json!(2)), "{receipt}"); + let (status, opened) = request_protocol(&app, "POST", + &format!("/api/worker/workflow-tasks/{}/cancellation-scopes/open", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"],"sequence":2}), "1.20").await; + assert_eq!((status, opened["opened"].clone()), (StatusCode::OK, json!(true)), "{opened}"); +} + +// Core changes a run's status at completion (markRunWaiting), not at a +// prefix checkpoint: a timer in a prefix leaves the run as the claim found it. +#[tokio::test] +async fn a_prefix_leaves_the_run_status_until_completion() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "scope-worker", json!(["cooperative_cancellation"])).await; + start(&app, "scope-status").await; + let task = claim(&app, "scope-worker").await; + let status_of = || async { request(&app, "GET", "/api/workflows/scope-status", Value::Null).await.1["status"].clone() }; + let claimed = status_of().await; + assert_ne!(claimed, "waiting"); + let (status, receipt) = checkpoint(&app, &task, + json!({"checkpoint_id":"timer","start_sequence":1,"commands":[{"type":"start_timer","delay_seconds":3600}]}), "1.20").await; + assert_eq!(status, StatusCode::OK, "{receipt}"); + assert_eq!(status_of().await, claimed); + let finished = finish_task(&app, &task, json!([marker("after-timer")])).await; + assert_eq!(finished.0, StatusCode::OK, "{}", finished.1); + assert_eq!(status_of().await, "waiting"); +} + +// Core's applyCompleteUpdate records the update at the batch's next sequence +// and returns the one after it: a command after the update takes the +// following sequence, in a completion as in a prefix checkpoint. +#[tokio::test] +async fn a_command_after_an_update_completion_takes_the_following_sequence() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + contract_worker(&app, "update-worker").await; + let run = start(&app, "scope-updates").await; + let first = claim(&app, "update-worker").await; + let parked = finish_task(&app, &first, json!([{"type":"open_signal_wait","signal_name":"finish"}])).await; + assert_eq!(parked.0, StatusCode::OK, "{}", parked.1); + for (round, through) in [(1, "complete"), (2, "checkpoint")] { + let accepted = request(&app, "POST", "/api/workflows/scope-updates/update/set", + json!({"input":envelope(Payload::Array(vec![Payload::Long(round)])),"request_id":format!("set-{round}"),"wait_for":"accepted"})).await; + assert_eq!(accepted.0, StatusCode::ACCEPTED, "{}", accepted.1); + let task = claim(&app, "update-worker").await; + assert_eq!(task["workflow_update_id"], accepted.1["update_id"], "{task}"); + let commands = json!([{"type":"complete_update","update_id":accepted.1["update_id"],"result":envelope(Payload::Long(round))}, + marker(&format!("after-update-{round}"))]); + if through == "complete" { + let finished = finish_task(&app, &task, commands).await; + assert_eq!(finished.0, StatusCode::OK, "{}", finished.1); + } else { + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let next = events_of(&history, "VersionMarkerRecorded").iter().filter_map(|event| event["payload"]["sequence"].as_i64()).max().unwrap() + 1; + let (status, receipt) = checkpoint(&app, &task, json!({"checkpoint_id":"update","start_sequence":next,"commands":commands}), "1.20").await; + assert_eq!((status, receipt["next_sequence"].clone()), (StatusCode::OK, json!(next + 2)), "{receipt}"); + } + let history = run_history(&app, &run["workflow_id"], &run["run_id"]).await; + let update = *events_of(&history, "UpdateCompleted").last().unwrap(); + let marker_event = events_of(&history, "VersionMarkerRecorded").into_iter() + .find(|event| event["payload"]["change_id"] == format!("after-update-{round}")).unwrap(); + assert_eq!(marker_event["payload"]["sequence"].as_i64().unwrap(), update["payload"]["sequence"].as_i64().unwrap() + 1, + "{update} {marker_event}"); + } +} diff --git a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt index 7f61e043a..c653238ab 100644 --- a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt +++ b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt @@ -1,4 +1,5 @@ # Published Server 2.11.1 API routes that native answers with 501 # rust_capability_not_implemented, in route-table order. Checked by -# rust/tests/execution/route_coverage.rs. -POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint +# rust/tests/execution/route_coverage.rs. None remain: the last, +# POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint, +# is served since October 27, 2026. diff --git a/scripts/conformance/server-parity/cancellation-scopes-contract.mjs b/scripts/conformance/server-parity/cancellation-scopes-contract.mjs index 1243e0ca8..dc6208631 100644 --- a/scripts/conformance/server-parity/cancellation-scopes-contract.mjs +++ b/scripts/conformance/server-parity/cancellation-scopes-contract.mjs @@ -6,7 +6,13 @@ import assert from 'node:assert/strict'; // impose, recorded against published Server 2.11.1. The probe normalizes the // observation: identities and digests become labels, the run requests' // instants named anchors and other instants their format, and clock-dependent -// relations measured facts. The fixture's `expected` holds the whole +// relations measured facts. `scope_checkpoints`, recorded against published +// Server 2.11.5, adds the prefix checkpoint a cooperative worker sends +// before opening a scope (POST …/cancellation-scopes/checkpoint): its gates +// and validation, Core's refusals, the committed prefix, the retained +// receipt and its retry, the claim kept for the scope and the completion. +// Each receipt's fingerprint is a server's own digest of the prefix, named +// by the probe, so only its reuse is compared. The fixture's `expected` holds the whole // normalized observation, with the probe's workflow prefix as `{id}`. Every // step compares exactly. // diff --git a/scripts/conformance/server-parity/cancellation-scopes-probe.php b/scripts/conformance/server-parity/cancellation-scopes-probe.php index 813d5deef..c9bb64d5f 100644 --- a/scripts/conformance/server-parity/cancellation-scopes-probe.php +++ b/scripts/conformance/server-parity/cancellation-scopes-probe.php @@ -448,6 +448,53 @@ function finishHttpCancellationScopes(Client $admin, array $fixture, string $wor $state['legacy_claim_open'] = $route('open', $w, $own($w) + ['sequence' => 1]); $state['legacy_claim_prepare'] = $route('prepare', $w, $boundary($w, 'NO-SUCH-SCOPE', 'NO-SUCH-REQUEST', 1, 'timer')); + // ---- R5: a cooperative claim checkpoints the commands of its turn + // before opening a scope (POST …/cancellation-scopes/checkpoint). Every + // step sits under one key, which sorts after the others, and every new + // identity is named, so no earlier label changes. ---- + $marker = static fn (string $change): array => ['type' => 'record_version_marker', 'change_id' => $change, 'version' => 1, + 'min_supported' => 1, 'max_supported' => 1]; + $prefix = [['type' => 'record_side_effect', 'result' => $empty], $marker('before-scope')]; + $checkpoint = static fn (array $task, ?array $body, ?string $protocol = '1.20'): array => $route('checkpoint', $task, $body, $protocol); + $ck = []; + // Claims that cannot checkpoint: R3's plain claim and R4's protocol 1.19 claim. + $ck['plain_claim'] = $checkpoint($v, $own($v) + ['checkpoint_id' => 'ck-plain', 'start_sequence' => 1, 'commands' => $prefix]); + $ck['legacy_claim'] = $checkpoint($w, $own($w) + ['checkpoint_id' => 'ck-legacy', 'start_sequence' => 1, 'commands' => $prefix]); + $queue5 = $workflowId.'-r5-queue'; + $worker5 = $workflowId.'-r5-worker'; + $register($worker5, $queue5, ['cooperative_cancellation']); + $r5 = $start('R5', $queue5); + $x = $poll($worker5, $queue5, 'R5.T1'); + $first = $own($x) + ['checkpoint_id' => 'ck-1', 'start_sequence' => 1, 'commands' => $prefix]; + // Gates and validation, in the controller's order, then Core's refusals. + $ck['protocol_1_19'] = $checkpoint($x, $first, '1.19'); + $ck['missing_fields'] = $checkpoint($x, $own($x) + ['commands' => $prefix, 'sticky_cache' => ['ttl_seconds' => 1]]); + $ck['invalid_fields'] = $checkpoint($x, $own($x) + ['checkpoint_id' => ['ck'], 'start_sequence' => 0, 'commands' => $prefix, + 'message_stream_cursors' => [['stream_name' => 's', 'through_position' => 0]]]); + $ck['foreign_owner'] = $checkpoint($x, ['lease_owner' => 'someone-else'] + $first); + $ck['terminal'] = $checkpoint($x, ['commands' => [$marker('terminal'), ['type' => 'complete_workflow', 'result' => $empty]]] + $first); + $ck['signal_wait'] = $checkpoint($x, ['commands' => [['type' => 'open_signal_wait', 'signal_name' => 'go']]] + $first); + // The prefix commits at 1 and 2 under the claim; a retry answers its + // receipt, and the same id with other commands, or a new id at a used + // sequence, is refused. + $ck['prefix'] = $checkpoint($x, $first); + $name($ck['prefix']['response']['fingerprint'] ?? null, 'R5.prefix.fingerprint'); + $ck['prefix_again'] = $checkpoint($x, $first); + $ck['mismatch'] = $checkpoint($x, ['commands' => [$marker('other')]] + $first); + $ck['sequence_behind'] = $checkpoint($x, ['checkpoint_id' => 'ck-2', 'commands' => [$marker('late')]] + $first); + $ck['history_after_prefix'] = $history($r5); + // A second prefix replaces the receipt: a timer at 3, with its task. + $ck['timer'] = $checkpoint($x, ['checkpoint_id' => 'ck-2', 'start_sequence' => 3, 'commands' => [$timer(null)]] + $first); + $name($ck['timer']['response']['fingerprint'] ?? null, 'R5.timer.fingerprint'); + $name($ck['timer']['response']['created_task_ids'][0] ?? null, 'R5.timer.task'); + $ck['first_after_second'] = $checkpoint($x, $first); + // The claim is unchanged: the scope opens at 4 and the turn completes. + $ck['open_after'] = $route('open', $x, $own($x) + ['sequence' => 4]); + $name($ck['open_after']['response']['scope_id'] ?? null, 'R5.S1'); + $ck['complete'] = $complete($x, [['type' => 'complete_workflow', 'result' => $empty]]); + $ck['history'] = $history($r5); + $state['scope_checkpoints'] = $ck; + return cancellationScopesNormalize($state, $names); } diff --git a/tests/Fixtures/ServerParity/cancellation-scopes.json b/tests/Fixtures/ServerParity/cancellation-scopes.json index ff9e5540e..d3677a0a6 100644 --- a/tests/Fixtures/ServerParity/cancellation-scopes.json +++ b/tests/Fixtures/ServerParity/cancellation-scopes.json @@ -7409,6 +7409,341 @@ "task_id": "{R4.T1}" }, "status": 409 + }, + "scope_checkpoints": { + "complete": { + "response": { + "created_task_ids": [], + "outcome": "completed", + "protocol_version": "1.20", + "reason": null, + "recorded": true, + "run_id": "{R5}", + "run_status": "completed", + "server_capabilities": "{present}", + "task_id": "{R5.T1}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "first_after_second": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_sequence_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "foreign_owner": { + "response": { + "error": "Workflow task lease is owned by another worker.", + "lease_owner": "{id}-r5-worker", + "protocol_version": "1.20", + "reason": "lease_owner_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}", + "workflow_task_attempt": 1 + }, + "status": 409 + }, + "history": [ + { + "event_type": "StartAccepted", + "sequence": 1 + }, + { + "event_type": "WorkflowStarted", + "sequence": 2 + }, + { + "event_type": "SideEffectRecorded", + "sequence": 3 + }, + { + "event_type": "VersionMarkerRecorded", + "sequence": 4 + }, + { + "event_type": "TimerScheduled", + "payload": { + "cancellation_scope_id": "{absent}", + "sequence": 3 + }, + "sequence": 5 + }, + { + "event_type": "CancellationScopeOpened", + "payload": { + "parent_scope_id": "root", + "schema": "durable-workflow.cancellation-scope/v1", + "scope_id": "{R5.S1}", + "sequence": 4, + "shield_parent": false, + "task": { + "attempt_count": 1, + "id": "{R5.T1}", + "lease_owner": "{id}-r5-worker", + "status": "leased", + "type": "workflow" + }, + "workflow_run_id": "{R5}" + }, + "sequence": 6 + }, + { + "event_type": "WorkflowCompleted", + "sequence": 7 + } + ], + "history_after_prefix": [ + { + "event_type": "StartAccepted", + "sequence": 1 + }, + { + "event_type": "WorkflowStarted", + "sequence": 2 + }, + { + "event_type": "SideEffectRecorded", + "sequence": 3 + }, + { + "event_type": "VersionMarkerRecorded", + "sequence": 4 + } + ], + "invalid_fields": { + "response": { + "errors": { + "checkpoint_id": [ + "The checkpoint id field must be a string." + ], + "message_stream_cursors": [ + "The message stream cursors field is prohibited." + ], + "start_sequence": [ + "The start sequence field must be at least 1." + ] + }, + "message": "The checkpoint id field must be a string. (and 2 more errors)", + "protocol_version": "1.20", + "reason": "validation_failed", + "server_capabilities": "{present}", + "validation_errors": { + "checkpoint_id": [ + "The checkpoint id field must be a string." + ], + "message_stream_cursors": [ + "The message stream cursors field is prohibited." + ], + "start_sequence": [ + "The start sequence field must be at least 1." + ] + } + }, + "status": 422 + }, + "legacy_claim": { + "response": { + "checkpointed": false, + "protocol_version": "1.20", + "reason": "active_claim_cancellation_not_supported", + "server_capabilities": "{present}" + }, + "status": 409 + }, + "mismatch": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "missing_fields": { + "response": { + "errors": { + "checkpoint_id": [ + "The checkpoint id field is required." + ], + "start_sequence": [ + "The start sequence field is required." + ], + "sticky_cache": [ + "The sticky cache field is prohibited." + ] + }, + "message": "The checkpoint id field is required. (and 2 more errors)", + "protocol_version": "1.20", + "reason": "validation_failed", + "server_capabilities": "{present}", + "validation_errors": { + "checkpoint_id": [ + "The checkpoint id field is required." + ], + "start_sequence": [ + "The start sequence field is required." + ], + "sticky_cache": [ + "The sticky cache field is prohibited." + ] + } + }, + "status": 422 + }, + "open_after": { + "response": { + "claim_released": false, + "created_task_ids": [], + "duplicate": false, + "history_event_id": "{ulid:36}", + "history_refresh_page_token": "MA==", + "lease_owner": "{id}-r5-worker", + "opened": true, + "parent_scope_id": "root", + "protocol_version": "1.20", + "reason": null, + "scope_id": "{R5.S1}", + "sequence": 4, + "server_capabilities": "{present}", + "shield_parent": false, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "plain_claim": { + "response": { + "checkpointed": false, + "protocol_version": "1.20", + "reason": "active_claim_cancellation_not_supported", + "server_capabilities": "{present}" + }, + "status": 409 + }, + "prefix": { + "response": { + "checkpoint_id": "ck-1", + "checkpointed": true, + "created_task_ids": [], + "duplicate": false, + "fingerprint": "{R5.prefix.fingerprint}", + "history_refresh_page_token": "MA==", + "lease_expires_at": "{time:Z6}", + "lease_owner": "{id}-r5-worker", + "next_sequence": 3, + "protocol_version": "1.20", + "reason": null, + "recorded_at": "{time:Z6}", + "server_capabilities": "{present}", + "start_sequence": 1, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "prefix_again": { + "response": { + "checkpoint_id": "ck-1", + "checkpointed": true, + "created_task_ids": [], + "duplicate": true, + "fingerprint": "{R5.prefix.fingerprint}", + "history_refresh_page_token": "MA==", + "lease_expires_at": "{time:Z6}", + "lease_owner": "{id}-r5-worker", + "next_sequence": 3, + "protocol_version": "1.20", + "reason": null, + "recorded_at": "{time:Z6}", + "server_capabilities": "{present}", + "start_sequence": 1, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "protocol_1_19": { + "response": { + "checkpointed": false, + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_requires_protocol_1_20", + "server_capabilities": "{present}" + }, + "status": 409 + }, + "sequence_behind": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_sequence_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "signal_wait": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "invalid_cancellation_scope_checkpoint_commands", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "terminal": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "invalid_cancellation_scope_checkpoint_commands", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "timer": { + "response": { + "checkpoint_id": "ck-2", + "checkpointed": true, + "created_task_ids": [ + "{R5.timer.task}" + ], + "duplicate": false, + "fingerprint": "{R5.timer.fingerprint}", + "history_refresh_page_token": "MA==", + "lease_expires_at": "{time:Z6}", + "lease_owner": "{id}-r5-worker", + "next_sequence": 4, + "protocol_version": "1.20", + "reason": null, + "recorded_at": "{time:Z6}", + "server_capabilities": "{present}", + "start_sequence": 3, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + } } } } diff --git a/tests/Fixtures/ServerParityExpected/cancellation-scopes-observation.json b/tests/Fixtures/ServerParityExpected/cancellation-scopes-observation.json index bbe9adaad..dfcc908dd 100644 --- a/tests/Fixtures/ServerParityExpected/cancellation-scopes-observation.json +++ b/tests/Fixtures/ServerParityExpected/cancellation-scopes-observation.json @@ -7338,5 +7338,340 @@ "workflow_run_id": "{R1}" }, "status": 409 + }, + "scope_checkpoints": { + "complete": { + "response": { + "created_task_ids": [], + "outcome": "completed", + "protocol_version": "1.20", + "reason": null, + "recorded": true, + "run_id": "{R5}", + "run_status": "completed", + "server_capabilities": "{present}", + "task_id": "{R5.T1}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "first_after_second": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_sequence_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "foreign_owner": { + "response": { + "error": "Workflow task lease is owned by another worker.", + "lease_owner": "pr1-cancellation-scopes-r5-worker", + "protocol_version": "1.20", + "reason": "lease_owner_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}", + "workflow_task_attempt": 1 + }, + "status": 409 + }, + "history": [ + { + "event_type": "StartAccepted", + "sequence": 1 + }, + { + "event_type": "WorkflowStarted", + "sequence": 2 + }, + { + "event_type": "SideEffectRecorded", + "sequence": 3 + }, + { + "event_type": "VersionMarkerRecorded", + "sequence": 4 + }, + { + "event_type": "TimerScheduled", + "payload": { + "cancellation_scope_id": "{absent}", + "sequence": 3 + }, + "sequence": 5 + }, + { + "event_type": "CancellationScopeOpened", + "payload": { + "parent_scope_id": "root", + "schema": "durable-workflow.cancellation-scope/v1", + "scope_id": "{R5.S1}", + "sequence": 4, + "shield_parent": false, + "task": { + "attempt_count": 1, + "id": "{R5.T1}", + "lease_owner": "pr1-cancellation-scopes-r5-worker", + "status": "leased", + "type": "workflow" + }, + "workflow_run_id": "{R5}" + }, + "sequence": 6 + }, + { + "event_type": "WorkflowCompleted", + "sequence": 7 + } + ], + "history_after_prefix": [ + { + "event_type": "StartAccepted", + "sequence": 1 + }, + { + "event_type": "WorkflowStarted", + "sequence": 2 + }, + { + "event_type": "SideEffectRecorded", + "sequence": 3 + }, + { + "event_type": "VersionMarkerRecorded", + "sequence": 4 + } + ], + "invalid_fields": { + "response": { + "errors": { + "checkpoint_id": [ + "The checkpoint id field must be a string." + ], + "message_stream_cursors": [ + "The message stream cursors field is prohibited." + ], + "start_sequence": [ + "The start sequence field must be at least 1." + ] + }, + "message": "The checkpoint id field must be a string. (and 2 more errors)", + "protocol_version": "1.20", + "reason": "validation_failed", + "server_capabilities": "{present}", + "validation_errors": { + "checkpoint_id": [ + "The checkpoint id field must be a string." + ], + "message_stream_cursors": [ + "The message stream cursors field is prohibited." + ], + "start_sequence": [ + "The start sequence field must be at least 1." + ] + } + }, + "status": 422 + }, + "legacy_claim": { + "response": { + "checkpointed": false, + "protocol_version": "1.20", + "reason": "active_claim_cancellation_not_supported", + "server_capabilities": "{present}" + }, + "status": 409 + }, + "mismatch": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "missing_fields": { + "response": { + "errors": { + "checkpoint_id": [ + "The checkpoint id field is required." + ], + "start_sequence": [ + "The start sequence field is required." + ], + "sticky_cache": [ + "The sticky cache field is prohibited." + ] + }, + "message": "The checkpoint id field is required. (and 2 more errors)", + "protocol_version": "1.20", + "reason": "validation_failed", + "server_capabilities": "{present}", + "validation_errors": { + "checkpoint_id": [ + "The checkpoint id field is required." + ], + "start_sequence": [ + "The start sequence field is required." + ], + "sticky_cache": [ + "The sticky cache field is prohibited." + ] + } + }, + "status": 422 + }, + "open_after": { + "response": { + "claim_released": false, + "created_task_ids": [], + "duplicate": false, + "history_event_id": "{ulid:36}", + "history_refresh_page_token": "MA==", + "lease_owner": "pr1-cancellation-scopes-r5-worker", + "opened": true, + "parent_scope_id": "root", + "protocol_version": "1.20", + "reason": null, + "scope_id": "{R5.S1}", + "sequence": 4, + "server_capabilities": "{present}", + "shield_parent": false, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "plain_claim": { + "response": { + "checkpointed": false, + "protocol_version": "1.20", + "reason": "active_claim_cancellation_not_supported", + "server_capabilities": "{present}" + }, + "status": 409 + }, + "prefix": { + "response": { + "checkpoint_id": "ck-1", + "checkpointed": true, + "created_task_ids": [], + "duplicate": false, + "fingerprint": "{R5.prefix.fingerprint}", + "history_refresh_page_token": "MA==", + "lease_expires_at": "{time:Z6}", + "lease_owner": "pr1-cancellation-scopes-r5-worker", + "next_sequence": 3, + "protocol_version": "1.20", + "reason": null, + "recorded_at": "{time:Z6}", + "server_capabilities": "{present}", + "start_sequence": 1, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "prefix_again": { + "response": { + "checkpoint_id": "ck-1", + "checkpointed": true, + "created_task_ids": [], + "duplicate": true, + "fingerprint": "{R5.prefix.fingerprint}", + "history_refresh_page_token": "MA==", + "lease_expires_at": "{time:Z6}", + "lease_owner": "pr1-cancellation-scopes-r5-worker", + "next_sequence": 3, + "protocol_version": "1.20", + "reason": null, + "recorded_at": "{time:Z6}", + "server_capabilities": "{present}", + "start_sequence": 1, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + }, + "protocol_1_19": { + "response": { + "checkpointed": false, + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_requires_protocol_1_20", + "server_capabilities": "{present}" + }, + "status": 409 + }, + "sequence_behind": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "cancellation_scope_checkpoint_sequence_mismatch", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "signal_wait": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "invalid_cancellation_scope_checkpoint_commands", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "terminal": { + "response": { + "checkpointed": false, + "duplicate": false, + "history_refresh_page_token": "MA==", + "protocol_version": "1.20", + "reason": "invalid_cancellation_scope_checkpoint_commands", + "server_capabilities": "{present}", + "task_id": "{R5.T1}" + }, + "status": 409 + }, + "timer": { + "response": { + "checkpoint_id": "ck-2", + "checkpointed": true, + "created_task_ids": [ + "{R5.timer.task}" + ], + "duplicate": false, + "fingerprint": "{R5.timer.fingerprint}", + "history_refresh_page_token": "MA==", + "lease_expires_at": "{time:Z6}", + "lease_owner": "pr1-cancellation-scopes-r5-worker", + "next_sequence": 4, + "protocol_version": "1.20", + "reason": null, + "recorded_at": "{time:Z6}", + "server_capabilities": "{present}", + "start_sequence": 3, + "task_id": "{R5.T1}", + "workflow_run_id": "{R5}", + "workflow_task_attempt": 1 + }, + "status": 200 + } } } diff --git a/tests/Unit/ServerParityCancellationScopesContractTest.mjs b/tests/Unit/ServerParityCancellationScopesContractTest.mjs index 26321056c..541c2b304 100644 --- a/tests/Unit/ServerParityCancellationScopesContractTest.mjs +++ b/tests/Unit/ServerParityCancellationScopesContractTest.mjs @@ -4,7 +4,8 @@ import test from 'node:test'; import {checkCancellationScopes} from '../../scripts/conformance/server-parity/cancellation-scopes-contract.mjs'; const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/cancellation-scopes.json', import.meta.url))); -// The observation published Server 2.11.1 returned for this fixture. +// The observation published Server 2.11.1 returned for this fixture, with +// the scope_checkpoints steps published Server 2.11.5 returned. const recorded = JSON.parse(readFileSync( new URL('../Fixtures/ServerParityExpected/cancellation-scopes-observation.json', import.meta.url))); const id = 'pr1-cancellation-scopes'; @@ -18,7 +19,7 @@ const check = state => checkCancellationScopes(fixture, state, id); const intended = JSON.parse(JSON.stringify(fixture.cancellation_scopes.expected).replaceAll('{id}', id)); const event = (history, type) => history.find(item => item.event_type === type); -test('the published 2.11.1 recording satisfies the contract', () => { +test('the published 2.11.1 and 2.11.5 recording satisfies the contract', () => { assert.deepStrictEqual(check(observation()), {scope: 'cancellation_scopes'}); }); @@ -47,6 +48,22 @@ for (const [label, change, message] of [ state[step].response.reason = 'cancellation_scope_not_recorded'; } }, /cancellation scopes prepare_root_scope/], + // Native answered 501 for the checkpoint before it served it. + ['an unimplemented checkpoint', state => { + state.scope_checkpoints.prefix = {status: 501, response: {reason: 'rust_capability_not_implemented'}}; + }, /cancellation scopes scope_checkpoints/], + ['a checkpoint retry that commits again', state => { + state.scope_checkpoints.prefix_again.response.duplicate = false; + }, /cancellation scopes scope_checkpoints/], + ['a retry with another fingerprint', state => { + state.scope_checkpoints.prefix_again.response.fingerprint = '{sha256:1}'; + }, /cancellation scopes scope_checkpoints/], + ['a prefix that releases the claim', state => { + state.scope_checkpoints.open_after = {status: 409, response: {opened: false, reason: 'cancellation_scope_workflow_claim_mismatch'}}; + }, /cancellation scopes scope_checkpoints/], + ['a refusal answered without the refresh token', state => { + delete state.scope_checkpoints.mismatch.response.history_refresh_page_token; + }, /cancellation scopes scope_checkpoints/], ['a replayed open that records a second scope', state => { state.open_s1_again.response.duplicate = false; }, /cancellation scopes open_s1_again/],