From f43e6fae27897ebbdf37a784fc84928e1bb680f8 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 17:55:10 +0000 Subject: [PATCH 1/2] Refuse a lost workflow task claim with PHP's ownership guard fields A paused worker that resumed with workflow tasks already given to another worker got only a reason from native (task_not_leased, conflicting_completion) and exited. Published PHP refuses those requests through WorkflowTaskOwnership::guard, whose body carries the task status and, once the run has closed, can_continue false; current workers read those fields to drop a lost claim and keep running. Completion, failure, lease renewal and task history now share one guard in PHP's order, with PHP's refusal bodies. A completion that differs from the recorded one gets the guard's refusal instead of conflicting_completion. A completed workflow task clears its lease expiry and keeps its owner, as PHP's does. The late-workflow-completion fixture covers the replaced claim's requests while the new claim holds the lease, after it completes and after the run closes. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 4 +- docs/rust-server-port.md | 32 ++++ rust/src/runtime/mod.rs | 1 + rust/src/runtime/store.rs | 23 ++- rust/src/runtime/workflow_task_guard.rs | 138 ++++++++++++++++++ rust/tests/execution.rs | 3 + rust/tests/execution/continuations.rs | 5 +- rust/tests/execution/late_workflow_claims.rs | 113 ++++++++++++++ rust/tests/execution/workflow_failures.rs | 8 +- .../conformance/server-parity/contract.mjs | 3 + .../late-workflow-completion-contract.mjs | 62 ++++++++ .../late-workflow-completion-probe.php | 76 ++++++++++ scripts/conformance/server-parity/probe.php | 6 + .../late-workflow-completion.json | 47 ++++++ ...rityLateWorkflowCompletionContractTest.mjs | 55 +++++++ 15 files changed, 567 insertions(+), 9 deletions(-) create mode 100644 rust/src/runtime/workflow_task_guard.rs create mode 100644 rust/tests/execution/late_workflow_claims.rs create mode 100644 scripts/conformance/server-parity/late-workflow-completion-contract.mjs create mode 100644 scripts/conformance/server-parity/late-workflow-completion-probe.php create mode 100644 tests/Fixtures/ServerParity/late-workflow-completion.json create mode 100644 tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 31788a264..236f105f5 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -24,6 +24,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -61,6 +62,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -1420,7 +1422,7 @@ jobs: -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 349f23089..701a1a923 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -85,6 +85,13 @@ contract fixture before its correction. or whether the fixture counts a boundary replay as the same claim, then add the fixture before changing either side. +- **A valid workflow task claim held by a stale worker.** Published PHP + 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. + Add a fixture for each before changing them. + ## Contract and verification inventory The status column summarizes bounded native source and its recorded component @@ -152,6 +159,31 @@ support is not evidence of multi-node safety. Newest checkpoints first. Each entry is exact to the source, tree and runs it names, including unsuccessful observations. +October 20, lost workflow task claims: + +- **Found by.** The [multi-node runs](../benchmarks/server-port/multinode/results-2026-10-20.md). + A paused SDK worker resumed with workflow tasks the server had already + given to the other worker. Native refused its late renewals and completion + with only a reason (`task_not_leased`, `conflicting_completion`), and the + worker exited 6 times. Published PHP's ownership guard also returns the task + status and, after the run closes, `can_continue: false`. A current worker + reads those fields to drop a lost claim and keep running. +- **Source.** Workflow task completion, failure, lease renewal and task history + now share PHP's guard (`WorkflowTaskOwnership::guard`), in PHP's order. + A task that is no longer leased is refused with `run_closed` when its run + has closed, and otherwise with `task_not_leased` and its status. An open + lease is then checked for expiry, owner and attempt. Each refusal has PHP's + body: `error`, the reason's fields, `protocol_version` and + `server_capabilities`. A completion that differs from the recorded one now + gets this refusal instead of `conflicting_completion`. An identical replay + still returns the recorded `already_completed` receipt. A completed workflow + task now clears its lease expiry and keeps its owner, as PHP's does. +- **Fixture.** `late-workflow-completion` replaces a worker that holds a + claim and sends the replaced claim's completion, renewal and failure while + the new claim holds the lease, after it completes, and after the run + closes. Published PHP 2.5.14, embedded and native passed in a local copy of + the SQLite parity step. Before the change, native failed the same fixture. + October 19, before #478, #476, #494 and #460 merged: - **Open native PRs.** [#478](https://github.com/durable-workflow/server/pull/478) diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 9f5766b20..46f379fa9 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -67,6 +67,7 @@ mod wait_timeouts; mod worker_contracts; mod worker_incarnations; mod workflow_failures; +mod workflow_task_guard; pub use execution::Runtime; pub use external_config::ExternalPayloadConfig; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index f944c8c58..d784a43c6 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1286,10 +1286,10 @@ where pub(crate) async fn heartbeat_task(&self, task_id: &str, body: Value) -> Result { let mut tx = self.begin().await?; let task = self.task_row(&mut tx, task_id).await?; - Self::fence(&task, &body)?; if DB::string(&task, "task_type")? != "workflow" { return Err(refuse(StatusCode::NOT_FOUND, "task_not_found")); } + Self::workflow_task_guard(&mut tx, &task, &body).await?; let run = Self::active_run(&mut tx, &DB::string(&task, "workflow_run_id")?).await?; let expires = Self::cancellation_lease_expiry(&run)?; tx.summary_changed(&DB::string(&run, "id")?); @@ -1314,7 +1314,7 @@ where if DB::string(&task, "task_type")? != "workflow" { return Err(refuse(StatusCode::NOT_FOUND, "task_not_found")); } - Self::fence(&task, &body)?; + Self::workflow_task_guard(&mut tx, &task, &body).await?; let run_id = DB::string(&task, "workflow_run_id")?; let (events, next) = Self::history_page_connection(&mut tx, &run_id, after, 100).await?; let run = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") @@ -1412,9 +1412,9 @@ where if DB::string(&task, "task_type")? != "workflow" { return Err(refuse(StatusCode::NOT_FOUND, "task_not_found")); } + Self::workflow_task_guard(&mut tx, &task, &body).await?; let run_id = DB::string(&task, "workflow_run_id")?; let run = Self::active_run(&mut tx, &run_id).await?; - Self::fence(&task, &body)?; let mut payload = DB::document_row(&task, "payload")?; let signals = Self::scalar("SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$1 AND status='received'") .bind(&run_id).fetch_one(&mut *tx).await?; @@ -1685,7 +1685,17 @@ where if DB::string(&task, "task_type")? != "workflow" { return Err(refuse(StatusCode::NOT_FOUND, "task_not_found")); } - if Self::duplicate_receipt(&task, &body)? { + // 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) { + Err(super::RuntimeError::Refused { + reason: "conflicting_completion" | "task_not_leased", + .. + }) => false, + other => other?, + }; + if replay { let run_id = DB::string(&task, "workflow_run_id")?; let current = Self::query("SELECT status FROM workflow_runs WHERE id=$1") .bind(&run_id) @@ -1696,7 +1706,7 @@ where "outcome": "completed", "recorded": false, "reason": "already_completed"}), ); } - Self::fence(&task, &body)?; + Self::workflow_task_guard(&mut tx, &task, &body).await?; let run_id = DB::string(&task, "workflow_run_id")?; let run = Self::active_run(&mut tx, &run_id).await?; let task_payload = DB::document_row(&task, "payload")?; @@ -2027,7 +2037,8 @@ where } else { false }; - Self::query("UPDATE workflow_tasks SET status=$1 WHERE id=$2") + // PHP clears a completed task's lease expiry and keeps its owner. + Self::query("UPDATE workflow_tasks SET status=$1,lease_expires_at=NULL WHERE id=$2") .bind(if failed { "failed" } else { "completed" }) .bind(task_id) .execute(&mut *tx) diff --git a/rust/src/runtime/workflow_task_guard.rs b/rust/src/runtime/workflow_task_guard.rs new file mode 100644 index 000000000..3d0ee56d2 --- /dev/null +++ b/rust/src/runtime/workflow_task_guard.rs @@ -0,0 +1,138 @@ +//! PHP's workflow task ownership guard (WorkflowTaskOwnership::guard) and the +//! refusal bodies WorkerController renders from it. A worker reads these +//! fields, not just the reason, to tell that its claim was lost: the task's +//! status, whether the run can continue, and who holds the lease now. +use super::{ + Result, RuntimeError, + backend::Backend, + query_authority::wire, + store::{Store, now}, + transaction::Transaction, +}; +use axum::http::StatusCode; +use serde_json::{Value, json}; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; + +fn refused( + task_id: &str, + attempt: &Value, + error: &str, + reason: &str, + fields: Value, +) -> RuntimeError { + let mut response = json!({"task_id":task_id,"workflow_task_attempt":attempt,"error":error,"reason":reason, + "protocol_version":"1.20","server_capabilities":super::store::capabilities()}); + if let (Some(response), Value::Object(fields)) = (response.as_object_mut(), fields) { + response.extend(fields); + } + RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response, + } +} + +impl Store +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, + for<'q> &'q str: Encode<'q, DB> + Type, + for<'q> Option<&'q str>: Encode<'q, DB>, + for<'q> Option: Encode<'q, DB>, + for<'q> String: Encode<'q, DB> + Type, + for<'q> i64: Encode<'q, DB> + Type, + for<'r> i64: Decode<'r, DB>, + usize: ColumnIndex, +{ + /// The checks run in PHP's order: a task that is no longer leased first + /// (run_closed when its run has closed), then expiry, owner and attempt. + pub(super) async fn workflow_task_guard( + tx: &mut Transaction<'_, DB>, + task: &DB::Row, + body: &Value, + ) -> Result<()> { + let lease_owner = super::text(body, "lease_owner")?; + let task_id = DB::string(task, "id")?; + let attempt = body["workflow_task_attempt"] + .as_i64() + .map_or(Value::Null, Value::from); + let status = DB::string(task, "status")?; + let owner = DB::optional_string(task, "lease_owner")? + .map(|owner| owner.trim().to_owned()) + .filter(|owner| !owner.is_empty()); + let expires = DB::optional_instant(task, "lease_expires_at")?; + let expires_wire = expires.map(wire); + if status != "leased" { + let run = + Self::query("SELECT status,closed_reason,closed_at FROM workflow_runs WHERE id=$1") + .bind(DB::string(task, "workflow_run_id")?) + .fetch_one(&mut **tx) + .await?; + let run_status = DB::string(&run, "status")?; + if super::run_observation_store::terminal(&run_status) { + let stop_reason = match run_status.as_str() { + "cancelled" => "run_cancelled", + "terminated" => "run_terminated", + "completed" => "run_completed", + _ => "run_failed", + }; + return Err(refused( + &task_id, + &attempt, + "Workflow run is already closed.", + "run_closed", + json!({"stop_reason":stop_reason,"cancel_requested":matches!(run_status.as_str(),"cancelled"|"terminated"), + "can_continue":false,"run_status":run_status, + "run_closed_reason":DB::optional_string(&run,"closed_reason")?, + "run_closed_at":DB::optional_instant(&run,"closed_at")?.map(wire), + "task_status":status,"lease_owner":owner,"lease_expires_at":expires_wire}), + )); + } + return Err(refused( + &task_id, + &attempt, + "Workflow task is not currently leased.", + "task_not_leased", + json!({"task_status":status}), + )); + } + let (Some(owner), Some(expires)) = (owner, expires) else { + return Err(refused( + &task_id, + &attempt, + "Workflow task is not currently leased.", + "task_not_leased", + json!({"task_status":status}), + )); + }; + // An expired lease is claimed again by the next poll. + if expires <= now() { + return Err(refused( + &task_id, + &attempt, + "Workflow task lease has expired and is waiting for recovery.", + "lease_expired", + json!({"task_status":"leased","lease_owner":owner,"lease_expires_at":expires_wire}), + )); + } + if lease_owner != owner { + return Err(refused( + &task_id, + &attempt, + "Workflow task lease is owned by another worker.", + "lease_owner_mismatch", + json!({"lease_owner":owner}), + )); + } + let current = DB::number(task, "attempt_count")?; + if current > 0 && attempt.as_i64() != Some(current) { + return Err(refused( + &task_id, + &attempt, + "Workflow task lease attempt does not match the current claim.", + "workflow_task_attempt_mismatch", + json!({"current_attempt":current}), + )); + } + Ok(()) + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index a9097d460..8c7200652 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -37,6 +37,9 @@ mod role_tokens; #[path = "execution/namespaces.rs"] mod namespaces; +#[path = "execution/late_workflow_claims.rs"] +mod late_workflow_claims; + #[path = "execution/named_timers.rs"] mod named_timers; diff --git a/rust/tests/execution/continuations.rs b/rust/tests/execution/continuations.rs index 16fb45dad..db959d87c 100644 --- a/rust/tests/execution/continuations.rs +++ b/rust/tests/execution/continuations.rs @@ -485,7 +485,10 @@ async fn continuation_commits_one_successor_and_preserves_original_receipts_afte conflicting["commands"][0]["arguments"] = envelope(Payload::Array(vec![Payload::Long(8)])); let refused = request(&app, "POST", &path, conflicting).await; assert_eq!(refused.0, StatusCode::CONFLICT); - assert_eq!(refused.1["reason"], "conflicting_completion"); + // PHP's ownership guard: the original run closed when it continued. + assert_eq!(refused.1["reason"], "run_closed", "{}", refused.1); + assert_eq!(refused.1["can_continue"], false); + assert_eq!(refused.1["task_status"], "completed"); let runs = request( &app, "GET", diff --git a/rust/tests/execution/late_workflow_claims.rs b/rust/tests/execution/late_workflow_claims.rs new file mode 100644 index 000000000..3e34d0d06 --- /dev/null +++ b/rust/tests/execution/late_workflow_claims.rs @@ -0,0 +1,113 @@ +use super::*; + +async fn late(app: &Router, task: &Value, action: &str) -> (StatusCode, Value) { + let path = format!( + "/api/worker/workflow-tasks/{}/{action}", + task["task_id"].as_str().unwrap() + ); + let body = match action { + "complete" => completion( + task, + json!([{"type":"complete_workflow","result":envelope(Payload::Long(1))}]), + ), + "fail" => { + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "failure":{"message":"late","type":"RuntimeException"}}) + } + _ => { + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"]}) + } + }; + let (status, mut response) = request(app, "POST", &path, body).await; + assert_eq!(status, StatusCode::CONFLICT, "{action}: {response}"); + assert_eq!(response["protocol_version"], "1.20", "{action}"); + let object = response.as_object_mut().unwrap(); + assert!(object.remove("server_capabilities").is_some(), "{action}"); + object.remove("protocol_version"); + (status, response) +} + +/// A replaced claim's late requests are refused with PHP's ownership guard +/// bodies, which a worker reads to recognize that its claim was lost. +#[tokio::test] +async fn a_replaced_workflow_claim_is_refused_with_the_published_ownership_fields() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "late-worker", json!(["echo"]), json!([])).await; + let run = start(&app, "late-claim").await; + let replaced = poll(&app, "late-worker", "workflow").await; + register(&app, "late-worker", json!(["echo"]), json!([])).await; + let current = poll(&app, "late-worker", "workflow").await; + assert_eq!(current["task_id"], replaced["task_id"]); + assert_eq!( + ( + replaced["workflow_task_attempt"].as_i64(), + current["workflow_task_attempt"].as_i64() + ), + (Some(1), Some(2)) + ); + let fence = json!({"task_id":replaced["task_id"],"workflow_task_attempt":1}); + let with = |fields: Value| { + let mut body = fence.clone(); + body.as_object_mut() + .unwrap() + .extend(fields.as_object().unwrap().clone()); + body + }; + let mismatch = with( + json!({"error":"Workflow task lease attempt does not match the current claim.", + "reason":"workflow_task_attempt_mismatch","current_attempt":2}), + ); + for action in ["complete", "heartbeat", "history"] { + assert_eq!(late(&app, &replaced, action).await.1, mismatch, "{action}"); + } + let mut owner = replaced.clone(); + owner["lease_owner"] = json!("other-worker"); + owner["workflow_task_attempt"] = json!(2); + assert_eq!( + late(&app, &owner, "complete").await.1, + json!({"task_id":replaced["task_id"],"workflow_task_attempt":2,"error":"Workflow task lease is owned by another worker.", + "reason":"lease_owner_mismatch","lease_owner":"late-worker"}) + ); + let history_path = format!( + "/api/workflows/late-claim/runs/{}/history", + run["run_id"].as_str().unwrap() + ); + let before = request(&app, "GET", &history_path, Value::Null).await.1; + let done = finish_task( + &app, + ¤t, + json!([{"type":"complete_workflow","result":envelope(Payload::Long(2))}]), + ) + .await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + let after = request(&app, "GET", &history_path, Value::Null).await.1; + for action in ["heartbeat", "fail", "complete"] { + let (_, refused) = late(&app, &replaced, action).await; + let closed_at = refused["run_closed_at"] + .as_str() + .unwrap_or_default() + .to_owned(); + assert!( + closed_at.ends_with('Z') && closed_at.len() == 27, + "{action}: {refused}" + ); + assert_eq!( + refused, + with( + json!({"error":"Workflow run is already closed.","reason":"run_closed","stop_reason":"run_completed", + "cancel_requested":false,"can_continue":false,"run_status":"completed","run_closed_reason":"completed", + "run_closed_at":closed_at,"task_status":"completed","lease_owner":"late-worker","lease_expires_at":null}) + ), + "{action}" + ); + } + assert_eq!( + request(&app, "GET", &history_path, Value::Null).await.1, + after + ); + assert_ne!(before, after); + runtime.close().await; + database.remove().await; +} diff --git a/rust/tests/execution/workflow_failures.rs b/rust/tests/execution/workflow_failures.rs index 72d00e3ff..4a5751d3e 100644 --- a/rust/tests/execution/workflow_failures.rs +++ b/rust/tests/execution/workflow_failures.rs @@ -88,7 +88,13 @@ async fn workflow_failure_is_terminal_atomic_and_retains_one_receipt_after_cold_ ) .await; assert_eq!(changed.0, StatusCode::CONFLICT); - assert_eq!(changed.1["reason"], "conflicting_completion"); + // PHP's ownership guard: the task is no longer leased and its run closed. + assert_eq!(changed.1["reason"], "run_closed", "{}", changed.1); + assert_eq!(changed.1["can_continue"], false); + // A task that failed its run is stored as failed, as PHP stores it. + assert_eq!(changed.1["task_status"], "failed"); + assert_eq!(changed.1["run_status"], "failed"); + assert_eq!(changed.1["stop_reason"], "run_failed"); assert_eq!(history(&cold_app, &run).await, original); snapshot(&database, &cold_app, &run).await; assert!( diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index e60868655..4ff23ea44 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -9,6 +9,7 @@ import {checkVisibility} from './visibility-contract.mjs'; import {checkAdmission} from './admission-contract.mjs'; import {checkNamespaces} from './namespace-contract.mjs'; import {checkWorkerDeregistration} from './worker-deregistration-contract.mjs'; +import {checkLateWorkflowCompletion} from './late-workflow-completion-contract.mjs'; import {checkPatchDeployment, checkPublishedPatchArtifacts, checkPatchPackageObservation} from './patch-deployment-contract.mjs'; import {checkLegacyMarkerArtifacts, checkLegacyMarkerPackageObservation, checkLegacyMarkerDeployment} from './legacy-marker-contract.mjs'; import {checkWaitingHistory} from './waiting-history-contract.mjs'; @@ -128,6 +129,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const admission = fixture.admission ? checkAdmission(fixture, observation, workflowId) : null; const namespaces = fixture.namespace_isolation ? checkNamespaces(fixture, observation, workflowId, checkObservation) : null; const workerDeregistration = fixture.worker_deregistration ? checkWorkerDeregistration(fixture, observation, workflowId) : null; + const lateWorkflowCompletion = fixture.late_workflow_completion ? checkLateWorkflowCompletion(fixture, observation, workflowId) : null; const patchDeployment = fixture.patch_deployment ? (fixture.patch_deployment.legacy_marker_history ? checkLegacyMarkerDeployment(fixture, observation, workflowId) : checkPatchDeployment(fixture, observation, workflowId)) : null; const waitingHistory = fixture.waiting_for_history ? checkWaitingHistory(fixture, observation, workflowId) : null; @@ -665,6 +667,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(admission ? {admission} : {}), ...(namespaces ? {namespace_isolation: namespaces} : {}), ...(workerDeregistration ? {worker_deregistration: workerDeregistration} : {}), + ...(lateWorkflowCompletion ? {late_workflow_completion: lateWorkflowCompletion} : {}), ...(patchDeployment ? {patch_deployment: patchDeployment} : {}), ...(waitingHistory ? {waiting_for_history: waitingHistory} : {}), ...(taskRetry ? {workflow_task_retry: taskRetry} : {}), diff --git a/scripts/conformance/server-parity/late-workflow-completion-contract.mjs b/scripts/conformance/server-parity/late-workflow-completion-contract.mjs new file mode 100644 index 000000000..e332d56ee --- /dev/null +++ b/scripts/conformance/server-parity/late-workflow-completion-contract.mjs @@ -0,0 +1,62 @@ +import assert from 'node:assert/strict'; + +// Reviewed late requests from a replaced workflow task claim +// (WorkflowTaskOwnership::guard and WorkerController::guardWorkflowTaskOwnership). +// A worker recognizes a lost claim from these fields, so the bodies are exact +// apart from server timestamps. Embedded applications have no HTTP worker +// registration, so the projection is the definition alone. +export function checkLateWorkflowCompletion(fixture, observation, workflowId) { + const definition = fixture.late_workflow_completion; + const state = observation.late_workflow_completion; + const projection = {scope: 'late_workflow_completion', expected_events: definition.expected_events}; + assert.ok(state && typeof state === 'object', 'actual late workflow completion observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_worker_registration_lifecycle'}, + 'embedded late workflow completion explicitly inapplicable'); + return projection; + } + const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); + const worker = `${workflowId}-late-worker`; + equal([state.applicable, state.worker_id, state.task_queue, state.peer_workflow_id], + [true, worker, `${workflowId}-late`, `${workflowId}-late-peer`], 'late completion identities'); + const timestamp = (value, label) => assert.match(value, /^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(\.\d{1,6})?Z$/, label); + const body = receipt => { + const {protocol_version, server_capabilities, ...rest} = receipt.response ?? {}; + assert.equal(protocol_version, '1.20', 'worker protocol refusal'); + assert.ok(server_capabilities && typeof server_capabilities === 'object', 'server capabilities'); + return rest; + }; + for (const [claims, label] of [[state.first_claims, 'first'], [state.last_claims, 'last']]) { + const [replaced, current] = claims; + equal([replaced.workflow_task_attempt, current.task_id, current.workflow_task_attempt, current.lease_owner], + [1, replaced.task_id, 2, worker], `${label} task claimed again as attempt 2 after re-registration`); + } + const [replaced] = state.first_claims; + const fence = {task_id: replaced.task_id, workflow_task_attempt: 1}; + + equal(state.while_leased.status, 409, 'late completion refused while the new claim holds the lease'); + equal(body(state.while_leased), {...fence, error: 'Workflow task lease attempt does not match the current claim.', + reason: 'workflow_task_attempt_mismatch', current_attempt: 2}, 'attempt mismatch refusal'); + equal(state.heartbeat_while_leased.status, 409, 'late lease renewal refused while the new claim holds the lease'); + equal(body(state.heartbeat_while_leased), body(state.while_leased), 'renewal refusal matches the completion refusal'); + + for (const key of ['after_completion', 'heartbeat_after_completion', 'failure_after_completion']) { + equal(state[key].status, 409, `${key} refused`); + equal(body(state[key]), {...fence, error: 'Workflow task is not currently leased.', reason: 'task_not_leased', + task_status: 'completed'}, `${key}: the completed task is not leased`); + } + equal(state.history_while_open.map(event => event.event_type), definition.expected_events.slice(0, 3), + 'late requests recorded nothing while the run was open'); + + const [lastReplaced] = state.last_claims; + const closed = body(state.after_close); + equal(state.after_close.status, 409, 'late completion refused after the run closed'); + timestamp(closed.run_closed_at, 'run closure time'); + const {run_closed_at, ...closedFields} = closed; + equal(closedFields, {task_id: lastReplaced.task_id, workflow_task_attempt: 1, error: 'Workflow run is already closed.', + reason: 'run_closed', stop_reason: 'run_completed', cancel_requested: false, can_continue: false, + run_status: 'completed', run_closed_reason: 'completed', task_status: 'completed', lease_owner: worker, + lease_expires_at: null}, 'run closed refusal: the completed task keeps its owner and has no lease expiry'); + equal(state.history.map(event => event.event_type), definition.expected_events, 'one completion recorded'); + return projection; +} diff --git a/scripts/conformance/server-parity/late-workflow-completion-probe.php b/scripts/conformance/server-parity/late-workflow-completion-probe.php new file mode 100644 index 000000000..de65b956d --- /dev/null +++ b/scripts/conformance/server-parity/late-workflow-completion-probe.php @@ -0,0 +1,76 @@ + 200, 'response' => $request()]; + } catch (ServerException $error) { + return ['status' => $error->status, 'response' => $error->details]; + } + }; + $worker = $workflowId.'-late-worker'; + $queue = $workflowId.'-late'; + $peerId = $workflowId.'-late-peer'; + // Registering the same worker_id again releases its workflow leases. + $register = static fn (): array => $client->registerWorker($worker, $queue, [$definition['workflow_type']], + [$definition['activity_type']]); + $claim = static function () use ($client, $worker, $queue): array { + $task = $client->pollWorkflowTask($worker, $queue, 5); + if (! is_array($task)) { + throw new RuntimeException('Expected a workflow task claim.'); + } + + return $task; + }; + $schedule = [[ + 'type' => 'schedule_activity', 'activity_type' => $definition['activity_type'], + 'arguments' => $client->payloadCodec()->envelope([$fixture['input']]), + 'retry_policy' => ['max_attempts' => 1, 'backoff_seconds' => []], + ]]; + $complete = [['type' => 'complete_workflow', 'result' => $client->payloadCodec()->envelope($fixture['input'])]]; + $late = static fn (array $task, array $commands): array => $receipt(static fn (): array => $client->completeWorkflowTask( + $task['task_id'], $worker, $task['workflow_task_attempt'], $commands)); + $state = ['applicable' => true, 'worker_id' => $worker, 'task_queue' => $queue, 'peer_workflow_id' => $peerId]; + $register(); + $handle = $client->startWorkflow($definition['workflow_type'], $peerId, $queue, [$fixture['input']]); + $state['peer_run_id'] = $handle->selectedRunId; + + $replaced = $claim(); + $register(); + $current = $claim(); + $state['first_claims'] = [$replaced, $current]; + $state['while_leased'] = $late($replaced, $schedule); + $state['heartbeat_while_leased'] = $receipt(static fn (): array => $client->heartbeatWorkflowTask( + $replaced['task_id'], $worker, $replaced['workflow_task_attempt'])); + $state['scheduled'] = $client->completeWorkflowTask($current['task_id'], $worker, $current['workflow_task_attempt'], $schedule); + $state['after_completion'] = $late($replaced, $schedule); + $state['heartbeat_after_completion'] = $receipt(static fn (): array => $client->heartbeatWorkflowTask( + $replaced['task_id'], $worker, $replaced['workflow_task_attempt'])); + $state['failure_after_completion'] = $receipt(static fn (): array => $client->failWorkflowTask( + $replaced['task_id'], $worker, $replaced['workflow_task_attempt'], 'late failure')); + $state['history_while_open'] = httpHistory($client, $peerId, $handle->selectedRunId); + + $activity = $client->pollActivityTask($worker, $queue, 5); + $client->completeActivityTask($activity['task_id'], $activity['activity_attempt_id'], $worker, $fixture['input']); + $replaced = $claim(); + $register(); + $current = $claim(); + $state['last_claims'] = [$replaced, $current]; + $state['closed'] = $client->completeWorkflowTask($current['task_id'], $worker, $current['workflow_task_attempt'], $complete); + $state['after_close'] = $late($replaced, $complete); + $state['history'] = httpHistory($client, $peerId, $handle->selectedRunId); + $client->deregisterWorker($worker); + + return $state; +} diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 060d26745..2361a1f10 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -427,6 +427,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $handle->selectedRunId, $queue, $url, $namespace) : null; $namespaces = isset($fixture['namespace_isolation']) ? finishHttpNamespaces($client, $fixture, $workflowId, $queue, $url) : null; $workerDeregistration = isset($fixture['worker_deregistration']) ? finishHttpWorkerDeregistration($client, $fixture, $workflowId, $queue, $url) : null; + $lateWorkflowCompletion = isset($fixture['late_workflow_completion']) ? finishHttpLateWorkflowCompletion($client, $fixture, $workflowId) : null; $waitingHistory = isset($fixture['waiting_for_history']) ? finishHttpWaitingHistory($client, $fixture, $workflowId, $queue, $url) : null; $taskRetry = isset($fixture['workflow_task_retry']) ? finishHttpTaskRetry($client, $fixture, $workflowId, $queue, $url) : null; if ($scheduleState !== null) { @@ -513,6 +514,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'admission' => $admission, 'namespace_isolation' => $namespaces, 'worker_deregistration' => $workerDeregistration, + 'late_workflow_completion' => $lateWorkflowCompletion, 'waiting_for_history' => $waitingHistory, 'workflow_task_retry' => $taskRetry, 'run_id' => $execution->runId, @@ -809,6 +811,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/admission-probe.php'; require __DIR__.'/namespace-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; + require __DIR__.'/late-workflow-completion-probe.php'; require __DIR__.'/waiting-history-probe.php'; require __DIR__.'/task-retry-probe.php'; require __DIR__.'/child-cancellation-probe.php'; @@ -942,6 +945,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['worker_deregistration'])) { $observation['worker_deregistration'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_worker_registration_lifecycle']; } + if ($mode === 'embedded' && isset($fixture['late_workflow_completion'])) { + $observation['late_workflow_completion'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_worker_registration_lifecycle']; + } $observation['mode'] = $mode; $observation['typed_output'] = typedValue($observation['output']); if ($mode === 'embedded' && isset($fixture['waiting_for_history'])) { diff --git a/tests/Fixtures/ServerParity/late-workflow-completion.json b/tests/Fixtures/ServerParity/late-workflow-completion.json new file mode 100644 index 000000000..42006e41d --- /dev/null +++ b/tests/Fixtures/ServerParity/late-workflow-completion.json @@ -0,0 +1,47 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "late-workflow-completion", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": { + "message": "late workflow completion λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "late workflow completion λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "WorkflowCompleted" + ], + "late_workflow_completion": { + "workflow_type": "parity.v1.late_completion", + "activity_type": "parity.v1.late_completion_activity", + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ActivityScheduled", + "ActivityStarted", + "ActivityCompleted", + "WorkflowCompleted" + ] + }, + "contract": "A worker process holding a workflow task claim is replaced by a new registration of the same worker_id, which releases the lease; the task is claimed again as attempt 2. Requests from the replaced attempt 1 are refused with the published ownership guard bodies a worker reads to recognize a lost claim. While attempt 2 holds the lease, completion and lease renewal are refused with 409 workflow_task_attempt_mismatch and the current attempt. After attempt 2 completes the task while the run is open, completion, renewal and failure are refused with 409 task_not_leased and task_status completed. After the run closes, a late completion is refused with 409 run_closed, can_continue false, the run status and closure, and the completed task, which keeps its lease owner and has no lease expiry. No late request records history; each task records one completion. Embedded applications have no HTTP worker registration." +} diff --git a/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs b/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs new file mode 100644 index 000000000..848d689f0 --- /dev/null +++ b/tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs @@ -0,0 +1,55 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkLateWorkflowCompletion} from '../../scripts/conformance/server-parity/late-workflow-completion-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/late-workflow-completion.json', import.meta.url))); +const worker = 'test-late-worker'; +function model() { + const protocol = {protocol_version: '1.20', server_capabilities: {supported_workflow_task_commands: []}}; + const claims = task => [1, 2].map(attempt => ({task_id: task, workflow_task_attempt: attempt, lease_owner: worker})); + const refusal = response => ({status: 409, response: {...protocol, task_id: 'task-1', workflow_task_attempt: 1, ...response}}); + const mismatch = refusal({error: 'Workflow task lease attempt does not match the current claim.', + reason: 'workflow_task_attempt_mismatch', current_attempt: 2}); + const notLeased = refusal({error: 'Workflow task is not currently leased.', reason: 'task_not_leased', task_status: 'completed'}); + const events = fixture.late_workflow_completion.expected_events.map(event_type => ({event_type})); + return structuredClone({mode: 'http', late_workflow_completion: {applicable: true, worker_id: worker, + task_queue: 'test-late', peer_workflow_id: 'test-late-peer', peer_run_id: 'peer-run', + first_claims: claims('task-1'), while_leased: mismatch, heartbeat_while_leased: mismatch, + scheduled: {recorded: true}, after_completion: notLeased, heartbeat_after_completion: notLeased, + failure_after_completion: notLeased, history_while_open: events.slice(0, 3), last_claims: claims('task-2'), + closed: {recorded: true}, after_close: {status: 409, response: {...protocol, task_id: 'task-2', workflow_task_attempt: 1, + error: 'Workflow run is already closed.', reason: 'run_closed', stop_reason: 'run_completed', cancel_requested: false, + can_continue: false, run_status: 'completed', run_closed_reason: 'completed', + run_closed_at: '2026-10-20T17:33:21.305246Z', task_status: 'completed', lease_owner: worker, lease_expires_at: null}}, + history: events}}); +} +test('published lost-claim refusals pass', () => + assert.equal(checkLateWorkflowCompletion(fixture, model(), 'test').scope, 'late_workflow_completion')); +test('embedded explicitly declares the HTTP worker lifecycle inapplicable', () => checkLateWorkflowCompletion(fixture, + {mode: 'embedded', late_workflow_completion: {applicable: false, reason: 'embedded_has_no_http_worker_registration_lifecycle'}}, 'test')); +for (const [label, mutate] of [ + ['late completion accepted while the new claim holds the lease', s => s.while_leased.status = 200], + ['attempt mismatch without the current attempt', s => delete s.while_leased.response.current_attempt], + ['renewal refused differently from completion', s => s.heartbeat_while_leased.response.reason = 'lease_owner_mismatch'], + ['native conflicting completion', s => s.after_completion.response.reason = 'conflicting_completion'], + ['refusal without the task status a worker reads', s => delete s.after_completion.response.task_status], + ['bare renewal refusal', s => s.heartbeat_after_completion.response = {reason: 'task_not_leased', message: 'task_not_leased'}], + ['late failure accepted', s => s.failure_after_completion.status = 200], + ['refusal without the published error text', s => delete s.failure_after_completion.response.error], + ['refusal without the worker protocol version', s => delete s.after_completion.response.protocol_version], + ['late request recorded history', s => s.history_while_open.push({event_type: 'WorkflowTaskFailed'})], + ['the released task is not claimed again', s => s.first_claims[1].task_id = 'other-task'], + ['closed run reported as an open task', s => s.after_close.response.reason = 'task_not_leased'], + ['closed run that can continue', s => s.after_close.response.can_continue = true], + ['closed task keeps a lease expiry', s => s.after_close.response.lease_expires_at = '2026-10-20T17:34:21.000000Z'], + ['closed task loses its owner', s => s.after_close.response.lease_owner = null], + ['closure time missing', s => s.after_close.response.run_closed_at = null], + ['second completion recorded', s => s.history.push({event_type: 'WorkflowCompleted'})], +]) { + test(`rejects ${label}`, () => { + const observation = model(); + mutate(observation.late_workflow_completion); + assert.throws(() => checkLateWorkflowCompletion(fixture, observation, 'test')); + }); +} From 2fcefc4ee6e36256c294f39cea62d7574faefec0 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 19:26:47 +0000 Subject: [PATCH 2/2] Keep the workflow task guard beside the fence it replaces The guard needed its own copy of the store's where-clause in a separate module, which Sonar counted as duplicated new code. It now lives in the store's existing impl next to the fence; the module keeps only the refusal body builder. The two task_not_leased branches share one refusal. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/store.rs | 80 +++++++++++++++ rust/src/runtime/workflow_task_guard.rs | 126 ++---------------------- 2 files changed, 87 insertions(+), 119 deletions(-) diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index dcdef896f..6134e9006 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -2443,6 +2443,86 @@ where Err(refuse(StatusCode::CONFLICT, "conflicting_completion")) } + /// PHP's workflow task ownership guard, in PHP's order: a task that is + /// no longer leased first (run_closed when its run has closed), then the + /// lease's expiry, owner and attempt. + pub(super) async fn workflow_task_guard( + tx: &mut Transaction<'_, DB>, + task: &DB::Row, + body: &Value, + ) -> Result<()> { + use super::query_authority::wire; + let lease_owner = text(body, "lease_owner")?; + let task_id = DB::string(task, "id")?; + let attempt = body["workflow_task_attempt"] + .as_i64() + .map_or(Value::Null, Value::from); + let refused = |error: &str, reason: &str, fields: Value| { + super::workflow_task_guard::refusal(&task_id, &attempt, error, reason, fields) + }; + let status = DB::string(task, "status")?; + let owner = DB::optional_string(task, "lease_owner")? + .map(|owner| owner.trim().to_owned()) + .filter(|owner| !owner.is_empty()); + let expires = DB::optional_instant(task, "lease_expires_at")?; + if status != "leased" { + let run = + Self::query("SELECT status,closed_reason,closed_at FROM workflow_runs WHERE id=$1") + .bind(DB::string(task, "workflow_run_id")?) + .fetch_one(&mut **tx) + .await?; + let run_status = DB::string(&run, "status")?; + if super::run_observation_store::terminal(&run_status) { + let stop_reason = match run_status.as_str() { + "cancelled" => "run_cancelled", + "terminated" => "run_terminated", + "completed" => "run_completed", + _ => "run_failed", + }; + return Err(refused( + "Workflow run is already closed.", + "run_closed", + json!({"stop_reason":stop_reason,"cancel_requested":matches!(run_status.as_str(),"cancelled"|"terminated"), + "can_continue":false,"run_status":run_status, + "run_closed_reason":DB::optional_string(&run,"closed_reason")?, + "run_closed_at":DB::optional_instant(&run,"closed_at")?.map(wire), + "task_status":status,"lease_owner":owner,"lease_expires_at":expires.map(wire)}), + )); + } + } + let (true, Some(owner), Some(expires)) = (status == "leased", owner, expires) else { + return Err(refused( + "Workflow task is not currently leased.", + "task_not_leased", + json!({"task_status":status}), + )); + }; + // An expired lease is claimed again by the next poll. + if expires <= now() { + return Err(refused( + "Workflow task lease has expired and is waiting for recovery.", + "lease_expired", + json!({"task_status":"leased","lease_owner":owner,"lease_expires_at":wire(expires)}), + )); + } + if lease_owner != owner { + return Err(refused( + "Workflow task lease is owned by another worker.", + "lease_owner_mismatch", + json!({"lease_owner":owner}), + )); + } + let current = DB::number(task, "attempt_count")?; + if current > 0 && attempt.as_i64() != Some(current) { + return Err(refused( + "Workflow task lease attempt does not match the current claim.", + "workflow_task_attempt_mismatch", + json!({"current_attempt":current}), + )); + } + Ok(()) + } + pub(super) fn fence(task: &DB::Row, body: &Value) -> Result<()> { if DB::string(task, "status")? != "leased" { return Err(refuse(StatusCode::CONFLICT, "task_not_leased")); diff --git a/rust/src/runtime/workflow_task_guard.rs b/rust/src/runtime/workflow_task_guard.rs index 3d0ee56d2..595b6754a 100644 --- a/rust/src/runtime/workflow_task_guard.rs +++ b/rust/src/runtime/workflow_task_guard.rs @@ -1,19 +1,13 @@ -//! PHP's workflow task ownership guard (WorkflowTaskOwnership::guard) and the -//! refusal bodies WorkerController renders from it. A worker reads these -//! fields, not just the reason, to tell that its claim was lost: the task's -//! status, whether the run can continue, and who holds the lease now. -use super::{ - Result, RuntimeError, - backend::Backend, - query_authority::wire, - store::{Store, now}, - transaction::Transaction, -}; +//! Refusal bodies of PHP's workflow task ownership guard +//! (WorkflowTaskOwnership::guard, rendered by WorkerController). A worker +//! reads these fields, not just the reason, to tell that its claim was lost: +//! the task's status, whether the run can continue, and who holds the lease +//! now. `Store::workflow_task_guard` decides which refusal applies. +use super::RuntimeError; use axum::http::StatusCode; use serde_json::{Value, json}; -use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; -fn refused( +pub(super) fn refusal( task_id: &str, attempt: &Value, error: &str, @@ -30,109 +24,3 @@ fn refused( response, } } - -impl Store -where - for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, - DB::Arguments: IntoArguments, - for<'q> &'q str: Encode<'q, DB> + Type, - for<'q> Option<&'q str>: Encode<'q, DB>, - for<'q> Option: Encode<'q, DB>, - for<'q> String: Encode<'q, DB> + Type, - for<'q> i64: Encode<'q, DB> + Type, - for<'r> i64: Decode<'r, DB>, - usize: ColumnIndex, -{ - /// The checks run in PHP's order: a task that is no longer leased first - /// (run_closed when its run has closed), then expiry, owner and attempt. - pub(super) async fn workflow_task_guard( - tx: &mut Transaction<'_, DB>, - task: &DB::Row, - body: &Value, - ) -> Result<()> { - let lease_owner = super::text(body, "lease_owner")?; - let task_id = DB::string(task, "id")?; - let attempt = body["workflow_task_attempt"] - .as_i64() - .map_or(Value::Null, Value::from); - let status = DB::string(task, "status")?; - let owner = DB::optional_string(task, "lease_owner")? - .map(|owner| owner.trim().to_owned()) - .filter(|owner| !owner.is_empty()); - let expires = DB::optional_instant(task, "lease_expires_at")?; - let expires_wire = expires.map(wire); - if status != "leased" { - let run = - Self::query("SELECT status,closed_reason,closed_at FROM workflow_runs WHERE id=$1") - .bind(DB::string(task, "workflow_run_id")?) - .fetch_one(&mut **tx) - .await?; - let run_status = DB::string(&run, "status")?; - if super::run_observation_store::terminal(&run_status) { - let stop_reason = match run_status.as_str() { - "cancelled" => "run_cancelled", - "terminated" => "run_terminated", - "completed" => "run_completed", - _ => "run_failed", - }; - return Err(refused( - &task_id, - &attempt, - "Workflow run is already closed.", - "run_closed", - json!({"stop_reason":stop_reason,"cancel_requested":matches!(run_status.as_str(),"cancelled"|"terminated"), - "can_continue":false,"run_status":run_status, - "run_closed_reason":DB::optional_string(&run,"closed_reason")?, - "run_closed_at":DB::optional_instant(&run,"closed_at")?.map(wire), - "task_status":status,"lease_owner":owner,"lease_expires_at":expires_wire}), - )); - } - return Err(refused( - &task_id, - &attempt, - "Workflow task is not currently leased.", - "task_not_leased", - json!({"task_status":status}), - )); - } - let (Some(owner), Some(expires)) = (owner, expires) else { - return Err(refused( - &task_id, - &attempt, - "Workflow task is not currently leased.", - "task_not_leased", - json!({"task_status":status}), - )); - }; - // An expired lease is claimed again by the next poll. - if expires <= now() { - return Err(refused( - &task_id, - &attempt, - "Workflow task lease has expired and is waiting for recovery.", - "lease_expired", - json!({"task_status":"leased","lease_owner":owner,"lease_expires_at":expires_wire}), - )); - } - if lease_owner != owner { - return Err(refused( - &task_id, - &attempt, - "Workflow task lease is owned by another worker.", - "lease_owner_mismatch", - json!({"lease_owner":owner}), - )); - } - let current = DB::number(task, "attempt_count")?; - if current > 0 && attempt.as_i64() != Some(current) { - return Err(refused( - &task_id, - &attempt, - "Workflow task lease attempt does not match the current claim.", - "workflow_task_attempt_mismatch", - json!({"current_attempt":current}), - )); - } - Ok(()) - } -}