From b90ebb12728d1bb0d130cff5b79ded09cbc7150a Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:11:03 +0000 Subject: [PATCH 1/5] Rust #325: serve cancellation scope prefix checkpoints POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint was the last published PHP route native answered 501. The published PHP SDK (3.6.1) calls it under cooperative cancellation, which native advertises, when a turn recorded commands before opening or delivering a scope, and it fails the workflow task on any refusal. Native now serves it as Server's checkpointCancellationScopePrefix over Core's checkpointCancellationScopePrefix: - protocol 1.20, then the checkpoint fields (checkpoint_id, start_sequence, prohibited stream cursors, waits and sticky cache) with Laravel's messages, then a completion's own validation and guards; - the claim must be the cooperative claim (409 active_claim_cancellation_not_supported, without the refresh token); - Core's checks in Core's order: the checkpoint arguments, terminal and turn-closing commands refused, the claim, a closed run or passed deadline, the retained receipt (an identical retry answers it with duplicate: true, anything else under its id is a mismatch), a root cancellation request (allowed only after its delivery, past the delivered boundary, before the cleanup deadline), the authored sequence, scope membership and update commands; - the commands apply as a completion applies them, metered, and the task stays leased with its lease unchanged; the task payload keeps the portable_scope_checkpoint receipt. Answers are 200, 404 for an unknown task, or 409, in the worker envelope with history_refresh_page_token. complete_workflow's non-terminal command arms move unchanged into apply_non_terminal, which the completion and the checkpoint share. The receipt's fingerprint is native's SHA-256 over the validated commands; Core hashes its own normalized command arrays, so the values differ. No client computes or compares it. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/cancellation_scopes.rs | 2 +- rust/src/runtime/execution.rs | 30 +- rust/src/runtime/http.rs | 81 ++- rust/src/runtime/mod.rs | 1 + rust/src/runtime/scope_checkpoint.rs | 342 ++++++++++ rust/src/runtime/store.rs | 622 ++++++++++-------- rust/tests/execution.rs | 2 + rust/tests/execution/scope_checkpoints.rs | 182 +++++ .../fixtures/unserved-php-routes.v2.11.1.txt | 5 +- 9 files changed, 966 insertions(+), 301 deletions(-) create mode 100644 rust/src/runtime/scope_checkpoint.rs create mode 100644 rust/tests/execution/scope_checkpoints.rs 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..8a9615f30 --- /dev/null +++ b/rust/src/runtime/scope_checkpoint.rs @@ -0,0 +1,342 @@ +//! `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 wait closes a +/// turn, a local report reconstructs an executed callback, and selection +/// cancellation requires activity fencing. +fn excluded(command: &Value) -> bool { + matches!( + command["type"].as_str(), + Some("record_local_activity" | "open_condition_wait" | "open_signal_wait" | "cancel_selection_operation") + ) || matches!(command["type"].as_str(), Some("complete_workflow" | "fail_workflow" | "continue_as_new")) +} + +/// The prefix's identity for retries: SHA-256 over its commands' JSON. +fn fingerprint(commands: &[Value]) -> String { + super::bundle_verifier::hex(&Sha256::digest(Value::Array(commands.to_vec()).to_string().as_bytes())) +} + +fn same_instant(left: &Value, right: Option>) -> bool { + let parsed = left.as_str().and_then(|text| DateTime::parse_from_rfc3339(text).ok()).map(|at| at.with_timezone(&Utc)); + parsed.is_some() && parsed == right +} + +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"] + && same_instant(&requested["requested_at"], same_or_none(&delivered["requested_at"])) + && same_instant(&requested["cleanup_deadline_at"], same_or_none(&delivered["cleanup_deadline_at"])) + && same_instant(&requested["cleanup_deadline_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(); + if attempt < 1 || owner.is_empty() || checkpoint.id.trim().is_empty() || checkpoint.id.len() > 255 + || commands.len() > 100 + { + 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"])); + Self::apply_task_signal(&mut tx, &run_id, &task_id, &task, transition_codec).await?; + 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), + }; + } + tx.summary_changed(&run_id); + 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) + } +} + +/// A delivered context's instant, for comparison with the request's. +fn same_or_none(value: &Value) -> Option> { + value.as_str().and_then(|text| DateTime::parse_from_rfc3339(text).ok()).map(|at| at.with_timezone(&Utc)) +} + +/// 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..54afe41a4 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1442,7 +1442,7 @@ where Ok(None) } - async fn apply_task_signal( + pub(super) async fn apply_task_signal( tx: &mut Transaction<'_, DB>, run_id: &str, task_id: &str, @@ -2637,7 +2637,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 +2650,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 +2810,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 +2887,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 { @@ -2969,6 +2980,17 @@ where Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec).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 +2999,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 +3008,8 @@ 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?; - } - Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") - .bind(&run_id) - .execute(&mut *tx) - .await?; + } else if command["type"] != "complete_workflow" { + self.apply_non_terminal(&mut tx, &apply, index, command, &mut sequence, &mut created).await?; } else { if Self::cancellation_request(&run)?.is_some() { Self::finish_root_cancellation(&mut tx, &run, Some(&task)).await?; @@ -3356,6 +3106,302 @@ 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") + ) { + Self::apply_update(tx, &run, &task, command, sequence + 1).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?; + 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(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?; + 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(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?; + } + Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") + .bind(&run_id) + .execute(&mut **tx) + .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>, @@ -4633,6 +4679,28 @@ pub(super) fn validate_arguments( /// 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. +/// 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> {} + const COMPLETION_RECEIPT: &str = "workflow_task_completion"; /// PHP's `TERMINAL_TYPES`: the commands that close the run. 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..d844313a2 --- /dev/null +++ b/rust/tests/execution/scope_checkpoints.rs @@ -0,0 +1,182 @@ +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}"); +} 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. From fb70c6aa738c6c87dfffa89f0ba42dcdc7450cda Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:14:38 +0000 Subject: [PATCH 2/5] Record cancellation scope prefix checkpoints in the shared fixture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The cancellation-scopes probe adds a fifth run, R5, whose cooperative claim checkpoints the commands of its turn before opening a scope: - protocol 1.19, missing and invalid fields, a foreign lease owner; - Core's refusals: a terminal command, a signal wait; - the prefix (a side effect and a version marker at 1 and 2), its identical retry, the same id with other commands, a new id at a used sequence; - a second prefix (a timer at 3, with its task), and the first prefix refused once the second replaced its receipt; - the scope opening at 4 on the unchanged claim, the completion, and the run's history. R3's plain claim and R4's protocol 1.19 claim are refused with active_claim_cancellation_not_supported. Every step sits under scope_checkpoints, which sorts after every earlier key, and every new identity and fingerprint is named, so no earlier label in the normalized observation changes. The expectations are published Server 2.11.5's answers (image sha256:bcc67285…, SQLite), where every existing step still matched. The contract unit test's stored observation gains the same steps, with checks that a retry does not commit again, keeps its fingerprint and the claim, and that refusals carry the refresh token. Co-Authored-By: Claude Opus 5.5 --- .../cancellation-scopes-contract.mjs | 8 +- .../cancellation-scopes-probe.php | 47 +++ .../ServerParity/cancellation-scopes.json | 335 ++++++++++++++++++ .../cancellation-scopes-observation.json | 335 ++++++++++++++++++ ...erParityCancellationScopesContractTest.mjs | 21 +- 5 files changed, 743 insertions(+), 3 deletions(-) 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/], From bb6e9c27727acddf339d2e57582df49b2b435da3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:28:27 +0000 Subject: [PATCH 3/5] Record the scope checkpoint port and its evidence in the port log The port log's parity surface now counts all 175 published API route methods, its missing-surface list marks cancellation scopes served with the prefix checkpoint, and the registration-freshness note says the checkpoint shares the completion's guard. The new checkpoint file records the contract, why the release candidate needed it, the fingerprint divergence, and what was verified. Co-Authored-By: Claude Opus 5.5 --- .../2026-10-27-scope-checkpoints.md | 93 +++++++++++++++++++ docs/rust-server-port.md | 17 +++- 2 files changed, 105 insertions(+), 5 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md 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..c0a6ac31c --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md @@ -0,0 +1,93 @@ +# 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. +- **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). + - `cancellation-scopes` recorded on SQLite against published 2.11.5 + (`sha256:bcc67285…`), a native image built from this branch, and + embedded 2.11.5. All three pass, and `compare` reports + `{"outcome":"pass","recordings":3}`. 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 From 51e7de54360052466481797bd913975954631de8 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:23:21 +0000 Subject: [PATCH 4/5] Align scope checkpoints and completions with Core's command application Review of the scope checkpoint found divergences from Core 2.15.2, three of them shared with the workflow task completion: - A signal-resumed claim's signal is applied once (recordAppliedSignalForSignalResume): an already-applied signal is left as it is instead of refused with signal_already_applied, so a claim that checkpointed its prefix still completes (completion and checkpoint). On the checkpoint path, a signal Core would not apply (command unavailable, its wait not open, another wait open) is skipped as Core skips it; the completion keeps native's refusals for those. - The authored sequence counts SearchAttributesUpserted, as nextDurableCommandSequence does: a turn or prefix ending in an upsert is followed at the next sequence (completion, checkpoint and scope open). - complete_update and fail_update take the batch's next sequence and advance it (applyCompleteUpdate, applyFailUpdate): a later command in the same batch takes the following sequence (completion and checkpoint). - A checkpoint changes no run status and no run summary: Core's markRunWaiting and projection run at completion. The completion keeps setting the run waiting after an activity, timer or wait, as before. - Core's own checkpoint refusals, signal application's included, answer as a refused checkpoint. Server's exception answers (structural limits, memo validation) keep their shapes, as PHP's catch block gives them for both. Also: the receipt's doc comment is back on COMPLETION_RECEIPT; the unreachable command-count and cancel_selection_operation checks are removed, with a note; one RFC 3339 parser serves the cancellation context checks; the fingerprint hashes serde_json::to_vec(commands), the same bytes. Each of the four new execution tests fails on the previous head and passes now. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/scope_checkpoint.rs | 57 +++++----- rust/src/runtime/store.rs | 65 +++++++---- rust/tests/execution/scope_checkpoints.rs | 125 ++++++++++++++++++++++ 3 files changed, 200 insertions(+), 47 deletions(-) diff --git a/rust/src/runtime/scope_checkpoint.rs b/rust/src/runtime/scope_checkpoint.rs index 8a9615f30..bce017bee 100644 --- a/rust/src/runtime/scope_checkpoint.rs +++ b/rust/src/runtime/scope_checkpoint.rs @@ -73,24 +73,28 @@ fn refused(task_id: &str, reason: &str) -> Value { const INVALID_COMMANDS: &str = "invalid_cancellation_scope_checkpoint_commands"; -/// Commands that cannot be part of a retained-claim prefix: a wait closes a -/// turn, a local report reconstructs an executed callback, and selection -/// cancellation requires activity fencing. +/// 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" | "cancel_selection_operation") - ) || matches!(command["type"].as_str(), Some("complete_workflow" | "fail_workflow" | "continue_as_new")) + 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]) -> String { - super::bundle_verifier::hex(&Sha256::digest(Value::Array(commands.to_vec()).to_string().as_bytes())) +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))) } -fn same_instant(left: &Value, right: Option>) -> bool { - let parsed = left.as_str().and_then(|text| DateTime::parse_from_rfc3339(text).ok()).map(|at| at.with_timezone(&Utc)); - parsed.is_some() && parsed == right +/// 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! { @@ -143,9 +147,10 @@ store_methods! { && requested["request_id"].as_str() == Some(request_id.as_str()) && requested["root_request_id"] == delivered["root_request_id"] && requested["request_id"] == delivered["request_id"] - && same_instant(&requested["requested_at"], same_or_none(&delivered["requested_at"])) - && same_instant(&requested["cleanup_deadline_at"], same_or_none(&delivered["cleanup_deadline_at"])) - && same_instant(&requested["cleanup_deadline_at"], deadline)) + && 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 @@ -174,9 +179,9 @@ store_methods! { 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(); - if attempt < 1 || owner.is_empty() || checkpoint.id.trim().is_empty() || checkpoint.id.len() > 255 - || commands.len() > 100 - { + // 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"); } @@ -184,7 +189,7 @@ store_methods! { tx.rollback().await?; return refusal(INVALID_COMMANDS); } - let fingerprint = fingerprint(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" @@ -259,7 +264,15 @@ store_methods! { } // Metered like a completion (Workflow #930). tx.meter = Some(super::history_meter::Meter::new(&limits, &task_id, &body["workflow_task_attempt"])); - Self::apply_task_signal(&mut tx, &run_id, &task_id, &task, transition_codec).await?; + // 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, @@ -322,7 +335,8 @@ store_methods! { error => Err(error), }; } - tx.summary_changed(&run_id); + // 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; @@ -333,10 +347,5 @@ store_methods! { } } -/// A delivered context's instant, for comparison with the request's. -fn same_or_none(value: &Value) -> Option> { - value.as_str().and_then(|text| DateTime::parse_from_rfc3339(text).ok()).map(|at| at.with_timezone(&Utc)) -} - /// 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 54afe41a4..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) } + /// `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); @@ -2977,7 +2991,7 @@ 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 { @@ -3010,6 +3024,18 @@ where } } 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?; + } } else { if Self::cancellation_request(&run)?.is_some() { Self::finish_root_cancellation(&mut tx, &run, Some(&task)).await?; @@ -3125,7 +3151,12 @@ where command["type"].as_str(), Some("complete_update" | "fail_update") ) { - Self::apply_update(tx, &run, &task, command, sequence + 1).await?; + // `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; @@ -3253,10 +3284,6 @@ where 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(tx, &run_id, super::memo_limits::Pending::Timers, &limits, task_id, attempt).await?; sequence += 1; @@ -3315,10 +3342,6 @@ where 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" { @@ -3391,10 +3414,6 @@ where if !direct || pending_wait_id.is_none() { Self::schedule_wait_timeout(tx, &run, &payload, task_id).await?; } - Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1") - .bind(&run_id) - .execute(&mut **tx) - .await?; } else { return Err(refuse(StatusCode::UNPROCESSABLE_ENTITY, "unsupported_workflow_command")); } @@ -4676,9 +4695,6 @@ pub(super) fn validate_arguments( Ok(()) } -/// 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. /// 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. @@ -4701,6 +4717,9 @@ impl Clone for Apply<'_, R> { } 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. const COMPLETION_RECEIPT: &str = "workflow_task_completion"; /// PHP's `TERMINAL_TYPES`: the commands that close the run. diff --git a/rust/tests/execution/scope_checkpoints.rs b/rust/tests/execution/scope_checkpoints.rs index d844313a2..b126a28a9 100644 --- a/rust/tests/execution/scope_checkpoints.rs +++ b/rust/tests/execution/scope_checkpoints.rs @@ -180,3 +180,128 @@ async fn a_prefix_waits_for_a_root_cancellation_to_be_delivered() { 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}"); + } +} From 0e7e955244d7a30fb19694bb04c263b90c44e582 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:54:04 +0000 Subject: [PATCH 5/5] Record the scope checkpoint review corrections and the deferred wait rows Co-Authored-By: Claude Opus 5.5 --- .../2026-10-27-scope-checkpoints.md | 45 +++++++++++++++++-- 1 file changed, 41 insertions(+), 4 deletions(-) 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 index c0a6ac31c..bf70b8580 100644 --- a/docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md +++ b/docs/rust-server-port-checkpoints/2026-10-27-scope-checkpoints.md @@ -58,6 +58,35 @@ 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. @@ -87,7 +116,15 @@ (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). - - `cancellation-scopes` recorded on SQLite against published 2.11.5 - (`sha256:bcc67285…`), a native image built from this branch, and - embedded 2.11.5. All three pass, and `compare` reports - `{"outcome":"pass","recordings":3}`. + - 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.