From a313004ae6fcd4f8405a596daf6e040df2e79ca6 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 15:10:10 +0000 Subject: [PATCH 1/6] List and describe workers natively, as published PHP does Native serves GET /api/workers and GET /api/workers/{id} with PHP's summary, filters, stale window and refusals. Registration and heartbeat now keep the facts PHP reports: capacities, bounded task slots, process metrics, heartbeat cadence and workflow definition fingerprints. Re-registering with a changed or omitted fingerprint is refused, as PHP refuses it. A worker-listing fixture drives this through the unchanged published SDK. Part of #325. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 4 +- docs/rust-server-port.md | 44 ++ rust/src/runtime/execution.rs | 34 +- rust/src/runtime/http.rs | 33 +- rust/src/runtime/memos.rs | 2 +- rust/src/runtime/mod.rs | 3 +- rust/src/runtime/query_authority.rs | 5 + rust/src/runtime/store.rs | 48 +- rust/src/runtime/worker_management.rs | 656 ++++++++++++++++++ rust/tests/execution.rs | 3 + rust/tests/execution/worker_listing.rs | 292 ++++++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../server-parity/worker-listing-contract.mjs | 112 +++ .../server-parity/worker-listing-probe.php | 55 ++ .../Fixtures/ServerParity/worker-listing.json | 28 + .../ServerParityWorkerListingContractTest.mjs | 85 +++ 17 files changed, 1391 insertions(+), 22 deletions(-) create mode 100644 rust/src/runtime/worker_management.rs create mode 100644 rust/tests/execution/worker_listing.rs create mode 100644 scripts/conformance/server-parity/worker-listing-contract.mjs create mode 100644 scripts/conformance/server-parity/worker-listing-probe.php create mode 100644 tests/Fixtures/ServerParity/worker-listing.json create mode 100644 tests/Unit/ServerParityWorkerListingContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index c4bb53b6c..e2409c4fe 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -24,6 +24,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityWorkerListingContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -61,6 +62,7 @@ on: - tests/Unit/ServerParityAdmissionContractTest.mjs - tests/Unit/ServerParityRoleTokenContractTest.mjs - tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs + - tests/Unit/ServerParityWorkerListingContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -1405,7 +1407,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/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs > parity-evidence/comparator-tests.txt docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \ --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \ "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 349f23089..8ae59bd2f 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -85,6 +85,16 @@ 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. +- **Activity leases of a replaced worker.** Re-registering a worker that holds + leases makes published PHP release its workflow *and* activity task leases. + Native releases only the workflow task leases. This needs a fixture with a + leased activity before the correction. + ## Contract and verification inventory The status column summarizes bounded native source and its recorded component @@ -152,6 +162,40 @@ 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. +- **Fixture.** `worker-listing` registers two workers through the unchanged + published SDK. It then lists, filters, describes and refuses them, and + re-registers and deregisters them. The contract checks 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 `worker-listing` and + `namespace-isolation`. All three targets passed and the comparison passed. + Native integration tests pass on SQLite, PostgreSQL, MySQL and MariaDB. + October 19, before #478, #476, #494 and #460 merged: - **Open native PRs.** [#478](https://github.com/durable-workflow/server/pull/478) diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 5715137b8..c029850ba 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 d09fe705f..39168a752 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -144,7 +144,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", @@ -266,7 +270,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") @@ -405,6 +409,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!( @@ -876,6 +887,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..7439be687 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 => { @@ -861,28 +883,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( diff --git a/rust/src/runtime/worker_management.rs b/rust/src/runtime/worker_management.rs new file mode 100644 index 000000000..41bfb10e8 --- /dev/null +++ b/rust/src/runtime/worker_management.rs @@ -0,0 +1,656 @@ +//! 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 + } + + 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 a9097d460..846c0db74 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..61960a9ef --- /dev/null +++ b/rust/tests/execution/worker_listing.rs @@ -0,0 +1,292 @@ +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; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index e60868655..3fd2bd003 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -9,6 +9,7 @@ import {checkVisibility} from './visibility-contract.mjs'; import {checkAdmission} from './admission-contract.mjs'; import {checkNamespaces} from './namespace-contract.mjs'; import {checkWorkerDeregistration} from './worker-deregistration-contract.mjs'; +import {checkWorkerListing} from './worker-listing-contract.mjs'; import {checkPatchDeployment, checkPublishedPatchArtifacts, checkPatchPackageObservation} from './patch-deployment-contract.mjs'; import {checkLegacyMarkerArtifacts, checkLegacyMarkerPackageObservation, checkLegacyMarkerDeployment} from './legacy-marker-contract.mjs'; import {checkWaitingHistory} from './waiting-history-contract.mjs'; @@ -128,6 +129,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const admission = fixture.admission ? checkAdmission(fixture, observation, workflowId) : null; const namespaces = fixture.namespace_isolation ? checkNamespaces(fixture, observation, workflowId, checkObservation) : null; const workerDeregistration = fixture.worker_deregistration ? checkWorkerDeregistration(fixture, observation, workflowId) : null; + const workerListing = fixture.worker_listing ? checkWorkerListing(fixture, observation, workflowId) : null; const patchDeployment = fixture.patch_deployment ? (fixture.patch_deployment.legacy_marker_history ? checkLegacyMarkerDeployment(fixture, observation, workflowId) : checkPatchDeployment(fixture, observation, workflowId)) : null; const waitingHistory = fixture.waiting_for_history ? checkWaitingHistory(fixture, observation, workflowId) : null; @@ -665,6 +667,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(admission ? {admission} : {}), ...(namespaces ? {namespace_isolation: namespaces} : {}), ...(workerDeregistration ? {worker_deregistration: workerDeregistration} : {}), + ...(workerListing ? {worker_listing: workerListing} : {}), ...(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 d63bc0f8d..b91e8dbdb 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -425,6 +425,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $handle->selectedRunId, $queue, $url, $namespace) : null; $namespaces = isset($fixture['namespace_isolation']) ? finishHttpNamespaces($client, $fixture, $workflowId, $queue, $url) : null; $workerDeregistration = isset($fixture['worker_deregistration']) ? finishHttpWorkerDeregistration($client, $fixture, $workflowId, $queue, $url) : null; + $workerListing = isset($fixture['worker_listing']) ? finishHttpWorkerListing($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) { @@ -511,6 +512,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'admission' => $admission, 'namespace_isolation' => $namespaces, 'worker_deregistration' => $workerDeregistration, + 'worker_listing' => $workerListing, 'waiting_for_history' => $waitingHistory, 'workflow_task_retry' => $taskRetry, 'run_id' => $execution->runId, @@ -807,6 +809,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/admission-probe.php'; require __DIR__.'/namespace-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; + require __DIR__.'/worker-listing-probe.php'; require __DIR__.'/waiting-history-probe.php'; require __DIR__.'/task-retry-probe.php'; require __DIR__.'/child-cancellation-probe.php'; @@ -937,6 +940,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa }; } $observation['typed_input'] = typedValue($observation['input']); + 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/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/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)); +}); From 0cba29252178858d07090ec86c5b8abfb123eaa2 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 15:41:38 +0000 Subject: [PATCH 2/6] Release a replaced worker's activity leases at once, as PHP does A new process registering a worker_id that holds activity leases now closes its running attempts as expired and returns the tasks to ready, so the next claim is attempt 2 of the same task. Native used to leave the task leased for the whole five-minute activity lease. The expired attempt's late completion gets PHP's whole stale receipt. A worker-replacement-activity-lease fixture checks this through the unchanged published SDK. Part of #325. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/server-parity.yml | 4 +- docs/rust-server-port.md | 32 +++++++--- rust/src/runtime/store.rs | 25 +++++++- rust/src/runtime/worker_management.rs | 31 ++++++++++ rust/tests/execution/worker_listing.rs | 61 +++++++++++++++++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 ++ .../worker-replacement-contract.mjs | 54 ++++++++++++++++ .../worker-replacement-probe.php | 54 ++++++++++++++++ .../worker-replacement-activity-lease.json | 20 ++++++ ...verParityWorkerReplacementContractTest.mjs | 56 +++++++++++++++++ 11 files changed, 334 insertions(+), 12 deletions(-) create mode 100644 scripts/conformance/server-parity/worker-replacement-contract.mjs create mode 100644 scripts/conformance/server-parity/worker-replacement-probe.php create mode 100644 tests/Fixtures/ServerParity/worker-replacement-activity-lease.json create mode 100644 tests/Unit/ServerParityWorkerReplacementContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index e2409c4fe..1e1c5f741 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -25,6 +25,7 @@ on: - 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 @@ -63,6 +64,7 @@ on: - 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 @@ -1407,7 +1409,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/ServerParityWorkerListingContractTest.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 09665ff4b..73402664b 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -90,10 +90,6 @@ contract fixture before its correction. 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. -- **Activity leases of a replaced worker.** Re-registering a worker that holds - leases makes published PHP release its workflow *and* activity task leases. - Native releases only the workflow task leases. This needs a fixture with a - leased activity before the correction. - **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` @@ -200,14 +196,30 @@ October 20, worker listing: 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. -- **Fixture.** `worker-listing` registers two workers through the unchanged - published SDK. It then lists, filters, describes and refuses them, and - re-registers and deregisters them. The contract checks each HTTP target - against the reviewed facts. Embedded mode is explicitly inapplicable. +- **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 `worker-listing` and + published PHP 2.5.14, embedded PHP and native for both fixtures and `namespace-isolation`. All three targets passed and the comparison passed. - Native integration tests pass on SQLite, PostgreSQL, MySQL and MariaDB. + 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: diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 7439be687..f294dfff8 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -864,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()) @@ -2209,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 index 41bfb10e8..9aa709ac6 100644 --- a/rust/src/runtime/worker_management.rs +++ b/rust/src/runtime/worker_management.rs @@ -377,6 +377,37 @@ where .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, diff --git a/rust/tests/execution/worker_listing.rs b/rust/tests/execution/worker_listing.rs index 61960a9ef..77ea5d0dd 100644 --- a/rust/tests/execution/worker_listing.rs +++ b/rust/tests/execution/worker_listing.rs @@ -290,3 +290,64 @@ async fn named_namespaces_list_only_their_own_workers() { 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 cc6708b40..d688dc6df 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -10,6 +10,7 @@ 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'; @@ -131,6 +132,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { 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; @@ -669,6 +671,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(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 b91e8dbdb..961a3a856 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -426,6 +426,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $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 +514,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'namespace_isolation' => $namespaces, 'worker_deregistration' => $workerDeregistration, 'worker_listing' => $workerListing, + 'worker_replacement' => $workerReplacement, 'waiting_for_history' => $waitingHistory, 'workflow_task_retry' => $taskRetry, 'run_id' => $execution->runId, @@ -810,6 +812,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa 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'; @@ -940,6 +943,9 @@ 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']; } 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-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/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)); +}); From afef428688c508640e0921bc8e4751d82fedcdf2 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 15:44:56 +0000 Subject: [PATCH 3/6] Record current PHP's workflow definition evolution as a gap to follow Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 73402664b..465987073 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -98,6 +98,14 @@ 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. - **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 From 05c5fc481fd110089aff9b98b95e5245125c7408 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 15:47:55 +0000 Subject: [PATCH 4/6] Bring the port log's current state and missing API surface up to date Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 92 ++++++++++++++++++++++++++++++---------- 1 file changed, 69 insertions(+), 23 deletions(-) diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 465987073..aa7d483c1 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`. +- **Open and qualified.** [#515](https://github.com/durable-workflow/server/pull/515) + retries the PHP reference's lock-pressure refusals in the probes. It passed + all 13 jobs and merges first. +- **Open, waiting for #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.** Merge #515. Then 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 From 3c112243f3ee75178b13cc5b84e6eb0bb4592ea9 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 16:03:36 +0000 Subject: [PATCH 5/6] Note #515 as merged in the port log Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index aa7d483c1..f798b1ff4 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -24,10 +24,10 @@ Updated October 20, 2026. - [#517](https://github.com/durable-workflow/server/pull/517) designs the backup-first takeover. Main's 13-job parity run passed at `ee88ac6ba`. -- **Open and qualified.** [#515](https://github.com/durable-workflow/server/pull/515) - retries the PHP reference's lock-pressure refusals in the probes. It passed - all 13 jobs and merges first. -- **Open, waiting for #515.** Each of these failed or deferred parity only in +- **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). @@ -45,7 +45,7 @@ Updated October 20, 2026. - **Parity surface.** With this branch, native serves 63 of published PHP's 173 API routes. The missing families are listed below under "API surface still missing". -- **Next action.** Merge #515. Then run parity once on each stack top (#510, +- **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 From aa351154b6b63ab8d81c6fbe779bc0d4422dbb5f Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 16:10:18 +0000 Subject: [PATCH 6/6] Record the agreed native boundary for Workflow trace context Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index f798b1ff4..9e8a2faab 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -152,6 +152,14 @@ contract fixture before its correction. 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