From 8ab5cc6be094e4476e3e5dc8364c1e4979a6df0a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 19:30:22 +0000 Subject: [PATCH 01/14] Show workers a consumed signal's sequence and keep full task snapshots, like Server 2.11.1 Worker polls and history pages add workflow_sequence to a consumed signal's SignalReceived from its record, as WorkflowTaskPoller does; the run history route still shows the event as recorded. Task snapshots keep compatibility, the dispatch and claim stamps and errors when they are set, as WorkflowHistoryEvent::taskSnapshot does. --- .../2026-10-23-worker-history-presentation.md | 28 ++++++++ rust/src/runtime/store.rs | 33 +++++++++- rust/src/runtime/worker_incarnations.rs | 10 ++- rust/tests/execution.rs | 2 + .../execution/worker_history_presentation.rs | 64 +++++++++++++++++++ 5 files changed, 133 insertions(+), 4 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md create mode 100644 rust/tests/execution/worker_history_presentation.rs diff --git a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md new file mode 100644 index 000000000..2109fcec2 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md @@ -0,0 +1,28 @@ +# October 23, what a worker reads in history + +- **Found by.** Comparing the message-stream fixture's worker polls against + published Server 2.11.1. +- **The gaps.** + - PHP's poll and history-page responses show a consumed signal's + `workflow_sequence` on its `SignalReceived` event, read from the signal's + record (`WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`). + Native showed the event as recorded. + - PHP's task snapshots in history keep every field that is set + (`WorkflowHistoryEvent::taskSnapshot`). Native's left out + `compatibility`, `last_dispatch_attempt_at`, `last_dispatched_at`, + `last_dispatch_error`, `last_claim_failed_at` and `last_claim_error`. +- **What native does now.** + - Worker polls and history pages add `workflow_sequence` to a consumed + signal's `SignalReceived`, when the payload does not have one. The + run's own history route still shows the events as recorded, as PHP's + does. + - Task snapshots keep those fields when they are set. +- **How it was checked.** A test runs on SQLite, PostgreSQL, MySQL and + MariaDB: + - a condition wait consumes an undeclared signal; + - the next poll and history page show that wait's sequence, and the + run's history does not; + - the completing task's snapshot has its dispatch stamps. +- **Not changed yet.** Some events PHP records with a task snapshot have + none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`. + That audit is a #325 follow-up. diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 93c299d12..03dfa699a 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1798,10 +1798,11 @@ where } } } - let (events, next) = match page { + let (mut events, next) = match page { Some(size) => Self::history_page_connection(&mut tx, &run_id, 0, size).await?, None => (Self::whole_history(&mut tx, &run_id).await?, None), }; + Self::with_signal_sequences(&mut tx, &run_id, &mut events).await?; let budget = Self::workflow_history_budget( &mut tx, &run_id, @@ -1865,7 +1866,8 @@ where } Self::workflow_task_guard(&mut tx, &task, &body).await?; let run_id = DB::string(&task, "workflow_run_id")?; - let (events, next) = Self::history_page_connection(&mut tx, &run_id, after, size).await?; + let (mut events, next) = Self::history_page_connection(&mut tx, &run_id, after, size).await?; + Self::with_signal_sequences(&mut tx, &run_id, &mut events).await?; let run = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") .bind(&run_id) .fetch_one(&mut *tx) @@ -3565,6 +3567,33 @@ where /// A poll's whole history, read in bounded pages: published PHP sends /// every event unless the worker asks for pages. + /// `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`: the + /// history a worker reads shows a consumed signal's workflow sequence on + /// its `SignalReceived`, from the signal's record. The run's own history + /// route shows the events as recorded. + async fn with_signal_sequences(connection: &mut DB::Connection, run_id: &str, events: &mut [Value]) -> Result<()> { + if !events.iter().any(|event| event["event_type"] == "SignalReceived") { + return Ok(()); + } + let rows = Self::query("SELECT id,workflow_sequence FROM workflow_signal_records WHERE workflow_run_id=$1 AND workflow_sequence IS NOT NULL") + .bind(run_id) + .fetch_all(&mut *connection) + .await?; + let mut sequences = std::collections::HashMap::new(); + for row in rows { + sequences.insert(DB::string(&row, "id")?, DB::number(&row, "workflow_sequence")?); + } + for event in events.iter_mut().filter(|event| event["event_type"] == "SignalReceived") { + let sequence = event["payload"]["signal_id"].as_str().and_then(|id| sequences.get(id)).copied(); + if let (Some(sequence), Some(payload)) = (sequence, event["payload"].as_object_mut()) { + if payload.get("workflow_sequence").is_none_or(Value::is_null) { + payload.insert("workflow_sequence".into(), json!(sequence)); + } + } + } + Ok(()) + } + async fn whole_history(connection: &mut DB::Connection, run_id: &str) -> Result> { let mut events = Vec::new(); let mut after = 0; diff --git a/rust/src/runtime/worker_incarnations.rs b/rust/src/runtime/worker_incarnations.rs index 2a83a3117..5685a92be 100644 --- a/rust/src/runtime/worker_incarnations.rs +++ b/rust/src/runtime/worker_incarnations.rs @@ -339,18 +339,24 @@ where "status":DB::string(task,"status")?,"attempt_count":DB::number(task,"attempt_count")?, "repair_count":DB::number(task,"repair_count")?,"queue":DB::string(task,"queue")?, "connection":DB::PHP_CONNECTION}); + // `WorkflowHistoryEvent::taskSnapshot` keeps every field that is set. for field in [ "available_at", "leased_at", "lease_expires_at", + "last_dispatch_attempt_at", + "last_dispatched_at", + "last_claim_failed_at", "repair_available_at", ] { if let Some(value) = DB::optional_instant(task, field)? { snapshot[field] = json!(value); } } - if let Some(owner) = DB::optional_string(task, "lease_owner")? { - snapshot["lease_owner"] = json!(owner); + for field in ["lease_owner", "compatibility", "last_dispatch_error", "last_claim_error"] { + if let Some(value) = DB::optional_string(task, field)? { + snapshot[field] = json!(value); + } } Ok(snapshot) } diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c3ce551e4..eabeeade2 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -109,6 +109,8 @@ mod parent_close_policies; mod history_pages; #[path = "execution/declared_signals.rs"] mod declared_signals; +#[path = "execution/worker_history_presentation.rs"] +mod worker_history_presentation; #[path = "execution/poll_waits.rs"] mod poll_waits; #[path = "execution/child_cascade.rs"] diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs new file mode 100644 index 000000000..db76a3d75 --- /dev/null +++ b/rust/tests/execution/worker_history_presentation.rs @@ -0,0 +1,64 @@ +use super::*; + +// Server 2.11.1 `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes` +// and `WorkflowHistoryEvent::taskSnapshot`: the history a worker reads shows a +// consumed signal's workflow sequence, and task snapshots keep their dispatch +// stamps. + +fn wait(key: &str) -> Value { + json!([{"type":"open_condition_wait","condition_key":key, + "condition_definition_fingerprint":format!("sha256:{}", "b".repeat(64))}]) +} + +fn received(history: &Value) -> Vec { + history.as_array().unwrap().iter().filter(|event| event["event_type"] == "SignalReceived") + .map(|event| event["payload"].clone()).collect() +} + +#[tokio::test] +async fn a_worker_reads_the_sequence_that_consumed_a_signal() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + // A run started before any worker declared a contract admits any signal. + let started = start(&app, "presented-signals").await; + let worker = json!({"worker_id":"presenter","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[]}); + assert_eq!(request(&app, "POST", "/api/worker/register", worker).await.0, StatusCode::CREATED); + let task = poll(&app, "presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let opened = history.as_array().unwrap().iter().find(|event| event["event_type"] == "ConditionWaitOpened").unwrap(); + let sequence = opened["payload"]["sequence"].clone(); + + let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/go", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "presenter", "workflow").await; + assert!(received(&task["history_events"])[0].get("workflow_sequence").is_none(), "not consumed yet: {task}"); + assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK); + + let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/again", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "presenter", "workflow").await; + let signals = received(&task["history_events"]); + assert_eq!(signals[0]["workflow_sequence"], sequence, "the consuming wait's sequence: {signals:?}"); + assert!(signals[1].get("workflow_sequence").is_none(), "{signals:?}"); + // A history page reads the same. + let (status, page) = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/history", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"]})).await; + assert_eq!(status, StatusCode::OK, "{page}"); + assert_eq!(received(&page["history_events"])[0]["workflow_sequence"], sequence, "{page}"); + // The run's own history shows the event as recorded. + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert!(received(&history)[0].get("workflow_sequence").is_none(), "{history}"); + + // The completing task was dispatched when the signal created it, and its + // snapshot keeps the dispatch stamps. + let done = json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]); + assert_eq!(finish_task(&app, &task, done).await.0, StatusCode::OK); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let completed = history.as_array().unwrap().iter().find(|event| event["event_type"] == "WorkflowCompleted").unwrap(); + for stamp in ["last_dispatch_attempt_at", "last_dispatched_at"] { + assert!(completed["payload"]["task"][stamp].is_string(), "{stamp}: {completed}"); + } + assert!(completed["payload"]["task"].get("last_claim_error").is_none(), "unset fields stay out: {completed}"); +} From 75c775edf1e5ab0e2b1af699014003e53f236e70 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 19:31:45 +0000 Subject: [PATCH 02/14] Record the dispatch-stamp difference the fuller task snapshots show --- .../2026-10-23-worker-history-presentation.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md index 2109fcec2..79271b917 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md @@ -26,3 +26,9 @@ - **Not changed yet.** Some events PHP records with a task snapshot have none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`. That audit is a #325 follow-up. +- **Known difference now visible in snapshots.** PHP stamps a dispatch only + on tasks a control-plane call dispatches, such as a signal's task. Tasks a + workflow task completion creates, such as the next message task, carry + none. Native stamps every new task, so those snapshots show stamps where + PHP's do not. This is the cause of the known `tasks_dispatched_last_minute` + difference too; fixing it is a #325 follow-up. From 8b0a8795d8488ea0414acc3350c6adff6191db76 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 20:28:48 +0000 Subject: [PATCH 03/14] Serve synchronous update validation like Server 2.11.1 An update whose durable contract declares a validator now waits for a validator-capable worker to approve it, as published PHP does: admission refusals in PHP's order, validation tasks keyed by run, update and request id, the dedicated and multiplexed worker polls, fenced approve and reject routes, lease expiry and re-claim, and 504 timeouts with their reasons. An accepted update's receipt no longer carries accepted: true, as PHP's result mapper drops it. The update-validation fixture records 49 steps against published PHP; native matches all of them. --- .github/workflows/server-parity.yml | 4 +- .../2026-10-23-update-validation.md | 85 ++ docs/rust-server-port.md | 7 +- rust/src/config_audit.rs | 31 +- rust/src/runtime/execution.rs | 91 +- rust/src/runtime/http.rs | 51 +- rust/src/runtime/mod.rs | 1 + rust/src/runtime/store.rs | 1 + rust/src/runtime/update_validation.rs | 1010 +++++++++++++++++ rust/src/runtime/updates.rs | 267 ++++- rust/tests/execution.rs | 2 + .../execution/external_payloads/commands.rs | 2 +- rust/tests/execution/rejected_updates.rs | 3 +- rust/tests/execution/update_validation.rs | 310 +++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../update-validation-contract.mjs | 64 ++ .../server-parity/update-validation-probe.php | 283 +++++ .../ServerParity/update-validation.json | 971 ++++++++++++++++ .../update-validation-observation.json | 1 + ...rverParityUpdateValidationContractTest.mjs | 66 ++ 21 files changed, 3225 insertions(+), 34 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-update-validation.md create mode 100644 rust/src/runtime/update_validation.rs create mode 100644 rust/tests/execution/update_validation.rs create mode 100644 scripts/conformance/server-parity/update-validation-contract.mjs create mode 100644 scripts/conformance/server-parity/update-validation-probe.php create mode 100644 tests/Fixtures/ServerParity/update-validation.json create mode 100644 tests/Fixtures/ServerParityExpected/update-validation-observation.json create mode 100644 tests/Unit/ServerParityUpdateValidationContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 679c1d450..d25a58262 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -41,6 +41,7 @@ on: - tests/Unit/ServerParityPayloadCleanupContractTest.mjs - tests/Unit/ServerParityRepairPassContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs + - tests/Unit/ServerParityUpdateValidationContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -95,6 +96,7 @@ on: - tests/Unit/ServerParityPayloadCleanupContractTest.mjs - tests/Unit/ServerParityRepairPassContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs + - tests/Unit/ServerParityUpdateValidationContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs - tests/Unit/ServerParityPublishedPatchContractTest.mjs @@ -1518,7 +1520,7 @@ jobs: -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \ install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \ - "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs > parity-evidence/comparator-tests.txt + "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityPayloadCleanupContractTest.mjs tests/Unit/ServerParityRepairPassContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs tests/Unit/ServerParityUpdateValidationContractTest.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-checkpoints/2026-10-23-update-validation.md b/docs/rust-server-port-checkpoints/2026-10-23-update-validation.md new file mode 100644 index 000000000..340999874 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-update-validation.md @@ -0,0 +1,85 @@ +# October 23, synchronous update validation + +- **The gap.** The Python SDK lets a workflow declare update validators. A + worker with validators registers only when the server advertises + `update_validation_tasks` and its `synchronous_update_validation` contract. + Native advertised neither, and refused a validated update with 422 + `update_validation_not_implemented`. So a Python worker with validators + could not run against native. +- **How PHP 2.11.1 does it.** Learned from `ValidatedExternalWorkflowUpdateAdmission` + and `WorkflowUpdateValidationTaskBroker`, then observed live against the + published image. + - A worker declares `update_validators` per workflow type and advertises + the `update_validation_tasks` capability. The run's `WorkflowStarted` + records `declared_update_validators` when the worker's contract has that + field. + - For a live run with a complete durable contract and a declared update, + admission checks, in order: + 1. a recorded receipt for the `request_id`; + 2. whether the contract states validators at all, else 409 + `update_validator_contract_missing`, for any remote update; + 3. the update's arguments; + 4. a `request_id`, which a validated update requires. + + None of these refusals is recorded. + - The validation task is keyed by run, update name and `request_id`, and + holds a hash of the arguments. Different arguments under the same key + conflict with 409. A task that has finished replays its result. The + queue's fresh workers must include one that is capable and matches the + run, else 409 with a routing reason. + - The caller waits up to `DW_UPDATE_VALIDATION_TIMEOUT` seconds (10). A + worker claims the task through `POST /api/worker/update-validation-tasks/poll` + or the workflow poll with `task_kinds`. The claim lasts + `DW_UPDATE_VALIDATION_LEASE_TIMEOUT` seconds (5). The task carries the + run's replay history and history export. + - Approval admits the update with its id equal to the task id. A validator + rejection records a rejected command and update under the task id, with + `UpdateRejected`, and answers 422. Any other worker failure answers 409 + and records nothing. Without an answer, the task times out with 504 and + a reason: not claimed, worker lost, or execution timeout. + - Approve and reject are fenced, in order: task found, not finished, + leased, owner, attempt, lease unexpired, worker fresh. An expired lease + can be claimed again with the next attempt. + - The multiplexed poll keeps a cursor per queue. It tries the kind the + cursor names first, and turns the cursor to the other kind after each + lease. Every workflow task names `task_kind: "workflow"`. +- **What native does now.** + - Admission follows the same order, with the same bodies, and waits + without holding a transaction. Approval re-enters admission with the + task id, after checking that the run is still current and the approval + still stands. + - The validation task poll, approve and reject routes, with Laravel's + request rules. + - The workflow poll validates `task_kinds`. With validation tasks + requested, it refuses a worker without the capability, answers + `no_workflow_capability` for a worker without workflow types, and + alternates kinds by the `workflow_task_poll_cursors` row. + - The capability, its contract and both timeouts (`DW_UPDATE_VALIDATION_*`, + now honored settings) are advertised. + - A validated task's payload carries the unsigned history export, as + `HistoryExport::forRun` builds it. +- **Native changes found on the way.** + - An accepted update's receipt no longer carries `accepted: true`. PHP's + result mapper sends `accepted` only when it is false. + - A worker contract without `update_validators` now makes a remote update + refuse with 409, as on PHP. Current PHP, Python and Rust SDK workers all + send the field. +- **How it was checked.** + - The `update-validation` fixture records 49 steps: refusals before + validation, a rejection through the multiplexed poll, an approval through + the dedicated poll with its fences, retries, three timeouts run side by + side, lease expiry and re-claim, a worker failure, multiplexed ordering + and every routing refusal. + - Two published PHP runs normalize identically. Native matches all 49 + steps on SQLite. The contract test refuses 19 changed observations. + - Native integration tests cover the same paths on SQLite, PostgreSQL and + MySQL. +- **Not done yet** (Server #325 follow-ups): + - **Poll request ids.** PHP replays a leased validation task for a retried + `poll_request_id` and binds the request id to its task kinds. Native + does neither, so a retried multiplexed poll waits for the lease to + expire before the task is offered again. + - **Cluster info.** PHP's `workflow_update_runtime_contract` manifest and + its `service_mode_update_validation` block are not published yet; that + belongs to the cluster info work. + - **Takeover.** PHP data with validation tasks is still refused. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 96247ce9f..223bc76bb 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -138,13 +138,18 @@ any other unknown path or method answers PHP's own 404 or 405 reported as its candidate's failure, with 207. - **Worker protocol features.** - Worker sessions. - - Update validation tasks. - Local activities (prepare, checkpoint, recover, heartbeat, control, outcome and cancellation acknowledgement). - Cancellation scopes. - Activity cancellation acknowledgement. (Lease renewal, from Server 2.11, is served: `POST /api/worker/activity-tasks/{task}/renew-lease`, with the `activity_lease_renewal` capability and the `activity-lease-renewal` fixture.) + - (Synchronous update validation is served: the validation task poll, + approve and reject routes, the multiplexed workflow poll with + `task_kinds`, and validator-gated update admission, with the + `update-validation` fixture. Native does not replay a validation task by + `poll_request_id` or bind a request id to its task kinds; see + `docs/rust-server-port-checkpoints/2026-10-23-update-validation.md`.) - **Run operations.** - Imports from embedded v2 and Waterline v1. - **Other families.** diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index 3eb4572f0..24915a09a 100644 --- a/rust/src/config_audit.rs +++ b/rust/src/config_audit.rs @@ -224,6 +224,21 @@ where .map_or(10, |value| value.int().max(1)) } +/// `server.update_validation`: how long an update request waits for its +/// validator (`timeout`, at least 0, 10 when unset) and how long a worker's +/// claim on a validation task lasts (`lease_timeout`, at least 1, 5 when +/// unset). The broker casts each with `(int)`. +pub fn update_validation_timeouts(lookup: F) -> (i64, i64) +where + F: Fn(&str) -> Option, +{ + let read = |name, legacy, default: i64| effective(&lookup, name, Some(legacy)).map_or(default, |value: PhpValue| value.int()); + ( + read("DW_UPDATE_VALIDATION_TIMEOUT", "WORKFLOW_SERVER_UPDATE_VALIDATION_TIMEOUT", 10).max(0), + read("DW_UPDATE_VALIDATION_LEASE_TIMEOUT", "WORKFLOW_SERVER_UPDATE_VALIDATION_LEASE_TIMEOUT", 5).max(1), + ) +} + /// `server.worker_protocol.history_page_size_default` and `_max`: `(int)` /// casts of the settings, 500 and 1000 when unset. The Workflow package's /// bridge then keeps every page between 1 and 1000 events. @@ -575,8 +590,8 @@ pub const SETTINGS: &[Setting] = &[ Setting { name: "DW_TASK_DISPATCH_MODE", legacy: Some("WORKFLOW_V2_TASK_DISPATCH_MODE"), disposition: Refused(Exact(&["poll"]), POLL_DISPATCH) }, Setting { name: "DW_TASK_QUEUE_ADMISSION_OVERRIDES", legacy: Some("WORKFLOW_SERVER_TASK_QUEUE_ADMISSION_OVERRIDES"), disposition: Refused(EmptyJson { strict: false }, TASK_ADMISSION) }, Setting { name: "DW_TRUST_FORWARDED_ATTRIBUTION_HEADERS", legacy: Some("WORKFLOW_SERVER_TRUST_FORWARDED_ATTRIBUTION_HEADERS"), disposition: Refused(FilterBool(false), FORWARDED) }, - Setting { name: "DW_UPDATE_VALIDATION_LEASE_TIMEOUT", legacy: Some("WORKFLOW_SERVER_UPDATE_VALIDATION_LEASE_TIMEOUT"), disposition: Ignored }, - Setting { name: "DW_UPDATE_VALIDATION_TIMEOUT", legacy: Some("WORKFLOW_SERVER_UPDATE_VALIDATION_TIMEOUT"), disposition: Ignored }, + Setting { name: "DW_UPDATE_VALIDATION_LEASE_TIMEOUT", legacy: Some("WORKFLOW_SERVER_UPDATE_VALIDATION_LEASE_TIMEOUT"), disposition: Honored }, + Setting { name: "DW_UPDATE_VALIDATION_TIMEOUT", legacy: Some("WORKFLOW_SERVER_UPDATE_VALIDATION_TIMEOUT"), disposition: Honored }, Setting { name: "DW_V1_ENABLED", legacy: Some("WORKFLOW_V1_ENABLED"), disposition: Refused(CastBool(false), V1) }, Setting { name: "DW_V2_CACHE_VALIDATION_MODE", legacy: Some("WORKFLOW_V2_CACHE_VALIDATION_MODE"), disposition: Ignored }, Setting { name: "DW_V2_COMPATIBILITY_HEARTBEAT_TTL", legacy: Some("WORKFLOW_V2_COMPATIBILITY_HEARTBEAT_TTL"), disposition: Refused(Int(30), COMPATIBILITY) }, @@ -808,6 +823,18 @@ mod tests { assert!(!cast("") && !cast("0")); } + #[test] + fn update_validation_timeouts_are_read_like_the_broker() { + let read = |values: &[(&str, &str)]| { + let environment = environment(values); + update_validation_timeouts(|key| environment.get(key).cloned()) + }; + assert_eq!(read(&[]), (10, 5)); + assert_eq!(read(&[("DW_UPDATE_VALIDATION_TIMEOUT", "3"), ("DW_UPDATE_VALIDATION_LEASE_TIMEOUT", "2")]), (3, 2)); + assert_eq!(read(&[("WORKFLOW_SERVER_UPDATE_VALIDATION_TIMEOUT", "4")]), (4, 5)); + assert_eq!(read(&[("DW_UPDATE_VALIDATION_TIMEOUT", "-1"), ("DW_UPDATE_VALIDATION_LEASE_TIMEOUT", "0")]), (0, 1)); + } + #[test] fn update_wait_is_read_like_the_package() { let read = |values: &[(&str, &str)]| { diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index fa6a57a4f..495b6a075 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -929,6 +929,90 @@ impl Runtime { } } } + /// `pollUpdateValidationTasks`. + pub(crate) async fn poll_update_validation(&self, body: Value) -> Result { + super::update_validation::poll_rules(&body)?; + let timeout = match body.get("timeout_seconds") { + None | Some(Value::Null) => self.worker_poll_wait(), + Some(value) => super::laravel_validation::integer(value).and_then(|value| u64::try_from(value).ok()).unwrap_or(0), + }; + let worker_id = super::text(&body, "worker_id")?; + let task_queue = super::text(&body, "task_queue")?; + let leased = match &*self.storage { + Storage::Sqlite(store) => store.update_validation_poll(worker_id, task_queue, timeout).await?, + Storage::Postgres(store) => store.update_validation_poll(worker_id, task_queue, timeout).await?, + Storage::MySql(store) => store.update_validation_poll(worker_id, task_queue, timeout).await?, + }; + let task = match leased { + Some(task_id) => Some(self.update_validation_task(&task_id).await?), + None => None, + }; + let status = if task.is_some() { "leased" } else { "empty" }; + Ok(json!({"task":task,"poll_status":status})) + } + /// `POST /worker/update-validation-tasks/{taskId}/approve` and `/reject`. + pub(crate) async fn complete_update_validation(&self, task_id: &str, body: Value, reject: bool) -> Result<(StatusCode, Value)> { + super::update_validation::completion_rules(&body, reject)?; + let failure = reject.then(|| body["failure"].clone()); + match &*self.storage { + Storage::Sqlite(store) => store.complete_update_validation(task_id, &body, failure.as_ref()).await, + Storage::Postgres(store) => store.complete_update_validation(task_id, &body, failure.as_ref()).await, + Storage::MySql(store) => store.complete_update_validation(task_id, &body, failure.as_ref()).await, + } + } + /// A leased validation task as a worker receives it, with the run's + /// replay bundle. + async fn update_validation_task(&self, task_id: &str) -> Result { + let mut task = match &*self.storage { + Storage::Sqlite(store) => store.update_validation_payload(task_id).await?, + Storage::Postgres(store) => store.update_validation_payload(task_id).await?, + Storage::MySql(store) => store.update_validation_payload(task_id).await?, + }; + if task.get("history_events").is_some() + && let (Some(workflow_id), Some(run_id)) = (task["workflow_id"].as_str(), task["run_id"].as_str()) + { + let export = self.history_bundle(&workflow_id.to_owned(), &run_id.to_owned(), false).await?; + task["history_export"] = export; + } + Ok(task) + } + /// `pollWorkflowTasks`: workflow tasks, validation tasks or both, as + /// `task_kinds` asks. Every task names its kind. + pub(crate) async fn poll_workflow_tasks(&self, body: Value, protocol_version: &str) -> Result { + let kinds = super::update_validation::task_kinds(&body)?; + let pages = self.history_pages(); + let wait = self.worker_poll_wait(); + let mark = |mut answer: Value| { + if let Some(task) = answer["task"].as_object_mut() { + task.insert("task_kind".into(), json!("workflow")); + } + answer + }; + if kinds == ["workflow"] { + return self.poll_with_pages(body, "workflow", protocol_version, pages, wait).await.map(mark); + } + let timeout = match body.get("timeout_seconds") { + None | Some(Value::Null) => wait, + Some(value) => super::laravel_validation::integer(value) + .filter(|seconds| (0..=60).contains(seconds)) + .and_then(|seconds| u64::try_from(seconds).ok()) + .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "invalid_poll_timeout"))?, + }; + let leased = match &*self.storage { + Storage::Sqlite(store) => store.multiplexed_poll(&body, &kinds, protocol_version, pages, timeout).await?, + Storage::Postgres(store) => store.multiplexed_poll(&body, &kinds, protocol_version, pages, timeout).await?, + Storage::MySql(store) => store.multiplexed_poll(&body, &kinds, protocol_version, pages, timeout).await?, + }; + match leased { + super::update_validation::Leased::Workflow(answer) => Ok(mark(answer)), + super::update_validation::Leased::Validation(task_id) => { + let mut task = self.update_validation_task(&task_id).await?; + // The multiplexed poll's history paging field. + task["next_history_page_token"] = Value::Null; + Ok(json!({"task":task,"poll_status":"leased","server_capabilities":super::store::capabilities()})) + } + } + } pub(crate) async fn poll_query(&self, body: Value) -> Result { match &*self.storage { Storage::Sqlite(store) => store.poll_query(body, &self.queries).await, @@ -1019,6 +1103,11 @@ impl Runtime { ); /// `HistoryController::export`: one run's replay bundle. pub(crate) async fn history_export(&self, workflow_id: &str, run_id: &str) -> Result { + self.history_bundle(workflow_id, run_id, true).await + } + /// One run's replay bundle, signed when asked and configured. + /// `HistoryExport::forRun` itself, as replay tasks carry it, is unsigned. + async fn history_bundle(&self, workflow_id: &str, run_id: &str, signed: bool) -> Result { let (mut rows, memo) = match &*self.storage { Storage::Sqlite(store) => store.history_export_rows(workflow_id, run_id).await?, Storage::Postgres(store) => store.history_export_rows(workflow_id, run_id).await?, @@ -1032,7 +1121,7 @@ impl Runtime { .map(|value| value.trim().to_owned()) .filter(|value| !value.is_empty()) }; - let signing = setting(["DW_V2_HISTORY_EXPORT_SIGNING_KEY", "WORKFLOW_V2_HISTORY_EXPORT_SIGNING_KEY"]).map(|key| { + let signing = setting(["DW_V2_HISTORY_EXPORT_SIGNING_KEY", "WORKFLOW_V2_HISTORY_EXPORT_SIGNING_KEY"]).filter(|_| signed).map(|key| { super::history_export::Signing { key, key_id: setting(["DW_V2_HISTORY_EXPORT_SIGNING_KEY_ID", "WORKFLOW_V2_HISTORY_EXPORT_SIGNING_KEY_ID"]), diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 98febdb4c..2ab6c055e 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -292,6 +292,15 @@ pub fn router(runtime: Runtime) -> Router { post(complete_query), ) .route("/api/worker/query-tasks/{task_id}/fail", post(fail_query)) + .route("/api/worker/update-validation-tasks/poll", post(poll_update_validation)) + .route( + "/api/worker/update-validation-tasks/{task_id}/approve", + post(approve_update_validation), + ) + .route( + "/api/worker/update-validation-tasks/{task_id}/reject", + post(reject_update_validation), + ) .fallback(route_fallback) .method_not_allowed_fallback(route_fallback) .layer(middleware::from_fn(empty_list_body)) @@ -751,6 +760,8 @@ fn worker_route(rest: &str) -> Option { | ["workflow-tasks" | "activity-tasks", "poll"] | ["query-tasks", "poll"] | ["query-tasks", _, "complete" | "fail" | "status"] + | ["update-validation-tasks", "poll"] + | ["update-validation-tasks", _, "approve" | "reject"] | ["workflow-tasks", _, "complete" | "fail" | "history" | "heartbeat"] | ["activity-tasks", _, "complete" | "fail" | "heartbeat" | "status" | "renew-lease"] ) @@ -1801,12 +1812,46 @@ async fn poll_workflow( headers: HeaderMap, Json(body): Json, ) -> Result> { - let pages = runtime.history_pages(); - let wait = runtime.worker_poll_wait(); runtime - .poll_with_pages(body, "workflow", request_protocol(&headers)?, pages, wait) + .poll_workflow_tasks(body, request_protocol(&headers)?) .await .map(Json) + .map_err(worker_refusal) +} + +async fn poll_update_validation( + Extension(runtime): Extension, + headers: HeaderMap, + Json(body): Json, +) -> Result> { + request_protocol(&headers)?; + runtime + .poll_update_validation(body) + .await + .map(|body| Json(worker_response(body))) + .map_err(worker_refusal) +} + +async fn approve_update_validation( + Extension(runtime): Extension, + Path(task_id): Path, + headers: HeaderMap, + Json(body): Json, +) -> Result<(StatusCode, Json)> { + request_protocol(&headers)?; + let (status, body) = runtime.complete_update_validation(&task_id, body, false).await.map_err(worker_refusal)?; + Ok((status, Json(worker_response(body)))) +} + +async fn reject_update_validation( + Extension(runtime): Extension, + Path(task_id): Path, + headers: HeaderMap, + Json(body): Json, +) -> Result<(StatusCode, Json)> { + request_protocol(&headers)?; + let (status, body) = runtime.complete_update_validation(&task_id, body, true).await.map_err(worker_refusal)?; + Ok((status, Json(worker_response(body)))) } async fn poll_activity( Extension(runtime): Extension, diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index d60cb0473..7af5afdf3 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -129,6 +129,7 @@ mod terminal_updates; mod timer_summary; mod transaction; mod update_summary; +mod update_validation; mod updates; pub mod upgrade; mod visibility; diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 93c299d12..4f7aa36fb 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -3961,6 +3961,7 @@ pub(crate) fn capabilities() -> Value { "worker_deregistration_fencing":{"schema":"durable-workflow.v2.worker-deregistration.v1","supported":true,"receipt_retention_seconds":600,"endpoint":"/worker/registrations/{workerId}/deregister"}, "workflow_memo_updates": false, "cooperative_cancellation": true, "prepared_local_activities": false, "worker_sessions": false, "sticky_execution": false, "local_activities": false, "message_streams": false, + "update_validation_tasks": true, "synchronous_update_validation": super::update_validation::capability(), "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true, "activity_lease_renewal": true}); value["supported_workflow_task_commands"] diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs new file mode 100644 index 000000000..b79de1e6f --- /dev/null +++ b/rust/src/runtime/update_validation.rs @@ -0,0 +1,1010 @@ +//! Published PHP 2.11.1's synchronous update validation +//! (`WorkflowUpdateValidationTaskBroker` and +//! `ValidatedExternalWorkflowUpdateAdmission`). +//! +//! An update whose durable contract declares a validator is not admitted until +//! a validator-capable worker approves it. The caller waits up to +//! `DW_UPDATE_VALIDATION_TIMEOUT` seconds. A worker claims the validation task +//! for `DW_UPDATE_VALIDATION_LEASE_TIMEOUT` seconds and approves or rejects it. +//! Every outcome stays on the task row, so a retry with the same `request_id` +//! and arguments gets the same answer. +use super::{Result, backend::Backend, refuse, store::{Store, id, now}, text, wire}; +use axum::http::StatusCode; +use serde_json::{Map, Value, json}; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; +use std::sync::OnceLock; +use std::time::Duration; + +pub(super) const CAPABILITY: &str = "update_validation_tasks"; +const CONTRACT_FIELD: &str = "update_validators"; +const TERMINAL: [&str; 4] = ["approved", "rejected", "failed", "timed_out"]; +pub(super) const VALIDATOR_REJECTED: &str = "update_validator_rejected"; +const TIMEOUT_MESSAGE: &str = "Synchronous update validation did not complete before the acceptance deadline."; +const NOT_ADVERTISED: &str = "Worker registration does not advertise synchronous update validation."; + +/// `server.update_validation`: how long a caller waits for a validator, and +/// how long a worker's claim lasts. +pub(super) fn timeouts() -> (i64, i64) { + static TIMEOUTS: OnceLock<(i64, i64)> = OnceLock::new(); + *TIMEOUTS.get_or_init(|| crate::config_audit::update_validation_timeouts(|key| std::env::var(key).ok())) +} + +/// `WorkerProtocol::serverCapabilities()['synchronous_update_validation']`. +pub(super) fn capability() -> Value { + let (timeout, lease) = timeouts(); + json!({"supported":true,"minimum_protocol_version":"1.13","acceptance_boundary":"validator_approved", + "worker_capability":CAPABILITY,"workflow_contract_field":CONTRACT_FIELD, + "validator_execution":"replay_without_handler_or_state_commit","request_id_required":true, + "duplicate_completion":"typed_rejection","stale_completion":"typed_rejection", + "missing_contract":"typed_rejection","unsupported_worker":"typed_rejection", + "task_poll":{"strategy":"multiplexed","endpoint":"/worker/workflow-tasks/poll","request_field":"task_kinds", + "task_kinds":["workflow","update_validation"],"default_task_kinds":["workflow"], + "response_discriminator":"task.task_kind","poll_request_id_binding":"normalized_task_kind_set", + "poll_request_id_conflict_reason":"poll_request_task_kinds_conflict"}, + "completion":{"approve_endpoint":"/worker/update-validation-tasks/{taskId}/approve", + "reject_endpoint":"/worker/update-validation-tasks/{taskId}/reject", + "fence_fields":["lease_owner","update_validation_attempt"], + "typed_failure_reasons":["update_validation_task_not_found","duplicate_update_validation_completion", + "update_validation_task_not_leased","update_validation_lease_owner_mismatch", + "stale_update_validation_completion","update_validation_lease_expired","update_validator_worker_lost"]}, + "timeout_seconds":timeout,"lease_timeout_seconds":lease, + "sdk_surfaces":{"sdk-python":"validator_capable","sdk-php":"validators_not_exposed","sdk-rust":"validators_not_exposed"}}) +} + +/// The 409 a poll gets from a registration without the capability. +pub(super) fn not_advertised() -> super::RuntimeError { + super::RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response: json!({"task":null,"poll_status":"unsupported", + "reason":"update_validation_capability_not_advertised","error":NOT_ADVERTISED}), + } +} + +/// `stringValue`: a trimmed, non-empty string. +fn string_value(value: &Value) -> Option<&str> { + value.as_str().map(str::trim).filter(|text| !text.is_empty()) +} + +/// `stringList`: trimmed, unique, non-empty strings. +fn string_list(value: &Value) -> Vec { + let mut list: Vec = value + .as_array() + .map(|items| items.iter().filter_map(string_value).map(str::to_owned).collect()) + .unwrap_or_default(); + list.sort(); + list.dedup(); + list +} + +/// `validationErrors`: field names mapped to their non-empty string messages. +pub(super) fn validation_errors(value: &Value) -> Map { + let fields: Vec<(String, &Value)> = match value { + Value::Object(map) => map.iter().map(|(field, messages)| (field.clone(), messages)).collect(), + Value::Array(list) => list.iter().enumerate().map(|(field, messages)| (field.to_string(), messages)).collect(), + _ => vec![], + }; + let mut errors = Map::new(); + for (field, messages) in fields { + let Some(messages) = messages.as_array() else { continue }; + let kept: Vec = messages + .iter() + .filter(|message| message.as_str().is_some_and(|text| !text.is_empty())) + .cloned() + .collect(); + if !kept.is_empty() { + errors.insert(field, Value::Array(kept)); + } + } + errors +} + +fn sha256(text: &str) -> String { + use sha2::{Digest, Sha256}; + Sha256::digest(text.as_bytes()).iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// `ValidatedExternalWorkflowUpdateAdmission::validationFailure`. +pub(super) fn admission_failure( + workflow_id: &str, + run_id: &str, + name: &str, + reason: &str, + message: &str, + retryable: bool, + validation_errors: Option, +) -> Value { + let mut body = json!({"workflow_id":workflow_id,"run_id":run_id,"update_name":name, + "outcome":"update_validation_failed","reason":reason,"rejection_reason":reason,"message":message}); + if let Some(errors) = validation_errors.filter(|errors| errors.as_object().is_some_and(|map| !map.is_empty())) { + body["validation_errors"] = errors; + } + body["retryable"] = json!(retryable); + body +} + +/// `WorkflowUpdateValidationTaskBroker::failure`. +#[allow(clippy::too_many_arguments)] +fn broker_failure( + workflow_id: &str, + run_id: &str, + name: &str, + reason: &str, + message: &str, + errors: Map, + update_id: &str, + retryable: bool, +) -> Value { + let rejected = reason == VALIDATOR_REJECTED; + let mut body = json!({"approved":false,"workflow_id":workflow_id,"run_id":run_id,"update_name":name,"update_id":update_id}); + if rejected { + body["update_status"] = json!("rejected"); + } + body["outcome"] = json!(if rejected { "rejected_invalid_arguments" } else { "update_validation_failed" }); + body["reason"] = json!(reason); + body["rejection_reason"] = json!(reason); + body["message"] = json!(message); + if !errors.is_empty() { + body["validation_errors"] = Value::Object(errors); + } + body["retryable"] = json!(retryable); + body +} + +/// `staleValidationFailure`: the run moved on between validation and commit. +pub(super) fn stale_failure(workflow_id: &str, run_id: &str, name: &str, update_id: &str, message: &str) -> (StatusCode, Value) { + let mut body = json!({"workflow_id":workflow_id}); + if !run_id.is_empty() { + body["run_id"] = json!(run_id); + } + body["update_name"] = json!(name); + body["update_id"] = json!(update_id); + body["outcome"] = json!("update_validation_failed"); + body["reason"] = json!("stale_update_validation_completion"); + body["rejection_reason"] = json!("stale_update_validation_completion"); + body["message"] = json!(message); + body["retryable"] = json!(true); + (StatusCode::CONFLICT, body) +} + +/// `RunCommandContract::forRun` from durable history: the start event holds +/// every declared contract field, and each declared query and update has a +/// handler contract. +pub(super) fn durable_contract(started: &Value) -> bool { + const FIELDS: [&str; 9] = ["declared_queries", "declared_query_contracts", "declared_signals", + "declared_signal_contracts", "declared_updates", "declared_update_contracts", "declared_entry_method", + "declared_entry_mode", "declared_entry_declaring_class"]; + if !FIELDS.iter().all(|field| started.get(*field).is_some()) { + return false; + } + let lists = ["declared_queries", "declared_query_contracts", "declared_signals", "declared_signal_contracts", + "declared_updates", "declared_update_contracts"]; + if !lists.iter().all(|field| started[*field].is_array()) { + return false; + } + if started["declared_entry_method"] != "handle" + || started["declared_entry_mode"] != "canonical" + || string_value(&started["declared_entry_declaring_class"]).is_none() + { + return false; + } + let covered = |names: &str, contracts: &str| { + let handled: Vec<&str> = started[contracts] + .as_array() + .map(|items| items.iter().filter_map(|item| string_value(&item["name"])).collect()) + .unwrap_or_default(); + string_list(&started[names]).iter().all(|name| handled.contains(&name.as_str())) + }; + covered("declared_queries", "declared_query_contracts") && covered("declared_updates", "declared_update_contracts") +} + +/// What one update's validation decided. +pub(super) enum Validation { + /// Admit the update under the task's id. + Approved { task_id: String, input_hash: String }, + /// The validator refused it; the rejection is recorded durably. + Rejected { failure: Value, validation_errors: Value }, + /// Anything else: answer with this status and body, recording nothing. + Failed(StatusCode, Value), +} + +/// The update a caller asked to validate. +pub(super) struct ValidationRequest<'a> { + pub(super) workflow_id: &'a str, + pub(super) run_id: &'a str, + pub(super) workflow_type: &'a str, + pub(super) task_queue: &'a str, + pub(super) compatibility: Option<&'a str>, + pub(super) fingerprint: Option<&'a str>, + pub(super) name: &'a str, + pub(super) request_id: &'a str, + pub(super) arguments: &'a str, + pub(super) context: Option<&'a Value>, +} + +/// `inputHash`: the codec and serialized arguments. +pub(super) fn input_hash(arguments: &str) -> String { + sha256(&format!("avro\0{arguments}")) +} + +/// One registration as validation routing reads it. +struct Registration { + worker_id: String, + fresh: bool, + capable: bool, + workflow_types: Vec, + contracts: Value, + build_id: Option, + fingerprints: Value, +} + +impl Registration { + /// `workerMatchesRun` and the type, contract, build and fingerprint + /// parts of `workerMatchesTask`. + fn serves(&self, workflow_type: &str, name: &str, compatibility: Option<&str>, fingerprint: Option<&str>) -> bool { + self.workflow_types.iter().any(|kind| kind == workflow_type) + && self.contracts[workflow_type].is_object() + && string_list(&self.contracts[workflow_type][CONTRACT_FIELD]).iter().any(|declared| declared == name) + && compatibility.is_none_or(|required| self.build_id.as_deref() == Some(required)) + && fingerprint.is_none_or(|recorded| self.fingerprints[workflow_type].as_str() == Some(recorded)) + } +} + +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, +{ + fn registration(&self, row: &DB::Row) -> Result { + let fresh_after = now() - chrono::Duration::seconds(self.worker_stale_after_seconds.max(1)); + // `WorkerPollFence::isFresh`: active, with a recent heartbeat. + let fresh = DB::optional_string(row, "status")?.as_deref() == Some("active") + && DB::optional_instant(row, "last_heartbeat_at")?.is_some_and(|heartbeat| heartbeat >= fresh_after); + let capabilities = DB::optional_document_row(row, "capabilities")?.unwrap_or(Value::Null); + Ok(Registration { + worker_id: DB::string(row, "worker_id")?, + fresh, + capable: capabilities.as_array().is_some_and(|names| names.iter().any(|name| name == CAPABILITY)), + workflow_types: string_list(&DB::optional_document_row(row, "supported_workflow_types")?.unwrap_or(Value::Null)), + contracts: DB::optional_document_row(row, "workflow_command_contracts")?.unwrap_or(Value::Null), + build_id: DB::optional_string(row, "build_id")?.filter(|build| !build.is_empty()), + fingerprints: DB::optional_document_row(row, "workflow_definition_fingerprints")?.unwrap_or(Value::Null), + }) + } + + /// Whether this registration may claim validation tasks at all. + pub(super) async fn worker_validates(&self, worker_id: &str) -> Result { + let 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?; + Ok(match row { + Some(row) => self.registration(&row)?.capable, + None => false, + }) + } + + /// `route`: the first missing link between the queue's workers and this + /// run's validator, with its message and whether a retry may help. + async fn validation_route(&self, request: &ValidationRequest<'_>) -> Result> { + let rows = Self::query("SELECT * FROM workflow_worker_registrations WHERE namespace=$1 AND task_queue=$2 AND status='active'") + .bind(self.namespace.as_ref()) + .bind(request.task_queue) + .fetch_all(&self.pool) + .await?; + let mut workers = Vec::with_capacity(rows.len()); + for row in &rows { + let worker = self.registration(row)?; + if worker.fresh { + workers.push(worker); + } + } + if workers.is_empty() { + return Ok(Some(("update_validator_worker_unavailable", + "No active worker is available to perform synchronous update validation.", true))); + } + workers.retain(|worker| worker.capable); + if workers.is_empty() { + return Ok(Some(("update_validation_capability_unsupported", + "Active workers do not advertise synchronous pre-accept update validation.", false))); + } + if !workers.iter().any(|worker| worker.serves(request.workflow_type, request.name, request.compatibility, request.fingerprint)) { + return Ok(Some(("update_validator_worker_incompatible", + "Active workers do not advertise a validator-capable implementation matching this workflow run.", true))); + } + Ok(None) + } + + /// `findOrCreateTask`: one task per run, update name and request id. + async fn validation_task(&self, request: &ValidationRequest<'_>, input_hash: &str) -> Result { + let key = sha256(&[self.namespace.as_ref(), request.run_id, request.name, request.request_id].join("\0")); + let mut tx = self.begin().await?; + let existing = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE idempotency_key=$1") + .bind(&key) + .fetch_optional(&mut *tx) + .await?; + if let Some(existing) = existing { + tx.commit().await?; + return Ok(existing); + } + let task_id = id(); + let at = now(); + Self::query("INSERT INTO workflow_update_validation_tasks(id,idempotency_key,namespace,workflow_instance_id,workflow_run_id,workflow_type,task_queue,compatibility,workflow_definition_fingerprint,update_name,request_id,input_hash,payload_codec,arguments,command_context,status,attempt_count,created_at,updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,'avro',$13,$14,'pending',0,$15,$16)") + .bind(&task_id).bind(&key).bind(self.namespace.as_ref()).bind(request.workflow_id).bind(request.run_id) + .bind(request.workflow_type).bind(request.task_queue).bind(request.compatibility).bind(request.fingerprint) + .bind(request.name).bind(request.request_id).bind(input_hash).bind(request.arguments) + .bind(request.context.map(DB::document)).bind(DB::bind_time(at)).bind(DB::bind_time(at)) + .execute(&mut *tx) + .await?; + let row = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE id=$1") + .bind(&task_id) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(row) + } + + /// `WorkflowUpdateValidationTaskBroker::validate`. + pub(super) async fn validate_update(&self, request: &ValidationRequest<'_>) -> Result { + let hash = input_hash(request.arguments); + let task = self.validation_task(request, &hash).await?; + let task_id = DB::string(&task, "id")?; + if DB::string(&task, "input_hash")? != hash { + return Ok(Validation::Failed(StatusCode::CONFLICT, broker_failure( + request.workflow_id, request.run_id, request.name, "update_validation_idempotency_conflict", + "The update request ID was already used with different arguments.", Map::new(), &task_id, false))); + } + if !TERMINAL.contains(&DB::string(&task, "status")?.as_str()) { + if let Some((reason, message, retryable)) = self.validation_route(request).await? { + return Ok(Validation::Failed(StatusCode::CONFLICT, broker_failure( + request.workflow_id, request.run_id, request.name, reason, message, Map::new(), &task_id, retryable))); + } + // Pollers on this node wake now; other nodes find the task on + // their next read. + self.wake.notify_waiters(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeouts().0 as u64); + loop { + let wake = self.wake.notified(); + let status = Self::query("SELECT status FROM workflow_update_validation_tasks WHERE id=$1") + .bind(&task_id) + .fetch_one(&self.pool) + .await?; + if TERMINAL.contains(&DB::string(&status, "status")?.as_str()) { + break; + } + if tokio::time::Instant::now() >= deadline { + self.time_out_validation(&task_id).await?; + break; + } + tokio::select! { + _ = wake => {}, + _ = tokio::time::sleep(Duration::from_millis(50)) => {}, + _ = tokio::time::sleep_until(deadline) => {}, + } + } + } + let task = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE id=$1") + .bind(&task_id) + .fetch_one(&self.pool) + .await?; + self.validation_result(&task, request) + } + + /// `result`: an approval, or the stored failure with its HTTP status. + fn validation_result(&self, task: &DB::Row, request: &ValidationRequest<'_>) -> Result { + let task_id = DB::string(task, "id")?; + let status = DB::string(task, "status")?; + if status == "approved" { + return Ok(Validation::Approved { task_id, input_hash: DB::string(task, "input_hash")? }); + } + let reason = DB::optional_string(task, "rejection_reason")?.unwrap_or_else(|| "update_validation_failed".into()); + let message = DB::optional_string(task, "rejection_message")?.unwrap_or_else(|| "Synchronous update validation failed.".into()); + let errors = validation_errors(&DB::optional_document_row(task, "validation_errors")?.unwrap_or(Value::Null)); + let mut failure = broker_failure(request.workflow_id, request.run_id, request.name, &reason, &message, errors.clone(), &task_id, false); + if let Some(kind) = DB::optional_string(task, "failure_type")? { + failure["failure_type"] = json!(kind); + } + failure["update_validation_attempt"] = json!(DB::number(task, "attempt_count")?); + if status == "rejected" && reason == VALIDATOR_REJECTED { + return Ok(Validation::Rejected { failure, validation_errors: Value::Object(errors) }); + } + let http = match status.as_str() { + "rejected" => StatusCode::UNPROCESSABLE_ENTITY, + "timed_out" => StatusCode::GATEWAY_TIMEOUT, + _ => StatusCode::CONFLICT, + }; + Ok(Validation::Failed(http, failure)) + } + + /// `markTimedOut`: the reason names what was still missing. + async fn time_out_validation(&self, task_id: &str) -> Result<()> { + let mut tx = self.begin().await?; + let task = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE id=$1") + .bind(task_id) + .fetch_one(&mut *tx) + .await?; + let status = DB::string(&task, "status")?; + if TERMINAL.contains(&status.as_str()) { + tx.commit().await?; + return Ok(()); + } + let owner = DB::optional_string(&task, "lease_owner")?; + let mut lost = false; + if status == "leased" { + let worker = match owner.as_deref() { + Some(owner) => Self::query("SELECT * FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") + .bind(DB::string(&task, "namespace")?) + .bind(owner) + .fetch_optional(&mut *tx) + .await?, + None => None, + }; + lost = match worker { + Some(worker) => !self.registration(&worker)?.fresh, + None => true, + }; + } + let (reason, message) = if lost { + ("update_validator_worker_lost", "The validator worker was lost before returning a validation result.") + } else if status == "pending" { + ("update_validation_task_not_claimed", TIMEOUT_MESSAGE) + } else { + ("update_validation_execution_timeout", TIMEOUT_MESSAGE) + }; + let at = now(); + Self::query("UPDATE workflow_update_validation_tasks SET status='timed_out',rejection_reason=$1,rejection_message=$2,timed_out_at=$3,lease_expires_at=NULL,updated_at=$4 WHERE id=$5") + .bind(reason).bind(message).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(task_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + /// `approvedTask`: the approval still stands for this run and input. + pub(super) async fn approved_validation( + tx: &mut super::transaction::Transaction<'_, DB>, + task_id: &str, + run_id: &str, + input_hash: &str, + ) -> Result { + Ok(Self::query("SELECT id FROM workflow_update_validation_tasks WHERE id=$1 AND workflow_run_id=$2 AND input_hash=$3 AND status='approved'") + .bind(task_id).bind(run_id).bind(input_hash) + .fetch_optional(&mut **tx) + .await? + .is_some()) + } + + /// `claim`: the oldest claimable task this worker can validate, leased + /// to it for the lease timeout. + pub(super) async fn claim_update_validation(&self, worker_id: &str, task_queue: &str) -> Result> { + let mut tx = self.begin().await?; + let Some(worker) = Self::query("SELECT * 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 { + tx.commit().await?; + return Ok(None); + }; + let worker = self.registration(&worker)?; + if !worker.fresh || !worker.capable { + tx.commit().await?; + return Ok(None); + } + let at = now(); + let tasks = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE namespace=$1 AND task_queue=$2 AND (status='pending' OR (status='leased' AND lease_expires_at<=$3)) ORDER BY created_at,id LIMIT 25") + .bind(self.namespace.as_ref()) + .bind(task_queue) + .bind(DB::bind_time(at)) + .fetch_all(&mut *tx) + .await?; + for task in tasks { + let compatibility = DB::optional_string(&task, "compatibility")?; + let fingerprint = DB::optional_string(&task, "workflow_definition_fingerprint")?; + if !worker.serves(&DB::string(&task, "workflow_type")?, &DB::string(&task, "update_name")?, + compatibility.as_deref(), fingerprint.as_deref()) + { + continue; + } + let task_id = DB::string(&task, "id")?; + let attempt = DB::number(&task, "attempt_count")?.saturating_add(1); + Self::query("UPDATE workflow_update_validation_tasks SET status='leased',attempt_count=$1,lease_owner=$2,lease_expires_at=$3,updated_at=$4 WHERE id=$5") + .bind(attempt).bind(&worker.worker_id) + .bind(DB::bind_time(at + chrono::Duration::seconds(timeouts().1))) + .bind(DB::bind_time(at)).bind(&task_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + return Ok(Some(task_id)); + } + tx.commit().await?; + Ok(None) + } + + /// `taskPayload` without its history export, which the runtime adds. + pub(super) async fn update_validation_payload(&self, task_id: &str) -> Result { + let task = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE id=$1") + .bind(task_id) + .fetch_one(&self.pool) + .await?; + let run_id = DB::string(&task, "workflow_run_id")?; + let mut payload = json!({}); + let mut connection = self.pool.acquire().await?; + if let Some(run) = Self::query("SELECT workflow_class,arguments,status,last_history_sequence FROM workflow_runs WHERE id=$1") + .bind(&run_id) + .fetch_optional(&mut *connection) + .await? + { + // `workflowReplayPayload`: the whole history, unpaged. + let mut history = Vec::new(); + let mut after = 0; + loop { + let (page, next) = Self::history_page_connection(&mut connection, &run_id, after, 1000).await?; + after = page.last().and_then(|event| event["sequence"].as_i64()).unwrap_or(after); + history.extend(page); + if next.is_none() { + break; + } + } + payload = json!({"workflow_class":DB::optional_string(&run, "workflow_class")?, + "workflow_arguments":DB::optional_string(&run, "arguments")?.as_deref().map(wire), + "run_status":DB::string(&run, "status")?, + "last_history_sequence":DB::optional_number(&run, "last_history_sequence")?.unwrap_or(0), + "history_events":history}); + } + let fields = json!({"task_kind":"update_validation","update_validation_task_id":task_id,"update_id":task_id, + "update_validation_attempt":DB::number(&task, "attempt_count")?, + "workflow_id":DB::string(&task, "workflow_instance_id")?,"run_id":run_id, + "workflow_type":DB::string(&task, "workflow_type")?,"compatibility":DB::optional_string(&task, "compatibility")?, + "update_name":DB::string(&task, "update_name")?,"update_arguments":wire(&DB::string(&task, "arguments")?), + "payload_codec":DB::string(&task, "payload_codec")?,"task_queue":DB::string(&task, "task_queue")?, + "lease_owner":DB::optional_string(&task, "lease_owner")?, + "lease_expires_at":super::task_queues::json_time(DB::optional_instant(&task, "lease_expires_at")?)}); + payload.as_object_mut().expect("payload").extend(fields.as_object().expect("fields").clone()); + Ok(payload) + } + + /// `guardCompletion`, then `approve` or `reject`. + pub(super) async fn complete_update_validation(&self, task_id: &str, body: &Value, failure: Option<&Value>) -> Result<(StatusCode, Value)> { + let owner = text(body, "lease_owner")?; + let attempt = super::laravel_validation::integer(&body["update_validation_attempt"]) + .ok_or_else(|| refuse(StatusCode::UNPROCESSABLE_ENTITY, "validation_failed"))?; + let mut tx = self.begin().await?; + let task = Self::query("SELECT * FROM workflow_update_validation_tasks WHERE id=$1") + .bind(task_id) + .fetch_optional(&mut *tx) + .await? + .filter(|task| DB::string(task, "namespace").ok().as_deref() == Some(self.namespace.as_ref())); + let refusal = |reason: &str, message: &str, status: StatusCode, current: Option| { + let mut body = json!({"update_validation_task_id":task_id,"update_id":task_id}); + if let Some(current) = current { + body["update_validation_attempt"] = json!(current); + } + body["outcome"] = json!("rejected"); + body["reason"] = json!(reason); + body["error"] = json!(message); + body["status"] = json!(status.as_u16()); + Ok((status, body)) + }; + let Some(task) = task else { + tx.commit().await?; + return refusal("update_validation_task_not_found", "Update validation task not found.", StatusCode::NOT_FOUND, None); + }; + let status = DB::string(&task, "status")?; + let current = DB::number(&task, "attempt_count")?; + let lease_owner = DB::optional_string(&task, "lease_owner")?; + let guard = if TERMINAL.contains(&status.as_str()) { + Some(if current != attempt || lease_owner.as_deref() != Some(owner) { + ("stale_update_validation_completion", "Update validation completion belongs to a stale delivery attempt.") + } else { + ("duplicate_update_validation_completion", "The update validation task already has a terminal result.") + }) + } else if status != "leased" { + Some(("update_validation_task_not_leased", "Update validation task is not leased.")) + } else if lease_owner.as_deref() != Some(owner) { + Some(("update_validation_lease_owner_mismatch", "Update validation lease is owned by another worker.")) + } else if current != attempt { + Some(("stale_update_validation_completion", "Update validation completion belongs to a stale delivery attempt.")) + } else if DB::optional_instant(&task, "lease_expires_at")?.is_none_or(|expires| expires <= now()) { + Some(("update_validation_lease_expired", "Update validation task lease has expired.")) + } else { + let worker = Self::query("SELECT * FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2") + .bind(self.namespace.as_ref()) + .bind(owner) + .fetch_optional(&mut *tx) + .await?; + let fresh = match worker { + Some(worker) => self.registration(&worker)?.fresh, + None => false, + }; + (!fresh).then_some(("update_validator_worker_lost", "The validator worker is no longer active.")) + }; + if let Some((reason, message)) = guard { + tx.commit().await?; + return refusal(reason, message, StatusCode::CONFLICT, Some(current)); + } + let at = now(); + let mut answer = json!({"update_validation_task_id":task_id,"update_id":task_id,"update_validation_attempt":attempt}); + match failure { + None => { + Self::query("UPDATE workflow_update_validation_tasks SET status='approved',approved_at=$1,lease_expires_at=NULL,updated_at=$2 WHERE id=$3") + .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(task_id) + .execute(&mut *tx) + .await?; + answer["outcome"] = json!("approved"); + } + Some(failure) => { + let reason = string_value(&failure["reason"]).unwrap_or(VALIDATOR_REJECTED).to_owned(); + let rejected = reason == VALIDATOR_REJECTED; + let message = string_value(&failure["message"]).map(str::to_owned).unwrap_or_else(|| { + if rejected { "The update validator rejected the request." } else { "Update validation failed." }.to_owned() + }); + let errors = Value::Object(validation_errors(&failure["validation_errors"])); + let sql = if rejected { + "UPDATE workflow_update_validation_tasks SET status='rejected',rejection_reason=$1,rejection_message=$2,failure_type=$3,validation_errors=$4,rejected_at=$5,failed_at=NULL,lease_expires_at=NULL,updated_at=$6 WHERE id=$7" + } else { + "UPDATE workflow_update_validation_tasks SET status='failed',rejection_reason=$1,rejection_message=$2,failure_type=$3,validation_errors=$4,rejected_at=NULL,failed_at=$5,lease_expires_at=NULL,updated_at=$6 WHERE id=$7" + }; + Self::query(sql) + .bind(reason.as_str()).bind(message).bind(string_value(&failure["type"])) + .bind(DB::document(&errors)).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(task_id) + .execute(&mut *tx) + .await?; + answer["outcome"] = json!(if rejected { "rejected" } else { "failed" }); + answer["reason"] = json!(reason); + } + } + answer["status"] = json!(200); + tx.commit().await?; + self.wake.notify_waiters(); + Ok((StatusCode::OK, answer)) + } +} + +/// The rules `pollUpdateValidationTasks` applies. +pub(super) fn poll_rules(body: &Value) -> Result<()> { + use super::laravel_validation::{Rule, failure_messages, rule_errors}; + let rules: &[(&str, &[Rule])] = &[ + ("worker_id", &[Rule::Required, Rule::String]), + ("task_queue", &[Rule::Required, Rule::String]), + ("timeout_seconds", &[Rule::Nullable, Rule::Integer, Rule::Min(0), Rule::Max(60)]), + ]; + failure_messages(rule_errors(body, rules, |_, _| None)).map_or(Ok(()), Err) +} + +/// The rules `approveUpdateValidationTask` and `rejectUpdateValidationTask` +/// apply, nested failure fields named as Laravel names them. +pub(super) fn completion_rules(body: &Value, reject: bool) -> Result<()> { + use super::laravel_validation::{Rule, failure_messages, rule_errors}; + let mut flat = Map::new(); + let mut rules: Vec<(String, Vec)> = vec![ + ("lease_owner".into(), vec![Rule::Required, Rule::String]), + ("update_validation_attempt".into(), vec![Rule::Required, Rule::Integer, Rule::Min(1)]), + ]; + for field in ["lease_owner", "update_validation_attempt"] { + if let Some(value) = body.get(field) { + flat.insert(field.into(), value.clone()); + } + } + if reject { + if let Some(failure) = body.get("failure") { + flat.insert("failure".into(), failure.clone()); + } + rules.push(("failure".into(), vec![Rule::Required, Rule::Array])); + let failure = body.get("failure").filter(|failure| failure.is_object()); + let nested = |field: &str| failure.and_then(|failure| failure.get(field)); + for (field, field_rules) in [ + ("message", vec![Rule::Required, Rule::String]), + ("reason", vec![Rule::Nullable, Rule::String]), + ("type", vec![Rule::Nullable, Rule::String]), + ("stack_trace", vec![Rule::Nullable, Rule::String]), + ("validation_errors", vec![Rule::Nullable, Rule::Array]), + ] { + let name = format!("failure.{field}"); + if let Some(value) = nested(field) { + flat.insert(name.clone(), value.clone()); + } + rules.push((name, field_rules)); + } + if let Some(errors) = nested("validation_errors").and_then(Value::as_object) { + for (field, messages) in errors { + let name = format!("failure.validation_errors.{field}"); + flat.insert(name.clone(), messages.clone()); + rules.push((name.clone(), vec![Rule::Array])); + for (index, message) in messages.as_array().into_iter().flatten().enumerate() { + let item = format!("{name}.{index}"); + flat.insert(item.clone(), message.clone()); + rules.push((item, vec![Rule::String])); + } + } + } + } + let borrowed: Vec<(&str, &[Rule])> = rules.iter().map(|(field, rules)| (field.as_str(), rules.as_slice())).collect(); + failure_messages(rule_errors(&Value::Object(flat), &borrowed, |_, _| None)).map_or(Ok(()), Err) +} + +/// `task_kinds`: `nullable|array|min:1|max:2`, each a distinct `workflow` or +/// `update_validation`. Absent or null means workflow tasks only. +pub(super) fn task_kinds(body: &Value) -> Result> { + use super::laravel_validation::{Rule, failure_messages, rule_errors}; + let Some(kinds) = body.get("task_kinds").filter(|kinds| !kinds.is_null()) else { + return Ok(vec!["workflow"]); + }; + let mut errors = rule_errors(body, &[("task_kinds", &[Rule::Nullable, Rule::Array, Rule::Min(1), Rule::Max(2)])], |_, _| None); + let items: Vec<(String, Value)> = match kinds { + Value::Array(items) => items.iter().enumerate().map(|(index, item)| (index.to_string(), item.clone())).collect(), + Value::Object(items) => items.iter().map(|(key, item)| (key.clone(), item.clone())).collect(), + _ => vec![], + }; + for (key, item) in &items { + let field = format!("task_kinds.{key}"); + let mut messages = Vec::new(); + if !item.is_string() { + messages.push(format!("The {field} field must be a string.")); + } + if items.iter().filter(|(_, other)| other == item).count() > 1 { + messages.push(format!("The {field} field has a duplicate value.")); + } + if !matches!(item.as_str(), Some("workflow" | "update_validation")) { + messages.push(format!("The selected {field} is invalid.")); + } + errors.push((field, messages)); + } + if let Some(error) = failure_messages(errors) { + return Err(error); + } + let mut requested = Vec::new(); + for kind in ["workflow", "update_validation"] { + if items.iter().any(|(_, item)| item == kind) { + requested.push(kind); + } + } + Ok(requested) +} + +/// What a multiplexed poll leased. +pub(super) enum Leased { + /// A workflow poll's answer, leased or not. + Workflow(Value), + /// A validation task's id, leased to the polling worker. + Validation(String), +} + +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, +{ + /// `resolveRegisteredWorker` for a validation poll, then its capability + /// check. Answers the registration's supported workflow types. + pub(super) async fn validation_poller(&self, worker_id: &str, task_queue: &str) -> Result> { + 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(super::RuntimeError::Protocol { + status: StatusCode::PRECONDITION_FAILED, + response: json!({"error":"Worker must be registered before polling. Call POST /worker/register first.", + "reason":"worker_not_registered","worker_id":worker_id}), + }); + }; + let registered = DB::string(&row, "task_queue")?; + if registered != task_queue { + return Err(super::RuntimeError::Protocol { + status: StatusCode::CONFLICT, + response: json!({"error":format!("Worker [{worker_id}] is registered for task queue [{registered}], not [{task_queue}]."), + "reason":"task_queue_mismatch","worker_id":worker_id,"registered_task_queue":registered, + "requested_task_queue":task_queue}), + }); + } + self.refuse_draining_worker(worker_id, task_queue).await?; + Ok(self.registration(&row)?.workflow_types) + } + + /// A dedicated validation poll: the registration, its capability, then + /// the wait for a claim. + pub(super) async fn update_validation_poll(&self, worker_id: &str, task_queue: &str, timeout: u64) -> Result> { + self.validation_poller(worker_id, task_queue).await?; + if !self.worker_validates(worker_id).await? { + return Err(not_advertised()); + } + self.poll_update_validation(worker_id, task_queue, timeout).await + } + + /// A workflow poll that asked for validation tasks: the registration, + /// a workflow worker, the capability, then both kinds. + pub(super) async fn multiplexed_poll( + &self, + body: &Value, + kinds: &[&str], + protocol_version: &str, + pages: super::history_pages::HistoryPages, + timeout: u64, + ) -> Result { + let worker_id = text(body, "worker_id")?; + let types = self.validation_poller(worker_id, text(body, "task_queue")?).await?; + if types.is_empty() { + return Ok(Leased::Workflow(json!({"task":null,"poll_status":"no_workflow_capability"}))); + } + if !self.worker_validates(worker_id).await? { + return Err(not_advertised()); + } + self.poll_task_kinds(body, kinds, protocol_version, pages, timeout).await + } + + /// Whether a claimable validation task waits on this queue. + async fn validation_waiting(&self, task_queue: &str) -> Result { + Ok(Self::query("SELECT id FROM workflow_update_validation_tasks WHERE namespace=$1 AND task_queue=$2 AND (status='pending' OR (status='leased' AND lease_expires_at<=$3)) LIMIT 1") + .bind(self.namespace.as_ref()) + .bind(task_queue) + .bind(DB::bind_time(now())) + .fetch_optional(&self.pool) + .await? + .is_some()) + } + + /// `poll`: claim a validation task, waiting up to `timeout` seconds. + pub(super) async fn poll_update_validation(&self, worker_id: &str, task_queue: &str, timeout: u64) -> Result> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout); + let mut pace = super::poll_readiness::ReadPace::new(); + let mut first = true; + loop { + let wake = self.wake.notified(); + if (std::mem::take(&mut first) || self.validation_waiting(task_queue).await?) + && let Some(task_id) = self.claim_update_validation(worker_id, task_queue).await? + { + return Ok(Some(task_id)); + } + if tokio::time::Instant::now() >= deadline { + return Ok(None); + } + pace.wait(wake, deadline).await; + } + } + + /// `WorkflowTaskKindSelector`'s cursor: which kind this queue tries first. + async fn first_task_kind(&self, task_queue: &str) -> Result { + Ok(Self::query("SELECT next_task_kind FROM workflow_task_poll_cursors WHERE namespace=$1 AND task_queue=$2") + .bind(self.namespace.as_ref()) + .bind(task_queue) + .fetch_optional(&self.pool) + .await? + .map(|row| DB::string(&row, "next_task_kind")) + .transpose()? + .unwrap_or_else(|| "update_validation".into())) + } + + /// After a lease, the queue tries the other kind first. + async fn turn_task_kind(&self, task_queue: &str, leased: &str) -> Result<()> { + let next = if leased == "workflow" { "update_validation" } else { "workflow" }; + let at = now(); + let mut tx = self.begin().await?; + // MySQL counts only changed rows, so look before writing. + let cursor = Self::query("SELECT next_task_kind FROM workflow_task_poll_cursors WHERE namespace=$1 AND task_queue=$2") + .bind(self.namespace.as_ref()).bind(task_queue) + .fetch_optional(&mut *tx) + .await?; + if cursor.is_some() { + Self::query("UPDATE workflow_task_poll_cursors SET next_task_kind=$1,updated_at=$2 WHERE namespace=$3 AND task_queue=$4") + .bind(next).bind(DB::bind_time(at)).bind(self.namespace.as_ref()).bind(task_queue) + .execute(&mut *tx) + .await?; + } else { + Self::query("INSERT INTO workflow_task_poll_cursors(namespace,task_queue,next_task_kind,created_at,updated_at) VALUES ($1,$2,$3,$4,$5)") + .bind(self.namespace.as_ref()).bind(task_queue).bind(next).bind(DB::bind_time(at)).bind(DB::bind_time(at)) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(()) + } + + /// `WorkflowTaskPoller` with both kinds requested: the queue's cursor + /// picks the kind tried first, and a lease turns it to the other kind. + /// When neither kind has work, the workflow poll's answer stands. + pub(super) async fn poll_task_kinds( + &self, + body: &Value, + kinds: &[&str], + protocol_version: &str, + pages: super::history_pages::HistoryPages, + timeout: u64, + ) -> Result { + let worker_id = text(body, "worker_id")?.to_owned(); + let task_queue = text(body, "task_queue")?.to_owned(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout); + let mut probe = body.clone(); + probe["timeout_seconds"] = json!(0); + let mut pace = super::poll_readiness::ReadPace::new(); + loop { + let wake = self.wake.notified(); + let first = self.first_task_kind(&task_queue).await?; + let mut order: Vec<&str> = kinds.to_vec(); + order.sort_by_key(|kind| *kind != first); + let mut workflow = None; + for kind in order { + if kind == "workflow" { + let answer = self.poll_with_pages(probe.clone(), "workflow", protocol_version, pages, 0).await?; + if !answer["task"].is_null() { + if kinds.len() > 1 { + self.turn_task_kind(&task_queue, "workflow").await?; + } + return Ok(Leased::Workflow(answer)); + } + workflow = Some(answer); + } else if let Some(task_id) = self.claim_update_validation(&worker_id, &task_queue).await? { + if kinds.len() > 1 { + self.turn_task_kind(&task_queue, "update_validation").await?; + } + return Ok(Leased::Validation(task_id)); + } + } + if tokio::time::Instant::now() >= deadline { + return Ok(Leased::Workflow(workflow.unwrap_or_else(|| { + json!({"task":null,"poll_status":"empty","server_capabilities":super::store::capabilities()}) + }))); + } + pace.wait(wake, deadline).await; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validation_errors_keep_non_empty_string_messages() { + let errors = validation_errors(&json!({"v":["too small", "", 3],"w":"not a list","x":[]})); + assert_eq!(Value::Object(errors), json!({"v":["too small"]})); + let listed = validation_errors(&json!([["first"]])); + assert_eq!(Value::Object(listed), json!({"0":["first"]})); + } + + #[test] + fn broker_failures_mark_only_validator_rejections_rejected() { + let rejected = broker_failure("w", "r", "u", VALIDATOR_REJECTED, "no", Map::new(), "t", false); + assert_eq!(rejected["update_status"], "rejected"); + assert_eq!(rejected["outcome"], "rejected_invalid_arguments"); + let failed = broker_failure("w", "r", "u", "validator_error", "no", Map::new(), "t", false); + assert!(failed.get("update_status").is_none()); + assert_eq!(failed["outcome"], "update_validation_failed"); + assert_eq!(failed["approved"], false); + } + + #[test] + fn registrations_serve_only_runs_their_contract_validates() { + let worker = Registration { + worker_id: "w".into(), + fresh: true, + capable: true, + workflow_types: vec!["t".into()], + contracts: json!({"t":{"update_validators":["check"]}}), + build_id: Some("b1".into()), + fingerprints: json!({"t":"sha256:f"}), + }; + assert!(worker.serves("t", "check", None, None)); + assert!(worker.serves("t", "check", Some("b1"), Some("sha256:f"))); + assert!(!worker.serves("t", "other", None, None)); + assert!(!worker.serves("u", "check", None, None)); + assert!(!worker.serves("t", "check", Some("b2"), None)); + assert!(!worker.serves("t", "check", None, Some("sha256:g"))); + } +} diff --git a/rust/src/runtime/updates.rs b/rust/src/runtime/updates.rs index 2521ddf90..7a78a7c1b 100644 --- a/rust/src/runtime/updates.rs +++ b/rust/src/runtime/updates.rs @@ -30,6 +30,44 @@ struct UpdateRejection<'a, R> { validation_errors: Value, /// A structural ceiling keeps no arguments and records its limit. exceeded: Option, + /// A validated update's rejection keeps its validation task's id. + update_id: Option, +} + +/// A validator's approval of one update on one run. +struct Approval { + task_id: String, + run_id: String, + input_hash: String, +} + +/// The request's own `wait_timeout_seconds`, as PHP echoes it. +fn request_body_timeout(body: &Value) -> Value { + body.get("wait_timeout_seconds").cloned().unwrap_or(Value::Null) +} + +/// A validation answer as the control plane sends it: the requester and the +/// update contract, with the wait the request asked for. +fn validation_response(mut body: Value, wait_for: &str, context: Option<&Value>, name: &str) -> Value { + if body.get("wait_for").is_none() { + body["wait_for"] = json!(wait_for); + } + if let Some(principal) = context.and_then(principal) { + body["principal"] = principal; + } + super::control_plane::Operation::Update(name).response(body) +} + +/// `principalPayload`: the requester's non-empty type, id and label. +fn principal(context: &Value) -> Option { + let principal: serde_json::Map = ["type", "id", "label"] + .iter() + .filter_map(|field| { + context["principal"][*field].as_str().filter(|value| !value.trim().is_empty()) + .map(|value| ((*field).to_owned(), json!(value))) + }) + .collect(); + (!principal.is_empty()).then_some(Value::Object(principal)) } /// `DefaultWorkflowControlPlane::updateStatus` for a rejected update. @@ -294,6 +332,7 @@ where for<'r> i64: Decode<'r, DB>, usize: ColumnIndex, { + #[allow(clippy::too_many_arguments)] pub(crate) async fn update_workflow( &self, workflow_id: &str, @@ -305,6 +344,26 @@ where codec: Arc, limits: super::memo_limits::StructuralLimits, external: Option<&super::external_resolution::PreparedExternal>, + ) -> Result<(StatusCode, Value)> { + self.admit_update(workflow_id, selected_run, name, body, prepared, context, codec, limits, external, None) + .await + } + + /// One update admission. A validator's approval admits the update under + /// the validation task's id, once its run is still the current run. + #[allow(clippy::too_many_arguments)] + async fn admit_update( + &self, + workflow_id: &str, + selected_run: Option<&str>, + name: &str, + body: Value, + prepared: PreparedControlArguments, + context: Option, + codec: Arc, + limits: super::memo_limits::StructuralLimits, + external: Option<&super::external_resolution::PreparedExternal>, + approval: Option, ) -> Result<(StatusCode, Value)> { // PHP's update_name column holds 255 bytes, one fewer than its // validation allows; native refuses what it cannot store. @@ -347,6 +406,13 @@ where "historical_run_command_rejected", )); } + if let Some(approval) = &approval + && approval.run_id != run_id + { + let (status, failure) = super::update_validation::stale_failure(workflow_id, &approval.run_id, name, &approval.task_id, + "The workflow advanced to a different run before validation could be committed."); + return Ok((status, validation_response(failure, wait_for, context.as_ref(), name))); + } let existing = if let Some(request_id) = request_id { Self::query("SELECT u.id,u.update_name,u.workflow_run_id FROM workflow_updates u JOIN workflow_commands c ON c.id=u.workflow_command_id WHERE c.workflow_instance_id=$1 AND c.command_type='update' AND c.request_id=$2") .bind(workflow_id).bind(request_id).fetch_optional(&mut *tx).await? @@ -386,23 +452,98 @@ where .bind(&run_id).fetch_one(&mut *tx).await .map_err(|error| at_update_lookup(error.into(), UpdateLookup::StartContract))?; let contract = DB::document_row(&started, "payload")?; + let declared = contract["declared_updates"] + .as_array() + .is_some_and(|names| names.iter().any(|value| value == name)); + if let Some(approval) = &approval + && (!active || !declared || !Self::approved_validation(&mut tx, &approval.task_id, &run_id, &approval.input_hash).await?) + { + let message = if active && declared { + "The validator approval no longer matches this update admission attempt." + } else { + "The validated update could not be committed to the current workflow run." + }; + let (status, failure) = super::update_validation::stale_failure(workflow_id, &run_id, name, &approval.task_id, message); + return Ok((status, validation_response(failure, wait_for, context.as_ref(), name))); + } + // `ValidatedExternalWorkflowUpdateAdmission::preAdmissionValidation`: + // a live run's declared update on a durable contract checks its + // validator declaration before anything is recorded. + if active && declared && approval.is_none() && super::update_validation::durable_contract(&contract) { + let Some(validators) = contract.get("declared_update_validators") else { + let failure = super::update_validation::admission_failure(workflow_id, &run_id, name, + "update_validator_contract_missing", + "The durable workflow contract does not state whether this update has a validator.", false, None); + return Ok((StatusCode::CONFLICT, validation_response(failure, wait_for, context.as_ref(), name))); + }; + if super::query_route::strings(validators).iter().any(|validator| validator == name) { + match validate_arguments(&contract, "declared_update_contracts", name, &prepared.values, + "unsupported_update_contract", "invalid_update_arguments") + { + Ok(()) => {} + Err(super::RuntimeError::Refused { reason: "invalid_update_arguments", .. }) => { + let failure = super::update_validation::admission_failure(workflow_id, &run_id, name, + "invalid_update_arguments", &format!("Workflow update [{name}] argument validation failed."), + false, Some(update_argument_errors(&contract, name, &prepared.values))); + return Ok((StatusCode::UNPROCESSABLE_ENTITY, validation_response(failure, wait_for, context.as_ref(), name))); + } + Err(error) => return Err(error), + } + let Some(request_id) = request_id else { + let failure = super::update_validation::admission_failure(workflow_id, &run_id, name, + "update_validation_request_id_required", + "Updates with declared validators require request_id so validation and application remain idempotent across retries.", + true, None); + return Ok((StatusCode::UNPROCESSABLE_ENTITY, validation_response(failure, wait_for, context.as_ref(), name))); + }; + // The caller waits for the validator without a transaction. + drop(tx); + let queue = DB::optional_string(&run, "queue")?.filter(|queue| !queue.trim().is_empty()); + let compatibility = DB::optional_string(&run, "compatibility")?.filter(|build| !build.is_empty()); + let fingerprint = super::query_route::trimmed(contract["workflow_definition_fingerprint"].as_str()); + let workflow_type = DB::string(&run, "workflow_type")?; + let request = super::update_validation::ValidationRequest { + workflow_id, + run_id: &run_id, + workflow_type: &workflow_type, + task_queue: queue.as_deref().unwrap_or("default"), + compatibility: compatibility.as_deref(), + fingerprint, + name, + request_id, + arguments: &prepared.arguments, + context: context.as_ref(), + }; + return match self.validate_update(&request).await? { + super::update_validation::Validation::Approved { task_id, input_hash } => { + let mut context = context.clone(); + if let Some(context) = context.as_mut().filter(|context| context["server"]["metadata"].is_object()) { + context["server"]["metadata"]["update_validation_task_id"] = json!(task_id); + context["server"]["metadata"]["update_validation_input_hash"] = json!(input_hash); + } + let approval = Approval { task_id, run_id: run_id.clone(), input_hash }; + Box::pin(self.admit_update(workflow_id, selected_run, name, body, prepared, context, codec, + limits, external, Some(approval))) + .await + } + super::update_validation::Validation::Rejected { failure, validation_errors } => { + let timeout = super::laravel_validation::integer(&request_body_timeout(&body)); + self.record_validator_rejection(workflow_id, &run_id, name, &prepared, request_id, + context.clone(), codec, failure, validation_errors, wait_for, timeout) + .await + } + super::update_validation::Validation::Failed(status, failure) => { + Ok((status, validation_response(failure, wait_for, context.as_ref(), name))) + } + }; + } + } // `WorkflowStub::submitUpdate`'s order: a closed run, an unknown // update, then invalid arguments. Each records a rejection. let mut rejection = if !active { Some(("run_not_active", Some("rejected_not_active"), json!([]))) - } else if !contract["declared_updates"] - .as_array() - .is_some_and(|names| names.iter().any(|value| value == name)) - { + } else if !declared { Some(("unknown_update", Some("rejected_unknown_update"), json!([]))) - } else if contract["declared_update_validators"] - .as_array() - .is_some_and(|names| names.iter().any(|value| value == name)) - { - return Err(refuse( - StatusCode::UNPROCESSABLE_ENTITY, - "update_validation_not_implemented", - )); } else { match validate_arguments( &contract, @@ -458,6 +599,7 @@ where outcome, validation_errors, exceeded, + update_id: approval.as_ref().map(|approval| approval.task_id.clone()), }, ) .await?; @@ -489,7 +631,9 @@ where .bind(&run_id).fetch_one(&mut *tx).await .map_err(|error| at_update_lookup(error.into(), UpdateLookup::Inventory))?; let command_id = id(); - let update_id = id(); + // `WorkflowUpdateValidationObserver`: an approved update keeps + // its validation task's id. + let update_id = approval.as_ref().map_or_else(id, |approval| approval.task_id.clone()); let command_sequence = DB::number(&run, "last_command_sequence")? .checked_add(1) .ok_or_else(|| refuse(StatusCode::CONFLICT, "command_sequence_exhausted"))?; @@ -587,6 +731,82 @@ where } } + /// `recordValidatorRejection`: the validator's refusal, recorded as a + /// rejected command and update under the validation task's id. + #[allow(clippy::too_many_arguments)] + async fn record_validator_rejection( + &self, + workflow_id: &str, + run_id: &str, + name: &str, + prepared: &PreparedControlArguments, + request_id: &str, + context: Option, + codec: Arc, + mut failure: Value, + validation_errors: Value, + wait_for: &str, + timeout: Option, + ) -> Result<(StatusCode, Value)> { + let task_id = failure["update_id"].as_str().unwrap_or_default().to_owned(); + let mut tx = self.begin().await?; + let instance = Self::query("SELECT current_run_id,last_message_sequence FROM workflow_instances WHERE id=$1 AND namespace=$2") + .bind(workflow_id).bind(self.namespace.as_ref()).fetch_optional(&mut *tx).await?; + let Some(instance) = instance.filter(|instance| DB::string(instance, "current_run_id").ok().as_deref() == Some(run_id)) else { + let (status, failure) = super::update_validation::stale_failure(workflow_id, run_id, name, &task_id, + "The workflow run changed before the validator rejection could be recorded."); + return Ok((status, validation_response(failure, wait_for, context.as_ref(), name))); + }; + let recorded = Self::query("SELECT workflow_command_id FROM workflow_updates WHERE id=$1") + .bind(&task_id).fetch_optional(&mut *tx).await?; + let command_id = match recorded { + Some(recorded) => DB::string(&recorded, "workflow_command_id")?, + None => { + let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1").bind(run_id).fetch_one(&mut *tx).await?; + Self::reject_update( + &mut tx, + codec, + UpdateRejection { + workflow_id, + instance: &instance, + run: &run, + name, + prepared, + request_id: Some(request_id), + context: context.clone(), + reason: super::update_validation::VALIDATOR_REJECTED, + outcome: Some("rejected_invalid_arguments"), + validation_errors, + exceeded: None, + update_id: Some(task_id.clone()), + }, + ) + .await?; + let recorded = Self::query("SELECT workflow_command_id FROM workflow_updates WHERE id=$1") + .bind(&task_id).fetch_one(&mut *tx).await?; + DB::string(&recorded, "workflow_command_id")? + } + }; + tx.commit().await?; + // `rejectionResponse`: the validator's failure with the rejection + // receipt's fields. + for (field, value) in [ + ("accepted", json!(false)), + ("command_id", json!(command_id)), + ("command_status", json!("rejected")), + ("outcome", json!("rejected_invalid_arguments")), + ("update_id", json!(task_id)), + ("update_status", json!("rejected")), + ("rejection_reason", json!(super::update_validation::VALIDATOR_REJECTED)), + ("wait_for", json!(wait_for)), + ("wait_timed_out", json!(false)), + ("wait_timeout_seconds", json!(timeout)), + ] { + failure[field] = value; + } + Ok((StatusCode::UNPROCESSABLE_ENTITY, validation_response(failure, wait_for, context.as_ref(), name))) + } + /// `WorkflowStub::rejectUpdateCommand`: a rejected command and update, /// and an `UpdateRejected` event on the run. Nothing is delivered. async fn reject_update( @@ -606,6 +826,7 @@ where outcome, validation_errors, exceeded, + update_id, } = rejection; let run_id = DB::string(run, "id")?; let (values, diagnostics, arguments) = match &exceeded { @@ -626,7 +847,7 @@ where ) .await?; let command_id = id(); - let update_id = id(); + let update_id = update_id.unwrap_or_else(id); let command_sequence = DB::number(run, "last_command_sequence")? .checked_add(1) .ok_or_else(|| refuse(StatusCode::CONFLICT, "command_sequence_exhausted"))?; @@ -743,7 +964,7 @@ where .filter(|errors| !errors.is_null()) .unwrap_or_else(|| json!([])); let mut response = json!({ - "accepted":!rejected,"workflow_id":DB::string(update,"workflow_instance_id")?,"run_id":DB::string(update,"workflow_run_id")?, + "workflow_id":DB::string(update,"workflow_instance_id")?,"run_id":DB::string(update,"workflow_run_id")?, "resolved_run_id":DB::string(update,"resolved_workflow_run_id")?,"requested_run_id":null,"target_scope":"instance", "workflow_type":DB::string(update,"workflow_type")?,"command_id":DB::string(update,"workflow_command_id")?, "command_sequence":DB::number(update,"command_sequence")?,"command_status":if rejected {"rejected"} else {"accepted"},"command_source":"control_plane", @@ -757,17 +978,15 @@ where "wait_for":wait_for,"wait_timed_out":timed_out,"wait_timeout_seconds":timeout, "rejected_at":DB::optional_instant(update,"rejected_at")? }); + // `ControlPlaneResultMapper` drops `accepted` and puts it back only + // when it is false. + if rejected { + response["accepted"] = json!(false); + } // The requester recorded on the command, as `principalPayload`. let context = DB::optional_document_row(update, "command_context")?.unwrap_or(Value::Null); - let principal: serde_json::Map = ["type", "id", "label"] - .iter() - .filter_map(|field| { - context["principal"][*field].as_str().filter(|value| !value.trim().is_empty()) - .map(|value| ((*field).to_owned(), json!(value))) - }) - .collect(); - if !principal.is_empty() { - response["principal"] = Value::Object(principal); + if let Some(principal) = principal(&context) { + response["principal"] = principal; } let name = response["update_name"].as_str().unwrap_or_default().to_owned(); Ok(super::control_plane::Operation::Update(&name).response(response)) diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c3ce551e4..96ecff906 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -87,6 +87,8 @@ mod rejected_updates; mod continue_messages; #[path = "execution/update_wait.rs"] mod update_wait; +#[path = "execution/update_validation.rs"] +mod update_validation; #[path = "execution/payload_limits.rs"] mod payload_limits; #[path = "execution/start_definition.rs"] diff --git a/rust/tests/execution/external_payloads/commands.rs b/rust/tests/execution/external_payloads/commands.rs index 82237259c..0038b91ed 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/rust/tests/execution/external_payloads/commands.rs @@ -70,7 +70,7 @@ async fn signal_update_and_query_input_resolve_runtime_references_like_php() { // An update resolves the reference to its bytes and keeps them inline. waiting_run(&app, "external-update", json!({"worker_id":"update-worker","task_queue":"external-update", "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":[],"capabilities":["workflow_updates"], - "workflow_command_contracts":{"echo":{"updates":["u"],"update_contracts":[{"name":"u","parameters":string_parameter()}]}}})).await; + "workflow_command_contracts":{"echo":{"updates":["u"],"update_validators":[],"update_contracts":[{"name":"u","parameters":string_parameter()}]}}})).await; let blob = argument("update"); let update = upload(&app, &blob).await; let (status, body) = json_request(&app, "POST", "/api/workflows/external-update/update/u", "operator-token", diff --git a/rust/tests/execution/rejected_updates.rs b/rust/tests/execution/rejected_updates.rs index 7c14dfbd9..90a17d058 100644 --- a/rust/tests/execution/rejected_updates.rs +++ b/rust/tests/execution/rejected_updates.rs @@ -80,7 +80,8 @@ async fn refused_updates_are_recorded_like_php() { // delivered to the waiting run. let (status, admitted) = update(&app, "rejected-update", "set", input(vec![Payload::Long(5)])).await; assert_eq!(status, StatusCode::ACCEPTED, "{admitted}"); - assert_eq!(admitted["accepted"], true, "{admitted}"); + // PHP's result mapper sends `accepted` only when it is false. + assert!(admitted.get("accepted").is_none(), "{admitted}"); assert!(admitted["command_sequence"].as_i64().unwrap() > unknown["command_sequence"].as_i64().unwrap()); let task = poll(&app, "update-a", "workflow").await; assert!(task["task_id"].is_string(), "{task}"); diff --git a/rust/tests/execution/update_validation.rs b/rust/tests/execution/update_validation.rs new file mode 100644 index 000000000..239222edd --- /dev/null +++ b/rust/tests/execution/update_validation.rs @@ -0,0 +1,310 @@ +use super::*; +use std::time::Duration; + +// Server 2.11.1's synchronous update validation, as observed against the +// published image: a declared validator decides an update before it is +// accepted, through validation tasks a capable worker claims. + +fn int_contract(name: &str) -> Value { + json!({"name":name,"parameters":[{"name":"value","type":"int","required":true,"variadic":false, + "default_available":false,"allows_null":false}]}) +} + +async fn validating_worker(app: &Router, worker: &str, queue: &str, validators: Option, capable: bool) { + let updates = ["approve_me", "plain", "reject_me"]; + let mut contract = json!({"signals":["finish"],"signal_contracts":[int_contract("finish")],"queries":[], + "updates":updates,"update_contracts":updates.iter().map(|name| int_contract(name)).collect::>()}); + if let Some(validators) = validators { + contract["update_validators"] = validators; + } + let mut capabilities = vec!["workflow_updates"]; + if capable { + capabilities.push("update_validation_tasks"); + } + let (status, body) = request(app, "POST", "/api/worker/register", json!({"worker_id":worker,"task_queue":queue, + "runtime":"python","supported_workflow_types":["echo"],"supported_activity_types":[], + "capabilities":capabilities,"workflow_command_contracts":{"echo":contract}})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); +} + +/// A run on `queue` waiting for a signal, so updates find it idle. +async fn waiting_run(app: &Router, workflow: &str, queue: &str, worker: &str) -> Value { + let (status, started) = request(app, "POST", "/api/workflows", json!({"workflow_id":workflow,"workflow_type":"echo", + "task_queue":queue,"input":envelope(Payload::Array(vec![]))})).await; + assert_eq!(status, StatusCode::CREATED, "{started}"); + let (status, polled) = request(app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":worker,"task_queue":queue,"timeout_seconds":0})).await; + assert_eq!(status, StatusCode::OK, "{polled}"); + let task = &polled["task"]; + assert_eq!(task["task_kind"], "workflow", "{polled}"); + assert_eq!(finish_task(app, task, json!([{"type":"open_signal_wait","signal_name":"finish"}])).await.0, StatusCode::OK); + started +} + +fn update_body(value: i64, request_id: Option<&str>) -> Value { + let mut body = json!({"input":envelope(Payload::Array(vec![Payload::Long(value)]))}); + if let Some(request_id) = request_id { + body["request_id"] = json!(request_id); + } + body +} + +/// An update request that waits for its validator, answered later. +fn send_update(app: &Router, workflow: &str, name: &str, body: Value) -> tokio::task::JoinHandle<(StatusCode, Value)> { + let app = app.clone(); + let path = format!("/api/workflows/{workflow}/update/{name}"); + tokio::spawn(async move { request(&app, "POST", &path, body).await }) +} + +async fn poll_validation(app: &Router, worker: &str, queue: &str) -> Value { + let (status, body) = request(app, "POST", "/api/worker/update-validation-tasks/poll", + json!({"worker_id":worker,"task_queue":queue,"timeout_seconds":5})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["poll_status"], "leased", "{body}"); + body["task"].clone() +} + +async fn complete(app: &Router, task: &str, verdict: &str, body: Value) -> (StatusCode, Value) { + request(app, "POST", &format!("/api/worker/update-validation-tasks/{task}/{verdict}"), body).await +} + +fn events<'a>(history: &'a Value, kind: &str) -> Vec<&'a Value> { + history.as_array().unwrap().iter().filter(|event| event["event_type"] == kind).collect() +} + +#[tokio::test] +async fn validators_approve_and_reject_updates_before_acceptance() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + validating_worker(&app, "validator", "vq", Some(json!(["approve_me", "reject_me"])), true).await; + let started = waiting_run(&app, "validated", "vq", "validator").await; + let history = || run_history(&app, &started["workflow_id"], &started["run_id"]); + let before = history().await; + let started_event = events(&before, "WorkflowStarted")[0]; + assert_eq!(started_event["payload"]["declared_update_validators"], json!(["approve_me", "reject_me"])); + + // Refusals before validation record nothing. + let (status, body) = request(&app, "POST", "/api/workflows/validated/update/approve_me", update_body(1, None)).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["reason"], "update_validation_request_id_required"); + assert_eq!(body["outcome"], "update_validation_failed"); + assert_eq!(body["retryable"], true); + assert_eq!(body["wait_for"], "accepted"); + assert!(body.get("update_id").is_none() && body.get("approved").is_none(), "{body}"); + let mut wrong = json!({"input":envelope(Payload::Array(vec![Payload::String("x".into())])),"request_id":"bad"}); + let (status, body) = request(&app, "POST", "/api/workflows/validated/update/approve_me", wrong.take()).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["reason"], "invalid_update_arguments"); + assert_eq!(body["message"], "Workflow update [approve_me] argument validation failed."); + assert_eq!(body["validation_errors"], json!({"value":["The value argument must be of type int."]})); + assert_eq!(history().await, before); + + // A rejection through the multiplexed poll. + let pending = send_update(&app, "validated", "reject_me", update_body(2, Some("r2"))); + let (status, polled) = request(&app, "POST", "/api/worker/workflow-tasks/poll", json!({"worker_id":"validator", + "task_queue":"vq","timeout_seconds":5,"task_kinds":["workflow","update_validation"]})).await; + assert_eq!(status, StatusCode::OK, "{polled}"); + let task = polled["task"].clone(); + assert_eq!(task["task_kind"], "update_validation", "{polled}"); + assert_eq!(task["update_name"], "reject_me"); + assert_eq!(task["update_validation_attempt"], 1); + assert_eq!(task["update_id"], task["update_validation_task_id"]); + assert_eq!(task["lease_owner"], "validator"); + assert!(task["next_history_page_token"].is_null() && task.get("next_history_page_token").is_some()); + assert_eq!(task["update_arguments"], update_body(2, None)["input"]); + assert!(task["history_events"].as_array().is_some_and(|events| !events.is_empty()), "{task}"); + assert!(task["history_export"].is_object(), "{task}"); + let id = task["update_validation_task_id"].as_str().unwrap().to_owned(); + let (status, body) = complete(&app, &id, "reject", json!({"lease_owner":"validator","update_validation_attempt":1})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["errors"], json!({"failure":["The failure field is required."],"failure.message":["The failure.message field is required."]})); + let (status, body) = complete(&app, &id, "reject", json!({"lease_owner":"validator","update_validation_attempt":1, + "failure":{"message":"too small","validation_errors":{"v":["too small"]}}})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["outcome"], "rejected"); + assert_eq!(body["reason"], "update_validator_rejected"); + let (status, rejected) = pending.await.unwrap(); + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{rejected}"); + for (field, expected) in [ + ("approved", json!(false)), + ("accepted", json!(false)), + ("outcome", json!("rejected_invalid_arguments")), + ("reason", json!("update_validator_rejected")), + ("update_status", json!("rejected")), + ("command_status", json!("rejected")), + ("message", json!("too small")), + ("validation_errors", json!({"v":["too small"]})), + ("update_validation_attempt", json!(1)), + ("update_id", json!(id)), + ("wait_timed_out", json!(false)), + ] { + assert_eq!(rejected[field], expected, "{field}: {rejected}"); + } + let (status, again) = complete(&app, &id, "reject", json!({"lease_owner":"validator","update_validation_attempt":1, + "failure":{"message":"again"}})).await; + assert_eq!(status, StatusCode::CONFLICT, "{again}"); + assert_eq!(again["reason"], "duplicate_update_validation_completion"); + + // An approval through the dedicated poll. + let pending = send_update(&app, "validated", "approve_me", update_body(3, Some("r1"))); + let task = poll_validation(&app, "validator", "vq").await; + let id = task["update_validation_task_id"].as_str().unwrap().to_owned(); + assert!(task.get("next_history_page_token").is_none(), "{task}"); + let (status, body) = complete(&app, &id, "approve", json!({"lease_owner":"someone","update_validation_attempt":1})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("update_validation_lease_owner_mismatch"))); + let (status, body) = complete(&app, &id, "approve", json!({"lease_owner":"validator","update_validation_attempt":2})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("stale_update_validation_completion"))); + let (status, body) = complete(&app, &id, "approve", json!({"lease_owner":"validator","update_validation_attempt":1})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body, json!({"update_validation_task_id":id,"update_id":id,"update_validation_attempt":1,"outcome":"approved", + "status":200,"protocol_version":body["protocol_version"],"server_capabilities":body["server_capabilities"]})); + let (status, approved) = pending.await.unwrap(); + assert_eq!(status, StatusCode::ACCEPTED, "{approved}"); + assert_eq!(approved["update_id"], json!(id)); + assert_eq!(approved["update_status"], "accepted"); + let (status, unknown) = complete(&app, "01ZZZZZZZZZZZZZZZZZZZZZZZZ", "approve", + json!({"lease_owner":"validator","update_validation_attempt":1})).await; + assert_eq!(status, StatusCode::NOT_FOUND, "{unknown}"); + assert!(unknown.get("update_validation_attempt").is_none(), "{unknown}"); + + // Retries return the recorded receipts; an undeclared validator is not asked. + let (status, retained) = request(&app, "POST", "/api/workflows/validated/update/approve_me", update_body(99, Some("r1"))).await; + assert_eq!((status, retained["update_id"].clone()), (StatusCode::ACCEPTED, json!(id))); + let (status, retained) = request(&app, "POST", "/api/workflows/validated/update/reject_me", update_body(2, Some("r2"))).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{retained}"); + assert_eq!(retained["reason"], "update_validator_rejected"); + assert_eq!(retained["validation_errors"], json!({"v":["too small"]})); + let (status, plain) = request(&app, "POST", "/api/workflows/validated/update/plain", update_body(4, Some("p1"))).await; + assert_eq!(status, StatusCode::ACCEPTED, "{plain}"); + + let after = history().await; + let rejected_events = events(&after, "UpdateRejected"); + assert_eq!(rejected_events.len(), 1, "{after}"); + assert_eq!(rejected_events[0]["payload"]["update_id"], rejected["update_id"]); + assert_eq!(rejected_events[0]["payload"]["validation_errors"], json!({"v":["too small"]})); + let accepted = events(&after, "UpdateAccepted"); + assert_eq!(accepted.len(), 2, "{after}"); + assert_eq!(accepted[0]["payload"]["update_id"], json!(id)); + assert_eq!(accepted[0]["payload"]["request_id"], "r1"); +} + +#[tokio::test] +async fn validation_routes_time_out_and_fence_like_php() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + // A worker contract without validators makes every remote update refuse. + validating_worker(&app, "silent", "q-missing", None, true).await; + waiting_run(&app, "missing", "q-missing", "silent").await; + let (status, body) = request(&app, "POST", "/api/workflows/missing/update/plain", update_body(1, None)).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "update_validator_contract_missing"); + assert_eq!(body["retryable"], false); + + // No capable worker, or none matching the run. + validating_worker(&app, "incapable", "q-cap", Some(json!(["approve_me"])), false).await; + waiting_run(&app, "incapable-run", "q-cap", "incapable").await; + let (status, body) = request(&app, "POST", "/api/workflows/incapable-run/update/approve_me", update_body(1, Some("c1"))).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!((body["reason"].clone(), body["retryable"].clone()), (json!("update_validation_capability_unsupported"), json!(false))); + assert_eq!(body["approved"], false); + assert!(body["update_id"].is_string(), "{body}"); + let (status, body) = request(&app, "POST", "/api/worker/update-validation-tasks/poll", + json!({"worker_id":"incapable","task_queue":"q-cap","timeout_seconds":0})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("update_validation_capability_not_advertised"))); + let (status, body) = request(&app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":"incapable","task_queue":"q-cap","timeout_seconds":0,"task_kinds":["update_validation"]})).await; + assert_eq!((status, body["poll_status"].clone()), (StatusCode::CONFLICT, json!("unsupported"))); + validating_worker(&app, "changed", "q-changed", Some(json!(["approve_me"])), true).await; + waiting_run(&app, "changed-run", "q-changed", "changed").await; + validating_worker(&app, "changed", "q-changed", Some(json!([])), true).await; + let (status, body) = request(&app, "POST", "/api/workflows/changed-run/update/approve_me", update_body(1, Some("i1"))).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!((body["reason"].clone(), body["retryable"].clone()), (json!("update_validator_worker_incompatible"), json!(true))); + let (status, body) = request(&app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":"changed","task_queue":"q-changed","timeout_seconds":0,"task_kinds":["workflow","activity"]})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["errors"], json!({"task_kinds.1":["The selected task_kinds.1 is invalid."]})); + + // Nobody claims: the request waits out the timeout, and retries replay it. + validating_worker(&app, "idle", "q-idle", Some(json!(["approve_me"])), true).await; + waiting_run(&app, "idle-run", "q-idle", "idle").await; + let unclaimed = send_update(&app, "idle-run", "approve_me", update_body(5, Some("r3"))); + + // A claimed task whose worker leaves. + validating_worker(&app, "leaver", "q-lost", Some(json!(["approve_me"])), true).await; + waiting_run(&app, "lost-run", "q-lost", "leaver").await; + let lost = send_update(&app, "lost-run", "approve_me", update_body(6, Some("l1"))); + let task = poll_validation(&app, "leaver", "q-lost").await; + let lost_id = task["update_validation_task_id"].as_str().unwrap().to_owned(); + assert_eq!(request(&app, "DELETE", "/api/workers/leaver", Value::Null).await.0, StatusCode::OK); + let (status, body) = complete(&app, &lost_id, "approve", json!({"lease_owner":"leaver","update_validation_attempt":1})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("update_validator_worker_lost"))); + + let (status, body) = unclaimed.await.unwrap(); + assert_eq!(status, StatusCode::GATEWAY_TIMEOUT, "{body}"); + assert_eq!(body["reason"], "update_validation_task_not_claimed"); + assert_eq!(body["update_validation_attempt"], 0); + let (status, again) = request(&app, "POST", "/api/workflows/idle-run/update/approve_me", update_body(5, Some("r3"))).await; + assert_eq!((status, again["update_id"].clone()), (StatusCode::GATEWAY_TIMEOUT, body["update_id"].clone())); + let (status, conflict) = request(&app, "POST", "/api/workflows/idle-run/update/approve_me", update_body(6, Some("r3"))).await; + assert_eq!((status, conflict["reason"].clone()), (StatusCode::CONFLICT, json!("update_validation_idempotency_conflict"))); + let (status, body) = lost.await.unwrap(); + assert_eq!(status, StatusCode::GATEWAY_TIMEOUT, "{body}"); + assert_eq!(body["reason"], "update_validator_worker_lost"); + + // With no worker left on the queue. + let (status, body) = request(&app, "POST", "/api/workflows/lost-run/update/approve_me", update_body(7, Some("u1"))).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("update_validator_worker_unavailable"))); +} + +#[tokio::test] +async fn expired_validation_leases_are_claimed_again() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + validating_worker(&app, "slow", "q-slow", Some(json!(["approve_me", "reject_me"])), true).await; + waiting_run(&app, "slow-run", "q-slow", "slow").await; + let pending = send_update(&app, "slow-run", "approve_me", update_body(8, Some("r4"))); + let first = poll_validation(&app, "slow", "q-slow").await; + let id = first["update_validation_task_id"].as_str().unwrap().to_owned(); + tokio::time::sleep(Duration::from_secs(6)).await; + let (status, body) = complete(&app, &id, "approve", json!({"lease_owner":"slow","update_validation_attempt":1})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("update_validation_lease_expired"))); + let second = poll_validation(&app, "slow", "q-slow").await; + assert_eq!((second["update_validation_task_id"].clone(), second["update_validation_attempt"].clone()), (json!(id), json!(2))); + let (status, body) = complete(&app, &id, "approve", json!({"lease_owner":"slow","update_validation_attempt":1})).await; + assert_eq!((status, body["reason"].clone()), (StatusCode::CONFLICT, json!("stale_update_validation_completion"))); + assert_eq!(complete(&app, &id, "approve", json!({"lease_owner":"slow","update_validation_attempt":2})).await.0, StatusCode::OK); + let (status, body) = pending.await.unwrap(); + assert_eq!((status, body["update_id"].clone()), (StatusCode::ACCEPTED, json!(id))); + + // A failure other than a validator rejection records no update. + let pending = send_update(&app, "slow-run", "reject_me", update_body(9, Some("r5"))); + let task = poll_validation(&app, "slow", "q-slow").await; + let id = task["update_validation_task_id"].as_str().unwrap().to_owned(); + let (status, body) = complete(&app, &id, "reject", json!({"lease_owner":"slow","update_validation_attempt":1, + "failure":{"message":"validator crashed","reason":"validator_error","type":"RuntimeError","stack_trace":"trace"}})).await; + assert_eq!((status, body["outcome"].clone()), (StatusCode::OK, json!("failed"))); + let (status, failed) = pending.await.unwrap(); + assert_eq!(status, StatusCode::CONFLICT, "{failed}"); + assert_eq!((failed["reason"].clone(), failed["failure_type"].clone()), (json!("validator_error"), json!("RuntimeError"))); + assert!(failed.get("update_status").is_none(), "{failed}"); + + // The multiplexed poll alternates kinds after each lease. + assert_eq!(request(&app, "POST", "/api/workflows/slow-run/update/plain", update_body(10, Some("p2"))).await.0, StatusCode::ACCEPTED); + let pending = send_update(&app, "slow-run", "approve_me", update_body(11, Some("r6"))); + tokio::time::sleep(Duration::from_millis(300)).await; + let mut kinds = vec![]; + for _ in 0..2 { + let (status, body) = request(&app, "POST", "/api/worker/workflow-tasks/poll", json!({"worker_id":"slow", + "task_queue":"q-slow","timeout_seconds":1,"task_kinds":["update_validation","workflow"]})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let task = &body["task"]; + kinds.push(task["task_kind"].clone()); + if task["task_kind"] == "update_validation" { + let id = task["update_validation_task_id"].as_str().unwrap(); + assert_eq!(complete(&app, id, "approve", json!({"lease_owner":"slow","update_validation_attempt":1})).await.0, StatusCode::OK); + } + } + assert_eq!(kinds, vec![json!("update_validation"), json!("workflow")]); + assert_eq!(pending.await.unwrap().0, StatusCode::ACCEPTED); +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 85f39f1ac..a182befeb 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -19,6 +19,7 @@ import {checkPrometheusMetrics} from './prometheus-metrics-contract.mjs'; import {checkPayloadCleanup} from './payload-cleanup-contract.mjs'; import {checkRepairPass} from './repair-pass-contract.mjs'; import {checkNamespaceDelete} from './namespace-delete-contract.mjs'; +import {checkUpdateValidation} from './update-validation-contract.mjs'; import {checkSchedulePolicies} from './schedule-policies-contract.mjs'; import {checkScheduleCatalog} from './schedule-catalog-contract.mjs'; import {checkSignalsBeforeWorker} from './signals-before-worker-contract.mjs'; @@ -155,6 +156,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const payloadCleanup = fixture.payload_cleanup ? checkPayloadCleanup(fixture, observation, workflowId) : null; const repairPass = fixture.repair_pass ? checkRepairPass(fixture, observation, workflowId) : null; const namespaceDelete = fixture.namespace_delete ? checkNamespaceDelete(fixture, observation, workflowId) : null; + const updateValidation = fixture.update_validation ? checkUpdateValidation(fixture, observation, workflowId) : null; const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null; const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId) : null; const signalsBeforeWorker = fixture.signals_before_worker ? checkSignalsBeforeWorker(fixture, observation, workflowId) : null; @@ -708,6 +710,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(payloadCleanup ? {payload_cleanup: payloadCleanup} : {}), ...(repairPass ? {repair_pass: repairPass} : {}), ...(namespaceDelete ? {namespace_delete: namespaceDelete} : {}), + ...(updateValidation ? {update_validation: updateValidation} : {}), ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}), ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}), ...(signalsBeforeWorker ? {signals_before_worker: signalsBeforeWorker} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index b6339da20..61f926c11 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -437,6 +437,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $payloadCleanup = isset($fixture['payload_cleanup']) ? finishHttpPayloadCleanup($client, $fixture, $workflowId, $url) : null; $repairPass = isset($fixture['repair_pass']) ? finishHttpRepairPass($client, $fixture, $workflowId, $url) : null; $namespaceDelete = isset($fixture['namespace_delete']) ? finishHttpNamespaceDelete($client, $fixture, $workflowId, $queue, $url) : null; + $updateValidation = isset($fixture['update_validation']) ? finishHttpUpdateValidation($client, $fixture, $workflowId, $url) : null; $schedulePolicies = isset($fixture['schedule_policies']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url) : null; $scheduleCatalog = isset($fixture['schedule_catalog']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url, 'schedule_catalog') @@ -543,6 +544,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'payload_cleanup' => $payloadCleanup, 'repair_pass' => $repairPass, 'namespace_delete' => $namespaceDelete, + 'update_validation' => $updateValidation, 'schedule_policies' => $schedulePolicies, 'schedule_catalog' => $scheduleCatalog, 'signals_before_worker' => $signalsBeforeWorker, @@ -853,6 +855,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/payload-cleanup-probe.php'; require __DIR__.'/repair-pass-probe.php'; require __DIR__.'/namespace-delete-probe.php'; + require __DIR__.'/update-validation-probe.php'; require __DIR__.'/schedule-policies-probe.php'; require __DIR__.'/signals-before-worker-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; @@ -1032,6 +1035,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['namespace_delete'])) { $observation['namespace_delete'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_namespaces']; } + if ($mode === 'embedded' && isset($fixture['update_validation'])) { + $observation['update_validation'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_update_validation']; + } if ($mode === 'embedded' && isset($fixture['schedule_policies'])) { $observation['schedule_policies'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_schedules']; } diff --git a/scripts/conformance/server-parity/update-validation-contract.mjs b/scripts/conformance/server-parity/update-validation-contract.mjs new file mode 100644 index 000000000..56007ac98 --- /dev/null +++ b/scripts/conformance/server-parity/update-validation-contract.mjs @@ -0,0 +1,64 @@ +import assert from 'node:assert/strict'; + +// Synchronous update validation, as published Server 2.11.1 serves it. Every +// recorded step must equal the reviewed fixture after three normalizations: +// - identifiers become labels in the order they first appear, walking object +// keys in sorted order so either server's key order reads alike; a reused +// identifier (an update keeping its validation task's id, a retry returning +// its receipt) must be reused, and a new one must be new; +// - timestamps become "