diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index ec9ffeae7..9e14ac03e 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/ServerParitySearchAttributeContractTest.mjs - tests/Unit/ServerParityWorkerListingContractTest.mjs - tests/Unit/ServerParityWorkerReplacementContractTest.mjs @@ -64,6 +65,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs - tests/Unit/ServerParitySearchAttributeContractTest.mjs - tests/Unit/ServerParityWorkerListingContractTest.mjs - tests/Unit/ServerParityWorkerReplacementContractTest.mjs @@ -1426,7 +1428,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/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.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/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.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 tests/Unit/ServerParityLateWorkflowCompletionContractTest.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 12083668b..0600015c9 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -121,6 +121,12 @@ contract fixture before its correction. above the request timeout, still inside the 9-second load wave, needs its own reviewed profile change. +- **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. - **History retention.** Published PHP's worker heartbeat runs an inline history retention pass, and the default namespace retention is bounded at 30 days, so PHP prunes old closed runs. Native has no retention yet. Its @@ -223,6 +229,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 20, search attribute registry: - **Source.** Native serves `GET` and `POST /api/search-attributes` and diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 5d75385b8..678bc41ac 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -69,6 +69,7 @@ mod worker_contracts; mod worker_incarnations; mod worker_management; 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 21b8e8834..5dcb683b1 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1309,10 +1309,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")?); @@ -1337,7 +1337,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") @@ -1435,9 +1435,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?; @@ -1708,7 +1708,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) @@ -1719,7 +1729,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")?; @@ -2050,7 +2060,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) @@ -2476,6 +2487,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 new file mode 100644 index 000000000..595b6754a --- /dev/null +++ b/rust/src/runtime/workflow_task_guard.rs @@ -0,0 +1,26 @@ +//! 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}; + +pub(super) fn refusal( + 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, + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 41183324c..0c82502d0 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -37,6 +37,8 @@ mod role_tokens; #[path = "execution/namespaces.rs"] mod namespaces; +#[path = "execution/late_workflow_claims.rs"] +mod late_workflow_claims; #[path = "execution/search_attributes.rs"] mod search_attributes; 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 a27a7ae14..1dd49d018 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 {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkWorkerListing} from './worker-listing-contract.mjs'; import {checkWorkerReplacement} from './worker-replacement-contract.mjs'; @@ -132,6 +133,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 searchAttributes = fixture.search_attributes ? checkSearchAttributes(fixture, observation, workflowId) : null; const workerListing = fixture.worker_listing ? checkWorkerListing(fixture, observation, workflowId) : null; const workerReplacement = fixture.worker_replacement ? checkWorkerReplacement(fixture, observation, workflowId) : null; @@ -672,6 +674,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(admission ? {admission} : {}), ...(namespaces ? {namespace_isolation: namespaces} : {}), ...(workerDeregistration ? {worker_deregistration: workerDeregistration} : {}), + ...(lateWorkflowCompletion ? {late_workflow_completion: lateWorkflowCompletion} : {}), ...(searchAttributes ? {search_attributes: searchAttributes} : {}), ...(workerListing ? {worker_listing: workerListing} : {}), ...(workerReplacement ? {worker_replacement: workerReplacement} : {}), 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 797e0c75e..85d00a6a1 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; $searchAttributes = isset($fixture['search_attributes']) ? finishHttpSearchAttributes($client, $fixture, $workflowId, $url) : null; $workerListing = isset($fixture['worker_listing']) ? finishHttpWorkerListing($client, $fixture, $workflowId) : null; $workerReplacement = isset($fixture['worker_replacement']) ? finishHttpWorkerReplacement($client, $fixture, $workflowId) : null; @@ -516,6 +517,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'admission' => $admission, 'namespace_isolation' => $namespaces, 'worker_deregistration' => $workerDeregistration, + 'late_workflow_completion' => $lateWorkflowCompletion, 'search_attributes' => $searchAttributes, 'worker_listing' => $workerListing, 'worker_replacement' => $workerReplacement, @@ -816,6 +818,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/namespace-probe.php'; require __DIR__.'/search-attribute-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; + require __DIR__.'/late-workflow-completion-probe.php'; require __DIR__.'/worker-listing-probe.php'; require __DIR__.'/worker-replacement-probe.php'; require __DIR__.'/waiting-history-probe.php'; @@ -957,6 +960,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']; + } if ($mode === 'embedded' && isset($fixture['search_attributes'])) { $observation['search_attributes'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_search_attribute_registry']; } 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')); + }); +}