From b20a98e1dded88510da2e5c2db4409ab8b7c003b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 00:02:50 +0000 Subject: [PATCH] Serve the task repair pass like published PHP 2.11 POST /api/system/repair/pass recovers overdue ready tasks and expired leases, creates lost workflow tasks and deadline tasks, and enforces expired activity deadlines, with PHP's request rules, scope-fair selection and loop throttle. Recoveries native does not perform yet (an expired running activity lease, a missing task for a run waiting on an activity, timer, update, signal or child) are reported as candidate failures with 207. Repair status now shares the pass's candidate query, with PHP 2.11's availability rule and microsecond cutoff. --- .github/workflows/server-parity.yml | 4 +- .../2026-10-23-repair-pass.md | 65 ++ docs/rust-server-port.md | 27 +- rust/src/runtime/execution.rs | 1 + rust/src/runtime/http.rs | 15 +- rust/src/runtime/mod.rs | 1 + rust/src/runtime/operator_metrics.rs | 13 +- rust/src/runtime/repair_candidates.rs | 56 +- rust/src/runtime/repair_pass.rs | 800 ++++++++++++++++++ rust/src/runtime/store.rs | 5 + rust/src/runtime/task_repair.rs | 22 +- rust/tests/execution.rs | 2 + rust/tests/execution/repair_pass.rs | 151 ++++ .../conformance/server-parity/contract.mjs | 3 + scripts/conformance/server-parity/probe.php | 6 + .../server-parity/repair-pass-contract.mjs | 28 + .../server-parity/repair-pass-probe.php | 36 + tests/Fixtures/ServerParity/repair-pass.json | 417 +++++++++ .../ServerParityRepairPassContractTest.mjs | 68 ++ 19 files changed, 1653 insertions(+), 67 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-repair-pass.md create mode 100644 rust/src/runtime/repair_pass.rs create mode 100644 rust/tests/execution/repair_pass.rs create mode 100644 scripts/conformance/server-parity/repair-pass-contract.mjs create mode 100644 scripts/conformance/server-parity/repair-pass-probe.php create mode 100644 tests/Fixtures/ServerParity/repair-pass.json create mode 100644 tests/Unit/ServerParityRepairPassContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 5ba5b0809..ac38bb173 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -39,6 +39,7 @@ on: - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs - tests/Unit/ServerParityPayloadCleanupContractTest.mjs + - tests/Unit/ServerParityRepairPassContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs @@ -92,6 +93,7 @@ on: - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs - tests/Unit/ServerParityPayloadCleanupContractTest.mjs + - tests/Unit/ServerParityRepairPassContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs @@ -1510,7 +1512,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/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 > 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-repair-pass.md b/docs/rust-server-port-checkpoints/2026-10-23-repair-pass.md new file mode 100644 index 000000000..b3058c0e8 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-repair-pass.md @@ -0,0 +1,65 @@ +# October 23, task repair pass + +- **Source.** Server 2.11.1 `SystemController::repairPass`, + `TaskWatchdog::runPass`, `TaskRepairCandidates`, `TaskRepair`, + `TaskRepairPolicy`, `RequestedTaskScope` and + `ActivityTimeoutEnforcer::expiredExecutionIds`. +- **Served.** Native answers admin-only `POST /api/system/repair/pass` in the + default or a named namespace. Like PHP's, the pass is server-wide. + - The request takes `run_ids` (at most 100 strings of 1 to 128 + characters), `instance_id`, `connection`, `queue` (comma-separated + values; `default` also matches a null or empty column) and + `respect_throttle`. Strings are trimmed and empty ones read as null. + - A pass that respects the throttle runs nothing within 5 seconds of the + previous pass and reports `throttled: true`. Any other pass runs and + restarts the throttle. The throttle is shared by the server's namespaces. + - Selection is PHP's poll-mode selection. It takes up to 25 repairable + tasks and up to 25 runs that need a task. Candidates come one from each + connection, queue and compatibility scope in turn, oldest first. + - A ready task is repairable once it is due and its dispatch failed, its + claim failed, or its dispatch and availability are more than 3 seconds + old. A leased task is repairable once its lease expired. + - Each selected task is recovered under its run. A ready task counts one + more repair. An expired lease returns its task to ready and closes the + lease's activity attempt. A closed run's task is settled instead. The + task is then marked dispatched. + - A run whose projected summary still needs repair gets its recoverable + task back. A replay-blocked task comes first. Otherwise the run gets a + new workflow task. + - An open run past its execution or run deadline, with no open workflow + task, gets a task with reason `deadline_expired`; native closes the run + when that task runs. An expired cancellation cleanup deadline is + enforced under a control task. + - Expired activity deadlines in the requested connections and queues are + enforced as by the activity timeout pass. +- **Decisions.** + - Two recoveries are not ported. They are reported as their candidate's + failure, so the pass answers 207 rather than claiming a repair: + - An expired lease on a running activity. PHP consumes the activity's + retry policy. The failure is also kept as the task's last error, as + PHP keeps a recovery exception. + - A missing task for a run waiting on an activity, a timer, a timer-backed + condition, an update, a signal or a child. + - Native has no service calls and no command contracts to backfill, so + those counts are always zero. + - A closed run leaves an activity open only when it completed without + waiting for it. Native treats such an open activity as abandoned, as + `ActivityAbandonment::allows` would. + - Native runs no repair loop. The pass runs only when requested. +- **Shared rule.** Repair status and the pass now share one candidate query. + The overdue branch also requires availability before the redispatch + cutoff. The cutoff binds with microseconds, as PHP 2.11 binds it. Native's + per-task recheck takes availability after the last dispatch as the + redispatch reference, as `dispatchOverdue` does. +- **Fixture.** `repair-pass` checks nine requests. Three passes are scoped to + run ids and a queue that do not exist, so each selects nothing and echoes + its trimmed filters, with empty strings read as null. Six invalid requests + answer Laravel's 422. The published PHP answers were checked against a + fresh 2.11.1 server. Native tests cover the recoveries on SQLite, + PostgreSQL and MySQL: + - the queue filter; + - the throttle; + - an overdue ready task and an expired workflow lease; + - a lost workflow task; + - a deadline task; + - the 207 report for an expired running activity lease. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index cb7e5b5a4..6b3b448e1 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -102,19 +102,22 @@ These published PHP routes on this tree have no native handler yet, by family. - **Used by the CLI or Waterline.** - Standalone activities: list, describe and start. - - The repair pass. (Namespace delete is served with the - `namespace-delete` fixture. The JSON - `GET|POST /api/system/metrics` is served with the `system-metrics` - fixture: namespace failure, durable-state and payload counts are exact; - native enforces no admission, durable-state or payload quota, so limits - are null and rejections zero, as on an unconfigured PHP server. - Projection drift is server-wide, and native's only drift source is a run - without its summary. The Prometheus summary - `GET /api/system/prometheus-metrics` is served with the - `prometheus-metrics` fixture, with PHP's default limit of 100 series per - family. External payload cleanup and its pass are served with the - `payload-cleanup` fixture.) - Task queue priority and fairness. + + Served from this family since: namespace delete, with the + `namespace-delete` fixture. The JSON + `GET|POST /api/system/metrics` is served with the `system-metrics` + fixture: namespace failure, durable-state and payload counts are exact; + native enforces no admission, durable-state or payload quota, so limits + are null and rejections zero, as on an unconfigured PHP server. + Projection drift is server-wide, and native's only drift source is a run + without its summary. The Prometheus summary + `GET /api/system/prometheus-metrics` is served with the + `prometheus-metrics` fixture, with PHP's default limit of 100 series per + family. External payload cleanup and its pass are served with the + `payload-cleanup` fixture. The task repair pass is served with the + `repair-pass` fixture; a recovery native does not perform yet is + reported as its candidate's failure, with 207. - **Worker protocol features.** - Worker sessions. - Update validation tasks. diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 3e18a769f..cabfc5d84 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -339,6 +339,7 @@ impl Runtime { delegate!(system_health(config: &super::operator_metrics::MetricsConfig) -> Result<(bool, Value)>); delegate!(operator_dashboard(config: &super::operator_metrics::MetricsConfig, bounded: bool) -> Result); delegate!(repair_status() -> Result); + delegate!(repair_pass(request: &super::repair_pass::PassRequest) -> Result); delegate!(system_metrics() -> Result); delegate!(prometheus_metrics() -> Result); delegate!(payload_cleanup_status(config: &super::operator_metrics::MetricsConfig) -> Result); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index 8089cea58..a27e18bff 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -214,6 +214,7 @@ pub fn router(runtime: Runtime) -> Router { get(workflow_type_operator_dashboard), ) .route("/api/system/repair", get(repair_status)) + .route("/api/system/repair/pass", post(repair_pass)) .route("/api/system/retention", get(retention_status)) .route("/api/system/retention/pass", post(retention_pass)) .route("/api/system/activity-timeouts", get(activity_timeout_status)) @@ -646,7 +647,11 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { { return method == "GET"; } - if path == "/api/system/retention/pass" || path == "/api/system/activity-timeouts/pass" { + // The repair pass is server-wide in PHP too. + if path == "/api/system/retention/pass" + || path == "/api/system/activity-timeouts/pass" + || path == "/api/system/repair/pass" + { return method == "POST"; } // The metrics describe the namespace's own state, read by GET or POST. @@ -1447,6 +1452,14 @@ async fn repair_status(Extension(runtime): Extension) -> Result, body: axum::body::Bytes) -> Result { + let request = super::repair_pass::pass_request(&request_json(&body)?)?; + let report = runtime.repair_pass(&request).await?; + let failed = super::repair_pass::failed(&report); + Ok(pass_response((report, failed))) +} + async fn retention_status(Extension(runtime): Extension, Query(query): QueryMap) -> Result> { let limit = super::retention::status_limit(query.get("limit").map(String::as_str))?; Ok(Json(runtime.retention_status(limit).await?)) diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index e93b05bdc..25c3668a5 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -78,6 +78,7 @@ mod prometheus_metrics; mod queries; mod query_authority; mod repair_candidates; +mod repair_pass; mod retention; mod runtime_credentials; mod run_observation; diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs index 0e3fb648c..ebbfed839 100644 --- a/rust/src/runtime/operator_metrics.rs +++ b/rust/src/runtime/operator_metrics.rs @@ -1084,14 +1084,9 @@ where /// published PHP's repair pass would select across every namespace. pub(super) async fn repair_status(&self) -> Result { let at = now(); - let bound = whole_second(at); - let cutoff = whole_second(at - Duration::seconds(REDISPATCH_AFTER_SECONDS)); - let mut tasks = Self::query(repair_candidates::TASK_SCOPES); - for time in [bound, bound, bound, cutoff, cutoff, cutoff, bound] { - tasks = tasks.bind(DB::bind_time(time)); - } - let tasks = scope_rows::(tasks.fetch_all(&self.pool).await?)?; - let runs = scope_rows::(Self::query(repair_candidates::RUN_SCOPES).fetch_all(&self.pool).await?)?; + let unfiltered = super::repair_pass::PassRequest::default(); + let tasks = self.repairable_task_scopes(&unfiltered, at).await?; + let runs = self.missing_task_scopes(&unfiltered).await?; let oldest_run = match Self::query(repair_candidates::OLDEST_RUN).fetch_optional(&self.pool).await? { Some(row) => DB::optional_instant(&row, "wait_started_at")? .or(DB::optional_instant(&row, "started_at")?) @@ -1450,7 +1445,7 @@ fn backend_capabilities(now: DateTime, connection: &str) -> Value { /// Native claims ready tasks on each poll; it runs no repair loop. /// The grouped candidate rows of one repair candidate query. -fn scope_rows(rows: Vec) -> Result> +pub(super) fn scope_rows(rows: Vec) -> Result> where usize: ColumnIndex, for<'r> i64: Decode<'r, DB>, diff --git a/rust/src/runtime/repair_candidates.rs b/rust/src/runtime/repair_candidates.rs index 323951192..5c8e02fe5 100644 --- a/rust/src/runtime/repair_candidates.rs +++ b/rust/src/runtime/repair_candidates.rs @@ -7,27 +7,6 @@ use serde_json::{Value, json}; /// `TaskRepairPolicy::SCAN_LIMIT`. pub(super) const SCAN_LIMIT: usize = 25; -/// `existingTaskQuery`: ready tasks that are due and whose dispatch failed, -/// whose claim failed, or whose dispatch is overdue, and leased tasks whose -/// lease expired. `$1`–`$3` and `$7` are "now"; `$4`–`$6` are the redispatch -/// cutoff, both in whole seconds as Laravel binds them. -macro_rules! repairable_tasks { - () => { - "((status='ready' AND (available_at IS NULL OR available_at<=$1) AND (\ - (last_dispatch_attempt_at IS NOT NULL AND last_dispatch_error IS NOT NULL AND last_dispatch_error<>'' \ - AND (last_dispatched_at IS NULL OR last_dispatch_attempt_at>last_dispatched_at) \ - AND (repair_available_at IS NULL OR repair_available_at<=$2)) \ - OR (last_claim_failed_at IS NOT NULL AND last_claim_error IS NOT NULL AND last_claim_error<>'' \ - AND ((repair_available_at IS NOT NULL AND repair_available_at<=$3) \ - OR (repair_available_at IS NULL AND last_claim_failed_at<=$4))) \ - OR ((last_dispatch_attempt_at IS NULL OR last_dispatch_error IS NULL OR last_dispatch_error='' \ - OR (last_dispatched_at IS NOT NULL AND last_dispatch_attempt_at<=last_dispatched_at)) \ - AND (last_claim_failed_at IS NULL OR last_claim_error IS NULL OR last_claim_error='') \ - AND (last_dispatched_at<=$5 OR (last_dispatched_at IS NULL AND created_at<=$6))))) \ - OR (status='leased' AND lease_expires_at IS NOT NULL AND lease_expires_at<=$7))" - }; -} - /// `missingTaskRunQuery`: open runs without a next task that need repair, or /// that wait for children which have all closed while no workflow task is open. macro_rules! missing_task_runs { @@ -45,18 +24,7 @@ macro_rules! missing_task_runs { }; } -pub(super) const TASK_SCOPES: &str = concat!( - "SELECT connection,queue,compatibility,COUNT(*) AS candidate_count,MIN(created_at) AS oldest_candidate_at \ - FROM workflow_tasks WHERE ", - repairable_tasks!(), - " GROUP BY connection,queue,compatibility" -); -pub(super) const RUN_SCOPES: &str = concat!( - "SELECT connection,queue,compatibility,COUNT(*) AS candidate_count,\ - MIN(COALESCE(wait_started_at,started_at,created_at)) AS oldest_candidate_at FROM workflow_run_summaries WHERE ", - missing_task_runs!(), - " GROUP BY connection,queue,compatibility" -); +pub(super) const MISSING_TASK_RUNS: &str = missing_task_runs!(); /// `oldestMissingRunCandidate`, in the backend's own null ordering, as PHP's. pub(super) const OLDEST_RUN: &str = concat!( "SELECT wait_started_at,started_at,created_at FROM workflow_run_summaries WHERE ", @@ -75,7 +43,7 @@ pub(super) struct ScopeRow { impl ScopeRow { /// `nullableString` of each scope column. - fn values(&self) -> [Option<&str>; 3] { + pub(super) fn values(&self) -> [Option<&str>; 3] { [&self.connection, &self.queue, &self.compatibility] .map(|value| value.as_deref().filter(|value| !value.is_empty())) } @@ -100,6 +68,19 @@ impl ScopeRow { /// `normalizedScopes` with each scope's selected ids counted: oldest scope /// first, each holding at most `limit` candidates. fn buckets(rows: &[ScopeRow], limit: usize) -> Vec<(String, usize)> { + normalized(rows) + .into_iter() + .map(|scope| { + let count: usize = rows.iter().filter(|row| row.within(scope)).map(|row| row.count).sum(); + (scope.key(), count.min(limit)) + }) + .filter(|(_, count)| *count > 0) + .collect() +} + +/// `normalizedScopes`: one row per scope key, keeping its oldest, ordered by +/// oldest candidate and then key. +pub(super) fn normalized(rows: &[ScopeRow]) -> Vec<&ScopeRow> { let mut chosen: Vec<&ScopeRow> = Vec::new(); for row in rows { match chosen.iter().position(|scope| scope.key() == row.key()) { @@ -116,13 +97,6 @@ fn buckets(rows: &[ScopeRow], limit: usize) -> Vec<(String, usize)> { order(left).cmp(&order(right)).then_with(|| left.key().cmp(&right.key())) }); chosen - .into_iter() - .map(|scope| { - let count: usize = rows.iter().filter(|row| row.within(scope)).map(|row| row.count).sum(); - (scope.key(), count.min(limit)) - }) - .filter(|(_, count)| *count > 0) - .collect() } /// `selectedScopeCounts`: one candidate from each scope in turn, up to diff --git a/rust/src/runtime/repair_pass.rs b/rust/src/runtime/repair_pass.rs new file mode 100644 index 000000000..8a429ced9 --- /dev/null +++ b/rust/src/runtime/repair_pass.rs @@ -0,0 +1,800 @@ +//! `POST /api/system/repair/pass`, as published Server 2.11 serves it +//! (`SystemController::repairPass` over `TaskWatchdog::runPass`): one +//! server-wide task repair pass, optionally scoped to runs, an instance, and +//! connections or queues. +//! +//! The pass recovers selected existing tasks (a ready task overdue for +//! dispatch is redispatched; an expired lease returns its task to ready), +//! creates a missing workflow task for a run that needs one, gives a run whose +//! deadline passed a task that closes it (or enforces an expired cancellation +//! cleanup deadline), and enforces expired activity deadlines. Selection is +//! PHP's poll-mode selection: up to 25 candidates of each kind, one from each +//! connection, queue and compatibility scope in turn, oldest first. +//! +//! Recoveries native does not perform yet are reported as failures of their +//! candidate, so the pass answers 207 rather than claiming a repair: an +//! expired lease on a running activity (PHP consumes its retry policy), and a +//! missing task for a run waiting on an activity, a timer, an update, a signal +//! or a child. +use super::Result; +use super::laravel_validation::{Rule, failure_messages, rule_errors}; +use super::operator_metrics::{REDISPATCH_AFTER_SECONDS, scope_rows, whole_second}; +use super::repair_candidates::{ScopeRow, normalized}; +use super::store::now; +use super::transaction::Transaction; +use chrono::{DateTime, Duration, Utc}; +use serde_json::{Map, Value, json}; +use sqlx::AssertSqlSafe; +use std::collections::VecDeque; +use std::time::Instant; + +/// `TaskRepairPolicy::scanLimit()`. +const SCAN_LIMIT: usize = 25; +/// `TaskRepairPolicy::loopThrottleSeconds()`. +const LOOP_THROTTLE_SECONDS: u64 = 5; + +/// The pass's request, after PHP's validation and trimming. +#[derive(Default)] +pub(crate) struct PassRequest { + pub(crate) run_ids: Vec, + pub(crate) instance_id: Option, + pub(crate) connection: Option, + pub(crate) queue: Option, + pub(crate) respect_throttle: bool, +} + +/// `repairPass`'s rules, in PHP's order. Laravel trims strings and reads +/// empty ones as null before validating. +pub(super) fn pass_request(body: &Value) -> Result { + let body = &super::start_request::normalized(body.clone()); + let mut errors = rule_errors(body, &[("run_ids", &[Rule::Nullable, Rule::Array, Rule::Max(100)])], |_, _| None); + let items: Vec<(String, Value)> = match &body["run_ids"] { + Value::Array(items) => items.iter().enumerate().map(|(index, item)| (index.to_string(), item.clone())).collect(), + Value::Object(fields) => fields.iter().map(|(key, item)| (key.clone(), item.clone())).collect(), + _ => Vec::new(), + }; + for (key, item) in &items { + let name = format!("run_ids.{key}"); + let mut value = Map::new(); + value.insert(name.clone(), item.clone()); + let rules: &[Rule] = &[Rule::String, Rule::Min(1), Rule::Max(128)]; + errors.extend(rule_errors(&Value::Object(value), &[(name.as_str(), rules)], |_, _| None)); + } + errors.extend(rule_errors( + body, + &[ + ("instance_id", &[Rule::Nullable, Rule::String, Rule::Min(1), Rule::Max(128)]), + ("connection", &[Rule::Nullable, Rule::String, Rule::Max(128)]), + ("queue", &[Rule::Nullable, Rule::String, Rule::Max(128)]), + ("respect_throttle", &[Rule::Nullable, Rule::Boolean]), + ], + |_, _| None, + )); + if let Some(error) = failure_messages(errors) { + return Err(error); + } + let text = |field: &str| body[field].as_str().map(str::to_owned); + Ok(PassRequest { + run_ids: items.iter().filter_map(|(_, id)| id.as_str().map(str::to_owned)).collect(), + instance_id: text("instance_id"), + connection: text("connection"), + queue: text("queue"), + // `$request->boolean()`: true, 1 and "1" are true. + respect_throttle: matches!(&body["respect_throttle"], Value::Bool(true)) + || body["respect_throttle"].as_i64() == Some(1) + || body["respect_throttle"].as_str() == Some("1"), + }) +} + +/// `emptyReport` with the controller's command-contract defaults. +fn empty_report(request: &PassRequest, throttled: bool) -> Map { + let value = json!({ + "connection": request.connection, "queue": request.queue, "run_ids": request.run_ids, + "instance_id": request.instance_id, "respect_throttle": request.respect_throttle, + "selected_command_contract_candidates": 0, "backfilled_command_contracts": 0, + "command_contract_backfill_unavailable": 0, "command_contract_failures": [], + "existing_task_failures": [], "missing_run_failures": [], "throttled": throttled, + "selected_existing_task_candidates": 0, "selected_missing_task_candidates": 0, "selected_total_candidates": 0, + "repaired_existing_tasks": 0, "repaired_missing_tasks": 0, "dispatched_tasks": 0, + "deadline_expired_candidates": 0, "deadline_expired_tasks_created": 0, "cancellation_deadlines_enforced": 0, + "deadline_expired_failures": [], "activity_timeout_candidates": 0, "activity_timeouts_enforced": 0, + "activity_timeout_failures": [], "service_call_timeout_candidates": 0, "service_call_timeouts_enforced": 0, + "service_call_timeout_failures": [], + }); + match value { + Value::Object(map) => map, + _ => unreachable!("the report is an object"), + } +} + +/// Whether the report holds a failure that PHP answers with 207. +pub(super) fn failed(report: &Value) -> bool { + ["existing_task_failures", "missing_run_failures", "command_contract_failures"] + .iter() + .any(|field| report[*field].as_array().is_some_and(|failures| !failures.is_empty())) +} + +fn bump(report: &mut Map, field: &str) { + let current = report[field].as_i64().unwrap_or_default(); + report.insert(field.to_owned(), json!(current + 1)); +} + +fn record_failure(report: &mut Map, field: &str, key: &str, id: &str, message: String) { + if let Some(Value::Array(failures)) = report.get_mut(field) { + failures.push(json!({key: id, "message": message})); + } +} + +/// `RequestedTaskScope::normalizeValues`: comma-separated, trimmed, unique. +fn scope_values(value: Option<&str>) -> Vec { + let mut values: Vec = Vec::new(); + for candidate in value.unwrap_or_default().split(',').map(str::trim).filter(|value| !value.is_empty()) { + if !values.iter().any(|value| value == candidate) { + values.push(candidate.to_owned()); + } + } + values +} + +enum Bind { + Text(String), + Time(DateTime), +} + +/// A statement built from fixed SQL text, with every value bound and numbered +/// for the backend. +pub(super) struct Sql { + text: String, + binds: Vec, + mysql: bool, +} + +impl Sql { + pub(super) fn new(text: &str, mysql: bool) -> Self { + Self { text: text.to_owned(), binds: Vec::new(), mysql } + } + + pub(super) fn push(&mut self, text: &str) -> &mut Self { + self.text.push_str(text); + self + } + + fn bind(&mut self, value: Bind) -> &mut Self { + self.binds.push(value); + if self.mysql { + self.text.push('?'); + } else { + self.text.push_str(&format!("${}", self.binds.len())); + } + self + } + + fn text(&mut self, value: &str) -> &mut Self { + self.bind(Bind::Text(value.to_owned())) + } + + fn time(&mut self, value: DateTime) -> &mut Self { + self.bind(Bind::Time(value)) + } + + fn list(&mut self, values: &[String]) -> &mut Self { + self.push("("); + for (index, value) in values.iter().enumerate() { + if index > 0 { + self.push(","); + } + self.text(value); + } + self.push(")") + } + + /// `RequestedTaskScope::applyValue` for one column: `default` matches a + /// null, empty or `default` value, and the default queue connection also + /// matches the default. + fn requested(&mut self, column: &str, value: Option<&str>) -> &mut Self { + let mut values = scope_values(value); + if values.is_empty() { + return self; + } + let default_connection = std::env::var("QUEUE_CONNECTION").unwrap_or_else(|_| "database".to_owned()); + if column == "connection" && values.contains(&default_connection) && !values.iter().any(|value| value == "default") { + values.push("default".to_owned()); + } + let exact: Vec = values.iter().filter(|value| *value != "default").cloned().collect(); + self.push(" AND ("); + if values.len() > exact.len() { + self.push(column).push(" IS NULL OR ").push(column).push("='' OR ").push(column).push("='default'"); + if !exact.is_empty() { + self.push(" OR ").push(column).push(" IN ").list(&exact); + } + } else { + self.push(column).push(" IN ").list(&exact); + } + self.push(")") + } + + /// `applyScope`: an absent scope value matches null, empty or the public + /// default; a present one matches exactly. + fn scope(&mut self, scope: &ScopeRow) -> &mut Self { + let [connection, queue, compatibility] = scope.values(); + for (column, value, default) in [("connection", connection, "default"), ("queue", queue, "default"), ("compatibility", compatibility, "any")] { + match value { + None => { + self.push(" AND (").push(column).push(" IS NULL OR ").push(column).push("='' OR ").push(column); + self.push("='").push(default).push("')"); + } + Some(value) => { + self.push(" AND ").push(column).push("=").text(value); + } + } + } + self + } + + /// `existingTaskQuery`'s repairable tasks: a due ready task whose dispatch + /// failed, whose claim failed, or whose dispatch and availability are + /// overdue, and a leased task whose lease expired. "Now" binds in whole + /// seconds, as Laravel binds a Carbon value; the redispatch cutoff binds + /// with microseconds, as PHP formats it. + fn repairable(&mut self, at: DateTime) -> &mut Self { + let bound = whole_second(at); + let cutoff = at - Duration::seconds(REDISPATCH_AFTER_SECONDS); + self.push("((status='ready' AND (available_at IS NULL OR available_at<=").time(bound).push(") AND (("); + self.push("last_dispatch_attempt_at IS NOT NULL AND last_dispatch_error IS NOT NULL AND last_dispatch_error<>'' "); + self.push("AND (last_dispatched_at IS NULL OR last_dispatch_attempt_at>last_dispatched_at) "); + self.push("AND (repair_available_at IS NULL OR repair_available_at<=").time(bound).push("))"); + self.push(" OR (last_claim_failed_at IS NOT NULL AND last_claim_error IS NOT NULL AND last_claim_error<>'' "); + self.push("AND ((repair_available_at IS NOT NULL AND repair_available_at<=").time(bound); + self.push(") OR (repair_available_at IS NULL AND last_claim_failed_at<=").time(cutoff).push(")))"); + self.push(" OR ((available_at IS NULL OR available_at<=").time(cutoff).push(") "); + self.push("AND (last_dispatch_attempt_at IS NULL OR last_dispatch_error IS NULL OR last_dispatch_error='' "); + self.push("OR (last_dispatched_at IS NOT NULL AND last_dispatch_attempt_at<=last_dispatched_at)) "); + self.push("AND (last_claim_failed_at IS NULL OR last_claim_error IS NULL OR last_claim_error='') "); + self.push("AND (last_dispatched_at<=").time(cutoff).push(" OR (last_dispatched_at IS NULL AND created_at<=").time(cutoff); + self.push("))))) OR (status='leased' AND lease_expires_at IS NOT NULL AND lease_expires_at<=").time(bound).push("))") + } + + /// `existingTaskQuery` with the pass's run, instance and scope filters. + fn repairable_tasks(&mut self, request: &PassRequest, at: DateTime) -> &mut Self { + self.repairable(at); + if !request.run_ids.is_empty() { + self.push(" AND workflow_run_id IN ").list(&request.run_ids); + } + if let Some(instance) = &request.instance_id { + self.push(" AND workflow_run_id IN (SELECT id FROM workflow_runs WHERE workflow_instance_id=").text(instance).push(")"); + } + self.requested("connection", request.connection.as_deref()).requested("queue", request.queue.as_deref()) + } + + /// `missingTaskRunQuery` with the pass's filters. + fn missing_task_runs(&mut self, request: &PassRequest) -> &mut Self { + self.push(super::repair_candidates::MISSING_TASK_RUNS); + if !request.run_ids.is_empty() { + self.push(" AND id IN ").list(&request.run_ids); + } + if let Some(instance) = &request.instance_id { + self.push(" AND workflow_instance_id=").text(instance); + } + self.requested("connection", request.connection.as_deref()).requested("queue", request.queue.as_deref()) + } +} + +/// `roundRobinCandidates`: one id from each scope in turn, up to the limit. +fn round_robin(mut buckets: Vec>, limit: usize) -> Vec { + let mut selected = Vec::new(); + while selected.len() < limit { + let mut progressed = false; + for ids in buckets.iter_mut() { + let Some(id) = ids.pop_front() else { + continue; + }; + selected.push(id); + progressed = true; + if selected.len() >= limit { + break; + } + } + if !progressed { + break; + } + } + selected +} + +/// What one candidate's recovery did. +enum Recovered { + /// A task is ready and was marked dispatched. + Task, + /// The candidate needed nothing, or a closed run's task was settled. + Nothing, + /// Native cannot perform this recovery yet; nothing was changed. + Unported(String), +} + +store_methods! { + async fn rows(&self, sql: Sql) -> Result> { + let mut query = sqlx::query::(AssertSqlSafe(sql.text)); + for bind in sql.binds { + query = match bind { + Bind::Text(text) => query.bind(text), + Bind::Time(time) => query.bind(DB::bind_time(time)), + }; + } + Ok(query.fetch_all(&self.pool).await?) + } + + async fn ids(&self, sql: Sql) -> Result> { + self.rows(sql).await?.iter().map(|row| DB::string(row, "id")).collect() + } + + /// `existingTaskScopeRows`. + pub(super) async fn repairable_task_scopes(&self, request: &PassRequest, at: DateTime) -> Result> { + let mut sql = Sql::new("SELECT connection,queue,compatibility,COUNT(*) AS candidate_count,MIN(created_at) AS oldest_candidate_at FROM workflow_tasks WHERE ", DB::NAME == "MySQL"); + sql.repairable_tasks(request, at).push(" GROUP BY connection,queue,compatibility"); + scope_rows::(self.rows(sql).await?) + } + + /// `missingTaskScopeRows`. + pub(super) async fn missing_task_scopes(&self, request: &PassRequest) -> Result> { + let mut sql = Sql::new("SELECT connection,queue,compatibility,COUNT(*) AS candidate_count,MIN(COALESCE(wait_started_at,started_at,created_at)) AS oldest_candidate_at FROM workflow_run_summaries WHERE ", DB::NAME == "MySQL"); + sql.missing_task_runs(request).push(" GROUP BY connection,queue,compatibility"); + scope_rows::(self.rows(sql).await?) + } + + /// `TaskRepairCandidates::taskIds` in poll mode: each scope's oldest + /// repairable tasks, by availability, creation and id, taken in turn. + async fn repair_task_ids(&self, request: &PassRequest, at: DateTime) -> Result> { + let scopes = self.repairable_task_scopes(request, at).await?; + let mut buckets = Vec::new(); + for scope in normalized(&scopes) { + let mut sql = Sql::new("SELECT id FROM workflow_tasks WHERE ", DB::NAME == "MySQL"); + sql.repairable_tasks(request, at).scope(scope); + sql.push(&format!(" ORDER BY available_at,created_at,id LIMIT {SCAN_LIMIT}")); + let ids = self.ids(sql).await?; + if !ids.is_empty() { + buckets.push(ids.into()); + } + } + Ok(round_robin(buckets, SCAN_LIMIT)) + } + + /// `TaskRepairCandidates::runIds`: each scope's runs that need a task, by + /// wait, start and id, taken in turn. + async fn repair_run_ids(&self, request: &PassRequest) -> Result> { + let scopes = self.missing_task_scopes(request).await?; + let mut buckets = Vec::new(); + for scope in normalized(&scopes) { + let mut sql = Sql::new("SELECT id FROM workflow_run_summaries WHERE ", DB::NAME == "MySQL"); + sql.missing_task_runs(request).scope(scope); + sql.push(&format!(" ORDER BY wait_started_at,started_at,id LIMIT {SCAN_LIMIT}")); + let ids = self.ids(sql).await?; + if !ids.is_empty() { + buckets.push(ids.into()); + } + } + Ok(round_robin(buckets, SCAN_LIMIT)) + } + + /// `deadlineExpiredRunIds`: open runs past their execution or run + /// deadline without an open workflow task, and runs past their + /// cancellation cleanup deadline. + async fn deadline_expired_run_ids(&self, request: &PassRequest, at: DateTime) -> Result> { + let mut sql = Sql::new("SELECT id FROM workflow_runs WHERE status IN ('pending','running','waiting') AND ((execution_deadline_at IS NOT NULL AND execution_deadline_at<=", DB::NAME == "MySQL"); + sql.time(at).push(") OR (run_deadline_at IS NOT NULL AND run_deadline_at<=").time(at); + sql.push(") OR (cancellation_request_command_id IS NOT NULL AND cancellation_deadline_at IS NOT NULL AND cancellation_deadline_at<=").time(at); + sql.push(")) AND (NOT EXISTS (SELECT 1 FROM workflow_tasks t WHERE t.workflow_run_id=workflow_runs.id AND t.task_type='workflow' AND t.status IN ('ready','leased'))"); + sql.push(" OR (cancellation_request_command_id IS NOT NULL AND cancellation_deadline_at IS NOT NULL AND cancellation_deadline_at<=").time(at).push("))"); + if !request.run_ids.is_empty() { + sql.push(" AND id IN ").list(&request.run_ids); + } + if let Some(instance) = &request.instance_id { + sql.push(" AND workflow_instance_id=").text(instance); + } + sql.requested("connection", request.connection.as_deref()).requested("queue", request.queue.as_deref()); + sql.push(&format!(" ORDER BY id LIMIT {SCAN_LIMIT}")); + self.ids(sql).await + } + + /// `ActivityTimeoutEnforcer::expiredExecutionIds` within the requested + /// connections and queues. + async fn scoped_expired_activity_ids(&self, request: &PassRequest, at: DateTime) -> Result> { + let mut sql = Sql::new("SELECT id FROM activity_executions WHERE status IN ('pending','running') AND ((status='pending' AND schedule_deadline_at IS NOT NULL AND schedule_deadline_at<=", DB::NAME == "MySQL"); + sql.time(at).push(") OR (status='running' AND close_deadline_at IS NOT NULL AND close_deadline_at<=").time(at); + sql.push(") OR (schedule_to_close_deadline_at IS NOT NULL AND schedule_to_close_deadline_at<=").time(at); + sql.push(") OR (status='running' AND heartbeat_deadline_at IS NOT NULL AND heartbeat_deadline_at<=").time(at).push("))"); + sql.requested("connection", request.connection.as_deref()).requested("queue", request.queue.as_deref()); + sql.push(&format!(" ORDER BY id LIMIT {SCAN_LIMIT}")); + self.ids(sql).await + } + + /// `TaskWatchdog::runPass`, with its loop throttle shared by this + /// server's namespaces. + pub(crate) async fn repair_pass(&self, request: &PassRequest) -> Result { + { + let mut last = self.repair_pass_at.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let recent = last.is_some_and(|at| at.elapsed().as_secs() < LOOP_THROTTLE_SECONDS); + if request.respect_throttle && recent { + return Ok(Value::Object(empty_report(request, true))); + } + *last = Some(Instant::now()); + } + let mut report = empty_report(request, false); + let at = now(); + let existing = self.repair_task_ids(request, at).await?; + let missing = self.repair_run_ids(request).await?; + report.insert("selected_existing_task_candidates".into(), json!(existing.len())); + report.insert("selected_missing_task_candidates".into(), json!(missing.len())); + report.insert("selected_total_candidates".into(), json!(existing.len() + missing.len())); + for task_id in &existing { + let message = match self.recover_candidate_task(task_id).await { + Ok(Recovered::Task) => { + bump(&mut report, "repaired_existing_tasks"); + bump(&mut report, "dispatched_tasks"); + continue; + } + Ok(Recovered::Nothing) => continue, + Ok(Recovered::Unported(message)) => message, + Err(error) => error.to_string(), + }; + self.record_task_error(task_id, &message).await; + record_failure(&mut report, "existing_task_failures", "candidate_id", task_id, message); + } + for run_id in &missing { + let message = match self.recover_missing_run(run_id).await { + Ok(Recovered::Task) => { + bump(&mut report, "repaired_missing_tasks"); + bump(&mut report, "dispatched_tasks"); + continue; + } + Ok(Recovered::Nothing) => continue, + Ok(Recovered::Unported(message)) => message, + Err(error) => error.to_string(), + }; + record_failure(&mut report, "missing_run_failures", "run_id", run_id, message); + } + let deadlines = self.deadline_expired_run_ids(request, now()).await?; + report.insert("deadline_expired_candidates".into(), json!(deadlines.len())); + for run_id in &deadlines { + match self.deadline_expired_task(run_id).await { + Ok((created, enforced)) => { + if created { + bump(&mut report, "deadline_expired_tasks_created"); + bump(&mut report, "dispatched_tasks"); + } + if enforced { + bump(&mut report, "cancellation_deadlines_enforced"); + } + } + Err(error) => record_failure(&mut report, "deadline_expired_failures", "run_id", run_id, error.to_string()), + } + } + let activities = self.scoped_expired_activity_ids(request, now()).await?; + report.insert("activity_timeout_candidates".into(), json!(activities.len())); + for activity_id in &activities { + match self.enforce_activity_deadline(activity_id).await { + Ok(super::activity_deadlines::Enforcement::Enforced { .. }) => { + // Native always opens the next task: the retry, or the + // workflow task that sees the timeout. + bump(&mut report, "activity_timeouts_enforced"); + bump(&mut report, "dispatched_tasks"); + } + Ok(super::activity_deadlines::Enforcement::Skipped(reason)) => { + record_failure(&mut report, "activity_timeout_failures", "execution_id", activity_id, reason.to_owned()); + } + Err(error) => record_failure(&mut report, "activity_timeout_failures", "execution_id", activity_id, error.to_string()), + } + } + self.wake.notify_waiters(); + Ok(Value::Object(report)) + } + + /// `failedExistingTask`: the failure is kept on the task for operators. + /// Recording it must not stop the pass. + async fn record_task_error(&self, task_id: &str, message: &str) { + let _ = Self::query("UPDATE workflow_tasks SET last_error=$1 WHERE id=$2") + .bind(message) + .bind(task_id) + .execute(&self.pool) + .await; + } + + /// `recoverExistingTask` for one candidate, under its run's lock. + async fn recover_candidate_task(&self, task_id: &str) -> Result { + let mut tx = self.begin().await?; + let Some(snapshot) = Self::query("SELECT workflow_run_id FROM workflow_tasks WHERE id=$1").bind(task_id).fetch_optional(&mut *tx).await? else { + tx.commit().await?; + return Ok(Recovered::Nothing); + }; + let run_id = DB::string(&snapshot, "workflow_run_id")?; + let Some(run) = Self::query("SELECT * FROM workflow_runs WHERE id=$1").bind(&run_id).fetch_optional(&mut *tx).await? else { + tx.commit().await?; + return Ok(Recovered::Nothing); + }; + let Some(task) = Self::query("SELECT * FROM workflow_tasks WHERE id=$1 AND workflow_run_id=$2") + .bind(task_id) + .bind(&run_id) + .fetch_optional(&mut *tx) + .await? + else { + tx.commit().await?; + return Ok(Recovered::Nothing); + }; + let at = now(); + if !super::task_repair::needs_recovery::(&task, at)? { + tx.commit().await?; + return Ok(Recovered::Nothing); + } + let recovered = Self::recover_task_row(&mut tx, &run, &task, at).await?; + if let Recovered::Unported(_) = recovered { + return Ok(recovered); + } + Self::project_run_summary(&mut tx, &run_id).await?; + if let Recovered::Task = recovered { + Self::mark_dispatched(&mut tx, task_id).await?; + } + tx.commit().await?; + Ok(recovered) + } + + /// `TaskRepair::recoverExistingTask` for a task that needs recovery. + async fn recover_task_row(tx: &mut Transaction<'_, DB>, run: &DB::Row, task: &DB::Row, at: DateTime) -> Result { + let task_id = DB::string(task, "id")?; + let task_type = DB::string(task, "task_type")?; + let run_status = DB::string(run, "status")?; + let closed = matches!(run_status.as_str(), "completed" | "failed" | "cancelled" | "terminated"); + let execution = if task_type == "activity" { + match DB::document_row(task, "payload")?["activity_execution_id"].as_str() { + Some(execution_id) => Self::query("SELECT * FROM activity_executions WHERE id=$1") + .bind(execution_id.to_owned()) + .fetch_optional(&mut **tx) + .await?, + None => None, + } + } else { + None + }; + let execution_status = execution.as_ref().map(|row| DB::string(row, "status")).transpose()?; + // Native leaves an activity open after its run closes only when the + // run completed without waiting for it (`ActivityAbandonment::allows`). + let abandoned = closed && matches!(execution_status.as_deref(), Some("pending" | "running")); + if closed && !abandoned { + // `settleTerminalTask`, closing a cancelled or terminated run's + // running attempt. + if matches!(run_status.as_str(), "cancelled" | "terminated") + && let Some(execution) = &execution + { + Self::close_current_attempt(tx, execution, "cancelled", at).await?; + } + let status = if DB::string(task, "status")? == "cancelled" { + "cancelled" + } else if run_status == "failed" { + "failed" + } else { + "completed" + }; + Self::query("UPDATE workflow_tasks SET status=$1,leased_at=NULL,lease_owner=NULL,lease_expires_at=NULL,updated_at=$2 WHERE id=$3") + .bind(status).bind(DB::bind_time(at)).bind(&task_id).execute(&mut **tx).await?; + return Ok(Recovered::Nothing); + } + if abandoned + && let Some(execution) = &execution + && super::activity_deadlines::expired_kind::(execution, at)?.is_some() + { + // Timeout enforcement owns the abandoned activity's outcome. + return Ok(Recovered::Nothing); + } + let repair_count = DB::number(task, "repair_count")? + .checked_add(1) + .filter(|count| *count <= DB::REPAIR_COUNTER_MAX) + .ok_or_else(|| super::refuse(axum::http::StatusCode::CONFLICT, "repair_count_exhausted"))?; + match DB::string(task, "status")?.as_str() { + "ready" => { + Self::query("UPDATE workflow_tasks SET repair_count=$1,repair_available_at=NULL,last_error=NULL,updated_at=$2 WHERE id=$3") + .bind(repair_count).bind(DB::bind_time(at)).bind(&task_id).execute(&mut **tx).await?; + } + "leased" => { + if execution_status.as_deref() == Some("running") { + return Ok(Recovered::Unported( + "Native does not yet recover an expired activity lease; the activity's retry policy is unchanged.".into(), + )); + } + // `closeExpiredActivityAttempt`. + if let Some(execution) = &execution { + Self::close_current_attempt(tx, execution, "expired", at).await?; + } + Self::query("UPDATE workflow_tasks SET status='ready',leased_at=NULL,lease_owner=NULL,lease_expires_at=NULL,repair_count=$1,repair_available_at=NULL,last_error=NULL,updated_at=$2 WHERE id=$3") + .bind(repair_count).bind(DB::bind_time(at)).bind(&task_id).execute(&mut **tx).await?; + } + "failed" if super::task_repair::replay_blocked::(task)? => { + // `recoverFailedReplayTask`. + let mut payload = DB::document_row(task, "payload")?; + if let Some(fields) = payload.as_object_mut() { + fields.retain(|key, _| !key.starts_with("replay_blocked")); + } + Self::query("UPDATE workflow_tasks SET status='ready',payload=$1,leased_at=NULL,lease_owner=NULL,lease_expires_at=NULL,repair_count=$2,repair_available_at=NULL,last_error=NULL,updated_at=$3 WHERE id=$4") + .bind(DB::document(&payload)).bind(repair_count).bind(DB::bind_time(at)).bind(&task_id).execute(&mut **tx).await?; + } + _ => return Ok(Recovered::Nothing), + } + tx.summary_changed(&DB::string(run, "id")?); + Ok(Recovered::Task) + } + + /// `closeActivityAttempt` for an execution's current running attempt. + async fn close_current_attempt(tx: &mut Transaction<'_, DB>, execution: &DB::Row, status: &str, at: DateTime) -> Result<()> { + if let Some(attempt) = DB::optional_string(execution, "current_attempt_id")? { + Self::query("UPDATE activity_attempts SET status=$1,lease_expires_at=NULL,closed_at=COALESCE(closed_at,$2),updated_at=$3 WHERE id=$4 AND status='running'") + .bind(status).bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(attempt).execute(&mut **tx).await?; + } + Ok(()) + } + + /// `TaskDispatcher::markDispatched` in poll mode. + async fn mark_dispatched(tx: &mut Transaction<'_, DB>, task_id: &str) -> Result<()> { + let at = now(); + Self::query("UPDATE workflow_tasks SET last_dispatch_attempt_at=$1,last_dispatched_at=$2,last_dispatch_error=NULL,repair_available_at=NULL WHERE id=$3 AND status='ready'") + .bind(DB::bind_time(at)).bind(DB::bind_time(at)).bind(task_id).execute(&mut **tx).await?; + Ok(()) + } + + /// `recoverMissingTask`: a run whose projected summary still needs repair + /// and names no next task gets its recoverable task back, or a new one. + async fn recover_missing_run(&self, run_id: &str) -> Result { + let mut tx = self.begin().await?; + let Some(run) = Self::query("SELECT * FROM workflow_runs WHERE id=$1").bind(run_id).fetch_optional(&mut *tx).await? else { + tx.commit().await?; + return Ok(Recovered::Nothing); + }; + Self::project_run_summary(&mut tx, run_id).await?; + let Some(summary) = Self::query("SELECT liveness_state,next_task_id,wait_kind,resume_source_kind FROM workflow_run_summaries WHERE id=$1") + .bind(run_id) + .fetch_optional(&mut *tx) + .await? + else { + tx.commit().await?; + return Ok(Recovered::Nothing); + }; + if DB::optional_string(&summary, "liveness_state")?.as_deref() != Some("repair_needed") + || DB::optional_string(&summary, "next_task_id")?.is_some() + { + tx.commit().await?; + return Ok(Recovered::Nothing); + } + let at = now(); + // `recoverableTask`: a replay-blocked task first, then the earliest. + let mut recoverable = Vec::new(); + for task in Self::query("SELECT * FROM workflow_tasks WHERE workflow_run_id=$1 AND status IN ('ready','leased','failed')") + .bind(run_id) + .fetch_all(&mut *tx) + .await? + { + let blocked = super::task_repair::replay_blocked::(&task)?; + if blocked || super::task_repair::needs_recovery::(&task, at)? { + recoverable.push(((!blocked, super::task_repair::recovery_order::(&task)?), task)); + } + } + recoverable.sort_by(|left, right| left.0.cmp(&right.0)); + let (recovered, task_id) = match recoverable.into_iter().next() { + Some((_, task)) => { + let task_id = DB::string(&task, "id")?; + (Self::recover_task_row(&mut tx, &run, &task, at).await?, task_id) + } + None => { + let wait_kind = DB::optional_string(&summary, "wait_kind")?.unwrap_or_default(); + let timer_resume = DB::optional_string(&summary, "resume_source_kind")?.as_deref() == Some("timer"); + if matches!(wait_kind.as_str(), "activity" | "timer" | "update" | "signal" | "child") + || (wait_kind == "condition" && timer_resume) + { + return Ok(Recovered::Unported(format!( + "Native does not yet create a missing task for a run waiting on {wait_kind}." + ))); + } + // `createMissingTask`'s workflow task, with + // `forMissingWorkflowTask`'s empty payload. + let queue = DB::string(&run, "queue")?; + let task_id = Self::create_task_dispatched(&mut tx, run_id, &queue, "workflow", json!({}), false).await?; + Self::query("UPDATE workflow_tasks SET repair_count=1 WHERE id=$1").bind(&task_id).execute(&mut *tx).await?; + (Recovered::Task, task_id) + } + }; + if let Recovered::Unported(_) = recovered { + return Ok(recovered); + } + Self::project_run_summary(&mut tx, run_id).await?; + if let Recovered::Task = recovered { + Self::mark_dispatched(&mut tx, &task_id).await?; + } + tx.commit().await?; + Ok(recovered) + } + + /// `createDeadlineExpiredTask`: an expired cancellation cleanup deadline + /// is enforced under a control task; another expired deadline gets a + /// workflow task, and native closes the run when that task runs. Returns + /// whether a task was created and whether a cleanup deadline was enforced. + async fn deadline_expired_task(&self, run_id: &str) -> Result<(bool, bool)> { + let mut tx = self.begin().await?; + let Some(run) = Self::query("SELECT * FROM workflow_runs WHERE id=$1").bind(run_id).fetch_optional(&mut *tx).await? else { + tx.commit().await?; + return Ok((false, false)); + }; + if !matches!(DB::string(&run, "status")?.as_str(), "pending" | "running" | "waiting") { + tx.commit().await?; + return Ok((false, false)); + } + let open = Self::query("SELECT * FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status IN ('ready','leased') ORDER BY id LIMIT 1") + .bind(run_id) + .fetch_optional(&mut *tx) + .await?; + let at = now(); + let expired = |field: &str| -> Result { Ok(DB::optional_instant(&run, field)?.is_some_and(|deadline| at >= deadline)) }; + let cleanup = DB::optional_string(&run, "cancellation_request_command_id")?.is_some() && expired("cancellation_deadline_at")?; + if !(expired("execution_deadline_at")? || expired("run_deadline_at")? || cleanup) { + tx.commit().await?; + return Ok((false, false)); + } + let queue = DB::string(&run, "queue")?; + if cleanup { + // Expiry is runtime authority; an absent worker must not keep the + // run open. + let control = match open { + Some(task) => task, + None => { + let payload = json!({"reason": "cleanup_deadline_expired"}); + let task_id = Self::create_task_dispatched(&mut tx, run_id, &queue, "workflow", payload, false).await?; + Self::query("UPDATE workflow_tasks SET repair_count=1 WHERE id=$1").bind(&task_id).execute(&mut *tx).await?; + Self::query("SELECT * FROM workflow_tasks WHERE id=$1").bind(&task_id).fetch_one(&mut *tx).await? + } + }; + Self::finish_root_cancellation(&mut tx, &run, Some(&control)).await?; + tx.summary_changed(run_id); + tx.commit().await?; + return Ok((false, true)); + } + if open.is_some() { + tx.commit().await?; + return Ok((false, false)); + } + let task_id = Self::create_task_dispatched(&mut tx, run_id, &queue, "workflow", json!({"reason": "deadline_expired"}), false).await?; + Self::query("UPDATE workflow_tasks SET repair_count=1 WHERE id=$1").bind(&task_id).execute(&mut *tx).await?; + Self::project_run_summary(&mut tx, run_id).await?; + Self::mark_dispatched(&mut tx, &task_id).await?; + tx.commit().await?; + Ok((true, false)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_robin_takes_one_from_each_scope_in_turn() { + let bucket = |ids: &[&str]| ids.iter().map(|id| (*id).to_owned()).collect::>(); + let buckets = vec![bucket(&["a1", "a2", "a3"]), bucket(&["b1"]), bucket(&["c1", "c2"])]; + assert_eq!(round_robin(buckets, 5), ["a1", "b1", "c1", "a2", "c2"]); + } + + #[test] + fn requested_scopes_match_php() { + let mut sql = Sql::new("", false); + sql.requested("queue", Some(" a , default, a ,b")); + assert_eq!(sql.text, " AND (queue IS NULL OR queue='' OR queue='default' OR queue IN ($1,$2))"); + let mut sql = Sql::new("", true); + sql.requested("connection", Some("redis")); + assert_eq!(sql.text, " AND (connection IN (?))"); + let mut sql = Sql::new("", false); + sql.requested("connection", Some(" , ")); + assert_eq!(sql.text, ""); + } + + #[test] + fn request_rules_and_trimming() { + let request = pass_request(&json!({"run_ids": [" r "], "queue": "", "respect_throttle": "1"})).unwrap(); + assert_eq!((request.run_ids, request.queue, request.respect_throttle), (vec!["r".to_owned()], None, true)); + assert!(pass_request(&json!({"run_ids": "x"})).is_err()); + assert!(pass_request(&json!({"run_ids": [""]})).is_err()); + assert!(pass_request(&json!({"respect_throttle": "yes"})).is_err()); + assert!(pass_request(&Value::Null).is_ok()); + } +} diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 4c3b50966..678154c08 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -39,6 +39,9 @@ pub(super) struct Store { pub(super) search_attribute_limit: i64, /// When a silent registration stops counting (`server.workers.stale_after_seconds`). pub(super) worker_stale_after_seconds: i64, + /// When this server last ran a task repair pass, shared by its namespaces + /// as PHP shares its loop throttle key. + pub(super) repair_pass_at: Arc>>, } /// The `closed_reason` a terminal history event gives its run, or `None` for /// an event that does not close the run. @@ -118,6 +121,7 @@ where update_wait_seconds: 10, search_attribute_limit: 100, worker_stale_after_seconds: 30, + repair_pass_at: Arc::default(), } } pub(super) fn in_namespace(&self, namespace: &str) -> Self { @@ -133,6 +137,7 @@ where update_wait_seconds: self.update_wait_seconds, search_attribute_limit: self.search_attribute_limit, worker_stale_after_seconds: self.worker_stale_after_seconds, + repair_pass_at: self.repair_pass_at.clone(), } } pub(super) async fn database_live(&self) -> bool { diff --git a/rust/src/runtime/task_repair.rs b/rust/src/runtime/task_repair.rs index d82373fab..90403d69f 100644 --- a/rust/src/runtime/task_repair.rs +++ b/rust/src/runtime/task_repair.rs @@ -18,7 +18,7 @@ use tokio::sync::Semaphore; /// `TaskRepairPolicy::readyTaskNeedsRedispatch` or `leaseExpired` for one /// task row, as of `at`. -fn needs_recovery(task: &DB::Row, at: DateTime) -> Result +pub(super) fn needs_recovery(task: &DB::Row, at: DateTime) -> Result where usize: ColumnIndex, for<'r> i64: Decode<'r, DB>, @@ -50,13 +50,29 @@ where None => claim_failed_at.is_some_and(|failed| failed <= cutoff), }); } + // `dispatchOverdue`: a task made available after its last dispatch is + // overdue from its availability. let available = DB::optional_instant(task, "available_at")?; - let reference = dispatched.or(DB::optional_instant(task, "created_at")?); + let reference = dispatched + .or(DB::optional_instant(task, "created_at")?) + .map(|reference| available.filter(|available| *available > reference).unwrap_or(reference)); Ok(available.is_none_or(|available| available <= at) && reference.is_some_and(|reference| reference <= cutoff)) } +/// `TaskRepair::replayBlocked`: a failed workflow task whose payload marks +/// its replay blocked. +pub(super) fn replay_blocked(task: &DB::Row) -> Result +where + usize: ColumnIndex, + for<'r> i64: Decode<'r, DB>, +{ + Ok(DB::string(task, "task_type")? == "workflow" + && DB::string(task, "status")? == "failed" + && DB::optional_document_row(task, "payload")?.is_some_and(|payload| payload["replay_blocked"] == true)) +} + /// `recoverableTask`'s order: earliest available, then created, then id. -fn recovery_order(task: &DB::Row) -> Result<(i64, i64, String)> +pub(super) fn recovery_order(task: &DB::Row) -> Result<(i64, i64, String)> where usize: ColumnIndex, for<'r> i64: Decode<'r, DB>, diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 631096079..6add46d17 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -158,6 +158,8 @@ mod workflow_repair; mod history_retention; #[path = "execution/repair_status.rs"] mod repair_status; +#[path = "execution/repair_pass.rs"] +mod repair_pass; #[path = "execution/activity_timeout_pass.rs"] mod activity_timeout_pass; #[path = "execution/start_request.rs"] diff --git a/rust/tests/execution/repair_pass.rs b/rust/tests/execution/repair_pass.rs new file mode 100644 index 000000000..83697efa4 --- /dev/null +++ b/rust/tests/execution/repair_pass.rs @@ -0,0 +1,151 @@ +use super::*; + +const PASS: &str = "/api/system/repair/pass"; +const LONG_AGO: &str = "'2020-01-02 03:04:05.000000'"; + +async fn started_on(app: &Router, workflow: &str, queue: &str) -> Value { + let (status, body) = request(app, "POST", "/api/workflows", json!({"workflow_id":workflow,"workflow_type":"echo","task_queue":queue,"input":[]})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + body +} + +/// Counts rows with a query built for this test. +async fn count(database: &TestDatabase, sql: String) -> i64 { + task_retry::rows(database, Box::leak(sql.into_boxed_str())).await +} + +async fn execute(database: &TestDatabase, sql: String) { + database.execute(sqlx::AssertSqlSafe(sql)).await; +} + +fn counts(report: &Value, fields: &[&str]) -> Vec { + fields.iter().map(|field| report[*field].clone()).collect() +} + +// Published PHP's repair pass redispatches overdue ready tasks, returns tasks +// whose lease expired, and creates a lost workflow task. Its filters and loop +// throttle shape what one pass selects. +#[tokio::test] +async fn repair_pass_recovers_tasks_and_runs_like_php() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let (status, empty) = request(&app, "POST", PASS, Value::Null).await; + assert_eq!(status, StatusCode::OK, "{empty}"); + assert_eq!( + counts(&empty, &["throttled", "selected_total_candidates", "dispatched_tasks", "service_call_timeout_candidates"]), + [json!(false), json!(0), json!(0), json!(0)] + ); + assert_eq!(empty["command_contract_failures"], json!([])); + + let overdue = started_on(&app, "overdue", "repair-a").await; + let leased = started_on(&app, "leased", "repair-b").await; + let stalled = started_on(&app, "stalled", "repair-c").await; + let (overdue, leased, stalled) = + (overdue["run_id"].as_str().unwrap(), leased["run_id"].as_str().unwrap(), stalled["run_id"].as_str().unwrap()); + // Each task was created and dispatched long before the redispatch cutoff. + execute(&database, format!("UPDATE workflow_tasks SET created_at={LONG_AGO},available_at={LONG_AGO},last_dispatched_at={LONG_AGO}")).await; + execute(&database, format!("UPDATE workflow_tasks SET status='leased',lease_owner='gone',lease_expires_at={LONG_AGO} WHERE workflow_run_id='{leased}'")).await; + // The stalled run lost its task. + execute(&database, format!("DELETE FROM workflow_tasks WHERE workflow_run_id='{stalled}'")).await; + execute(&database, format!("UPDATE workflow_run_summaries SET liveness_state='repair_needed',next_task_id=NULL WHERE id='{stalled}'")).await; + + // A pass scoped to one queue selects only its candidates. + let (status, scoped) = request(&app, "POST", PASS, json!({"queue": " repair-a "})).await; + assert_eq!(status, StatusCode::OK, "{scoped}"); + assert_eq!(scoped["queue"], "repair-a"); + assert_eq!( + counts(&scoped, &["selected_existing_task_candidates", "selected_missing_task_candidates", "repaired_existing_tasks", "dispatched_tasks"]), + [json!(1), json!(0), json!(1), json!(1)], + "{scoped}" + ); + assert_eq!( + count(&database, format!("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id='{overdue}' AND status='ready' AND repair_count=1 AND last_dispatched_at>'2021-01-01 00:00:00'")).await, + 1 + ); + + // A pass that respects the loop throttle runs nothing right after another. + let (status, throttled) = request(&app, "POST", PASS, json!({"respect_throttle": true})).await; + assert_eq!(status, StatusCode::OK, "{throttled}"); + assert_eq!(counts(&throttled, &["throttled", "selected_total_candidates"]), [json!(true), json!(0)]); + + // An unthrottled pass returns the expired lease and creates the lost task. + let (status, full) = request(&app, "POST", PASS, json!({"respect_throttle": "0"})).await; + assert_eq!(status, StatusCode::OK, "{full}"); + assert_eq!( + counts(&full, &["throttled", "selected_existing_task_candidates", "selected_missing_task_candidates", "repaired_existing_tasks", "repaired_missing_tasks", "dispatched_tasks"]), + [json!(false), json!(1), json!(1), json!(1), json!(1), json!(2)], + "{full}" + ); + assert_eq!( + count(&database, format!("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id='{leased}' AND status='ready' AND lease_owner IS NULL AND lease_expires_at IS NULL AND repair_count=1")).await, + 1 + ); + assert_eq!( + count(&database, format!("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id='{stalled}' AND task_type='workflow' AND status='ready' AND repair_count=1 AND last_dispatched_at IS NOT NULL")).await, + 1 + ); + assert_eq!( + count(&database, format!("SELECT COUNT(*) FROM workflow_run_summaries WHERE id='{stalled}' AND next_task_id IS NOT NULL")).await, + 1 + ); + + // Nothing is left to repair. + let (status, quiet) = request(&app, "POST", PASS, json!({})).await; + assert_eq!(status, StatusCode::OK, "{quiet}"); + assert_eq!(quiet["selected_total_candidates"], 0, "{quiet}"); +} + +// A run past its deadline gets the task that closes it. A recovery native +// does not perform yet is reported as the candidate's failure, with 207, +// and kept on the task as its last error. +#[tokio::test] +async fn repair_pass_creates_deadline_tasks_and_reports_unported_recoveries() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!(["act"])).await; + + let deadline = started_on(&app, "deadline", "repair-d").await; + let deadline = deadline["run_id"].as_str().unwrap(); + execute(&database, format!("UPDATE workflow_tasks SET status='completed' WHERE workflow_run_id='{deadline}'")).await; + execute(&database, format!("UPDATE workflow_runs SET execution_deadline_at={LONG_AGO} WHERE id='{deadline}'")).await; + + // A running activity whose lease expired: PHP consumes its retry policy. + let started = start(&app, "activity").await; + let task = poll(&app, "worker", "workflow").await; + let command = json!({"type":"schedule_activity","activity_type":"act","arguments":envelope(Payload::Array(vec![]))}); + assert_eq!(finish_task(&app, &task, json!([command])).await.0, StatusCode::OK); + let activity = poll(&app, "worker", "activity").await; + assert!(activity["task_id"].is_string(), "{activity}"); + let run = started["run_id"].as_str().unwrap(); + execute(&database, format!("UPDATE workflow_tasks SET lease_expires_at={LONG_AGO} WHERE workflow_run_id='{run}' AND task_type='activity'")).await; + + let (status, report) = request(&app, "POST", PASS, json!({})).await; + assert_eq!(status, StatusCode::MULTI_STATUS, "{report}"); + assert_eq!( + counts(&report, &["deadline_expired_candidates", "deadline_expired_tasks_created", "selected_existing_task_candidates", "repaired_existing_tasks", "dispatched_tasks"]), + [json!(1), json!(1), json!(1), json!(0), json!(1)], + "{report}" + ); + let failures = report["existing_task_failures"].as_array().unwrap(); + assert_eq!(failures.len(), 1, "{report}"); + assert_eq!(failures[0]["candidate_id"], activity["task_id"]); + assert!(failures[0]["message"].as_str().unwrap().starts_with("Native does not yet recover an expired activity lease"), "{report}"); + let task_id = activity["task_id"].as_str().unwrap(); + assert_eq!( + count(&database, format!("SELECT COUNT(*) FROM workflow_tasks WHERE id='{task_id}' AND status='leased' AND repair_count=0 AND last_error LIKE 'Native does not yet%'")).await, + 1 + ); + assert_eq!( + count(&database, format!("SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id='{deadline}' AND task_type='workflow' AND status='ready' AND repair_count=1")).await, + 1 + ); + + // The deadline task is open now, so the run is no longer a candidate. + let (_, again) = request(&app, "POST", PASS, json!({"run_ids": [deadline]})).await; + assert_eq!(again["deadline_expired_candidates"], 0, "{again}"); + + let (status, invalid) = request(&app, "POST", PASS, json!({"run_ids": [""], "queue": "q".repeat(129)})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{invalid}"); + assert_eq!(invalid["errors"]["run_ids.0"], json!(["The run_ids.0 field must be a string.", "The run_ids.0 field must be at least 1 characters."])); + assert_eq!(invalid["errors"]["queue"], json!(["The queue field must not be greater than 128 characters."])); +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 08932d03e..85f39f1ac 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -17,6 +17,7 @@ import {checkServiceCatalog} from './service-catalog-contract.mjs'; import {checkSystemMetrics} from './system-metrics-contract.mjs'; 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 {checkSchedulePolicies} from './schedule-policies-contract.mjs'; import {checkScheduleCatalog} from './schedule-catalog-contract.mjs'; @@ -152,6 +153,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const systemMetrics = fixture.system_metrics ? checkSystemMetrics(fixture, observation, workflowId) : null; const prometheusMetrics = fixture.prometheus_metrics ? checkPrometheusMetrics(fixture, observation, workflowId) : null; 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 schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null; const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId) : null; @@ -704,6 +706,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(systemMetrics ? {system_metrics: systemMetrics} : {}), ...(prometheusMetrics ? {prometheus_metrics: prometheusMetrics} : {}), ...(payloadCleanup ? {payload_cleanup: payloadCleanup} : {}), + ...(repairPass ? {repair_pass: repairPass} : {}), ...(namespaceDelete ? {namespace_delete: namespaceDelete} : {}), ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}), ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}), diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index c5d205f5b..b6339da20 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -435,6 +435,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $systemMetrics = isset($fixture['system_metrics']) ? finishHttpSystemMetrics($client, $fixture, $workflowId, $url) : null; $prometheusMetrics = isset($fixture['prometheus_metrics']) ? finishHttpPrometheusMetrics($client, $fixture, $workflowId, $url) : null; $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; $schedulePolicies = isset($fixture['schedule_policies']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url) : null; $scheduleCatalog = isset($fixture['schedule_catalog']) @@ -540,6 +541,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'system_metrics' => $systemMetrics, 'prometheus_metrics' => $prometheusMetrics, 'payload_cleanup' => $payloadCleanup, + 'repair_pass' => $repairPass, 'namespace_delete' => $namespaceDelete, 'schedule_policies' => $schedulePolicies, 'schedule_catalog' => $scheduleCatalog, @@ -849,6 +851,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/system-metrics-probe.php'; require __DIR__.'/prometheus-metrics-probe.php'; require __DIR__.'/payload-cleanup-probe.php'; + require __DIR__.'/repair-pass-probe.php'; require __DIR__.'/namespace-delete-probe.php'; require __DIR__.'/schedule-policies-probe.php'; require __DIR__.'/signals-before-worker-probe.php'; @@ -1023,6 +1026,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['payload_cleanup'])) { $observation['payload_cleanup'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_system_routes']; } + if ($mode === 'embedded' && isset($fixture['repair_pass'])) { + $observation['repair_pass'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_system_routes']; + } if ($mode === 'embedded' && isset($fixture['namespace_delete'])) { $observation['namespace_delete'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_namespaces']; } diff --git a/scripts/conformance/server-parity/repair-pass-contract.mjs b/scripts/conformance/server-parity/repair-pass-contract.mjs new file mode 100644 index 000000000..cbf6348d0 --- /dev/null +++ b/scripts/conformance/server-parity/repair-pass-contract.mjs @@ -0,0 +1,28 @@ +import assert from 'node:assert/strict'; + +// The fixture's requests and answers with its namespace in place of "@own". +export function repairPassCases(fixture, namespace) { + const own = value => JSON.parse(JSON.stringify(value).split('@own').join(namespace)); + const definition = fixture.repair_pass; + return definition.requests.map(([label, body]) => [label, own(body), own(definition.expected[label])]); +} + +// Reviewed repair pass answers (SystemController::repairPass), exact. +export function checkRepairPass(fixture, observation, workflowId) { + const state = observation.repair_pass; + const projection = {scope: 'repair_pass', cases: fixture.repair_pass.requests.length}; + assert.ok(state && typeof state === 'object', 'actual repair pass observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_system_routes'}, 'embedded repair pass explicitly inapplicable'); + return projection; + } + const namespace = `${workflowId}-repair`.toLowerCase(); + const cases = repairPassCases(fixture, namespace); + assert.deepStrictEqual([state.applicable, state.namespace], [true, namespace], 'repair namespace'); + assert.deepStrictEqual(Object.keys(state.responses), cases.map(([label]) => label), 'every request recorded in order'); + for (const [label, , [status, body]] of cases) { + const recorded = state.responses[label]; + assert.deepStrictEqual([recorded.status, recorded.response], [status, body], label); + } + return projection; +} diff --git a/scripts/conformance/server-parity/repair-pass-probe.php b/scripts/conformance/server-parity/repair-pass-probe.php new file mode 100644 index 000000000..6d894369e --- /dev/null +++ b/scripts/conformance/server-parity/repair-pass-probe.php @@ -0,0 +1,36 @@ +createNamespace($namespace, 'repair pass λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); + $http = new RepairPassHttp(['base_uri' => rtrim($url, '/').'/api/system/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); + $own = static fn (mixed $value): mixed => json_decode( + str_replace('@own', $namespace, json_encode($value, JSON_THROW_ON_ERROR)), true, flags: JSON_THROW_ON_ERROR); + $state = ['applicable' => true, 'namespace' => $namespace, 'responses' => []]; + foreach ($fixture['repair_pass']['requests'] as [$label, $body]) { + $response = $http->request('POST', 'repair/pass', ['headers' => array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', + ]), 'json' => $own($body)]); + $state['responses'][$label] = ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + } + + return $state; +} diff --git a/tests/Fixtures/ServerParity/repair-pass.json b/tests/Fixtures/ServerParity/repair-pass.json new file mode 100644 index 000000000..8bc98874c --- /dev/null +++ b/tests/Fixtures/ServerParity/repair-pass.json @@ -0,0 +1,417 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "repair-pass", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "repair pass λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "repair pass λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "repair_pass": { + "requests": [ + [ + "filtered pass", + { + "run_ids": [ + "@own-missing" + ], + "queue": "@own-queue" + } + ], + [ + "trimmed filters", + { + "run_ids": [ + " @own-missing ", + "@own-other" + ], + "instance_id": " @own-instance ", + "connection": " @own-connection ", + "queue": " @own-queue " + } + ], + [ + "empty instance", + { + "run_ids": [ + "@own-missing" + ], + "queue": "@own-queue", + "instance_id": "", + "connection": "" + } + ], + [ + "run ids not a list", + { + "run_ids": "x", + "queue": "@own-queue" + } + ], + [ + "empty run id", + { + "run_ids": [ + "" + ], + "queue": "@own-queue" + } + ], + [ + "too many run ids", + { + "run_ids": [ + "@own-r0", + "@own-r1", + "@own-r2", + "@own-r3", + "@own-r4", + "@own-r5", + "@own-r6", + "@own-r7", + "@own-r8", + "@own-r9", + "@own-r10", + "@own-r11", + "@own-r12", + "@own-r13", + "@own-r14", + "@own-r15", + "@own-r16", + "@own-r17", + "@own-r18", + "@own-r19", + "@own-r20", + "@own-r21", + "@own-r22", + "@own-r23", + "@own-r24", + "@own-r25", + "@own-r26", + "@own-r27", + "@own-r28", + "@own-r29", + "@own-r30", + "@own-r31", + "@own-r32", + "@own-r33", + "@own-r34", + "@own-r35", + "@own-r36", + "@own-r37", + "@own-r38", + "@own-r39", + "@own-r40", + "@own-r41", + "@own-r42", + "@own-r43", + "@own-r44", + "@own-r45", + "@own-r46", + "@own-r47", + "@own-r48", + "@own-r49", + "@own-r50", + "@own-r51", + "@own-r52", + "@own-r53", + "@own-r54", + "@own-r55", + "@own-r56", + "@own-r57", + "@own-r58", + "@own-r59", + "@own-r60", + "@own-r61", + "@own-r62", + "@own-r63", + "@own-r64", + "@own-r65", + "@own-r66", + "@own-r67", + "@own-r68", + "@own-r69", + "@own-r70", + "@own-r71", + "@own-r72", + "@own-r73", + "@own-r74", + "@own-r75", + "@own-r76", + "@own-r77", + "@own-r78", + "@own-r79", + "@own-r80", + "@own-r81", + "@own-r82", + "@own-r83", + "@own-r84", + "@own-r85", + "@own-r86", + "@own-r87", + "@own-r88", + "@own-r89", + "@own-r90", + "@own-r91", + "@own-r92", + "@own-r93", + "@own-r94", + "@own-r95", + "@own-r96", + "@own-r97", + "@own-r98", + "@own-r99", + "@own-r100" + ], + "queue": "@own-queue" + } + ], + [ + "throttle not boolean", + { + "run_ids": [ + "@own-missing" + ], + "queue": "@own-queue", + "respect_throttle": "yes" + } + ], + [ + "queue too long", + { + "run_ids": [ + "@own-missing" + ], + "queue": "qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq" + } + ], + [ + "connection too long", + { + "run_ids": [ + "@own-missing" + ], + "queue": "@own-queue", + "connection": "ccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" + } + ] + ], + "expected": { + "filtered pass": [ + 200, + { + "connection": null, + "queue": "@own-queue", + "run_ids": [ + "@own-missing" + ], + "instance_id": null, + "respect_throttle": false, + "selected_command_contract_candidates": 0, + "backfilled_command_contracts": 0, + "command_contract_backfill_unavailable": 0, + "command_contract_failures": [], + "existing_task_failures": [], + "missing_run_failures": [], + "throttled": false, + "selected_existing_task_candidates": 0, + "selected_missing_task_candidates": 0, + "selected_total_candidates": 0, + "repaired_existing_tasks": 0, + "repaired_missing_tasks": 0, + "dispatched_tasks": 0, + "deadline_expired_candidates": 0, + "deadline_expired_tasks_created": 0, + "cancellation_deadlines_enforced": 0, + "deadline_expired_failures": [], + "activity_timeout_candidates": 0, + "activity_timeouts_enforced": 0, + "activity_timeout_failures": [], + "service_call_timeout_candidates": 0, + "service_call_timeouts_enforced": 0, + "service_call_timeout_failures": [] + } + ], + "trimmed filters": [ + 200, + { + "connection": "@own-connection", + "queue": "@own-queue", + "run_ids": [ + "@own-missing", + "@own-other" + ], + "instance_id": "@own-instance", + "respect_throttle": false, + "selected_command_contract_candidates": 0, + "backfilled_command_contracts": 0, + "command_contract_backfill_unavailable": 0, + "command_contract_failures": [], + "existing_task_failures": [], + "missing_run_failures": [], + "throttled": false, + "selected_existing_task_candidates": 0, + "selected_missing_task_candidates": 0, + "selected_total_candidates": 0, + "repaired_existing_tasks": 0, + "repaired_missing_tasks": 0, + "dispatched_tasks": 0, + "deadline_expired_candidates": 0, + "deadline_expired_tasks_created": 0, + "cancellation_deadlines_enforced": 0, + "deadline_expired_failures": [], + "activity_timeout_candidates": 0, + "activity_timeouts_enforced": 0, + "activity_timeout_failures": [], + "service_call_timeout_candidates": 0, + "service_call_timeouts_enforced": 0, + "service_call_timeout_failures": [] + } + ], + "empty instance": [ + 200, + { + "connection": null, + "queue": "@own-queue", + "run_ids": [ + "@own-missing" + ], + "instance_id": null, + "respect_throttle": false, + "selected_command_contract_candidates": 0, + "backfilled_command_contracts": 0, + "command_contract_backfill_unavailable": 0, + "command_contract_failures": [], + "existing_task_failures": [], + "missing_run_failures": [], + "throttled": false, + "selected_existing_task_candidates": 0, + "selected_missing_task_candidates": 0, + "selected_total_candidates": 0, + "repaired_existing_tasks": 0, + "repaired_missing_tasks": 0, + "dispatched_tasks": 0, + "deadline_expired_candidates": 0, + "deadline_expired_tasks_created": 0, + "cancellation_deadlines_enforced": 0, + "deadline_expired_failures": [], + "activity_timeout_candidates": 0, + "activity_timeouts_enforced": 0, + "activity_timeout_failures": [], + "service_call_timeout_candidates": 0, + "service_call_timeouts_enforced": 0, + "service_call_timeout_failures": [] + } + ], + "run ids not a list": [ + 422, + { + "message": "The run ids field must be an array.", + "reason": "validation_failed", + "errors": { + "run_ids": [ + "The run ids field must be an array." + ] + }, + "validation_errors": { + "run_ids": [ + "The run ids field must be an array." + ] + } + } + ], + "empty run id": [ + 422, + { + "message": "The run_ids.0 field must be a string. (and 1 more error)", + "reason": "validation_failed", + "errors": { + "run_ids.0": [ + "The run_ids.0 field must be a string.", + "The run_ids.0 field must be at least 1 characters." + ] + }, + "validation_errors": { + "run_ids.0": [ + "The run_ids.0 field must be a string.", + "The run_ids.0 field must be at least 1 characters." + ] + } + } + ], + "too many run ids": [ + 422, + { + "message": "The run ids field must not have more than 100 items.", + "reason": "validation_failed", + "errors": { + "run_ids": [ + "The run ids field must not have more than 100 items." + ] + }, + "validation_errors": { + "run_ids": [ + "The run ids field must not have more than 100 items." + ] + } + } + ], + "throttle not boolean": [ + 422, + { + "message": "The respect throttle field must be true or false.", + "reason": "validation_failed", + "errors": { + "respect_throttle": [ + "The respect throttle field must be true or false." + ] + }, + "validation_errors": { + "respect_throttle": [ + "The respect throttle field must be true or false." + ] + } + } + ], + "queue too long": [ + 422, + { + "message": "The queue field must not be greater than 128 characters.", + "reason": "validation_failed", + "errors": { + "queue": [ + "The queue field must not be greater than 128 characters." + ] + }, + "validation_errors": { + "queue": [ + "The queue field must not be greater than 128 characters." + ] + } + } + ], + "connection too long": [ + 422, + { + "message": "The connection field must not be greater than 128 characters.", + "reason": "validation_failed", + "errors": { + "connection": [ + "The connection field must not be greater than 128 characters." + ] + }, + "validation_errors": { + "connection": [ + "The connection field must not be greater than 128 characters." + ] + } + } + ] + } + }, + "contract": "POST /api/system/repair/pass (published Server 2.11) runs one task repair pass: it recovers ready tasks that need redispatch and expired leases, creates missing workflow tasks, enforces expired run, activity and service-call deadlines, and reports each count with its failures, answering 207 when any failed. The pass is server-wide, so the fixture scopes every pass to run ids and a task queue that do not exist; each then reports zero candidates and echoes its trimmed filters, with empty strings read as null. Run ids are at most 100 non-empty strings; respect_throttle is a boolean; connection and queue are at most 128 characters. Invalid requests answer Laravel's 422. Embedded applications have no HTTP system routes." +} diff --git a/tests/Unit/ServerParityRepairPassContractTest.mjs b/tests/Unit/ServerParityRepairPassContractTest.mjs new file mode 100644 index 000000000..5fbb73f75 --- /dev/null +++ b/tests/Unit/ServerParityRepairPassContractTest.mjs @@ -0,0 +1,68 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import {checkRepairPass, repairPassCases} from '../../scripts/conformance/server-parity/repair-pass-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/repair-pass.json', import.meta.url))); +const definition = fixture.repair_pass; +const id = 'parity-v1-repair-pass'; +const own = `${id}-repair`; + +const observation = () => { + const responses = {}; + for (const [label, , [status, response]] of repairPassCases(fixture, own)) { + responses[label] = {status, response: structuredClone(response)}; + } + return {mode: 'http', repair_pass: {applicable: true, namespace: own, responses}}; +}; + +test('the repair pass accepts the published answers', () => { + assert.deepStrictEqual(checkRepairPass(fixture, observation(), id), {scope: 'repair_pass', cases: 9}); +}); + +test('the fixture scopes every pass away from other work', () => { + for (const [label, body] of definition.requests) { + assert.ok(Array.isArray(body.run_ids) || typeof body.run_ids === 'string', `${label} names run ids`); + assert.ok(typeof body.queue !== 'string' || body.queue.includes('@own') || body.queue.length > 128, `${label} names its own queue`); + } + const passes = Object.values(definition.expected).filter(([status]) => status === 200); + assert.equal(passes.length, 3); + for (const [, report] of passes) { + assert.equal(report.selected_total_candidates, 0); + assert.equal(report.throttled, false); + } +}); + +test('the placeholders become the probe namespace', () => { + const [, body, [, report]] = repairPassCases(fixture, own).find(([label]) => label === 'trimmed filters'); + assert.equal(body.instance_id, ` ${own}-instance `); + assert.equal(report.instance_id, `${own}-instance`); + assert.ok(!JSON.stringify(repairPassCases(fixture, own)).includes('@own')); +}); + +test('the embedded repair pass is explicitly inapplicable', () => { + checkRepairPass(fixture, {mode: 'embedded', repair_pass: {applicable: false, reason: 'embedded_has_no_http_system_routes'}}, id); + assert.throws(() => checkRepairPass(fixture, {mode: 'embedded', repair_pass: {applicable: true}}, id)); +}); + +const report = (state, label) => state.responses[label].response; +for (const [label, mutate] of [ + ['an unimplemented route', state => { state.responses['filtered pass'] = {status: 501, response: {reason: 'rust_capability_not_implemented'}}; }], + ['a missing request', state => { delete state.responses['queue too long']; }], + ['another namespace', state => { state.namespace = 'default'; }], + ['untrimmed filters', state => { report(state, 'trimmed filters').queue = ` ${own}-queue `; }], + ['an empty instance kept as text', state => { report(state, 'empty instance').instance_id = ''; }], + ['a throttled pass', state => { report(state, 'filtered pass').throttled = true; }], + ['a selected candidate', state => { report(state, 'filtered pass').selected_total_candidates = 1; }], + ['a missing service call count', state => { delete report(state, 'filtered pass').service_call_timeout_candidates; }], + ['a partial pass', state => { state.responses['filtered pass'].status = 207; }], + ['an accepted text throttle', state => { state.responses['throttle not boolean'] = {status: 200, response: {}}; }], + ['another run id message', state => { report(state, 'empty run id').errors['run_ids.0'] = ['The run_ids.0 field must be a string.']; }], + ['an accepted run id list over 100', state => { state.responses['too many run ids'].status = 200; }], +]) { + test(`the repair pass rejects ${label}`, () => { + const value = observation(); + mutate(value.repair_pass); + assert.throws(() => checkRepairPass(fixture, value, id)); + }); +}