Serve the task repair pass like published PHP 2.11

.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

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.

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.

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<Value>);
     delegate!(repair_status() -> Result<Value>);
+    delegate!(repair_pass(request: &super::repair_pass::PassRequest) -> Result<Value>);
     delegate!(system_metrics() -> Result<Value>);
     delegate!(prometheus_metrics() -> Result<Value>);
     delegate!(payload_cleanup_status(config: &super::operator_metrics::MetricsConfig) -> Result<Value>);

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<Runtime>) -> Result<Json<Va
     Ok(Json(runtime.repair_status().await?))
 }
 
+/// `SystemController::repairPass`: 207 when a candidate failed.
+async fn repair_pass(Extension(runtime): Extension<Runtime>, body: axum::body::Bytes) -> Result<Response> {
+    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<Runtime>, Query(query): QueryMap) -> Result<Json<Value>> {
     let limit = super::retention::status_limit(query.get("limit").map(String::as_str))?;
     Ok(Json(runtime.retention_status(limit).await?))

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;

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<Value> {
         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::<DB>(tasks.fetch_all(&self.pool).await?)?;
-        let runs = scope_rows::<DB>(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<Utc>, 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<DB: Backend>(rows: Vec<DB::Row>) -> Result<Vec<ScopeRow>>
+pub(super) fn scope_rows<DB: Backend>(rows: Vec<DB::Row>) -> Result<Vec<ScopeRow>>
 where
     usize: ColumnIndex<DB::Row>,
     for<'r> i64: Decode<'r, DB>,

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