diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 31788a264..0a01b6373 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -24,6 +24,8 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityWorkerListingContractTest.mjs + - tests/Unit/ServerParityWorkerReplacementContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -61,6 +63,8 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityWorkerListingContractTest.mjs + - tests/Unit/ServerParityWorkerReplacementContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -1420,7 +1424,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/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 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 dd7884a60..84290f3e9 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -14,29 +14,75 @@ qualify a later tree. ## Current state -Updated October 19, 2026, after merging #460. - -- **Merged native slices.** [#478](https://github.com/durable-workflow/server/pull/478) - (run observation, timed waits, `all` barriers, terminal update closure and - cooperative cancellation with abandoned and independently cancelled - children), [#476](https://github.com/durable-workflow/server/pull/476) - (run activity metadata listing), [#494](https://github.com/durable-workflow/server/pull/494) - (CPU deadline-health observer timing) and [#460](https://github.com/durable-workflow/server/pull/460) - (read-only inspection of an existing PHP database). No native PR is open. -- **Shared-suite reliability.** Each 13-job run still tends to show one - intermittent failure in a step its change does not touch, and qualification - records classify each one. Diagnosed and corrected: published PHP SQLite - reference lock pressure, CPU observer client timing, and native replay of a - live query lease to a fresh poll, which published PHP never does without a - matching poll request id. - Not yet diagnosed: one native-test failure in `execution-and-codec` step 8 on - main's post-#478 run, which four local repetitions under the CI limits did - not reproduce. -- **Next action.** Correct native timer dispatch under CPU load with a - deadline-health control. Then continue the remaining parity families, the - backup-first in-place takeover and the performance baseline on disposable - hosts. - +Updated October 20, 2026. + +- **Merged on October 20.** + - [#507](https://github.com/durable-workflow/server/pull/507) gives codec work under the transition lock its own permit. + - [#508](https://github.com/durable-workflow/server/pull/508) keeps explicit worker command declarations. + - [#509](https://github.com/durable-workflow/server/pull/509) reports poll statuses as the worker protocol does. + - [#515](https://github.com/durable-workflow/server/pull/515) retries the PHP reference's lock-pressure refusals in the probes. + - [#516](https://github.com/durable-workflow/server/pull/516) lets held long polls wait without the transition lock. + - [#517](https://github.com/durable-workflow/server/pull/517) designs the backup-first takeover. + - [#499](https://github.com/durable-workflow/server/pull/499) and [#510](https://github.com/durable-workflow/server/pull/510) replay a live query lease only to its own poll request and remember query poll answers per request. + - [#542](https://github.com/durable-workflow/server/pull/542) records repeated matched PHP/native performance pairs. + - This branch: [#543](https://github.com/durable-workflow/server/pull/543) (immediate termination) with [#544](https://github.com/durable-workflow/server/pull/544) (worker listing and registration facts). +- **Open.** + - The backup-first takeover stack: [#523](https://github.com/durable-workflow/server/pull/523) + (fingerprint), [#529](https://github.com/durable-workflow/server/pull/529) + (layouts) and [#532](https://github.com/durable-workflow/server/pull/532), + [#533](https://github.com/durable-workflow/server/pull/533) and + [#536](https://github.com/durable-workflow/server/pull/536) (SQLite, + PostgreSQL and MySQL/MariaDB takeover). Then + [#538](https://github.com/durable-workflow/server/pull/538) (SQLite WAL + preparation) and [#539](https://github.com/durable-workflow/server/pull/539) + (a PHP-accepted signal and an expired PHP lease resumed under native, plus + refusal of unsupported features). Its top passed all 13 parity jobs. + - [#548](https://github.com/durable-workflow/server/pull/548): PHP's + ownership guard refusals for a lost workflow task claim. + - [#549](https://github.com/durable-workflow/server/pull/549): the + multi-node kill and restart qualification and its results. + - [#550](https://github.com/durable-workflow/server/pull/550): the search + attribute registry. [#551](https://github.com/durable-workflow/server/pull/551): + namespace update. +- **Parity surface.** With this branch, native serves 63 of published PHP's 173 API + routes. The missing families are listed below under "API surface still + missing". +- **Next action.** Merge the open PRs as their checks pass. Then port history + export, a large bundle the CLI and both SDKs use for replay. Packaging the + native binary into the server image remains an open gate. + +## API surface still missing + +These published PHP 2.9.8 routes have no native handler yet, by family. The +CLI and Waterline use the first group. + +- **Used by the CLI or Waterline.** + - Schedule listing, update and backfill. + - Task queues and their build IDs. + - Namespace update, delete and external storage. + - Search attributes. + - `system` repair, retention and activity-timeout passes, metrics and + operator dashboards. + - Storage test. + - Activities (standalone). +- **Worker protocol features.** + - Worker sessions. + - Update validation tasks. + - Prepared local activities. + - Cancellation scopes. + - Activity cancellation acknowledgement. +- **Run operations.** + - History export. + - Archive. + - Redrive. + - Imports from embedded v2 and Waterline v1. +- **Other families.** + - Message streams. + - Service catalogs and endpoints. + - Nexus operations. + - Runtime credentials. + - Deployments. + - Bridge adapters. ## Known gaps found in review These are confirmed differences from the published PHP reference and open @@ -75,6 +121,41 @@ contract fixture before its correction. above the request timeout, still inside the 9-second load wave, needs its own reviewed profile change. +- **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 + heartbeat omits PHP's `retention` report rather than claim a pass. The + `system/retention` endpoints are absent too. +- **Python worker shutdown after a late synchronous query handler (not yet + diagnosed).** In the SQLite query-authority cohort on #515's run + 39431491496, the published Python SDK 2.5.15 client in `late-handler-sync` + mode failed against native. Its 6-second handler returned and was fenced + with `query_execution_budget_expired` after one claim. The SDK's query thread + then did not stop within the 5-second shutdown budget. Forty local + repetitions of that case alone passed. Find what holds the query thread + after the fence before changing either side. +- **Workflow definition evolution (PHP main after #525, not yet released).** + Current PHP routes a run whose recorded workflow definition is gone to a + replacement definition that passes strict replay. It then reports + `definition_routing` on task queues and run diagnostics. It also records + each worker's protocol version under `_worker_protocol_version` in its + stored process metrics. That entry shows up in the worker listing. Native + follows the frozen published reference, which has none of this. Port it + with its own fixture once it is published. +- **Trace context (Workflow #773 / workflow#788, not yet published).** A + Workflow minor will add nullable `observability_context` columns on runs, + commands and tasks, plus an entry in schedule metadata. Native-owned + databases are unaffected. A Server release carrying the columns is a new + takeover source layout, which native refuses with + `upgrade_source_catalog_unqualified` until its recipe lands. The recipe must + also refuse rows that carry trace context until native implements the + contract. The boundary is agreed on #325. +- **Terminating a child run.** Published Workflow resolves the parent with + `ChildRunTerminated` and the `terminated` child call status. Native refuses + this with `422 termination_propagation_not_available` until the parent-side + wait, child summary and failure projections for `ChildRunTerminated` have a + fixture. + ## Contract and verification inventory The status column summarizes bounded native source and its recorded component @@ -142,6 +223,80 @@ 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, worker listing: + +- **Source.** Native serves `GET /api/workers` and `GET /api/workers/{id}`, + as published PHP's WorkerManagementController does: + - the summary fields; + - the `task_queue` and `status` filters, with stale workers hidden unless + asked for; + - the stale window; + - `404 worker_not_found`. + + Operators may read these; only `DELETE` needs an admin. Named namespaces may + read them. +- **Registration facts.** Registration and heartbeat now keep what PHP keeps: + - capacities, defaulting to 100, 100 and 10; + - task slots, bounded by those capacities; + - normalized process metrics; + - the heartbeat cadence; + - workflow definition fingerprints. + + Re-registering an active worker with a changed fingerprint is refused with + `409 workflow_definition_changed`. Omitting an advertised one is refused with + `409 workflow_definition_fingerprint_missing`, and neither refusal changes + the registration. Omitting the fingerprints keeps the advertised one. A + heartbeat changes only what it reports. Its acknowledgement carries the + configured cadence and stale window. +- **A replaced worker's activity leases.** A new process registering a + worker_id that holds activity leases now releases them at once, as PHP + does. Its running attempts close as `expired`, and the tasks return to + ready, so the next claim is attempt 2 of the same task. This holds even under + a one-attempt retry policy, since PHP checks no policy at claim time. Before + this, native left the task leased for the five-minute activity lease. The + expired attempt's completion now gets PHP's whole stale receipt: + `409 stale_attempt`, `recorded: false`, and the activity, attempt and task + statuses. +- **Fixtures.** + - `worker-listing` registers two workers through the unchanged published + SDK. It then lists, filters, describes and refuses them, and + re-registers and deregisters them. + - `worker-replacement-activity-lease` replaces a worker holding an activity + lease, then completes the original and the replacement attempts. + + The contracts check each HTTP target against the reviewed facts. Embedded + mode is explicitly inapplicable. +- **Local evidence.** A local copy of the SQLite parity step recorded + published PHP 2.5.14, embedded PHP and native for both fixtures and + `namespace-isolation`. All three targets passed and the comparison passed. + Before the release, native failed the replacement fixture: its second claim + found no task. Native integration tests pass on SQLite, PostgreSQL, MySQL + and MariaDB. + +October 20, immediate termination: + +- **Source.** Native serves `POST /api/workflows/{id}/terminate` and its + selected-run form. Published Workflow runs cancel and terminate through one + `attemptTerminalCommand`, so native now does the same. Terminate records + `TerminateRequested` and `WorkflowTerminated`, the `terminated` status, + closed reason, failure category and propagation kind, and + `WorkflowTerminatedException`. It cancels open tasks, activities and timers + exactly as cancel does. A late activity outcome is refused with + `run_terminated`. The response carries the `terminate` control-plane + contract, which matches cancel's. +- **Refused.** Terminating a child run returns + `422 termination_propagation_not_available` and changes nothing (see known + gaps). The other propagation refusals are cancel's. +- **Fixtures.** `terminate-before-claim`, `terminate-pending-timer` and + `terminate-leased-activity` reuse the cancellation workflows with + `immediate_cancellation.command = "terminate"`. The shared contract and its + runner unit tests check the termination identity in place of cancellation's. +- **Local evidence.** A local copy of the SQLite parity step recorded published + PHP 2.5.14 over HTTP, embedded PHP and native for the three cancel and three + terminate fixtures. All six passed on each target, and the comparison passed. + Native integration tests for both commands pass on SQLite, PostgreSQL and + MySQL. + October 20, namespace update: - **Source.** Native serves `PUT /api/namespaces/{name}`, as published PHP's diff --git a/rust/src/runtime/cancellation.rs b/rust/src/runtime/cancellation.rs index 39635b999..2093ed53d 100644 --- a/rust/src/runtime/cancellation.rs +++ b/rust/src/runtime/cancellation.rs @@ -1,5 +1,6 @@ -//! Immediate cancellation of root or direct child runs. Propagation trees and -//! broader propagation dependencies remain refused until their fixtures qualify. +//! Immediate cancellation or termination of root runs and immediate +//! cancellation of direct child runs. Propagation trees and broader +//! propagation dependencies remain refused until their fixtures qualify. use super::transaction::Transaction; use super::{ Result, RuntimeError, @@ -30,6 +31,53 @@ pub(super) fn normalize_http_reason(reason: &str) -> &str { }) } +/// The two immediate terminal commands. Published PHP runs both through one +/// implementation; they differ only in the identity they record. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum TerminalCommand { + Cancel, + Terminate, +} + +impl TerminalCommand { + fn command_type(self) -> &'static str { + match self { + Self::Cancel => "cancel", + Self::Terminate => "terminate", + } + } + + /// The outcome, run status, closed reason, failure category and + /// propagation kind, which PHP records as one word. + pub(super) fn closed(self) -> &'static str { + match self { + Self::Cancel => "cancelled", + Self::Terminate => "terminated", + } + } + + fn events(self) -> (&'static str, &'static str) { + match self { + Self::Cancel => ("CancelRequested", "WorkflowCancelled"), + Self::Terminate => ("TerminateRequested", "WorkflowTerminated"), + } + } + + fn exception(self) -> &'static str { + match self { + Self::Cancel => "Workflow\\V2\\Exceptions\\WorkflowCancelledException", + Self::Terminate => "Workflow\\V2\\Exceptions\\WorkflowTerminatedException", + } + } + + fn propagation_unavailable(self) -> &'static str { + match self { + Self::Cancel => "cancellation_propagation_not_available", + Self::Terminate => "termination_propagation_not_available", + } + } +} + pub(super) async fn prepare_reason( body: &Value, semaphore: Arc, @@ -164,8 +212,9 @@ where for<'r> i64: Decode<'r, DB>, usize: ColumnIndex, { - pub(crate) async fn cancel_workflow( + pub(crate) async fn close_workflow( &self, + command: TerminalCommand, workflow_id: &str, selected_run: Option<&str>, reason: Option, @@ -188,11 +237,16 @@ where "pending" | "running" | "waiting" ); if !closed { - let unsupported_links = - Self::scalar("SELECT COUNT(*) FROM workflow_links WHERE child_workflow_run_id=$1 AND link_type!='child_workflow'") - .bind(&run_id) - .fetch_one(&mut *tx) - .await?; + // A terminated child resolves its parent with ChildRunTerminated, + // whose parent projections are not qualified yet. Only + // cancellation resolves a parent here. + let unsupported_links = Self::scalar(match command { + TerminalCommand::Cancel => "SELECT COUNT(*) FROM workflow_links WHERE child_workflow_run_id=$1 AND link_type!='child_workflow'", + TerminalCommand::Terminate => "SELECT COUNT(*) FROM workflow_links WHERE child_workflow_run_id=$1", + }) + .bind(&run_id) + .fetch_one(&mut *tx) + .await?; let pending_children = Self::scalar("SELECT COUNT(*) FROM workflow_child_calls WHERE parent_workflow_run_id=$1 AND status NOT IN ('completed','cancelled','failed')") .bind(&run_id).fetch_one(&mut *tx).await?; let pending_queries = @@ -206,7 +260,7 @@ where if unsupported_links != 0 || pending_children != 0 || pending_queries { return Err(refuse( StatusCode::UNPROCESSABLE_ENTITY, - "cancellation_propagation_not_available", + command.propagation_unavailable(), )); } } @@ -227,11 +281,11 @@ where let at = now(); let status = if closed { "rejected" } else { "accepted" }; let rejection = if closed { Some("run_not_active") } else { None }; - let outcome = if closed { None } else { Some("cancelled") }; + let outcome = if closed { None } else { Some(command.closed()) }; // The frozen HTTP selected-run route guards the current run but records // an instance-scoped command. Preserve this public receipt representation. - Self::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,resolved_workflow_run_id,command_type,target_scope,source,status,outcome,rejection_reason,workflow_type,payload_codec,payload,command_sequence,accepted_at,applied_at,rejected_at) VALUES ($1,$2,$3,$4,'cancel','instance','control_plane',$5,$6,$7,$8,'avro',$9,$10,CASE WHEN $11='accepted' THEN $12 ELSE NULL END,CASE WHEN $13='accepted' THEN $14 ELSE NULL END,CASE WHEN $15='rejected' THEN $16 ELSE NULL END)") - .bind(&command_id).bind(workflow_id).bind(&run_id).bind(&run_id).bind(status).bind(outcome).bind(rejection) + Self::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,resolved_workflow_run_id,command_type,target_scope,source,status,outcome,rejection_reason,workflow_type,payload_codec,payload,command_sequence,accepted_at,applied_at,rejected_at) VALUES ($1,$2,$3,$4,$5,'instance','control_plane',$6,$7,$8,$9,'avro',$10,$11,CASE WHEN $12='accepted' THEN $13 ELSE NULL END,CASE WHEN $14='accepted' THEN $15 ELSE NULL END,CASE WHEN $16='rejected' THEN $17 ELSE NULL END)") + .bind(&command_id).bind(workflow_id).bind(&run_id).bind(&run_id).bind(command.command_type()).bind(status).bind(outcome).bind(rejection) .bind(DB::string(&run,"workflow_type")?).bind(if closed {None} else {blob.as_deref()}).bind(command_sequence) .bind(status).bind(DB::bind_time(at)).bind(status).bind(DB::bind_time(at)) .bind(status).bind(DB::bind_time(at)).execute(&mut *tx).await?; @@ -252,12 +306,13 @@ where if let Some(reason) = &reason { identity["reason"] = json!(reason); } + let (requested_event, terminal_event) = command.events(); let mut requested = identity.clone(); - requested["command_type"] = json!("cancel"); + requested["command_type"] = json!(command.command_type()); Self::append( &mut tx, &run_id, - "CancelRequested", + requested_event, requested, None, Some(&command_id), @@ -284,9 +339,10 @@ where .await?; } let failure_id = id(); + let closed_as = command.closed(); let message = reason.as_ref().map_or_else( - || "Workflow cancelled.".to_owned(), - |reason| format!("Workflow cancelled: {reason}"), + || format!("Workflow {closed_as}."), + |reason| format!("Workflow {closed_as}: {reason}"), ); // Corrected PHP stores the entire diagnostic as one JSON string // literal when NUL would truncate or invalidate a SQL text value. @@ -296,29 +352,31 @@ where } else { message }; - const EXCEPTION: &str = "Workflow\\V2\\Exceptions\\WorkflowCancelledException"; - Self::query("INSERT INTO workflow_failures(id,workflow_run_id,source_kind,source_id,propagation_kind,failure_category,handled,exception_class,message,file,line,trace_preview,created_at,updated_at) VALUES ($1,$2,'workflow_run',$3,'cancelled','cancelled',FALSE,$4,$5,'',0,'',$6,$7)") - .bind(&failure_id).bind(&run_id).bind(&run_id).bind(EXCEPTION).bind(&message) + let exception = command.exception(); + Self::query("INSERT INTO workflow_failures(id,workflow_run_id,source_kind,source_id,propagation_kind,failure_category,handled,exception_class,message,file,line,trace_preview,created_at,updated_at) VALUES ($1,$2,'workflow_run',$3,$4,$5,FALSE,$6,$7,'',0,'',$8,$9)") + .bind(&failure_id).bind(&run_id).bind(&run_id).bind(closed_as).bind(closed_as).bind(exception).bind(&message) .bind(DB::bind_time(at)).bind(DB::bind_time(at)).execute(&mut *tx).await?; let mut terminal = identity; terminal["failure_id"] = json!(failure_id); - terminal["failure_category"] = json!("cancelled"); - terminal["closed_reason"] = json!("cancelled"); - terminal["exception_class"] = json!(EXCEPTION); + terminal["failure_category"] = json!(closed_as); + terminal["closed_reason"] = json!(closed_as); + terminal["exception_class"] = json!(exception); terminal["message"] = json!(message); - Self::query("UPDATE workflow_runs SET status='cancelled',closed_reason='cancelled',closed_at=$1,last_progress_at=$2 WHERE id=$3") - .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(&run_id).execute(&mut *tx).await?; + Self::query("UPDATE workflow_runs SET status=$1,closed_reason=$2,closed_at=$3,last_progress_at=$4 WHERE id=$5") + .bind(closed_as).bind(closed_as).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(&run_id).execute(&mut *tx).await?; Self::append( &mut tx, &run_id, - "WorkflowCancelled", + terminal_event, terminal.clone(), None, Some(&command_id), ) .await?; Self::close_pending_updates(&mut tx, &run_id, None).await?; - Self::resolve_child(&mut tx, &run, ChildResolution::Cancelled(&terminal), at).await?; + if command == TerminalCommand::Cancel { + Self::resolve_child(&mut tx, &run, ChildResolution::Cancelled(&terminal), at).await?; + } tx.summary_changed(&run_id); tx.commit().await?; self.wake.notify_waiters(); @@ -438,9 +496,14 @@ where .await?; let attempt = Self::query("SELECT * FROM activity_attempts WHERE id=$1 AND workflow_task_id=$2 AND workflow_run_id=$3") .bind(attempt_id).bind(DB::string(task,"id")?).bind(&run_id).fetch_optional(&mut **tx).await?; + let run_status = DB::string(&run, "status")?; + let reason = match run_status.as_str() { + "cancelled" => "run_cancelled", + "terminated" => "run_terminated", + _ => return Err(refuse(StatusCode::CONFLICT, "stale_attempt")), + }; if let Some(attempt) = attempt { - if DB::string(&run, "status")? == "cancelled" - && DB::string(&attempt, "status")? == "cancelled" + if DB::string(&attempt, "status")? == "cancelled" && DB::optional_string(&attempt, "lease_owner")?.as_deref() == Some(owner) { let activity = Self::query( @@ -453,8 +516,8 @@ where return Err(RuntimeError::Protocol { status: StatusCode::CONFLICT, response: json!({"task_id":DB::string(task,"id")?,"activity_attempt_id":attempt_id, - "outcome":"ignored","recorded":false,"reason":"run_cancelled","next_task_id":null, - "cancel_requested":true,"can_continue":false,"run_status":"cancelled", + "outcome":"ignored","recorded":false,"reason":reason,"next_task_id":null, + "cancel_requested":true,"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")?, "activity_status":DB::string(&activity,"status")?,"attempt_status":"cancelled","task_status":"cancelled", "lease_owner":owner,"lease_expires_at":null}), diff --git a/rust/src/runtime/control_plane.rs b/rust/src/runtime/control_plane.rs index 571177e92..48d1aa449 100644 --- a/rust/src/runtime/control_plane.rs +++ b/rust/src/runtime/control_plane.rs @@ -11,6 +11,7 @@ pub(super) enum Operation<'a> { Observe, History, Cancel, + Terminate, Repair, Debug, Update, @@ -90,6 +91,7 @@ impl Operation<'_> { ), Self::History => ("history", &["workflow_id", "run_id"], &["next_page_token"]), Self::Cancel => ("cancel", &["workflow_id"], &["outcome"]), + Self::Terminate => ("terminate", &["workflow_id"], &["outcome"]), Self::Repair => ("repair", &["workflow_id"], &["outcome"]), Self::Debug => ("debug", &["workflow_id"], &[]), Self::Update => ( @@ -136,7 +138,7 @@ impl Operation<'_> { ]); metadata["contract"]["rejection_reasons"] = json!(["control_plane_internal_error"]); } - if matches!(self, Self::Cancel | Self::Repair) { + if matches!(self, Self::Cancel | Self::Terminate | Self::Repair) { metadata["contract"]["rejection_fields"] = json!(["workflow_id", "run_id", "reason", "message", "remediation"]); metadata["contract"]["rejection_reasons"] = json!(["v1_projection_read_only"]); @@ -296,6 +298,8 @@ pub(super) async fn admission_response(request: Request, mut body: Value) -> Val | ("GET", ["api", "workflows", _, "runs", _, "observation"]) => Operation::Observe, ("POST", ["api", "workflows", _, "cancel"]) | ("POST", ["api", "workflows", _, "runs", _, "cancel"]) => Operation::Cancel, + ("POST", ["api", "workflows", _, "terminate"]) + | ("POST", ["api", "workflows", _, "runs", _, "terminate"]) => Operation::Terminate, ("POST", ["api", "workflows", _, "repair"]) | ("POST", ["api", "workflows", _, "runs", _, "repair"]) => Operation::Repair, ("GET", ["api", "workflows", _, "debug"]) diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 024b6037f..3a54ecc0a 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -706,8 +706,38 @@ impl Runtime { page_size: i64, ) -> Result ); - delegate!(register(body: Value) -> Result); - delegate!(worker_heartbeat(body: Value) -> Result); + pub(crate) async fn register(&self, body: Value) -> Result { + match &*self.storage { + Storage::Sqlite(store) => store.register(body, &self.queries).await, + Storage::Postgres(store) => store.register(body, &self.queries).await, + Storage::MySql(store) => store.register(body, &self.queries).await, + } + } + pub(crate) async fn worker_heartbeat(&self, body: Value) -> Result { + match &*self.storage { + Storage::Sqlite(store) => store.worker_heartbeat(body, &self.queries).await, + Storage::Postgres(store) => store.worker_heartbeat(body, &self.queries).await, + Storage::MySql(store) => store.worker_heartbeat(body, &self.queries).await, + } + } + pub(crate) async fn list_workers( + &self, + task_queue: Option<&str>, + status: Option<&str>, + ) -> Result { + match &*self.storage { + Storage::Sqlite(store) => store.list_workers(task_queue, status, &self.queries).await, + Storage::Postgres(store) => store.list_workers(task_queue, status, &self.queries).await, + Storage::MySql(store) => store.list_workers(task_queue, status, &self.queries).await, + } + } + pub(crate) async fn describe_worker(&self, worker_id: &str) -> Result { + match &*self.storage { + Storage::Sqlite(store) => store.describe_worker(worker_id, &self.queries).await, + Storage::Postgres(store) => store.describe_worker(worker_id, &self.queries).await, + Storage::MySql(store) => store.describe_worker(worker_id, &self.queries).await, + } + } delegate!(deregister(worker_id: &str) -> Result); pub(crate) async fn deregister_fenced(&self, worker_id: &str, body: Value) -> Result { let codec = self.transition_codec.clone(); @@ -786,8 +816,9 @@ impl Runtime { } } } - pub(crate) async fn cancel_workflow( + pub(crate) async fn close_workflow( &self, + command: super::cancellation::TerminalCommand, workflow_id: &str, run_id: Option<&str>, body: Value, @@ -797,17 +828,17 @@ impl Runtime { match &*self.storage { Storage::Sqlite(store) => { store - .cancel_workflow(workflow_id, run_id, reason, blob) + .close_workflow(command, workflow_id, run_id, reason, blob) .await } Storage::Postgres(store) => { store - .cancel_workflow(workflow_id, run_id, reason, blob) + .close_workflow(command, workflow_id, run_id, reason, blob) .await } Storage::MySql(store) => { store - .cancel_workflow(workflow_id, run_id, reason, blob) + .close_workflow(command, workflow_id, run_id, reason, blob) .await } } diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index e931a08ab..37288d86a 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -11,7 +11,7 @@ use serde::Deserialize; use serde_json::{Value, json}; use subtle::ConstantTimeEq; -use super::{Result, Runtime, refuse, store::capabilities}; +use super::{Result, Runtime, cancellation::TerminalCommand, refuse, store::capabilities}; #[derive(Clone)] pub(super) struct AuthenticatedPrincipal { @@ -78,6 +78,10 @@ pub fn router(runtime: Runtime) -> Router { ) .route("/api/workflows/{workflow_id}/runs", get(list_runs)) .route("/api/workflows/{workflow_id}/cancel", post(cancel_current)) + .route( + "/api/workflows/{workflow_id}/terminate", + post(terminate_current), + ) .route("/api/workflows/{workflow_id}/repair", post(repair_current)) .route( "/api/workflows/{workflow_id}/runs/{run_id}/repair", @@ -100,6 +104,10 @@ pub fn router(runtime: Runtime) -> Router { "/api/workflows/{workflow_id}/runs/{run_id}/cancel", post(cancel_run), ) + .route( + "/api/workflows/{workflow_id}/runs/{run_id}/terminate", + post(terminate_run), + ) .route( "/api/workflows/{workflow_id}/query/{query_name}", post(query_current), @@ -147,7 +155,11 @@ pub fn router(runtime: Runtime) -> Router { "/api/worker/registrations/{worker_id}/deregister", post(deregister_fenced), ) - .route("/api/workers/{worker_id}", delete(deregister)) + .route("/api/workers", get(list_workers)) + .route( + "/api/workers/{worker_id}", + get(describe_worker).delete(deregister), + ) .route("/api/worker/workflow-tasks/poll", post(poll_workflow)) .route( "/api/worker/workflow-tasks/{task_id}/deliver-cancellation", @@ -269,7 +281,7 @@ async fn guard(State(runtime): State, mut request: Request, next: Next) &["worker", "operator", "admin"] } else if worker { &["worker"] - } else if path.starts_with("/api/workers/") + } else if path.starts_with("/api/workers/") && request.method() != Method::GET || path == "/api/namespaces" && request.method() == Method::POST // Updating a namespace or its external storage needs an admin. || path.starts_with("/api/namespaces/") && request.method() == Method::PUT @@ -407,6 +419,13 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { if path == "/api/workflows" { return method == "POST"; } + if path == "/api/workers" + || path + .strip_prefix("/api/workers/") + .is_some_and(|id| !id.is_empty() && !id.contains('/')) + { + return method == "GET"; + } if let Some(path) = path.strip_prefix("/api/workflows/") { let parts: Vec<_> = path.split('/').collect(); if matches!( @@ -432,9 +451,14 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { } return matches!( parts.as_slice(), - [_] | [_, "cancel"] + [_] | [_, "cancel" | "terminate"] | [_, "runs", _] - | [_, "runs", _, "history" | "cancel" | "activities"] + | [ + _, + "runs", + _, + "history" | "cancel" | "terminate" | "activities" + ] ); } if let Some(path) = path.strip_prefix("/api/worker/") { @@ -596,18 +620,45 @@ async fn list_runs( Ok(Json(runtime.list_runs(&workflow).await?)) } +async fn close( + runtime: Runtime, + command: TerminalCommand, + workflow_id: &str, + run_id: Option<&str>, + body: Value, +) -> Result<(StatusCode, Json)> { + let operation = || match command { + TerminalCommand::Cancel => super::control_plane::Operation::Cancel, + TerminalCommand::Terminate => super::control_plane::Operation::Terminate, + }; + let (status, response) = runtime + .close_workflow(command, workflow_id, run_id, body) + .await + .map_err(|error| operation().refusal(error, workflow_id, run_id))?; + Ok((status, Json(response))) +} + async fn cancel_current( Extension(runtime): Extension, Path(workflow_id): Path, Json(body): Json, ) -> Result<(StatusCode, Json)> { - let (status, response) = runtime - .cancel_workflow(&workflow_id, None, body) - .await - .map_err(|error| { - super::control_plane::Operation::Cancel.refusal(error, &workflow_id, None) - })?; - Ok((status, Json(response))) + close(runtime, TerminalCommand::Cancel, &workflow_id, None, body).await +} + +async fn terminate_current( + Extension(runtime): Extension, + Path(workflow_id): Path, + Json(body): Json, +) -> Result<(StatusCode, Json)> { + close( + runtime, + TerminalCommand::Terminate, + &workflow_id, + None, + body, + ) + .await } async fn repair_current( @@ -696,13 +747,29 @@ async fn cancel_run( Path((workflow_id, run_id)): Path<(String, String)>, Json(body): Json, ) -> Result<(StatusCode, Json)> { - let (status, response) = runtime - .cancel_workflow(&workflow_id, Some(&run_id), body) - .await - .map_err(|error| { - super::control_plane::Operation::Cancel.refusal(error, &workflow_id, Some(&run_id)) - })?; - Ok((status, Json(response))) + close( + runtime, + TerminalCommand::Cancel, + &workflow_id, + Some(&run_id), + body, + ) + .await +} + +async fn terminate_run( + Extension(runtime): Extension, + Path((workflow_id, run_id)): Path<(String, String)>, + Json(body): Json, +) -> Result<(StatusCode, Json)> { + close( + runtime, + TerminalCommand::Terminate, + &workflow_id, + Some(&run_id), + body, + ) + .await } async fn request_cancellation_current( @@ -885,6 +952,24 @@ async fn worker_heartbeat( ) -> Result> { runtime.worker_heartbeat(body).await.map(Json) } +async fn list_workers( + Extension(runtime): Extension, + Query(query): Query>, +) -> Result> { + runtime + .list_workers( + query.get("task_queue").map(String::as_str), + query.get("status").map(String::as_str), + ) + .await + .map(Json) +} +async fn describe_worker( + Extension(runtime): Extension, + Path(worker_id): Path, +) -> Result> { + runtime.describe_worker(&worker_id).await.map(Json) +} async fn deregister( Extension(runtime): Extension, Path(worker_id): Path, diff --git a/rust/src/runtime/memos.rs b/rust/src/runtime/memos.rs index c44b1a975..0ddfaeb83 100644 --- a/rust/src/runtime/memos.rs +++ b/rust/src/runtime/memos.rs @@ -427,7 +427,7 @@ mod tests { .unwrap(), ); store.register(json!({"worker_id":"memo-unit-worker", "task_queue":"test", "runtime":"php", - "supported_workflow_types":["echo"], "supported_activity_types":[], "capabilities":["memo_upserts"]})).await.unwrap(); + "supported_workflow_types":["echo"], "supported_activity_types":[], "capabilities":["memo_upserts"]}), &crate::runtime::QueryTaskConfig::default()).await.unwrap(); let original = store .start( json!({"workflow_id":"memo-unit", "workflow_type":"echo", "task_queue":"test", diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 9f5766b20..2e8893020 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -34,8 +34,8 @@ mod memo_limits; mod memos; pub mod mysql; mod namespaces; -pub mod postgres; mod poll_readiness; +pub mod postgres; mod previews; mod queries; mod query_authority; @@ -66,6 +66,7 @@ mod wait_projection; mod wait_timeouts; mod worker_contracts; mod worker_incarnations; +mod worker_management; mod workflow_failures; pub use execution::Runtime; diff --git a/rust/src/runtime/query_authority.rs b/rust/src/runtime/query_authority.rs index 26bea17c4..8140262e9 100644 --- a/rust/src/runtime/query_authority.rs +++ b/rust/src/runtime/query_authority.rs @@ -13,6 +13,8 @@ pub struct QueryTaskConfig { pub lease_timeout_seconds: i64, pub receipt_ttl_seconds: i64, pub worker_stale_after_seconds: i64, + /// The worker heartbeat cadence advertised on registration and heartbeat. + pub worker_heartbeat_interval_seconds: i64, pub poll_timeout_seconds: i64, } @@ -23,6 +25,7 @@ impl Default for QueryTaskConfig { lease_timeout_seconds: 60, receipt_ttl_seconds: 180, worker_stale_after_seconds: 30, + worker_heartbeat_interval_seconds: 10, poll_timeout_seconds: 5, } } @@ -85,6 +88,8 @@ impl QueryTaskConfig { heartbeat.saturating_mul(3).max(30), )? .max(1), + // PHP clamps the advertised cadence to one second through an hour. + worker_heartbeat_interval_seconds: heartbeat.clamp(1, 3600), poll_timeout_seconds: integer( "DW_QUERY_TASK_POLL_TIMEOUT", "WORKFLOW_SERVER_QUERY_TASK_POLL_TIMEOUT", diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 31014d9f0..6059cc28c 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -796,7 +796,11 @@ where Ok(()) } - pub(crate) async fn register(&self, body: Value) -> Result { + pub(crate) async fn register( + &self, + body: Value, + config: &super::QueryTaskConfig, + ) -> Result { let worker_id = text(&body, "worker_id")?; let queue = text(&body, "task_queue")?; for field in ["supported_workflow_types", "supported_activity_types"] { @@ -813,6 +817,7 @@ where } } let runtime = text(&body, "runtime")?; + let registration = super::worker_management::Registration::parse(&body)?; let supplied_contracts = body .get("workflow_command_contracts") .map(|contracts| { @@ -821,8 +826,25 @@ where .transpose()?; let token = super::worker_incarnations::token().await?; let mut tx = self.begin().await?; - let previous = Self::query("SELECT worker_id,workflow_command_contracts FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") + let previous = Self::query("SELECT worker_id,status,workflow_command_contracts,workflow_definition_fingerprints FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") .bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut *tx).await?; + // Only an active registration guards its workflow definitions. + let current = match &previous { + Some(worker) if DB::optional_string(worker, "status")?.as_deref() == Some("active") => { + Some( + DB::optional_document_row(worker, "workflow_definition_fingerprints")? + .unwrap_or(Value::Null), + ) + } + _ => None, + }; + let supported_types: Vec = body["supported_workflow_types"] + .as_array() + .into_iter() + .flatten() + .filter_map(|t| t.as_str().map(str::to_owned)) + .collect(); + let fingerprints = registration.reconcile(current.as_ref(), &supported_types)?; let contracts = match supplied_contracts { Some(contracts) => contracts, None => { @@ -842,6 +864,8 @@ where // and lease release commit together; failure restores the old attempt. Self::query("UPDATE workflow_tasks SET status='ready',leased_at=NULL,lease_owner=NULL,lease_expires_at=NULL WHERE namespace=$1 AND lease_owner=$2 AND task_type='workflow' AND status='leased'") .bind(self.namespace.as_ref()).bind(worker_id).execute(&mut *tx).await?; + Self::release_replaced_activity_leases(&mut tx, self.namespace.as_ref(), worker_id) + .await?; } Self::query(DB::WORKER_REGISTRATION_SQL) .bind(self.namespace.as_ref()) @@ -861,28 +885,24 @@ where .await?; Self::query("UPDATE workflow_worker_registrations SET status='active',workflow_command_contracts=$1 WHERE worker_id=$2 AND namespace=$3") .bind(DB::document(&contracts)).bind(worker_id).bind(self.namespace.as_ref()).execute(&mut *tx).await?; + Self::record_registration( + &mut tx, + self.namespace.as_ref(), + worker_id, + ®istration, + &fingerprints, + ) + .await?; self.rotate_incarnation(&mut tx, worker_id, &token).await?; tx.commit().await?; self.wake.notify_waiters(); Ok( json!({"worker_id": worker_id, "registration_token":token, "registered": true, "namespace": self.namespace.as_ref(), "task_queue": queue, - "runtime": body["runtime"], "build_id": body["build_id"], "heartbeat_interval_seconds": 10, + "runtime": body["runtime"], "build_id": body["build_id"], "heartbeat_interval_seconds": config.worker_heartbeat_interval_seconds, "status": "active", "capabilities": body["capabilities"], "capability_manifest": body["capability_manifest"]}), ) } - pub(crate) async fn worker_heartbeat(&self, body: Value) -> Result { - let worker_id = text(&body, "worker_id")?; - let result = Self::query("UPDATE workflow_worker_registrations SET last_heartbeat_at=$1 WHERE worker_id=$2 AND namespace=$3") - .bind(DB::bind_time(now())).bind(worker_id).bind(self.namespace.as_ref()).execute(&self.pool).await?; - if DB::affected(result) != 1 { - return Err(refuse(StatusCode::NOT_FOUND, "worker_not_registered")); - } - Ok( - json!({"worker_id": worker_id, "heartbeat_recorded": true, "acknowledged":true, "heartbeat_interval_seconds": 10}), - ) - } - pub(crate) async fn deregister(&self, worker_id: &str) -> Result { let mut tx = self.begin().await?; let deleted = Self::query( @@ -2191,7 +2211,28 @@ where if DB::string(&attempt, "status")? != "running" || DB::number(&attempt, "attempt_number")? != DB::number(&task, "attempt_count")? { - return Err(refuse(StatusCode::CONFLICT, "stale_attempt")); + // Published PHP's stale completion receipt: nothing is recorded, + // and the owner of that attempt learns where its work stands. + let mut receipt = json!({"task_id": task_id, "activity_attempt_id": attempt_id, + "outcome": "completed", "recorded": false, "reason": "stale_attempt", "next_task_id": null}); + let owner = text(&body, "lease_owner")?; + if DB::optional_string(&attempt, "lease_owner")?.as_deref() == Some(owner) { + let activity = Self::query("SELECT status FROM activity_executions WHERE id=$1") + .bind(DB::string(&attempt, "activity_execution_id")?) + .fetch_optional(&mut *tx) + .await?; + receipt["lease_owner"] = json!(owner); + receipt["activity_status"] = match activity { + Some(activity) => json!(DB::string(&activity, "status")?), + None => Value::Null, + }; + receipt["attempt_status"] = json!(DB::string(&attempt, "status")?); + receipt["task_status"] = json!(DB::string(&task, "status")?); + } + return Err(super::RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response: receipt, + }); } let run_id = DB::string(&task, "workflow_run_id")?; let run = Self::active_run(&mut tx, &run_id).await?; diff --git a/rust/src/runtime/worker_management.rs b/rust/src/runtime/worker_management.rs new file mode 100644 index 000000000..9aa709ac6 --- /dev/null +++ b/rust/src/runtime/worker_management.rs @@ -0,0 +1,687 @@ +//! Worker registration facts and the worker management reads, as published +//! PHP's WorkerController and WorkerManagementController report them. +use super::transaction::Transaction; +use super::{ + QueryTaskConfig, Result, RuntimeError, + backend::Backend, + refuse, + store::{Store, now}, +}; +use axum::http::StatusCode; +use chrono::{DateTime, SecondsFormat, Utc}; +use serde_json::{Map, Value, json}; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; +use std::collections::BTreeMap; + +const PROCESS_METRICS: [&str; 6] = [ + "cpu_percent", + "memory_bytes", + "process_uptime_seconds", + "process_id", + "host", + "process_started_at", +]; +/// A nullable count column: its field in task_slots, then the static +/// statements that set it and clear it. Values are always bound. +struct Count { + field: &'static str, + set: &'static str, + clear: &'static str, +} +const SLOTS: [Count; 3] = [ + Count { + field: "workflow_available", + set: "UPDATE workflow_worker_registrations SET available_workflow_slots=$1 WHERE worker_id=$2 AND namespace=$3", + clear: "UPDATE workflow_worker_registrations SET available_workflow_slots=NULL WHERE worker_id=$1 AND namespace=$2", + }, + Count { + field: "activity_available", + set: "UPDATE workflow_worker_registrations SET available_activity_slots=$1 WHERE worker_id=$2 AND namespace=$3", + clear: "UPDATE workflow_worker_registrations SET available_activity_slots=NULL WHERE worker_id=$1 AND namespace=$2", + }, + Count { + field: "session_available", + set: "UPDATE workflow_worker_registrations SET available_session_slots=$1 WHERE worker_id=$2 AND namespace=$3", + clear: "UPDATE workflow_worker_registrations SET available_session_slots=NULL WHERE worker_id=$1 AND namespace=$2", + }, +]; +const INTERVAL: Count = Count { + field: "heartbeat_interval_seconds", + set: "UPDATE workflow_worker_registrations SET heartbeat_interval_seconds=$1 WHERE worker_id=$2 AND namespace=$3", + clear: "UPDATE workflow_worker_registrations SET heartbeat_interval_seconds=NULL WHERE worker_id=$1 AND namespace=$2", +}; + +fn invalid() -> RuntimeError { + refuse( + StatusCode::UNPROCESSABLE_ENTITY, + "invalid_worker_definition", + ) +} + +/// Laravel's `integer` rule: a JSON integer or a string of digits. +fn integer(value: &Value) -> Option { + value.as_i64().or_else(|| { + let text = value.as_str()?; + let digits = text.strip_prefix('-').unwrap_or(text); + (!digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit())) + .then(|| text.parse().ok()) + .flatten() + }) +} + +/// A nullable integer field with inclusive bounds; absent and null are None. +fn bounded(body: &Value, field: &str, min: i64, max: i64) -> Result> { + match body.get(field) { + None | Some(Value::Null) => Ok(None), + Some(value) => integer(value) + .filter(|n| (min..=max).contains(n)) + .map(Some) + .ok_or_else(invalid), + } +} + +/// PHP's boundedSlotCount: only an integer counts, clamped to the capacity. +fn slot(value: &Value, capacity: i64) -> Option { + let count = value.as_i64().or_else(|| { + let text = value.as_str()?; + (!text.is_empty() && text.bytes().all(|b| b.is_ascii_digit())) + .then(|| text.parse().ok()) + .flatten() + })?; + Some(count.max(0).min(capacity.max(0))) +} + +/// Validates the process metrics a worker reports and keeps PHP's normalized +/// subset. Absent means "not reported"; the outer Option is presence. +fn process_metrics(body: &Value) -> Result>> { + let Some(value) = body.get("process_metrics") else { + return Ok(None); + }; + if value.is_null() { + return Ok(Some(None)); + } + // A JSON list is a PHP array too; it holds none of the named metrics. + let empty = Map::new(); + let metrics = match value { + Value::Object(metrics) => metrics, + Value::Array(_) => &empty, + _ => return Err(invalid()), + }; + let mut normalized = Map::new(); + for key in PROCESS_METRICS { + let Some(entry) = metrics.get(key).filter(|v| !v.is_null()) else { + continue; + }; + match key { + "host" | "process_started_at" => { + let limit = if key == "host" { 255 } else { 64 }; + let text = entry.as_str().ok_or_else(invalid)?; + if text.chars().count() > limit { + return Err(invalid()); + } + let trimmed = text.trim(); + if !trimmed.is_empty() { + normalized.insert( + key.into(), + json!(trimmed.chars().take(255).collect::()), + ); + } + } + "cpu_percent" => { + // PHP validates a numeric string but stores only a number. + let number = entry + .as_f64() + .or_else(|| entry.as_str()?.trim().parse::().ok()) + .filter(|n| n.is_finite() && *n >= 0.0) + .ok_or_else(invalid)?; + if entry.is_number() { + normalized.insert(key.into(), json!(number)); + } + } + _ => { + let number = integer(entry).filter(|n| *n >= 0).ok_or_else(invalid)?; + normalized.insert(key.into(), json!(number)); + } + } + } + Ok(Some( + (!normalized.is_empty()).then(|| Value::Object(normalized)), + )) +} + +/// PHP's workflowDefinitionFingerprints: trimmed non-empty string pairs. +fn fingerprints(value: &Value) -> BTreeMap { + let pairs: Vec<(String, &Value)> = match value { + Value::Object(map) => map.iter().map(|(k, v)| (k.clone(), v)).collect(), + Value::Array(list) => list + .iter() + .enumerate() + .map(|(i, v)| (i.to_string(), v)) + .collect(), + _ => Vec::new(), + }; + pairs + .into_iter() + .filter_map(|(workflow_type, fingerprint)| { + let workflow_type = workflow_type.trim(); + let fingerprint = fingerprint.as_str()?.trim(); + (!workflow_type.is_empty() && !fingerprint.is_empty()) + .then(|| (workflow_type.to_owned(), fingerprint.to_owned())) + }) + .collect() +} + +/// Published PHP stores and renders an empty map as a JSON list. +fn php_map(map: &BTreeMap) -> Value { + if map.is_empty() { + json!([]) + } else { + json!(map) + } +} + +fn definition_refusal(reason: &str, workflow_type: &str) -> RuntimeError { + let (error, remediation) = if reason == "workflow_definition_changed" { + ( + "Worker attempted to re-register a changed workflow definition.", + "Restart the worker with a new worker_id before registering a changed workflow class definition.", + ) + } else { + ( + "Worker omitted a previously advertised workflow definition fingerprint.", + "Advertise the matching definition fingerprint or restart with a new worker_id.", + ) + }; + RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response: json!({"error": error, "reason": reason, "workflow_type": workflow_type, + "remediation": remediation}), + } +} + +/// The registration facts PHP keeps beside a worker's identity. +pub(super) struct Registration { + fingerprints: Option>, + capacity: [i64; 3], + slots: [Option; 3], + process_metrics: Option, + heartbeat_interval: Option, +} + +impl Registration { + pub(super) fn parse(body: &Value) -> Result { + let fingerprints = match body.get("workflow_definition_fingerprints") { + None => None, + Some(Value::Null) => Some(BTreeMap::new()), + Some(value @ (Value::Object(_) | Value::Array(_))) => { + let entries: Vec<&Value> = match value { + Value::Object(map) => map.values().collect(), + Value::Array(list) => list.iter().collect(), + _ => unreachable!(), + }; + if entries + .iter() + .any(|v| v.as_str().is_none_or(|s| s.chars().count() > 255)) + { + return Err(invalid()); + } + Some(fingerprints(value)) + } + Some(_) => return Err(invalid()), + }; + let capacity = [ + bounded( + body, + "max_concurrent_workflow_tasks", + 0, + i64::from(u32::MAX), + )? + .unwrap_or(100), + bounded( + body, + "max_concurrent_activity_tasks", + 0, + i64::from(u32::MAX), + )? + .unwrap_or(100), + bounded( + body, + "max_concurrent_worker_sessions", + 1, + i64::from(u32::MAX), + )? + .unwrap_or(10), + ]; + if capacity[0] == 0 && capacity[1] == 0 { + return Err(invalid()); + } + let slots = task_slots(body, capacity)?.unwrap_or([None; 3]); + Ok(Self { + fingerprints, + capacity, + slots, + process_metrics: process_metrics(body)?.flatten(), + heartbeat_interval: bounded(body, "heartbeat_interval_seconds", 1, 3600)?, + }) + } + + /// Re-registering an active worker may not change a workflow definition, + /// nor omit one it advertised. Returns the fingerprints to store. + pub(super) fn reconcile( + &self, + current: Option<&Value>, + supported_types: &[String], + ) -> Result { + let mut incoming = self.fingerprints.clone().unwrap_or_default(); + if let Some(current) = current.map(fingerprints) { + if self.fingerprints.is_some() { + for workflow_type in supported_types { + if current.contains_key(workflow_type) && !incoming.contains_key(workflow_type) + { + return Err(definition_refusal( + "workflow_definition_fingerprint_missing", + workflow_type, + )); + } + } + } + for (workflow_type, fingerprint) in &incoming { + if current.get(workflow_type).is_some_and(|c| c != fingerprint) { + return Err(definition_refusal( + "workflow_definition_changed", + workflow_type, + )); + } + } + for (workflow_type, fingerprint) in current { + if supported_types.iter().any(|t| t.trim() == workflow_type) { + incoming.entry(workflow_type).or_insert(fingerprint); + } + } + } + Ok(php_map(&incoming)) + } +} + +/// task_slots as PHP validates it, each count bounded by its capacity. +fn task_slots(body: &Value, capacity: [i64; 3]) -> Result; 3]>> { + let slots = match body.get("task_slots") { + None | Some(Value::Null) => return Ok(None), + Some(Value::Object(slots)) => slots, + // A JSON list is a PHP array without the named slots. + Some(Value::Array(_)) => return Ok(Some([None; 3])), + Some(_) => return Err(invalid()), + }; + let mut bounded = [None; 3]; + for (index, count) in SLOTS.iter().enumerate() { + match slots.get(count.field) { + None | Some(Value::Null) => {} + Some(value) => { + if integer(value).is_none_or(|n| n < 0) { + return Err(invalid()); + } + bounded[index] = slot(value, capacity[index]); + } + } + } + Ok(Some(bounded)) +} + +fn time(at: Option>) -> Value { + json!(at.map(|at| at.to_rfc3339_opts(SecondsFormat::Micros, true))) +} + +fn status_filter(status: Option<&str>) -> Option { + status + .map(|s| s.trim().to_lowercase()) + .filter(|s| !s.is_empty()) +} + +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, +{ + /// Stores the facts a registration reports. Slots and process metrics it + /// does not report are cleared, as PHP's updateOrCreate does. + pub(super) async fn record_registration( + tx: &mut Transaction<'_, DB>, + namespace: &str, + worker_id: &str, + registration: &Registration, + fingerprints: &Value, + ) -> Result<()> { + let [workflow, activity, sessions] = registration.capacity; + Self::query("UPDATE workflow_worker_registrations SET workflow_definition_fingerprints=$1,max_concurrent_workflow_tasks=$2,max_concurrent_activity_tasks=$3,max_concurrent_worker_sessions=$4,process_metrics=$5 WHERE worker_id=$6 AND namespace=$7") + .bind(DB::document(fingerprints)).bind(workflow).bind(activity).bind(sessions) + .bind(registration.process_metrics.as_ref().map(DB::document)) + .bind(worker_id).bind(namespace) + .execute(&mut **tx).await?; + for (count, value) in SLOTS.iter().zip(registration.slots) { + Self::set_count(tx, count, value, worker_id, namespace).await?; + } + Self::set_count( + tx, + &INTERVAL, + registration.heartbeat_interval, + worker_id, + namespace, + ) + .await + } + + /// A new process registering a worker_id that holds activity leases + /// replaces a process that may have died mid-task. As PHP does, close its + /// running attempts and return the tasks at once, rather than waiting out + /// the activity lease. + pub(super) async fn release_replaced_activity_leases( + tx: &mut Transaction<'_, DB>, + namespace: &str, + worker_id: &str, + ) -> Result<()> { + let tasks = Self::query("SELECT id,payload FROM workflow_tasks WHERE namespace=$1 AND task_type='activity' AND status='leased' AND lease_owner=$2 ORDER BY id") + .bind(namespace).bind(worker_id).fetch_all(&mut **tx).await?; + for task in tasks { + let task_id = DB::string(&task, "id")?; + let execution = DB::optional_document_row(&task, "payload")? + .and_then(|payload| payload["activity_execution_id"].as_str().map(str::to_owned)) + .filter(|execution| !execution.is_empty()); + if let Some(execution) = execution { + let attempt = Self::query("SELECT a.id FROM activity_attempts a JOIN activity_executions e ON e.id=a.activity_execution_id WHERE a.workflow_task_id=$1 AND a.activity_execution_id=$2 AND a.lease_owner=$3 AND a.status='running' ORDER BY a.attempt_number DESC LIMIT 1") + .bind(&task_id).bind(&execution).bind(worker_id).fetch_optional(&mut **tx).await?; + if let Some(attempt) = attempt { + Self::query("UPDATE activity_attempts SET status='expired',lease_expires_at=NULL,closed_at=COALESCE(closed_at,$1) WHERE id=$2") + .bind(DB::bind_time(now())).bind(DB::string(&attempt, "id")?) + .execute(&mut **tx).await?; + } + } + Self::query("UPDATE workflow_tasks SET status='ready',leased_at=NULL,lease_owner=NULL,lease_expires_at=NULL,last_claim_failed_at=NULL,last_claim_error=NULL WHERE id=$1") + .bind(&task_id).execute(&mut **tx).await?; + } + Ok(()) + } + + async fn set_count( + tx: &mut Transaction<'_, DB>, + count: &Count, + value: Option, + worker_id: &str, + namespace: &str, + ) -> Result<()> { + match value { + Some(value) => { + Self::query(count.set) + .bind(value) + .bind(worker_id) + .bind(namespace) + .execute(&mut **tx) + .await? + } + None => { + Self::query(count.clear) + .bind(worker_id) + .bind(namespace) + .execute(&mut **tx) + .await? + } + }; + Ok(()) + } + + pub(crate) async fn worker_heartbeat( + &self, + body: Value, + config: &QueryTaskConfig, + ) -> Result { + let worker_id = super::text(&body, "worker_id")?; + let metrics = process_metrics(&body)?; + let interval = bounded(&body, "heartbeat_interval_seconds", 1, 3600)?; + let mut tx = self.begin().await?; + let Some(worker) = Self::query("SELECT max_concurrent_workflow_tasks,max_concurrent_activity_tasks,max_concurrent_worker_sessions FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2") + .bind(worker_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await? + else { + return Err(RuntimeError::Protocol { + status: StatusCode::NOT_FOUND, + response: json!({"error": "Worker not registered.", "reason": "worker_not_registered", + "worker_id": worker_id}), + }); + }; + let capacity = [ + DB::number(&worker, "max_concurrent_workflow_tasks")?, + DB::number(&worker, "max_concurrent_activity_tasks")?, + DB::number(&worker, "max_concurrent_worker_sessions")?, + ]; + Self::query("UPDATE workflow_worker_registrations SET last_heartbeat_at=$1,updated_at=$2 WHERE worker_id=$3 AND namespace=$4") + .bind(DB::bind_time(now())).bind(DB::bind_time(now())).bind(worker_id).bind(self.namespace.as_ref()) + .execute(&mut *tx).await?; + // Only the slots, metrics and cadence the heartbeat reports change. + let reported = task_slots(&body, capacity)?; + let namespace = self.namespace.as_ref(); + if let (Some(bounded), Some(slots)) = (reported, body["task_slots"].as_object()) { + for (count, value) in SLOTS.iter().zip(bounded) { + if slots.contains_key(count.field) { + Self::set_count(&mut tx, count, value, worker_id, namespace).await?; + } + } + } + if let Some(metrics) = metrics { + Self::query("UPDATE workflow_worker_registrations SET process_metrics=$1 WHERE worker_id=$2 AND namespace=$3") + .bind(metrics.as_ref().map(DB::document)).bind(worker_id).bind(namespace) + .execute(&mut *tx).await?; + } + if interval.is_some() { + Self::set_count(&mut tx, &INTERVAL, interval, worker_id, namespace).await?; + } + tx.commit().await?; + // PHP also reports an inline history retention pass here. Native has + // no retention yet, so it does not claim one. + Ok(json!({"worker_id": worker_id, "acknowledged": true, + "heartbeat_interval_seconds": config.worker_heartbeat_interval_seconds, + "stale_after_seconds": config.worker_stale_after_seconds})) + } + + fn worker_summary(row: &DB::Row, stale_after: i64, at: DateTime) -> Result { + let document = |field: &str| -> Result { + Ok(DB::optional_document_row(row, field)? + .filter(|v| !v.is_null()) + .unwrap_or_else(|| json!([]))) + }; + let empty_map_as_list = |value: Value| match value { + Value::Object(map) if map.is_empty() => json!([]), + other => other, + }; + let capabilities: Vec = document("capabilities")? + .as_array() + .map(|flags| { + flags + .iter() + .filter_map(|f| f.as_str().map(str::trim).filter(|f| !f.is_empty())) + .map(str::to_owned) + .collect() + }) + .unwrap_or_default(); + let heartbeat = DB::optional_instant(row, "last_heartbeat_at")?; + let stale = heartbeat.is_none_or(|h| h < at - chrono::Duration::seconds(stale_after)); + let stored = DB::optional_string(row, "status")? + .map(|s| s.trim().to_lowercase()) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "active".into()); + let capacity = [ + DB::number(row, "max_concurrent_workflow_tasks")?, + DB::number(row, "max_concurrent_activity_tasks")?, + DB::number(row, "max_concurrent_worker_sessions")?, + ]; + Ok(json!({ + "worker_id": DB::string(row, "worker_id")?, + "namespace": DB::string(row, "namespace")?, + "task_queue": DB::string(row, "task_queue")?, + "runtime": DB::string(row, "runtime")?, + "sdk_version": DB::optional_string(row, "sdk_version")?, + "build_id": DB::optional_string(row, "build_id")?, + "supported_workflow_types": document("supported_workflow_types")?, + "workflow_definition_fingerprints": empty_map_as_list(document("workflow_definition_fingerprints")?), + "workflow_command_contracts": empty_map_as_list(document("workflow_command_contracts")?), + "supported_activity_types": document("supported_activity_types")?, + "capabilities": capabilities, + "capability_manifest": document("capability_manifest")?, + "max_concurrent_workflow_tasks": capacity[0], + "max_concurrent_activity_tasks": capacity[1], + "max_concurrent_worker_sessions": capacity[2], + "task_slots": { + "workflow_available": DB::optional_number(row, "available_workflow_slots")?, + "activity_available": DB::optional_number(row, "available_activity_slots")?, + "session_available": DB::optional_number(row, "available_session_slots")?, + "workflow_capacity": capacity[0], + "activity_capacity": capacity[1], + "session_capacity": capacity[2], + }, + "process_metrics": DB::optional_document_row(row, "process_metrics")?.filter(|v| !v.is_null()), + "heartbeat_interval_seconds": DB::optional_number(row, "heartbeat_interval_seconds")?, + "status": if stale { "stale".to_owned() } else { stored }, + "last_heartbeat_at": time(heartbeat), + "registered_at": time(DB::optional_instant(row, "created_at")?), + })) + } + + pub(crate) async fn list_workers( + &self, + task_queue: Option<&str>, + status: Option<&str>, + config: &QueryTaskConfig, + ) -> Result { + let stale_after = config.worker_stale_after_seconds.max(1); + let at = now(); + // PHP filters only on a truthy queue, and "0" is not truthy. + let queue = task_queue.filter(|q| !q.is_empty() && *q != "0"); + let rows = match queue { + Some(queue) => Self::query("SELECT * FROM workflow_worker_registrations WHERE namespace=$1 AND task_queue=$2 ORDER BY last_heartbeat_at DESC") + .bind(self.namespace.as_ref()).bind(queue).fetch_all(&self.pool).await?, + None => Self::query("SELECT * FROM workflow_worker_registrations WHERE namespace=$1 ORDER BY last_heartbeat_at DESC") + .bind(self.namespace.as_ref()).fetch_all(&self.pool).await?, + }; + let filter = status_filter(status); + let mut workers = Vec::new(); + for row in &rows { + let summary = Self::worker_summary(row, stale_after, at)?; + let stale = summary["status"] == "stale"; + let keep = match filter.as_deref() { + None => !stale, + Some("stale") => stale, + Some(wanted) => !stale && summary["status"] == wanted, + }; + if keep { + workers.push(summary); + } + } + Ok(json!({"workers": workers, "stale_after_seconds": stale_after})) + } + + pub(crate) async fn describe_worker( + &self, + worker_id: &str, + config: &QueryTaskConfig, + ) -> Result { + let stale_after = config.worker_stale_after_seconds.max(1); + let Some(row) = Self::query( + "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2", + ) + .bind(worker_id) + .bind(self.namespace.as_ref()) + .fetch_optional(&self.pool) + .await? + else { + return Err(RuntimeError::Protocol { + status: StatusCode::NOT_FOUND, + response: json!({"message": format!("Worker [{worker_id}] not found in namespace [{}].", self.namespace), + "reason": "worker_not_found"}), + }); + }; + let mut summary = Self::worker_summary(&row, stale_after, now())?; + summary["updated_at"] = time(DB::optional_instant(&row, "updated_at")?); + summary["stale_after_seconds"] = json!(stale_after); + Ok(summary) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registration_keeps_php_facts_and_bounds_slots() { + let parsed = Registration::parse(&json!({"max_concurrent_workflow_tasks": 3, + "max_concurrent_activity_tasks": "5", "task_slots": {"workflow_available": 2, + "activity_available": 9, "session_available": "1"}, "process_metrics": {"process_id": 7, + "process_uptime_seconds": "0", "cpu_percent": 1.5, "host": " h ", "invented": 1}, + "heartbeat_interval_seconds": 15, "workflow_definition_fingerprints": {" a ": " f ", "b": " "}})) + .unwrap(); + assert_eq!(parsed.capacity, [3, 5, 10]); + assert_eq!(parsed.slots, [Some(2), Some(5), Some(1)]); + assert_eq!( + parsed.process_metrics, + Some( + json!({"cpu_percent": 1.5, "process_uptime_seconds": 0, "process_id": 7, "host": "h"}) + ) + ); + assert_eq!(parsed.heartbeat_interval, Some(15)); + assert_eq!( + parsed.fingerprints, + Some(BTreeMap::from([("a".into(), "f".into())])) + ); + for invalid in [ + json!({"max_concurrent_workflow_tasks": 0, "max_concurrent_activity_tasks": 0}), + json!({"max_concurrent_worker_sessions": 0}), + json!({"task_slots": {"workflow_available": -1}}), + json!({"heartbeat_interval_seconds": 3601}), + json!({"process_metrics": {"memory_bytes": -1}}), + json!({"workflow_definition_fingerprints": {"a": 1}}), + ] { + assert!(Registration::parse(&invalid).is_err(), "{invalid}"); + } + let defaults = Registration::parse(&json!({})).unwrap(); + assert_eq!(defaults.capacity, [100, 100, 10]); + assert_eq!(defaults.slots, [None; 3]); + assert_eq!(defaults.process_metrics, None); + } + + #[test] + fn re_registration_guards_and_keeps_advertised_fingerprints() { + let types = ["a".to_owned(), "b".to_owned()]; + let current = json!({"a": "1", "b": "2", "gone": "3"}); + let reason = |body: Value| match Registration::parse(&body) + .unwrap() + .reconcile(Some(¤t), &types) + { + Err(RuntimeError::Protocol { response, .. }) => ( + response["reason"].clone(), + response["workflow_type"].clone(), + ), + other => panic!("{:?}", other.map(|_| ())), + }; + assert_eq!( + reason(json!({"workflow_definition_fingerprints": {"a": "1", "b": "changed"}})), + (json!("workflow_definition_changed"), json!("b")) + ); + assert_eq!( + reason(json!({"workflow_definition_fingerprints": {"a": "1"}})), + (json!("workflow_definition_fingerprint_missing"), json!("b")) + ); + let omitted = Registration::parse(&json!({})).unwrap(); + assert_eq!( + omitted.reconcile(Some(¤t), &types).unwrap(), + json!({"a": "1", "b": "2"}) + ); + assert_eq!(omitted.reconcile(None, &types).unwrap(), json!([])); + assert_eq!( + omitted.reconcile(Some(&json!([])), &types).unwrap(), + json!([]) + ); + } +} diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 8acc969bc..79bbb11f0 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -55,6 +55,9 @@ mod worker_contracts; #[path = "execution/worker_deregistration.rs"] mod worker_deregistration; +#[path = "execution/worker_listing.rs"] +mod worker_listing; + #[path = "execution/worker_pressure.rs"] mod worker_pressure; @@ -271,7 +274,11 @@ impl TestDatabase { } async fn cancellation_failure_count(&self, run: &str, failure: &str) -> i64 { - let query = "SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$1 AND source_id=$2 AND id=$3 AND source_kind='workflow_run' AND propagation_kind='cancelled' AND failure_category='cancelled' AND handled=false"; + self.terminal_failure_count(run, failure, "cancelled").await + } + + async fn terminal_failure_count(&self, run: &str, failure: &str, kind: &str) -> i64 { + let query = "SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$1 AND source_id=$2 AND id=$3 AND source_kind='workflow_run' AND propagation_kind=$4 AND failure_category=$5 AND handled=false"; match self { Self::Sqlite(dir) => { let pool = sqlx::SqlitePool::connect(&format!( @@ -284,6 +291,8 @@ impl TestDatabase { .bind(run) .bind(run) .bind(failure) + .bind(kind) + .bind(kind) .fetch_one(&pool) .await .unwrap(); @@ -299,6 +308,8 @@ impl TestDatabase { .bind(run) .bind(run) .bind(failure) + .bind(kind) + .bind(kind) .fetch_one(&pool) .await .unwrap(); @@ -310,8 +321,8 @@ impl TestDatabase { .connect_with(options.as_ref().clone()) .await .unwrap(); - let count = sqlx::query_scalar("SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=? AND source_id=? AND id=? AND source_kind='workflow_run' AND propagation_kind='cancelled' AND failure_category='cancelled' AND handled=false") - .bind(run).bind(run).bind(failure).fetch_one(&pool).await.unwrap(); + let count = sqlx::query_scalar("SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=? AND source_id=? AND id=? AND source_kind='workflow_run' AND propagation_kind=? AND failure_category=? AND handled=false") + .bind(run).bind(run).bind(failure).bind(kind).bind(kind).fetch_one(&pool).await.unwrap(); pool.close().await; count } @@ -1406,6 +1417,27 @@ async fn direct_child_cancellation_preserves_original_failure_and_one_parent_res child_id.as_str().unwrap(), child_run.as_str().unwrap() ); + // A terminated child would resolve its parent with ChildRunTerminated, + // which is not qualified yet. The refusal leaves both runs unchanged. + let before = run_history(&app, &child_id, &child_run).await; + let terminate = request( + &app, + "POST", + &path.replace("/cancel", "/terminate"), + json!({"reason":"direct child termination"}), + ) + .await; + assert_eq!( + terminate.0, + StatusCode::UNPROCESSABLE_ENTITY, + "{}", + terminate.1 + ); + assert_eq!( + terminate.1["reason"], + "termination_propagation_not_available" + ); + assert_eq!(run_history(&app, &child_id, &child_run).await, before); let accepted = request( &app, "POST", @@ -3794,8 +3826,26 @@ async fn cooperative_delivery_cleanup_and_original_deadline_survive_fresh_runtim } #[tokio::test] -async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_pools() { - for phase in ["before_claim", "pending_timer", "leased_activity"] { +async fn immediate_cancellation_and_termination_close_original_work_once_and_survive_fresh_pools() { + for (command, phase) in ["cancel", "terminate"].into_iter().flat_map(|command| { + ["before_claim", "pending_timer", "leased_activity"].map(|phase| (command, phase)) + }) { + // Published PHP runs both commands through one implementation. + let (closed, requested_event, terminal_event, exception) = if command == "cancel" { + ( + "cancelled", + "CancelRequested", + "WorkflowCancelled", + "Workflow\\V2\\Exceptions\\WorkflowCancelledException", + ) + } else { + ( + "terminated", + "TerminateRequested", + "WorkflowTerminated", + "Workflow\\V2\\Exceptions\\WorkflowTerminatedException", + ) + }; let database = TestDatabase::new().await; let runtime_a = database.open().await.unwrap(); let runtime_b = database.open().await.unwrap(); @@ -3819,18 +3869,18 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo let workflow = json!(phase); let run = started["run_id"].clone(); let path = format!( - "/api/workflows/{phase}/runs/{}/cancel", + "/api/workflows/{phase}/runs/{}/{command}", run.as_str().unwrap() ); let mut activity_task = Value::Null; if phase != "before_claim" { let task = poll(&app_a, "cancel-a", "workflow").await; - let command = if phase == "pending_timer" { + let scheduling = if phase == "pending_timer" { json!({"type":"start_timer","delay_seconds":60}) } else { json!({"type":"schedule_activity","activity_type":"echo-activity","arguments":envelope(Payload::Array(vec![Payload::Long(9007199254740993)]))}) }; - let completed = finish_task(&app_a, &task, json!([command])).await; + let completed = finish_task(&app_a, &task, json!([scheduling])).await; assert_eq!(completed.0, StatusCode::OK, "{}", completed.1); if phase == "leased_activity" { activity_task = poll(&app_b, "cancel-b", "activity").await; @@ -3841,7 +3891,7 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo let wrong_run = request( &app_b, "POST", - &format!("/api/workflows/{phase}/runs/another-run/cancel"), + &format!("/api/workflows/{phase}/runs/another-run/{command}"), json!({"reason":"replacement"}), ) .await; @@ -3878,7 +3928,7 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo } else { right.1 }; - assert_eq!(accepted["outcome"], "cancelled"); + assert_eq!(accepted["outcome"], closed); let history = run_history(&app_a, &workflow, &run).await; let events = history.as_array().unwrap(); assert_eq!( @@ -3887,8 +3937,15 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo ); let requested = &events[prefix.as_array().unwrap().len()]; let terminal = events.last().unwrap(); - assert_eq!(requested["event_type"], "CancelRequested"); - assert_eq!(terminal["event_type"], "WorkflowCancelled"); + assert_eq!(requested["event_type"], requested_event); + assert_eq!(requested["payload"]["command_type"], command); + assert_eq!(terminal["event_type"], terminal_event); + assert_eq!(terminal["payload"]["closed_reason"], closed); + assert_eq!(terminal["payload"]["failure_category"], closed); + assert_eq!( + terminal["payload"]["message"], + format!("Workflow {closed}: parity cancellation λ") + ); assert_eq!( requested["payload"]["workflow_command_id"], accepted["command_id"] @@ -3898,15 +3955,13 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo accepted["command_id"] ); assert_eq!(terminal["payload"]["reason"], "parity cancellation λ"); - assert_eq!( - terminal["payload"]["exception_class"], - "Workflow\\V2\\Exceptions\\WorkflowCancelledException" - ); + assert_eq!(terminal["payload"]["exception_class"], exception); assert_eq!( database - .cancellation_failure_count( + .terminal_failure_count( run.as_str().unwrap(), - terminal["payload"]["failure_id"].as_str().unwrap() + terminal["payload"]["failure_id"].as_str().unwrap(), + closed ) .await, 1 @@ -3945,7 +4000,8 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo ); let stale = request(&app_a, "POST", &stale_path, stale_body.clone()).await; assert_eq!(stale.0, StatusCode::CONFLICT, "{}", stale.1); - assert_eq!(stale.1["reason"], "run_cancelled"); + assert_eq!(stale.1["reason"], format!("run_{closed}")); + assert_eq!(stale.1["run_status"], closed); assert_eq!(stale.1["recorded"], false); assert_eq!(stale.1["task_id"], activity_task["task_id"]); assert_eq!( @@ -3974,10 +4030,10 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo ) .await .1; - assert_eq!(description["status"], "cancelled"); + assert_eq!(description["status"], closed); assert_eq!(description["status_bucket"], "failed"); assert_eq!(description["is_terminal"], true); - assert_eq!(description["closed_reason"], "cancelled"); + assert_eq!(description["closed_reason"], closed); assert_eq!( ValueCodec::new() .unwrap() @@ -3996,9 +4052,10 @@ async fn immediate_cancellation_closes_original_work_once_and_survives_fresh_poo assert_eq!(run_history(&app_c, &workflow, &run).await, history); assert_eq!( database - .cancellation_failure_count( + .terminal_failure_count( run.as_str().unwrap(), - terminal["payload"]["failure_id"].as_str().unwrap() + terminal["payload"]["failure_id"].as_str().unwrap(), + closed ) .await, 1 diff --git a/rust/tests/execution/namespaces.rs b/rust/tests/execution/namespaces.rs index 2d20f8d0d..986aa32f3 100644 --- a/rust/tests/execution/namespaces.rs +++ b/rust/tests/execution/namespaces.rs @@ -377,6 +377,7 @@ async fn named_namespaces_preserve_original_runs_and_global_workflow_id_reservat ("GET", path.clone(), "run_not_found"), ("GET", format!("{path}/history"), "run_not_found"), ("POST", format!("{path}/cancel"), "instance_not_found"), + ("POST", format!("{path}/terminate"), "instance_not_found"), ] { let (status, body) = scoped( &app, diff --git a/rust/tests/execution/worker_listing.rs b/rust/tests/execution/worker_listing.rs new file mode 100644 index 000000000..77ea5d0dd --- /dev/null +++ b/rust/tests/execution/worker_listing.rs @@ -0,0 +1,353 @@ +use super::*; + +async fn worker_request( + app: &Router, + method: &str, + path: &str, + body: Value, +) -> (StatusCode, Value) { + request(app, method, path, body).await +} + +fn registration(worker: &str, queue: &str, extra: Value) -> Value { + let mut body = json!({"worker_id": worker, "task_queue": queue, "runtime": "php", + "sdk_version": "durable-workflow-php/2.2.6", "supported_workflow_types": ["listing"], + "supported_activity_types": ["listing-activity"]}); + for (key, value) in extra.as_object().unwrap() { + body[key] = value.clone(); + } + body +} + +async fn ours(app: &Router, query: &str) -> Vec { + let (status, body) = request(app, "GET", &format!("/api/workers{query}"), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["stale_after_seconds"], 30); + let mut workers: Vec = body["workers"].as_array().unwrap().clone(); + workers.sort_by(|a, b| a["worker_id"].as_str().cmp(&b["worker_id"].as_str())); + workers +} + +#[tokio::test] +async fn workers_are_listed_with_their_registration_and_heartbeat_facts() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let fingerprint = "sha256:".to_owned() + &"1".repeat(64); + let first = registration( + "listing-first", + "queue-a", + json!({ + "max_concurrent_workflow_tasks": 3, "max_concurrent_activity_tasks": 5, + "max_concurrent_worker_sessions": 2, "build_id": "build-λ", + "workflow_definition_fingerprints": {"listing": fingerprint}}), + ); + let (status, receipt) = + worker_request(&app, "POST", "/api/worker/register", first.clone()).await; + assert_eq!(status, StatusCode::CREATED, "{receipt}"); + assert_eq!(receipt["heartbeat_interval_seconds"], 10); + let second = registration( + "listing-second", + "queue-b", + json!({"supported_activity_types": []}), + ); + assert_eq!( + worker_request(&app, "POST", "/api/worker/register", second) + .await + .0, + StatusCode::CREATED + ); + let (status, heartbeat) = worker_request(&app, "POST", "/api/worker/heartbeat", json!({ + "worker_id": "listing-first", "task_slots": {"workflow_available": 2, "activity_available": 9, + "session_available": 1}, "process_metrics": {"process_id": 17, "process_uptime_seconds": 0}})).await; + assert_eq!(status, StatusCode::OK, "{heartbeat}"); + assert_eq!(heartbeat["acknowledged"], true); + assert_eq!(heartbeat["stale_after_seconds"], 30); + assert_eq!(heartbeat["heartbeat_interval_seconds"], 10); + + let listed = ours(&app, "").await; + assert_eq!(listed.len(), 2); + let worker = &listed[0]; + assert_eq!(worker["worker_id"], "listing-first"); + assert_eq!( + worker["workflow_definition_fingerprints"], + json!({"listing": fingerprint}) + ); + assert_eq!(worker["workflow_command_contracts"], json!([])); + assert_eq!(worker["build_id"], "build-λ"); + assert_eq!( + worker["task_slots"], + json!({"workflow_available": 2, "activity_available": 5, + "session_available": 1, "workflow_capacity": 3, "activity_capacity": 5, "session_capacity": 2}) + ); + assert_eq!( + worker["process_metrics"], + json!({"process_uptime_seconds": 0, "process_id": 17}) + ); + assert_eq!(worker["status"], "active"); + assert_eq!(worker["heartbeat_interval_seconds"], Value::Null); + let defaults = &listed[1]; + assert_eq!(defaults["workflow_definition_fingerprints"], json!([])); + assert_eq!(defaults["supported_activity_types"], json!([])); + assert_eq!( + defaults["task_slots"], + json!({"workflow_available": null, "activity_available": null, + "session_available": null, "workflow_capacity": 100, "activity_capacity": 100, "session_capacity": 10}) + ); + assert_eq!(defaults["process_metrics"], Value::Null); + assert_eq!(ours(&app, "?task_queue=queue-a").await.len(), 1); + assert_eq!( + ours(&app, "?task_queue=0").await.len(), + 2, + "PHP ignores a falsy queue" + ); + assert_eq!(ours(&app, "?status=%20ACTIVE%20").await.len(), 2); + assert!(ours(&app, "?status=stale").await.is_empty()); + assert!(ours(&app, "?status=draining").await.is_empty()); + + let (status, described) = request(&app, "GET", "/api/workers/listing-first", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{described}"); + assert_eq!(described["stale_after_seconds"], 30); + assert!(described["updated_at"].as_str().is_some()); + let (status, missing) = request(&app, "GET", "/api/workers/listing-missing", Value::Null).await; + assert_eq!(status, StatusCode::NOT_FOUND); + assert_eq!(missing["reason"], "worker_not_found"); + assert_eq!( + missing["message"], + "Worker [listing-missing] not found in namespace [default]." + ); + + // A changed or omitted definition is refused and changes nothing. + for (fingerprints, reason) in [ + ( + json!({"listing": "sha256:changed"}), + "workflow_definition_changed", + ), + ( + json!({"other": "sha256:other"}), + "workflow_definition_fingerprint_missing", + ), + ] { + let mut changed = first.clone(); + changed["workflow_definition_fingerprints"] = fingerprints; + let (status, refusal) = worker_request(&app, "POST", "/api/worker/register", changed).await; + assert_eq!(status, StatusCode::CONFLICT, "{refusal}"); + assert_eq!(refusal["reason"], reason); + assert_eq!(refusal["workflow_type"], "listing"); + assert_eq!( + request(&app, "GET", "/api/workers/listing-first", Value::Null) + .await + .1, + described + ); + } + // Omitting the fingerprints keeps the advertised one and clears what the + // registration does not report. + let mut omitted = first.clone(); + omitted + .as_object_mut() + .unwrap() + .remove("workflow_definition_fingerprints"); + assert_eq!( + worker_request(&app, "POST", "/api/worker/register", omitted) + .await + .0, + StatusCode::CREATED + ); + let after = request(&app, "GET", "/api/workers/listing-first", Value::Null) + .await + .1; + assert_eq!( + after["workflow_definition_fingerprints"], + json!({"listing": fingerprint}) + ); + assert_eq!(after["task_slots"]["workflow_available"], Value::Null); + assert_eq!(after["process_metrics"], Value::Null); + assert_eq!(after["registered_at"], described["registered_at"]); + + // A heartbeat changes only what it reports. + assert_eq!( + worker_request( + &app, + "POST", + "/api/worker/heartbeat", + json!({"worker_id": "listing-first", + "task_slots": {"session_available": 7}, "heartbeat_interval_seconds": 15}) + ) + .await + .0, + StatusCode::OK + ); + let after = request(&app, "GET", "/api/workers/listing-first", Value::Null) + .await + .1; + assert_eq!(after["task_slots"]["session_available"], 2); + assert_eq!(after["task_slots"]["workflow_available"], Value::Null); + assert_eq!(after["heartbeat_interval_seconds"], 15); + let (status, unregistered) = worker_request( + &app, + "POST", + "/api/worker/heartbeat", + json!({"worker_id": "listing-missing"}), + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND); + assert_eq!(unregistered["reason"], "worker_not_registered"); + for invalid in [ + json!({"max_concurrent_workflow_tasks": 0, "max_concurrent_activity_tasks": 0}), + json!({"task_slots": {"workflow_available": -1}}), + json!({"heartbeat_interval_seconds": 0}), + ] { + let body = registration("listing-invalid", "queue-a", invalid); + assert_eq!( + worker_request(&app, "POST", "/api/worker/register", body) + .await + .0, + StatusCode::UNPROCESSABLE_ENTITY + ); + } + assert_eq!( + request(&app, "GET", "/api/workers/listing-invalid", Value::Null) + .await + .0, + StatusCode::NOT_FOUND + ); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn a_silent_worker_is_listed_only_as_stale() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap().with_query_task_config( + durable_workflow_server::runtime::QueryTaskConfig { + worker_stale_after_seconds: 1, + ..Default::default() + }, + ); + let app = router(runtime.clone()); + let body = registration("listing-silent", "queue-a", json!({})); + assert_eq!( + worker_request(&app, "POST", "/api/worker/register", body) + .await + .0, + StatusCode::CREATED + ); + // MySQL rounds these timestamps to whole seconds. + tokio::time::sleep(std::time::Duration::from_millis(2100)).await; + let (_, listed) = request(&app, "GET", "/api/workers", Value::Null).await; + assert_eq!(listed["workers"], json!([])); + assert_eq!(listed["stale_after_seconds"], 1); + let (_, stale) = request(&app, "GET", "/api/workers?status=stale", Value::Null).await; + assert_eq!(stale["workers"][0]["worker_id"], "listing-silent"); + assert_eq!(stale["workers"][0]["status"], "stale"); + assert_eq!( + request(&app, "GET", "/api/workers/listing-silent", Value::Null) + .await + .1["status"], + "stale" + ); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn named_namespaces_list_only_their_own_workers() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + for name in ["alpha", "beta"] { + namespaces::create(&app, name).await; + } + let (status, body) = namespaces::scoped( + &app, + "alpha", + "POST", + "/api/worker/register", + registration("listing-alpha", "queue-a", json!({})), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + let (status, alpha) = + namespaces::scoped(&app, "alpha", "GET", "/api/workers", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{alpha}"); + assert_eq!(alpha["workers"][0]["namespace"], "alpha"); + let (_, beta) = namespaces::scoped(&app, "beta", "GET", "/api/workers", Value::Null).await; + assert_eq!(beta["workers"], json!([])); + let (status, missing) = namespaces::scoped( + &app, + "beta", + "GET", + "/api/workers/listing-alpha", + Value::Null, + ) + .await; + assert_eq!(status, StatusCode::NOT_FOUND); + assert_eq!( + missing["message"], + "Worker [listing-alpha] not found in namespace [beta]." + ); + runtime.close().await; + database.remove().await; +} + +#[tokio::test] +async fn a_replacement_registration_releases_the_activity_lease_at_once() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "replaced", json!(["echo"]), json!(["echo-activity"])).await; + let started = start(&app, "replaced-peer").await; + let task = poll(&app, "replaced", "workflow").await; + let scheduled = finish_task( + &app, + &task, + json!([{"type":"schedule_activity","activity_type":"echo-activity", + "arguments":envelope(Payload::Array(vec![Payload::Long(7)])), + "retry_policy":{"max_attempts":1,"backoff_seconds":[]}}]), + ) + .await; + assert_eq!(scheduled.0, StatusCode::OK, "{}", scheduled.1); + let original = poll(&app, "replaced", "activity").await; + assert_eq!(original["attempt_number"], 1); + // The same worker_id registering again replaces a process that may have + // died holding the lease; the task does not wait out the activity lease. + register(&app, "replaced", json!(["echo"]), json!(["echo-activity"])).await; + let replacement = poll(&app, "replaced", "activity").await; + assert_eq!(replacement["task_id"], original["task_id"]); + assert_eq!(replacement["attempt_number"], 2); + assert_ne!( + replacement["activity_attempt_id"], + original["activity_attempt_id"] + ); + let complete = |claim: &Value| { + json!({"activity_attempt_id":claim["activity_attempt_id"],"lease_owner":"replaced", + "result":envelope(Payload::Long(7))}) + }; + let path = format!( + "/api/worker/activity-tasks/{}/complete", + original["task_id"].as_str().unwrap() + ); + let (status, stale) = request(&app, "POST", &path, complete(&original)).await; + assert_eq!(status, StatusCode::CONFLICT, "{stale}"); + assert_eq!( + stale, + json!({"task_id":original["task_id"],"activity_attempt_id":original["activity_attempt_id"], + "outcome":"completed","recorded":false,"reason":"stale_attempt","next_task_id":null, + "lease_owner":"replaced","activity_status":"running","attempt_status":"expired","task_status":"leased"}) + ); + let (status, done) = request(&app, "POST", &path, complete(&replacement)).await; + assert_eq!(status, StatusCode::OK, "{done}"); + assert_eq!(done["recorded"], true); + let history = run_history(&app, &json!("replaced-peer"), &started["run_id"]).await; + let started_attempts: Vec<_> = history + .as_array() + .unwrap() + .iter() + .filter(|event| event["event_type"] == "ActivityStarted") + .map(|event| event["payload"]["attempt_number"].clone()) + .collect(); + assert_eq!(started_attempts, vec![json!(1), json!(2)]); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/cancellation-contract.mjs b/scripts/conformance/server-parity/cancellation-contract.mjs index ba48cf5ce..b96309fca 100644 --- a/scripts/conformance/server-parity/cancellation-contract.mjs +++ b/scripts/conformance/server-parity/cancellation-contract.mjs @@ -1,8 +1,24 @@ import assert from 'node:assert/strict'; +// Published PHP closes a run immediately by cancel or terminate through one +// implementation. The reviewed difference is only the recorded identity. +const COMMANDS = { + cancel: {closed: 'cancelled', requested: 'CancelRequested', terminal: 'WorkflowCancelled', + exception: 'Workflow\\V2\\Exceptions\\WorkflowCancelledException', sdk: 'DurableWorkflow\\Exception\\WorkflowCancelled'}, + terminate: {closed: 'terminated', requested: 'TerminateRequested', terminal: 'WorkflowTerminated', + exception: 'Workflow\\V2\\Exceptions\\WorkflowTerminatedException', sdk: 'DurableWorkflow\\Exception\\WorkflowTerminated'}, +}; + +export function immediateCommand(fixture) { + const name = fixture.immediate_cancellation?.command ?? 'cancel'; + assert.ok(Object.hasOwn(COMMANDS, name), 'reviewed immediate terminal command'); + return {name, ...COMMANDS[name]}; +} + // Reviewed immediate-cancel semantics. Cooperative cleanup and propagation to // children/updates remain separate gates; raw representations are retained. export function checkImmediateCancellation(fixture, observation) { + const command = immediateCommand(fixture); const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); const nonempty = (value, label) => assert.ok(typeof value === 'string' && value.length > 0, label); const time = (value, label) => { const parsed = Date.parse(value); assert.ok(Number.isFinite(parsed), label); return parsed; }; @@ -13,7 +29,7 @@ export function checkImmediateCancellation(fixture, observation) { const events = observation.events; const rawEvents = events.map(({decoded, typed_decoded, ...event}) => event); const cancellation = observation.cancellation; - const requested = events.find(event => event.event_type === 'CancelRequested'); + const requested = events.find(event => event.event_type === command.requested); const terminal = events.at(-1); const commandId = requested.payload.workflow_command_id; const failureId = terminal.payload.failure_id; @@ -30,11 +46,12 @@ export function checkImmediateCancellation(fixture, observation) { equal(event.payload.workflow_command_id, commandId, 'original cancel command across lifecycle'); equal(event.payload.reason, reason, 'immutable original cancellation reason'); } - equal(requested.payload.command_type, 'cancel', 'immediate cancel command'); - equal(terminal.payload.failure_category, 'cancelled', 'typed cancellation failure category'); - equal(terminal.payload.closed_reason, 'cancelled', 'terminal cancellation reason'); - equal(terminal.payload.exception_class, 'Workflow\\V2\\Exceptions\\WorkflowCancelledException', 'published cancellation exception identity'); - const completeDiagnostic = `Workflow cancelled: ${reason}`; + equal(terminal.event_type, command.terminal, 'immediate terminal event'); + equal(requested.payload.command_type, command.name, 'immediate terminal command'); + equal(terminal.payload.failure_category, command.closed, 'typed terminal failure category'); + equal(terminal.payload.closed_reason, command.closed, 'terminal closed reason'); + equal(terminal.payload.exception_class, command.exception, 'published terminal exception identity'); + const completeDiagnostic = `Workflow ${command.closed}: ${reason}`; const encoding = fixture.immediate_cancellation.diagnostic_encoding; assert.ok(encoding === undefined || encoding === 'json_string_literal_if_nul', 'reviewed diagnostic encoding'); const encoded = encoding === 'json_string_literal_if_nul' && completeDiagnostic.includes('\0'); @@ -45,17 +62,17 @@ export function checkImmediateCancellation(fixture, observation) { } equal(Object.keys(terminal.decoded), [], 'cancellation commits no workflow result'); equal(Object.keys(terminal.typed_decoded), [], 'cancellation has no typed success result'); - equal(observation.execution.closed_reason, 'cancelled', 'durable run closure'); + equal(observation.execution.closed_reason, command.closed, 'durable run closure'); assert.ok(time(observation.execution.closed_at, 'durable cancellation close timestamp') >= time(observation.execution.started_at, 'original start time'), 'cancellation closes the original run'); const fresh = cancellation.fresh_execution; equal(fresh.run_id ?? fresh.id, observation.run_id, 'fresh original run identity'); - equal(fresh.status, 'cancelled', 'fresh durable cancellation'); + equal(fresh.status, command.closed, 'fresh durable closure'); equal(fresh.output, null, 'no fresh success output'); const accepted = cancellation.accepted; equal(accepted.status, 200, 'accepted cancellation transport'); equal(accepted.response.command_id, commandId, 'acknowledged original cancellation command'); equal(accepted.response.command_status, 'accepted', 'accepted cancellation command'); - equal(accepted.response.outcome, 'cancelled', 'terminal cancellation acknowledgment'); + equal(accepted.response.outcome, command.closed, 'terminal command acknowledgment'); equal(accepted.response.command_sequence, 2, 'cancel follows original start command'); const duplicate = cancellation.duplicate; equal(duplicate.status, 409, 'repeat cancellation refusal'); @@ -130,7 +147,7 @@ export function checkImmediateCancellation(fixture, observation) { for (const receipt of [report, cancellation.late_completion]) { equal(receipt.status, 409, 'late activity completion refused'); equal(receipt.response.recorded, false, 'late activity result is not durable'); - equal(receipt.response.reason, 'run_cancelled', 'late cancelled run fencing'); + equal(receipt.response.reason, `run_${command.closed}`, 'late closed run fencing'); equal(receipt.response.outcome, 'ignored', 'cancelled run ignores late outcome'); equal(receipt.response.cancel_requested, true, 'late outcome sees cancellation'); equal(receipt.response.can_continue, false, 'late cancelled activity must stop'); @@ -152,21 +169,22 @@ export function checkImmediateCancellation(fixture, observation) { } equal(new Set(identities).size, identities.length, 'all cancellation work identities remain distinct'); if (observation.mode === 'http') { - equal(cancellation.outcome, {type: 'DurableWorkflow\\Exception\\WorkflowCancelled', message: reason}, 'actual SDK typed cancelled result'); + equal(cancellation.outcome, {type: command.sdk, message: reason}, 'actual SDK typed closed result'); equal(cancellation.remaining_tasks, {workflow: null, activity: null}, 'cancelled run yields no remaining work'); equal(observation.workflow_polls.length, phase === 'before_claim' ? 0 : 1, 'actual workflow claim count'); equal(observation.workflow_completions.length, phase === 'before_claim' ? 0 : 1, 'one actual scheduling completion'); if (phase !== 'leased_activity') equal(observation.activity_polls, [], 'no unexpected activity claim'); } else { - equal(fresh.cancelled, true, 'embedded public cancellation outcome'); + equal(fresh.cancelled, command.name === 'cancel', 'embedded public cancellation outcome'); + equal(fresh.terminated, command.name === 'terminate', 'embedded public termination outcome'); equal(cancellation.failures.length, 1, 'one original embedded cancellation failure'); const failure = cancellation.failures[0]; equal(failure.id, failureId, 'original durable cancellation failure'); equal(failure.workflow_run_id, observation.run_id, 'failure original run'); equal(failure.source_kind, 'workflow_run', 'failure source kind'); equal(failure.source_id, observation.run_id, 'failure original source'); - equal(failure.propagation_kind, 'cancelled', 'failure cancellation propagation'); - equal(failure.failure_category, 'cancelled', 'failure typed category'); + equal(failure.propagation_kind, command.closed, 'failure propagation kind'); + equal(failure.failure_category, command.closed, 'failure typed category'); equal(failure.exception_class, terminal.payload.exception_class, 'failure original exception class'); equal(failure.message, terminal.payload.message, 'failure original message'); equal(failure.handled, false, 'cancellation failure not swallowed'); diff --git a/scripts/conformance/server-parity/cancellation-probe.php b/scripts/conformance/server-parity/cancellation-probe.php index 98f31d054..d2f13e175 100644 --- a/scripts/conformance/server-parity/cancellation-probe.php +++ b/scripts/conformance/server-parity/cancellation-probe.php @@ -8,6 +8,7 @@ use DurableWorkflow\Worker\WorkflowContext; use DurableWorkflow\Exception\ServerException; use DurableWorkflow\Exception\WorkflowCancelled; +use DurableWorkflow\Exception\WorkflowTerminated; use Illuminate\Contracts\Console\Kernel; use Illuminate\Support\Facades\Artisan; use Workflow\Serializers\Serializer; @@ -23,6 +24,7 @@ // mutate engine rows or synthesize a claim, cancellation or terminal outcome. final class CancellationProbeState { + public static string $command = 'cancel'; public static string $reason; public static ?array $receipt = null; public static array $before = []; @@ -45,13 +47,18 @@ function httpCancellationObservation(array $fixture, string $workflowId, string transport: observedChildTransport($polls, $completions, $activityPolls, $activityOutcomes)); $phase = $fixture['immediate_cancellation']['phase']; $reason = $fixture['immediate_cancellation']['reason']; + // Cancel and terminate are one published implementation; the fixture + // names which command closes the run. + $close = ($fixture['immediate_cancellation']['command'] ?? 'cancel') === 'terminate' + ? static fn (?string $reason, string $runId): array => $client->terminateWorkflow($workflowId, $reason, $runId) + : static fn (?string $reason, string $runId): array => $client->cancelWorkflow($workflowId, $reason, $runId); $handle = $worker = null; $before = []; $receipt = null; $deadline = microtime(true) + 30; - $cancel = static function () use ($client, $workflowId, $reason, &$handle, &$before, &$receipt): void { + $cancel = static function () use ($client, $workflowId, $reason, $close, &$handle, &$before, &$receipt): void { $before = httpHistory($client, $workflowId, $handle->selectedRunId); - $receipt = cancellationReceipt(static fn (): array => $client->cancelWorkflow($workflowId, $reason, $handle->selectedRunId)); + $receipt = cancellationReceipt(static fn (): array => $close($reason, $handle->selectedRunId)); }; $worker = (new Worker($client, $queue, clock: static function () use (&$worker, &$handle, &$receipt, $client, $workflowId, $phase, $cancel, $deadline): float { @@ -99,12 +106,12 @@ function httpCancellationObservation(array $fixture, string $workflowId, string unset($report); $execution = $handle->describeSelectedRun(); $events = httpHistory($client, $workflowId, $handle->selectedRunId); - $duplicate = cancellationReceipt(static fn (): array => $client->cancelWorkflow($workflowId, 'replacement reason', $handle->selectedRunId)); + $duplicate = cancellationReceipt(static fn (): array => $close('replacement reason', $handle->selectedRunId)); $afterDuplicate = httpHistory($client, $workflowId, $handle->selectedRunId); $outcome = null; try { $handle->resultOfSelectedRun(1); - } catch (WorkflowCancelled $error) { + } catch (WorkflowCancelled|WorkflowTerminated $error) { $outcome = ['type' => $error::class, 'message' => $error->getMessage()]; } $late = null; @@ -148,11 +155,16 @@ function embeddedCancellationReceipt($command): array 'response' => \Workflow\V2\Support\CommandResponse::payload($command)]; } +function closeEmbeddedRun(WorkflowStub $stub, string $reason) +{ + return CancellationProbeState::$command === 'terminate' ? $stub->attemptTerminate($reason) : $stub->attemptCancel($reason); +} + function cancelEmbeddedRun(string $runId): void { CancellationProbeState::$before = embeddedCancellationHistory($runId); CancellationProbeState::$receipt = embeddedCancellationReceipt( - WorkflowStub::loadRun($runId, 'default')->attemptCancel(CancellationProbeState::$reason)); + closeEmbeddedRun(WorkflowStub::loadRun($runId, 'default'), CancellationProbeState::$reason)); } function embeddedCancellationObservation(array $fixture, string $workflowId, string $namespace, string $queue, string $root): array @@ -174,6 +186,7 @@ function embeddedCancellationObservation(array $fixture, string $workflowId, str $class = config('workflows.v2.types.workflows')[$fixture['workflow_type']]; $stub = WorkflowStub::make($class, $workflowId); CancellationProbeState::$reason = $fixture['immediate_cancellation']['reason']; + CancellationProbeState::$command = $fixture['immediate_cancellation']['command'] ?? 'cancel'; $stub->start($fixture['input'], new WorkflowOptions(connection: 'database', queue: $queue), new StartOptions(executionTimeoutSeconds: 3600, runTimeoutSeconds: 600)); $phase = $fixture['immediate_cancellation']['phase']; @@ -187,7 +200,7 @@ function embeddedCancellationObservation(array $fixture, string $workflowId, str '--max-time' => 10, '--sleep' => 0, '--tries' => 1]); $run = WorkflowRun::query()->findOrFail($stub->runId()); $events = embeddedCancellationHistory($run->id); - $duplicate = embeddedCancellationReceipt(WorkflowStub::loadRun($run->id, $namespace)->attemptCancel('replacement reason')); + $duplicate = embeddedCancellationReceipt(closeEmbeddedRun(WorkflowStub::loadRun($run->id, $namespace), 'replacement reason')); $afterDuplicate = embeddedCancellationHistory($run->id); // Redeliver original closed activity work through its actual queue job. foreach (CancellationProbeState::$claims as $taskId) { @@ -205,7 +218,8 @@ function embeddedCancellationObservation(array $fixture, string $workflowId, str 'cancellation' => ['before' => CancellationProbeState::$before, 'accepted' => CancellationProbeState::$receipt, 'duplicate' => $duplicate, 'history_before_duplicate' => $events, 'history_after_duplicate' => $afterDuplicate, 'redelivered_task_ids' => CancellationProbeState::$claims, 'fresh_history' => embeddedCancellationHistory($run->id), - 'fresh_execution' => ['id' => $fresh->runId(), 'status' => $fresh->status(), 'cancelled' => $fresh->cancelled(), 'output' => $fresh->output()], + 'fresh_execution' => ['id' => $fresh->runId(), 'status' => $fresh->status(), 'cancelled' => $fresh->cancelled(), + 'terminated' => $fresh->terminated(), 'output' => $fresh->output()], 'failures' => $run->failures()->get()->map->toArray()->all(), 'attempts' => $run->activityAttempts()->get()->map->toArray()->all(), 'tasks' => $run->tasks()->get()->map->toArray()->all()]]; diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index e60868655..d688dc6df 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -2,13 +2,15 @@ import assert from 'node:assert/strict'; import {checkChildCancellation} from './child-cancellation-contract.mjs'; import {checkActivityRetry} from './retry-contract.mjs'; import {checkTerminalActivityFailure} from './failure-contract.mjs'; -import {checkImmediateCancellation} from './cancellation-contract.mjs'; +import {checkImmediateCancellation, immediateCommand} from './cancellation-contract.mjs'; import {checkCooperativeCancellation} from './cooperative-contract.mjs'; import {checkSchedule, scheduledWorkflowIdentity} from './schedule-contract.mjs'; 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 {checkWorkerListing} from './worker-listing-contract.mjs'; +import {checkWorkerReplacement} from './worker-replacement-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'; @@ -73,7 +75,8 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { equal(observation.namespace, fixture.namespace ?? 'default', 'namespace'); equal(observation.task_queue, fixture.sdk_update_direction?.task_queue ?? fixture.sdk_query_direction?.task_queue ?? fixture.sdk_signal_direction?.task_queue ?? fixture.sdk_timer_author?.task_queue ?? fixture.sdk_child_direction?.parent_queue ?? fixture.sdk_direction?.workflow_queue ?? 'server-parity-v1', 'task queue'); const cancelled = Boolean(fixture.immediate_cancellation || fixture.cooperative_cancellation); - equal(observation.status, cancelled ? 'cancelled' : 'completed', 'actual durable terminal state'); + const closedAs = fixture.immediate_cancellation ? immediateCommand(fixture).closed : 'cancelled'; + equal(observation.status, cancelled ? closedAs : 'completed', 'actual durable terminal state'); equal(observation.payload_codec, 'avro', 'payload codec'); const output = cancelled ? null : fixture.output ?? fixture.signal_value ?? fixture.input; const typedOutput = cancelled ? {type: 'null', value: null} : fixture.typed_output ?? fixture.typed_signal_value ?? fixture.typed_value; @@ -128,6 +131,8 @@ 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 workerListing = fixture.worker_listing ? checkWorkerListing(fixture, observation, workflowId) : null; + const workerReplacement = fixture.worker_replacement ? checkWorkerReplacement(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 +670,8 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(admission ? {admission} : {}), ...(namespaces ? {namespace_isolation: namespaces} : {}), ...(workerDeregistration ? {worker_deregistration: workerDeregistration} : {}), + ...(workerListing ? {worker_listing: workerListing} : {}), + ...(workerReplacement ? {worker_replacement: workerReplacement} : {}), ...(patchDeployment ? {patch_deployment: patchDeployment} : {}), ...(waitingHistory ? {waiting_for_history: waitingHistory} : {}), ...(taskRetry ? {workflow_task_retry: taskRetry} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 060d26745..a929634f4 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -427,6 +427,8 @@ 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; + $workerListing = isset($fixture['worker_listing']) ? finishHttpWorkerListing($client, $fixture, $workflowId) : null; + $workerReplacement = isset($fixture['worker_replacement']) ? finishHttpWorkerReplacement($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 +515,8 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'admission' => $admission, 'namespace_isolation' => $namespaces, 'worker_deregistration' => $workerDeregistration, + 'worker_listing' => $workerListing, + 'worker_replacement' => $workerReplacement, 'waiting_for_history' => $waitingHistory, 'workflow_task_retry' => $taskRetry, 'run_id' => $execution->runId, @@ -809,6 +813,8 @@ 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__.'/worker-listing-probe.php'; + require __DIR__.'/worker-replacement-probe.php'; require __DIR__.'/waiting-history-probe.php'; require __DIR__.'/task-retry-probe.php'; require __DIR__.'/child-cancellation-probe.php'; @@ -939,6 +945,12 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa }; } $observation['typed_input'] = typedValue($observation['input']); + if ($mode === 'embedded' && isset($fixture['worker_replacement'])) { + $observation['worker_replacement'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_worker_registration_lifecycle']; + } + if ($mode === 'embedded' && isset($fixture['worker_listing'])) { + $observation['worker_listing'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_worker_registration_lifecycle']; + } if ($mode === 'embedded' && isset($fixture['worker_deregistration'])) { $observation['worker_deregistration'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_worker_registration_lifecycle']; } diff --git a/scripts/conformance/server-parity/worker-listing-contract.mjs b/scripts/conformance/server-parity/worker-listing-contract.mjs new file mode 100644 index 000000000..f57dcf083 --- /dev/null +++ b/scripts/conformance/server-parity/worker-listing-contract.mjs @@ -0,0 +1,112 @@ +import assert from 'node:assert/strict'; + +const SUMMARY_FIELDS = ['worker_id', 'namespace', 'task_queue', 'runtime', 'sdk_version', 'build_id', + 'supported_workflow_types', 'workflow_definition_fingerprints', 'workflow_command_contracts', + 'supported_activity_types', 'capabilities', 'capability_manifest', 'max_concurrent_workflow_tasks', + 'max_concurrent_activity_tasks', 'max_concurrent_worker_sessions', 'task_slots', 'process_metrics', + 'heartbeat_interval_seconds', 'status', 'last_heartbeat_at', 'registered_at']; + +// Reviewed worker management contract (WorkerManagementController and the +// worker registration and heartbeat facts it reports). Embedded applications +// have no HTTP worker registration, so the projection is the definition alone. +export function checkWorkerListing(fixture, observation, workflowId) { + const definition = fixture.worker_listing; + const state = observation.worker_listing; + const projection = {scope: 'worker_listing', workflow_type: definition.workflow_type}; + assert.ok(state && typeof state === 'object', 'actual worker listing observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_worker_registration_lifecycle'}, + 'embedded worker listing explicitly inapplicable'); + return projection; + } + const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); + const time = (value, label) => assert.ok(typeof value === 'string' && Number.isFinite(Date.parse(value)), label); + const first = `${workflowId}-listing-first`; + const second = `${workflowId}-listing-second`; + const queues = [`${workflowId}-listing-a`, `${workflowId}-listing-b`]; + equal(state.applicable, true, 'HTTP worker listing actually executed'); + equal([state.first, state.second, state.queues], [first, second, queues], 'listing identities'); + assert.ok(Number.isInteger(state.probe_pid) && state.probe_pid > 0, 'probe process identity'); + + const registrations = [['first_registration', first, queues[0], definition.build_id], + ['second_registration', second, queues[1], null], ['reregistration', first, queues[0], definition.build_id]]; + for (const [key, workerId, queue, buildId] of registrations) { + const receipt = state[key]; + equal([receipt.worker_id, receipt.registered, receipt.namespace, receipt.task_queue, receipt.runtime, + receipt.build_id, receipt.status, receipt.heartbeat_interval_seconds], + [workerId, true, 'default', queue, 'php', buildId, 'active', definition.heartbeat_interval_seconds], `${key} receipt`); + } + const heartbeat = state.heartbeat; + equal([heartbeat.worker_id, heartbeat.acknowledged, heartbeat.heartbeat_interval_seconds, heartbeat.stale_after_seconds], + [first, true, definition.heartbeat_interval_seconds, definition.stale_after_seconds], 'heartbeat acknowledgement'); + + const capacity = definition.capacity; + const firstFacts = {worker_id: first, namespace: 'default', task_queue: queues[0], runtime: 'php', + build_id: definition.build_id, supported_workflow_types: [definition.workflow_type], + workflow_definition_fingerprints: {[definition.workflow_type]: definition.fingerprint}, + // Published PHP renders an empty map as a JSON list. + workflow_command_contracts: [], supported_activity_types: [definition.activity_type], + max_concurrent_workflow_tasks: capacity.workflow, max_concurrent_activity_tasks: capacity.activity, + max_concurrent_worker_sessions: capacity.sessions, + task_slots: {...definition.stored_task_slots, workflow_capacity: capacity.workflow, + activity_capacity: capacity.activity, session_capacity: capacity.sessions}, + process_metrics: {process_uptime_seconds: 0, process_id: state.probe_pid}, + heartbeat_interval_seconds: null, status: 'active'}; + // The published SDK's registration defaults. + const secondFacts = {worker_id: second, namespace: 'default', task_queue: queues[1], runtime: 'php', build_id: null, + supported_workflow_types: [definition.workflow_type], workflow_definition_fingerprints: [], + workflow_command_contracts: [], supported_activity_types: [], max_concurrent_workflow_tasks: 1, + max_concurrent_activity_tasks: 1, max_concurrent_worker_sessions: 10, + task_slots: {workflow_available: null, activity_available: null, session_available: null, + workflow_capacity: 1, activity_capacity: 1, session_capacity: 10}, + process_metrics: null, heartbeat_interval_seconds: null, status: 'active'}; + const registered = {[first]: state.first_registration, [second]: state.second_registration}; + const summary = (worker, facts, label) => { + equal(Object.keys(worker).sort(), [...SUMMARY_FIELDS].sort(), `${label} summary fields`); + for (const [field, value] of Object.entries(facts)) equal(worker[field], value, `${label} ${field}`); + assert.ok(typeof worker.sdk_version === 'string' && worker.sdk_version.startsWith('durable-workflow-php/'), `${label} SDK identity`); + equal(worker.capabilities, registered[worker.worker_id].capabilities, `${label} registered capabilities`); + equal(worker.capability_manifest, registered[worker.worker_id].capability_manifest, `${label} registered manifest`); + time(worker.last_heartbeat_at, `${label} heartbeat time`); + time(worker.registered_at, `${label} registration time`); + }; + const listing = (value, expected, label) => { + equal(value.stale_after_seconds, definition.stale_after_seconds, `${label} stale window`); + // Equal heartbeat times may tie, so order is not part of this contract. + const workers = [...value.workers].sort((a, b) => a.worker_id.localeCompare(b.worker_id)); + equal(workers.map(worker => worker.worker_id), expected.map(facts => facts.worker_id), `${label} workers`); + workers.forEach((worker, index) => summary(worker, expected[index], `${label} ${worker.worker_id}`)); + }; + listing(state.listed, [firstFacts, secondFacts], 'listed'); + listing(state.first_queue, [firstFacts], 'task_queue filter'); + listing(state.active, [firstFacts, secondFacts], 'status=active'); + listing(state.stale, [], 'status=stale'); + listing(state.after_deregistration, [], 'after deregistration'); + + const described = ({updated_at, stale_after_seconds, ...worker}, facts, label) => { + summary(worker, facts, label); + time(updated_at, `${label} update time`); + equal(stale_after_seconds, definition.stale_after_seconds, `${label} stale window`); + }; + described(state.described, firstFacts, 'described'); + equal(state.missing, {status: 404, response: {reason: 'worker_not_found', + message: `Worker [${workflowId}-listing-missing] not found in namespace [default].`}}, 'missing worker'); + const refusal = (receipt, reason, error, remediation, label) => { + equal(receipt.status, 409, `${label} status`); + const {protocol_version, server_capabilities, ...body} = receipt.response; + equal(body, {error, reason, workflow_type: definition.workflow_type, remediation}, `${label} refusal`); + }; + refusal(state.changed_definition, 'workflow_definition_changed', 'Worker attempted to re-register a changed workflow definition.', + 'Restart the worker with a new worker_id before registering a changed workflow class definition.', 'changed definition'); + refusal(state.missing_fingerprint, 'workflow_definition_fingerprint_missing', 'Worker omitted a previously advertised workflow definition fingerprint.', + 'Advertise the matching definition fingerprint or restart with a new worker_id.', 'omitted fingerprint'); + equal(state.after_refusals, state.described, 'refused registrations change nothing'); + + // A valid re-registration keeps the advertised fingerprint and resets the + // task slots and process metrics it did not report. + described(state.after_reregistration, {...firstFacts, + task_slots: {...firstFacts.task_slots, workflow_available: null, activity_available: null, session_available: null}, + process_metrics: null}, 'after re-registration'); + equal(state.after_reregistration.registered_at, state.described.registered_at, 're-registration keeps the registration time'); + return projection; +} diff --git a/scripts/conformance/server-parity/worker-listing-probe.php b/scripts/conformance/server-parity/worker-listing-probe.php new file mode 100644 index 000000000..e5ad91348 --- /dev/null +++ b/scripts/conformance/server-parity/worker-listing-probe.php @@ -0,0 +1,55 @@ + 200, 'response' => $request()]; + } catch (ServerException $error) { + return ['status' => $error->status, 'response' => $error->details]; + } + }; + $first = $workflowId.'-listing-first'; + $second = $workflowId.'-listing-second'; + $queues = [$workflowId.'-listing-a', $workflowId.'-listing-b']; + $type = $definition['workflow_type']; + $capacity = $definition['capacity']; + $registerFirst = static fn (?array $fingerprints): array => $client->registerWorker($first, $queues[0], [$type], + [$definition['activity_type']], maxConcurrentWorkflowTasks: $capacity['workflow'], + maxConcurrentActivityTasks: $capacity['activity'], buildId: $definition['build_id'], + maxConcurrentWorkerSessions: $capacity['sessions'], workflowDefinitionFingerprints: $fingerprints); + // The namespace is shared with other fixtures' workers; keep only these two. + $ours = static fn (array $listing): array => ['stale_after_seconds' => $listing['stale_after_seconds'] ?? null, + 'workers' => array_values(array_filter($listing['workers'] ?? [], + static fn (array $worker): bool => in_array($worker['worker_id'] ?? null, [$first, $second], true)))]; + $state = ['applicable' => true, 'first' => $first, 'second' => $second, 'queues' => $queues, 'probe_pid' => getmypid()]; + $state['first_registration'] = $registerFirst([$type => $definition['fingerprint']]); + $state['second_registration'] = $client->registerWorker($second, $queues[1], [$type], []); + $state['heartbeat'] = $client->heartbeatWorker($first, $definition['task_slots']); + $state['listed'] = $ours($client->listWorkers()); + $state['first_queue'] = $ours($client->listWorkers($queues[0])); + $state['active'] = $ours($client->listWorkers(null, 'active')); + $state['stale'] = $ours($client->listWorkers(null, 'stale')); + $state['described'] = $client->describeWorker($first); + $state['missing'] = $receipt(static fn (): array => $client->describeWorker($workflowId.'-listing-missing')); + $state['changed_definition'] = $receipt(static fn (): array => $registerFirst([$type => $definition['changed_fingerprint']])); + $state['missing_fingerprint'] = $receipt(static fn (): array => $registerFirst([$definition['other_workflow_type'] => $definition['fingerprint']])); + $state['after_refusals'] = $client->describeWorker($first); + // Omitting the fingerprints keeps the advertised one. + $state['reregistration'] = $registerFirst(null); + $state['after_reregistration'] = $client->describeWorker($first); + foreach ([$first, $second] as $workerId) { + $client->deregisterWorker($workerId); + } + $state['after_deregistration'] = $ours($client->listWorkers()); + + return $state; +} diff --git a/scripts/conformance/server-parity/worker-replacement-contract.mjs b/scripts/conformance/server-parity/worker-replacement-contract.mjs new file mode 100644 index 000000000..6d1b438e0 --- /dev/null +++ b/scripts/conformance/server-parity/worker-replacement-contract.mjs @@ -0,0 +1,54 @@ +import assert from 'node:assert/strict'; + +// Reviewed replacement of a worker process holding an activity lease +// (WorkerController::releaseLeasedActivityTasksForReplacedWorker and the +// stale completion receipt). Embedded applications have no HTTP worker +// registration, so the projection is the definition alone. +export function checkWorkerReplacement(fixture, observation, workflowId) { + const definition = fixture.worker_replacement; + const state = observation.worker_replacement; + const projection = {scope: 'worker_replacement', expected_events: definition.expected_events}; + assert.ok(state && typeof state === 'object', 'actual worker replacement observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_worker_registration_lifecycle'}, + 'embedded worker replacement explicitly inapplicable'); + return projection; + } + const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); + const worker = `${workflowId}-replaced-worker`; + equal([state.applicable, state.worker_id, state.task_queue, state.peer_workflow_id], + [true, worker, `${workflowId}-replacement`, `${workflowId}-replacement-peer`], 'replacement identities'); + for (const key of ['registration', 'replacement_registration']) { + equal([state[key].worker_id, state[key].registered], [worker, true], `${key} receipt`); + } + const original = state.original_claim; + const replacement = state.replacement_claim; + assert.ok(original && replacement, 'the released task is claimed again at once'); + equal([original.attempt_number, original.lease_owner], [1, worker], 'original attempt'); + equal([replacement.task_id, replacement.attempt_number, replacement.lease_owner], + [original.task_id, 2, worker], 'the same task claimed again as attempt 2 under a one-attempt policy'); + assert.notEqual(replacement.activity_attempt_id, original.activity_attempt_id, 'a new attempt identity'); + equal(state.history_after_replacement.map(event => event.event_type), definition.expected_events.slice(0, 4), + 'releasing the lease records no history'); + + const {protocol_version, server_capabilities, ...stale} = state.stale_completion.response; + equal(state.stale_completion.status, 409, 'stale completion refused'); + equal(stale, {task_id: original.task_id, activity_attempt_id: original.activity_attempt_id, outcome: 'completed', + recorded: false, reason: 'stale_attempt', next_task_id: null, lease_owner: worker, activity_status: 'running', + attempt_status: 'expired', task_status: 'leased'}, 'published stale completion receipt'); + const completion = state.completion; + equal(completion.status, 200, 'replacement completion accepted'); + equal([completion.response.task_id, completion.response.activity_attempt_id, completion.response.outcome, + completion.response.recorded, completion.response.reason], + [original.task_id, replacement.activity_attempt_id, 'completed', true, null], 'replacement completion recorded once'); + + const history = state.history; + equal(history.map(event => event.event_type), definition.expected_events, 'complete replacement history'); + const started = history.filter(event => event.event_type === 'ActivityStarted').map(event => event.payload); + equal(started.map(payload => [payload.activity_attempt_id, payload.attempt_number]), + [[original.activity_attempt_id, 1], [replacement.activity_attempt_id, 2]], 'each attempt started once'); + const completed = history.find(event => event.event_type === 'ActivityCompleted').payload; + equal([completed.activity_attempt_id, completed.attempt_number], [replacement.activity_attempt_id, 2], + 'only the replacement attempt completes'); + return projection; +} diff --git a/scripts/conformance/server-parity/worker-replacement-probe.php b/scripts/conformance/server-parity/worker-replacement-probe.php new file mode 100644 index 000000000..6090d35eb --- /dev/null +++ b/scripts/conformance/server-parity/worker-replacement-probe.php @@ -0,0 +1,54 @@ + 200, 'response' => $request()]; + } catch (ServerException $error) { + return ['status' => $error->status, 'response' => $error->details]; + } + }; + $worker = $workflowId.'-replaced-worker'; + $queue = $workflowId.'-replacement'; + $peerId = $workflowId.'-replacement-peer'; + $register = static fn (): array => $client->registerWorker($worker, $queue, [$definition['workflow_type']], + [$definition['activity_type']]); + $envelope = static fn (): array => $client->payloadCodec()->envelope($fixture['input']); + $state = ['applicable' => true, 'worker_id' => $worker, 'task_queue' => $queue, 'peer_workflow_id' => $peerId]; + $state['registration'] = $register(); + $handle = $client->startWorkflow($definition['workflow_type'], $peerId, $queue, [$fixture['input']]); + $state['peer_run_id'] = $handle->selectedRunId; + $task = $client->pollWorkflowTask($worker, $queue, 5); + $state['scheduled'] = $client->completeWorkflowTask($task['task_id'], $worker, $task['workflow_task_attempt'], [[ + 'type' => 'schedule_activity', 'activity_type' => $definition['activity_type'], + 'arguments' => $client->payloadCodec()->envelope([$fixture['input']]), + 'retry_policy' => ['max_attempts' => 1, 'backoff_seconds' => []], + ]]); + $state['original_claim'] = $client->pollActivityTask($worker, $queue, 5); + $state['replacement_registration'] = $register(); + $state['history_after_replacement'] = httpHistory($client, $peerId, $handle->selectedRunId); + $state['replacement_claim'] = $client->pollActivityTask($worker, $queue, 5); + $original = $state['original_claim']; + $state['stale_completion'] = $receipt(static fn (): array => $client->completeActivityTask($original['task_id'], + $original['activity_attempt_id'], $worker, $fixture['input'])); + $replacement = $state['replacement_claim']; + $state['completion'] = $receipt(static fn (): array => $client->completeActivityTask($replacement['task_id'], + $replacement['activity_attempt_id'], $worker, $fixture['input'])); + $task = $client->pollWorkflowTask($worker, $queue, 5); + $state['closed'] = $client->completeWorkflowTask($task['task_id'], $worker, $task['workflow_task_attempt'], + [['type' => 'complete_workflow', 'result' => $envelope()]]); + $state['history'] = httpHistory($client, $peerId, $handle->selectedRunId); + $client->deregisterWorker($worker); + + return $state; +} diff --git a/tests/Fixtures/ServerParity/terminate-before-claim.json b/tests/Fixtures/ServerParity/terminate-before-claim.json new file mode 100644 index 000000000..d7a5dd175 --- /dev/null +++ b/tests/Fixtures/ServerParity/terminate-before-claim.json @@ -0,0 +1,42 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "terminate-before-claim", + "workflow_type": "parity.v1.cancel_before_claim", + "input": { + "message": "cancel λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cancel λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "immediate_cancellation": { + "command": "terminate", + "phase": "before_claim", + "receipt_target_scope": {"http": "instance", "embedded": "run"}, + "reason": " \t\u00a0parity cancellation λ\u200b\ud834\udd73\r\n", + "http_reason": "parity cancellation λ" + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "TerminateRequested", + "WorkflowTerminated" + ], + "contract": "Immediate selected-run termination shares the published cancellation implementation and records the termination identity: TerminateRequested and WorkflowTerminated, the terminated run status, closed reason and failure category, and WorkflowTerminatedException. Open tasks, activities and timers are cancelled as for cancellation, and a late activity outcome is refused with run_terminated. Repeating termination is refused with run_not_active and unchanged history. These fixtures reuse the cancellation fixtures' workflows. Terminating a child run is not qualified." +} diff --git a/tests/Fixtures/ServerParity/terminate-leased-activity.json b/tests/Fixtures/ServerParity/terminate-leased-activity.json new file mode 100644 index 000000000..0b1e9891a --- /dev/null +++ b/tests/Fixtures/ServerParity/terminate-leased-activity.json @@ -0,0 +1,44 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "terminate-leased-activity", + "workflow_type": "parity.v1.cancel_leased_activity", + "input": { + "message": "cancel λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cancel λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "immediate_cancellation": { + "command": "terminate", + "phase": "leased_activity", + "receipt_target_scope": {"http": "instance", "embedded": "run"}, + "reason": "parity cancellation λ" + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "ActivityScheduled", + "ActivityStarted", + "TerminateRequested", + "ActivityCancelled", + "WorkflowTerminated" + ], + "contract": "Immediate selected-run termination shares the published cancellation implementation and records the termination identity: TerminateRequested and WorkflowTerminated, the terminated run status, closed reason and failure category, and WorkflowTerminatedException. Open tasks, activities and timers are cancelled as for cancellation, and a late activity outcome is refused with run_terminated. Repeating termination is refused with run_not_active and unchanged history. These fixtures reuse the cancellation fixtures' workflows. Terminating a child run is not qualified." +} diff --git a/tests/Fixtures/ServerParity/terminate-pending-timer.json b/tests/Fixtures/ServerParity/terminate-pending-timer.json new file mode 100644 index 000000000..68424ca96 --- /dev/null +++ b/tests/Fixtures/ServerParity/terminate-pending-timer.json @@ -0,0 +1,44 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "terminate-pending-timer", + "workflow_type": "parity.v1.cancel_pending_timer", + "input": { + "message": "cancel λ", + "count": 9007199254740993, + "optional": null + }, + "typed_value": { + "type": "map", + "value": { + "message": { + "type": "string", + "value": "cancel λ" + }, + "count": { + "type": "int64", + "value": "9007199254740993" + }, + "optional": { + "type": "null", + "value": null + } + } + }, + "activity": false, + "immediate_cancellation": { + "command": "terminate", + "phase": "pending_timer", + "receipt_target_scope": {"http": "instance", "embedded": "run"}, + "reason": "parity cancellation λ", + "delay_seconds": 60 + }, + "expected_events": [ + "StartAccepted", + "WorkflowStarted", + "TimerScheduled", + "TerminateRequested", + "TimerCancelled", + "WorkflowTerminated" + ], + "contract": "Immediate selected-run termination shares the published cancellation implementation and records the termination identity: TerminateRequested and WorkflowTerminated, the terminated run status, closed reason and failure category, and WorkflowTerminatedException. Open tasks, activities and timers are cancelled as for cancellation, and a late activity outcome is refused with run_terminated. Repeating termination is refused with run_not_active and unchanged history. These fixtures reuse the cancellation fixtures' workflows. Terminating a child run is not qualified." +} diff --git a/tests/Fixtures/ServerParity/worker-listing.json b/tests/Fixtures/ServerParity/worker-listing.json new file mode 100644 index 000000000..f92676d4f --- /dev/null +++ b/tests/Fixtures/ServerParity/worker-listing.json @@ -0,0 +1,28 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "worker-listing", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "worker listing λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "worker listing λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "worker_listing": { + "workflow_type": "parity.v1.listing", + "other_workflow_type": "parity.v1.listing_other", + "activity_type": "parity.v1.listing_activity", + "build_id": "listing-build-λ", + "fingerprint": "sha256:1111111111111111111111111111111111111111111111111111111111111111", + "changed_fingerprint": "sha256:2222222222222222222222222222222222222222222222222222222222222222", + "capacity": {"workflow": 3, "activity": 5, "sessions": 2}, + "task_slots": {"workflow_available": 2, "activity_available": 9, "session_available": 1}, + "stored_task_slots": {"workflow_available": 2, "activity_available": 5, "session_available": 1}, + "stale_after_seconds": 30, + "heartbeat_interval_seconds": 10 + }, + "contract": "Workers registered through the unchanged published SDK are listed and described with the registration and heartbeat facts they sent: capacities, task slots bounded by those capacities, process metrics, definition fingerprints and build ID. The task_queue and status filters select among live workers; a missing worker is 404 worker_not_found. Re-registering the same worker_id with a changed workflow definition fingerprint is refused with 409 workflow_definition_changed, and omitting a previously advertised fingerprint with 409 workflow_definition_fingerprint_missing; neither refusal changes the stored registration. A later valid re-registration keeps the advertised fingerprint and resets the task slots it did not report. Staleness after the 30-second window and the heartbeat's retention report need separate fixtures. Embedded applications have no HTTP worker registration." +} diff --git a/tests/Fixtures/ServerParity/worker-replacement-activity-lease.json b/tests/Fixtures/ServerParity/worker-replacement-activity-lease.json new file mode 100644 index 000000000..d75591ee0 --- /dev/null +++ b/tests/Fixtures/ServerParity/worker-replacement-activity-lease.json @@ -0,0 +1,20 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "worker-replacement-activity-lease", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "worker replacement λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "worker replacement λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "worker_replacement": { + "workflow_type": "parity.v1.replacement", + "activity_type": "parity.v1.replacement_activity", + "expected_events": ["StartAccepted", "WorkflowStarted", "ActivityScheduled", "ActivityStarted", "ActivityStarted", "ActivityCompleted", "WorkflowCompleted"] + }, + "contract": "A worker process that dies holding an activity lease is replaced by a new process registering the same worker_id. The registration releases the lease at once: the original attempt is closed and the same activity task is claimed again as attempt 2, even under a one-attempt retry policy, with a second ActivityStarted. The original attempt's late completion is refused and records nothing; the replacement attempt's completion is recorded once. Embedded applications have no HTTP worker registration." +} diff --git a/tests/Unit/ServerParityRunnerTest.mjs b/tests/Unit/ServerParityRunnerTest.mjs index 4c1629f60..081b0372b 100644 --- a/tests/Unit/ServerParityRunnerTest.mjs +++ b/tests/Unit/ServerParityRunnerTest.mjs @@ -3,6 +3,7 @@ import assert from 'node:assert/strict'; import {readFileSync} from 'node:fs'; import {checkObservation, compareRecords as compareCorpusRecords} from '../../scripts/conformance/server-parity/contract.mjs'; import {parseDatabaseResourceLimits} from '../../scripts/conformance/server-parity/database-resource-contract.mjs'; +import {immediateCommand} from '../../scripts/conformance/server-parity/cancellation-contract.mjs'; import './ServerParityCrossKindPollWakeTest.mjs'; const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/one-activity.json', import.meta.url))); @@ -10,8 +11,8 @@ const reviewed = {'one-activity.json': fixture}; const retryFixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/activity-retry.json', import.meta.url))); const exhaustedFixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/activity-failure-exhausted.json', import.meta.url))); const filteredFixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/activity-failure-filtered.json', import.meta.url))); -const cancellationFixtures = ['before-claim', 'pending-timer', 'leased-activity'].map(phase => - JSON.parse(readFileSync(new URL(`../Fixtures/ServerParity/cancel-${phase}.json`, import.meta.url)))); +const cancellationFixtures = ['cancel', 'terminate'].flatMap(command => ['before-claim', 'pending-timer', 'leased-activity'].map(phase => + JSON.parse(readFileSync(new URL(`../Fixtures/ServerParity/${command}-${phase}.json`, import.meta.url))))); const nulCancellationFixtures = ['leading', 'interior'].map(position => JSON.parse(readFileSync(new URL(`../Fixtures/ServerParity/cancel-${position}-nul-reason.json`, import.meta.url)))); const compareRecords = records => compareCorpusRecords(records, {'one-activity.json': 'fixture-sha'}, reviewed); @@ -70,10 +71,11 @@ test('database memory receipt rejects missing, inexact, unknown or swap-enabled function cancellationObservation(fixture, mode = 'http') { const raw = observation(); const {phase} = fixture.immediate_cancellation; + const closing = immediateCommand(fixture); const reason = mode === 'http' ? fixture.immediate_cancellation.http_reason ?? fixture.immediate_cancellation.reason : fixture.immediate_cancellation.reason; - Object.assign(raw, {mode, workflow_type: fixture.workflow_type, status: 'cancelled', + Object.assign(raw, {mode, workflow_type: fixture.workflow_type, status: closing.closed, input: [fixture.input], typed_input: {type: 'list', value: [fixture.typed_value]}, output: null, typed_output: {type: 'null', value: null}}); raw.events = raw.events.slice(0, 2); @@ -87,7 +89,7 @@ function cancellationObservation(fixture, mode = 'http') { {payload: {...structuredClone(activityBase), activity_attempt_id: 'cancel-attempt', attempt_number: 1, task: {id: 'cancel-task', lease_owner: 'cancel-owner'}}, decoded: {activity_arguments: [fixture.input]}, typed_decoded: {activity_arguments: {type: 'list', value: [fixture.typed_value]}}}); const common = {workflow_instance_id: raw.workflow_id, workflow_run_id: raw.run_id, workflow_command_id: 'cancel-command', reason}; - raw.events.push({payload: {...common, command_type: 'cancel'}, decoded: {}, typed_decoded: {}}); + raw.events.push({payload: {...common, command_type: closing.name}, decoded: {}, typed_decoded: {}}); if (phase === 'pending_timer') raw.events.push({payload: {...raw.events[2].payload, cancelled_at: '2026-01-01T00:00:04Z'}, decoded: {}, typed_decoded: {}}); if (phase === 'leased_activity') raw.events.push({payload: {...structuredClone(activityBase), workflow_command_id: 'cancel-command', activity_attempt_id: 'cancel-attempt', attempt_number: 1, @@ -95,27 +97,27 @@ function cancellationObservation(fixture, mode = 'http') { activity_attempt: {id: 'cancel-attempt', activity_execution_id: 'cancel-activity', task_id: 'cancel-task', lease_owner: 'cancel-owner', status: 'cancelled', closed_at: '2026-01-01T00:00:05Z'}}, decoded: {activity_arguments: [fixture.input]}, typed_decoded: {activity_arguments: {type: 'list', value: [fixture.typed_value]}}}); - raw.events.push({payload: {...common, failure_id: 'cancel-failure', failure_category: 'cancelled', closed_reason: 'cancelled', - exception_class: 'Workflow\\V2\\Exceptions\\WorkflowCancelledException', message: + raw.events.push({payload: {...common, failure_id: 'cancel-failure', failure_category: closing.closed, closed_reason: closing.closed, + exception_class: closing.exception, message: fixture.immediate_cancellation.diagnostic_encoding === 'json_string_literal_if_nul' && reason.includes('\0') - ? JSON.stringify(`Workflow cancelled: ${reason}`) : `Workflow cancelled: ${reason}`}, decoded: {}, typed_decoded: {}}); + ? JSON.stringify(`Workflow ${closing.closed}: ${reason}`) : `Workflow ${closing.closed}: ${reason}`}, decoded: {}, typed_decoded: {}}); raw.events = raw.events.map((event, index) => ({...event, sequence: index + 1, event_type: fixture.expected_events[index], timestamp: `2026-01-01T00:00:0${index}Z`})); - Object.assign(raw.execution, {closed_reason: 'cancelled', closed_at: raw.events.at(-1).timestamp}); + Object.assign(raw.execution, {closed_reason: closing.closed, closed_at: raw.events.at(-1).timestamp}); const history = raw.events.map(({decoded, typed_decoded, ...event}) => event); - const accepted = {command_id: 'cancel-command', command_status: 'accepted', outcome: 'cancelled', command_sequence: 2, + const accepted = {command_id: 'cancel-command', command_status: 'accepted', outcome: closing.closed, command_sequence: 2, workflow_id: raw.workflow_id, run_id: raw.run_id, target_scope: fixture.immediate_cancellation.receipt_target_scope[mode]}; - const stale = {status: 409, response: {reason: 'run_cancelled', outcome: 'ignored', cancel_requested: true, can_continue: false, + const stale = {status: 409, response: {reason: `run_${closing.closed}`, outcome: 'ignored', cancel_requested: true, can_continue: false, lease_owner: 'cancel-owner', lease_expires_at: null, recorded: false, task_id: 'cancel-task', activity_attempt_id: 'cancel-attempt', activity_status: 'cancelled', attempt_status: 'cancelled', task_status: 'cancelled'}}; - raw.cancellation = {before: history.slice(0, history.findIndex(event => event.event_type === 'CancelRequested')), + raw.cancellation = {before: history.slice(0, history.findIndex(event => event.event_type === closing.requested)), accepted: {status: 200, response: accepted}, duplicate: {status: 409, response: {...accepted, command_id: 'rejected-cancel', command_sequence: 3, command_status: 'rejected', rejection_reason: 'run_not_active'}}, history_before_duplicate: history, history_after_duplicate: history, fresh_history: history, - fresh_execution: {id: raw.run_id, status: 'cancelled', output: null, cancelled: true}, - outcome: {type: 'DurableWorkflow\\Exception\\WorkflowCancelled', message: reason}, remaining_tasks: {workflow: null, activity: null}, + fresh_execution: {id: raw.run_id, status: closing.closed, output: null, cancelled: closing.name === 'cancel', terminated: closing.name === 'terminate'}, + outcome: {type: closing.sdk, message: reason}, remaining_tasks: {workflow: null, activity: null}, late_completion: phase === 'leased_activity' ? stale : null, redelivered_task_ids: phase === 'leased_activity' ? ['cancel-task'] : [], failures: [{id: 'cancel-failure', workflow_run_id: raw.run_id, source_kind: 'workflow_run', source_id: raw.run_id, - propagation_kind: 'cancelled', failure_category: 'cancelled', exception_class: raw.events.at(-1).payload.exception_class, + propagation_kind: closing.closed, failure_category: closing.closed, exception_class: raw.events.at(-1).payload.exception_class, message: raw.events.at(-1).payload.message, handled: false}], tasks: [{id: 'cancel-task', status: 'cancelled', lease_expires_at: null}], attempts: [{id: 'cancel-attempt', workflow_task_id: 'cancel-task', status: 'cancelled', lease_expires_at: null}]}; raw.workflow_polls = phase === 'before_claim' ? [] : [{}]; @@ -183,6 +185,19 @@ for (const fixture of cancellationFixtures) { assert.throws(() => checkObservation(fixture, raw, 'test-one-activity')); }); } +for (const fixture of cancellationFixtures.filter(fixture => fixture.immediate_cancellation.command === 'terminate')) { + for (const [label, mutate] of [ + ['a cancel command', raw => { raw.events.find(event => event.event_type === 'TerminateRequested').payload.command_type = 'cancel'; }], + ['a cancelled failure category', raw => { raw.events.at(-1).payload.failure_category = 'cancelled'; }], + ['the cancellation exception', raw => { raw.events.at(-1).payload.exception_class = 'Workflow\\V2\\Exceptions\\WorkflowCancelledException'; }], + ['a cancelled run', raw => { raw.status = 'cancelled'; }], + ['a cancelled SDK outcome', raw => { raw.cancellation.outcome.type = 'DurableWorkflow\\Exception\\WorkflowCancelled'; }], + ]) test(`${fixture.id} rejects ${label}`, () => { + const raw = cancellationObservation(fixture); + mutate(raw); + assert.throws(() => checkObservation(fixture, raw, 'test-one-activity')); + }); +} for (const [label, mutate] of [ ['recomputed deadline', raw => { raw.events[4].payload.fire_at = '2026-01-01T00:02:02Z'; }], ['replacement timer', raw => { raw.events[4].payload.timer_id = 'replacement'; }], diff --git a/tests/Unit/ServerParityWorkerListingContractTest.mjs b/tests/Unit/ServerParityWorkerListingContractTest.mjs new file mode 100644 index 000000000..fe3c89e7a --- /dev/null +++ b/tests/Unit/ServerParityWorkerListingContractTest.mjs @@ -0,0 +1,85 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkWorkerListing} from '../../scripts/conformance/server-parity/worker-listing-contract.mjs'; + +// A modeled observation shaped like published PHP's. It checks the contract +// itself; only actual recordings qualify a server. +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/worker-listing.json', import.meta.url))); +const definition = fixture.worker_listing; +const id = 'listing-reference'; +const [first, second] = [`${id}-listing-first`, `${id}-listing-second`]; +const queues = [`${id}-listing-a`, `${id}-listing-b`]; +const capabilities = ['query_tasks', 'workflow_updates']; +const manifest = {local_activities: {supported: false, minimum_protocol_version: '1.18', reason: 'not_enabled_for_this_worker_registration'}}; +const at = '2026-10-20T14:49:45.000000Z'; +const receipt = (workerId, queue, buildId) => ({worker_id: workerId, registered: true, namespace: 'default', task_queue: queue, + runtime: 'php', build_id: buildId, capabilities, capability_manifest: manifest, status: 'active', + heartbeat_interval_seconds: definition.heartbeat_interval_seconds, protocol_version: '1.20'}); +const capacity = definition.capacity; +const firstWorker = {worker_id: first, namespace: 'default', task_queue: queues[0], runtime: 'php', + sdk_version: 'durable-workflow-php/2.2.6', build_id: definition.build_id, supported_workflow_types: [definition.workflow_type], + workflow_definition_fingerprints: {[definition.workflow_type]: definition.fingerprint}, workflow_command_contracts: [], + supported_activity_types: [definition.activity_type], capabilities, capability_manifest: manifest, + max_concurrent_workflow_tasks: capacity.workflow, max_concurrent_activity_tasks: capacity.activity, + max_concurrent_worker_sessions: capacity.sessions, task_slots: {...definition.stored_task_slots, + workflow_capacity: capacity.workflow, activity_capacity: capacity.activity, session_capacity: capacity.sessions}, + process_metrics: {process_uptime_seconds: 0, process_id: 17}, heartbeat_interval_seconds: null, status: 'active', + last_heartbeat_at: at, registered_at: at}; +const secondWorker = {...firstWorker, worker_id: second, task_queue: queues[1], build_id: null, + workflow_definition_fingerprints: [], supported_activity_types: [], max_concurrent_workflow_tasks: 1, + max_concurrent_activity_tasks: 1, max_concurrent_worker_sessions: 10, + task_slots: {workflow_available: null, activity_available: null, session_available: null, + workflow_capacity: 1, activity_capacity: 1, session_capacity: 10}, process_metrics: null}; +const listing = workers => ({stale_after_seconds: definition.stale_after_seconds, workers}); +const described = {...firstWorker, updated_at: at, stale_after_seconds: definition.stale_after_seconds}; +const refusal = (reason, error, remediation) => ({status: 409, response: {error, reason, + workflow_type: definition.workflow_type, remediation, protocol_version: '1.20'}}); +const observation = () => structuredClone({mode: 'http', worker_listing: {applicable: true, first, second, queues, probe_pid: 17, + first_registration: receipt(first, queues[0], definition.build_id), second_registration: receipt(second, queues[1], null), + reregistration: receipt(first, queues[0], definition.build_id), + heartbeat: {worker_id: first, acknowledged: true, heartbeat_interval_seconds: 10, stale_after_seconds: 30, protocol_version: '1.20'}, + listed: listing([secondWorker, firstWorker]), first_queue: listing([firstWorker]), active: listing([firstWorker, secondWorker]), + stale: listing([]), after_deregistration: listing([]), described, + missing: {status: 404, response: {reason: 'worker_not_found', message: `Worker [${id}-listing-missing] not found in namespace [default].`}}, + changed_definition: refusal('workflow_definition_changed', 'Worker attempted to re-register a changed workflow definition.', + 'Restart the worker with a new worker_id before registering a changed workflow class definition.'), + missing_fingerprint: refusal('workflow_definition_fingerprint_missing', 'Worker omitted a previously advertised workflow definition fingerprint.', + 'Advertise the matching definition fingerprint or restart with a new worker_id.'), + after_refusals: described, + after_reregistration: {...described, task_slots: {...described.task_slots, workflow_available: null, + activity_available: null, session_available: null}, process_metrics: null}}}); + +test('worker listing accepts the published facts in any equal-heartbeat order', () => { + assert.deepStrictEqual(checkWorkerListing(fixture, observation(), id), + {scope: 'worker_listing', workflow_type: definition.workflow_type}); +}); + +test('embedded worker listing is explicitly inapplicable', () => { + assert.deepStrictEqual(checkWorkerListing(fixture, {mode: 'embedded', worker_listing: + {applicable: false, reason: 'embedded_has_no_http_worker_registration_lifecycle'}}, id), + checkWorkerListing(fixture, observation(), id)); +}); + +for (const [label, mutate] of [ + ['an unbounded task slot', state => { state.listed.workers[1].task_slots.activity_available = 9; }], + ['default capacities in place of the registered ones', state => { state.described.max_concurrent_workflow_tasks = 100; }], + ['a dropped fingerprint', state => { state.listed.workers[1].workflow_definition_fingerprints = []; }], + ['an empty map rendered as an object', state => { state.listed.workers[0].workflow_definition_fingerprints = {}; }], + ['missing process metrics', state => { state.described.process_metrics = null; }], + ['an extra summary field', state => { state.listed.workers[0].invented = true; }], + ['an unfiltered task queue', state => { state.first_queue.workers.push(structuredClone(secondWorker)); }], + ['a live worker listed as stale', state => { state.stale.workers.push(structuredClone(firstWorker)); }], + ['an accepted definition change', state => { state.changed_definition = {status: 201, response: {}}; }], + ['a refusal that changed the registration', state => { state.after_refusals.task_slots.workflow_available = 1; }], + ['a lost advertised fingerprint', state => { state.after_reregistration.workflow_definition_fingerprints = []; }], + ['slots kept across re-registration', state => { state.after_reregistration.task_slots.workflow_available = 2; }], + ['a replaced registration time', state => { state.after_reregistration.registered_at = '2026-10-20T15:00:00.000000Z'; }], + ['a wrong stale window', state => { state.heartbeat.stale_after_seconds = 300; }], + ['a worker left after deregistration', state => { state.after_deregistration.workers.push(structuredClone(firstWorker)); }], + ['a different missing-worker message', state => { state.missing.response.message = 'Worker not found.'; }], +]) test(`worker listing rejects ${label}`, () => { + const value = observation(); + mutate(value.worker_listing); + assert.throws(() => checkWorkerListing(fixture, value, id)); +}); diff --git a/tests/Unit/ServerParityWorkerReplacementContractTest.mjs b/tests/Unit/ServerParityWorkerReplacementContractTest.mjs new file mode 100644 index 000000000..50c1277c5 --- /dev/null +++ b/tests/Unit/ServerParityWorkerReplacementContractTest.mjs @@ -0,0 +1,56 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkWorkerReplacement} from '../../scripts/conformance/server-parity/worker-replacement-contract.mjs'; + +// A modeled observation shaped like published PHP's. It checks the contract +// itself; only actual recordings qualify a server. +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/worker-replacement-activity-lease.json', import.meta.url))); +const id = 'replacement-reference'; +const worker = `${id}-replaced-worker`; +const [task, first, second] = ['task-1', 'attempt-1', 'attempt-2']; +const claim = (attempt, number) => ({task_id: task, activity_attempt_id: attempt, attempt_number: number, lease_owner: worker}); +const event = (event_type, payload = {}) => ({event_type, payload}); +const history = [event('StartAccepted'), event('WorkflowStarted'), event('ActivityScheduled'), + event('ActivityStarted', {activity_attempt_id: first, attempt_number: 1}), + event('ActivityStarted', {activity_attempt_id: second, attempt_number: 2}), + event('ActivityCompleted', {activity_attempt_id: second, attempt_number: 2}), event('WorkflowCompleted')]; +const observation = () => structuredClone({mode: 'http', worker_replacement: {applicable: true, worker_id: worker, + task_queue: `${id}-replacement`, peer_workflow_id: `${id}-replacement-peer`, + registration: {worker_id: worker, registered: true}, replacement_registration: {worker_id: worker, registered: true}, + original_claim: claim(first, 1), replacement_claim: claim(second, 2), history_after_replacement: history.slice(0, 4), + stale_completion: {status: 409, response: {task_id: task, activity_attempt_id: first, outcome: 'completed', recorded: false, + reason: 'stale_attempt', next_task_id: null, lease_owner: worker, activity_status: 'running', attempt_status: 'expired', + task_status: 'leased', protocol_version: '1.20'}}, + completion: {status: 200, response: {task_id: task, activity_attempt_id: second, outcome: 'completed', recorded: true, + reason: null, next_task_id: 'workflow-task'}}, + history}}); + +test('worker replacement accepts the published release and redelivery', () => { + assert.deepStrictEqual(checkWorkerReplacement(fixture, observation(), id), + {scope: 'worker_replacement', expected_events: fixture.worker_replacement.expected_events}); +}); + +test('embedded worker replacement is explicitly inapplicable', () => { + assert.deepStrictEqual(checkWorkerReplacement(fixture, {mode: 'embedded', worker_replacement: + {applicable: false, reason: 'embedded_has_no_http_worker_registration_lifecycle'}}, id), + checkWorkerReplacement(fixture, observation(), id)); +}); + +for (const [label, mutate] of [ + ['a lease that waits out its expiry', state => { state.replacement_claim = null; }], + ['a different task', state => { state.replacement_claim.task_id = 'task-2'; }], + ['a reused attempt number', state => { state.replacement_claim.attempt_number = 1; }], + ['a reused attempt identity', state => { state.replacement_claim.activity_attempt_id = first; }], + ['history recorded at release', state => { state.history_after_replacement.push(event('ActivityFailed')); }], + ['an accepted stale completion', state => { state.stale_completion = {status: 200, response: {recorded: true}}; }], + ['a bare stale refusal', state => { state.stale_completion.response = {reason: 'stale_attempt'}; }], + ['an open original attempt', state => { state.stale_completion.response.attempt_status = 'running'; }], + ['an unrecorded replacement', state => { state.completion.response.recorded = false; }], + ['the original attempt completing', state => { state.history[5].payload.activity_attempt_id = first; }], + ['a missing second start', state => { state.history.splice(4, 1); }], +]) test(`worker replacement rejects ${label}`, () => { + const value = observation(); + mutate(value.worker_replacement); + assert.throws(() => checkWorkerReplacement(fixture, value, id)); +});