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 76ad3655d..9e8a2faab 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, and CPU observer client timing. Diagnosed, decision - pending: the query-authority deadline-boundary replay (see known gaps). - 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.** Decide the query deadline-boundary replay contract and add - its fixture. Then 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. + - [#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. + + Main's 13-job parity run passed at `ee88ac6ba`. +- **Also merged.** [#515](https://github.com/durable-workflow/server/pull/515) + retries the PHP reference's lock-pressure refusals in the probes, after + passing all 13 parity jobs. +- **Open, qualifying on main with #515.** Each of these failed or deferred parity only in + steps #515 corrects or that it doesn't touch. + - The query pair: [#499](https://github.com/durable-workflow/server/pull/499), + then [#510](https://github.com/durable-workflow/server/pull/510). + - 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). + - Immediate termination, and worker listing with registration facts. +- **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.** Run parity once on each stack top (#510, + #539 and the worker listing branch) and merge each stack. Then port history + export, a large bundle the CLI and both SDKs use for replay. Multi-node + qualification and packaging the native binary into the server image remain + open gates. + +## 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 @@ -85,6 +131,11 @@ contract fixture before its correction. or whether the fixture counts a boundary replay as the same claim, then add the fixture before changing either side. +- **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` @@ -93,6 +144,22 @@ contract fixture before its correction. 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 @@ -166,6 +233,56 @@ 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 diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 10ce77203..ddf4cf92f 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -705,8 +705,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(); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index bff4d7e08..d2a4fd1a9 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -152,7 +152,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", @@ -274,7 +278,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 || path.starts_with("/api/namespaces/") && path.ends_with("/external-storage") @@ -413,6 +417,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!( @@ -932,6 +943,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 f944c8c58..f294dfff8 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 fd70a1286..35a1e6729 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; 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/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 461a5efe3..d688dc6df 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -9,6 +9,8 @@ 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'; @@ -129,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; @@ -666,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/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/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)); +});