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)); + }); +}