Rust server: prune expired run history like published PHP

docs/rust-server-port-checkpoints/2026-10-21-history-retention.md

@@ -0,0 +1,71 @@
+# October 21, history retention
+
+- **Found by.** The published dw 2.3.0 CLI sweep. `system:retention-status`,
+  `system:retention-pass` and `system:activity-timeout-status` received
+  native's 501. Native's worker heartbeat also left out PHP's `retention`
+  report. As a result, a bounded namespace never pruned its closed runs. The
+  default namespace keeps 30 days.
+- **Source.** Native serves admin-only `GET /api/system/retention`,
+  `POST /api/system/retention/pass` and `GET /api/system/activity-timeouts`.
+  They work in the default or a named namespace. They follow published PHP
+  2.5.14's `SystemController`, `HistoryRetentionEnforcer` and
+  `ActivityTimeoutScanner`:
+  - **Policy.** It comes from the namespace's `retention_mode` and
+    `retention_days`. Days that are not stored come from
+    `DW_HISTORY_RETENTION_DAYS` (30 by default). The cutoff is that many days
+    before now.
+  - **Expired runs.** These are the namespace's closed, unarchived, completed
+    or failed run summaries that closed before the cutoff, oldest first. The
+    scan returns at most `min(100, limit)`.
+  - **Pass.** It prunes the run ids it is given, or else the expired runs.
+    - It skips a run with PHP's reasons: `namespace_retention_forever`,
+      `run_not_found`, `run_not_terminal`, `run_archived` or
+      `detached_activity_still_open`.
+    - Otherwise, one transaction deletes the run's rows from PHP's twenty
+      detail tables, in PHP's order, and sets the run's `details_pruned_at`.
+      The run row itself stays.
+    - The report has PHP's per-table counts. The pass answers 207 when any
+      run failed.
+  - **Heartbeat.** Each worker heartbeat runs PHP's inline pass: one run, at
+    most once a minute per namespace. Its result is the heartbeat's
+    `retention` report.
+  - **Activity timeouts.** The scan reports open executions past a schedule,
+    close, schedule-to-close or heartbeat deadline. As in PHP, it covers
+    every namespace.
+  - **Validation.** `limit`, `run_ids` and `run_ids.*` get Laravel's messages,
+    including keyed items such as `run_ids.a`.
+- **Differences from PHP.**
+  - **Payload and stream cleanup.** Before pruning, PHP deletes the run's
+    external payload objects and releases its message stream items. Native
+    does not clean these up yet. It skips such a run with
+    `external_payload_cleanup_unavailable` or
+    `message_stream_cleanup_unavailable`, so it never deletes rows that still
+    point at storage it would leave behind. Its report's three
+    message-stream counts are therefore zero.
+  - **Open activities.** PHP holds a run whose abandoned activity is still
+    open. Native holds a run with any open activity. Native closes a run only
+    after cancelling its open activities, and it refuses a completion while
+    one is pending (`pending_durable_operations`). An open activity on a
+    closed run is therefore one that PHP abandoned.
+  - **Throttle scope.** PHP's inline throttle is a cache entry. Native's
+    throttle is per process, so several native nodes can each run an inline
+    pass in the same minute. A second prune of the same run deletes nothing
+    more.
+  - **Timeline counts.** Native writes no timeline entries, so
+    `timeline_entries_deleted` is zero for runs native wrote.
+- **Still open.**
+  - `POST /api/system/activity-timeouts/pass`. Native enforces activity
+    deadlines in its own loop.
+  - External payload and message stream cleanup.
+- **Verification.**
+  - The `history_retention` integration tests pass on SQLite, PostgreSQL and
+    MySQL.
+  - The Waterline harness ran 31 requests against published PHP 2.5.14 and
+    against native: registration, two runs, cancellation with an open
+    activity, the status, explicit and empty passes, two heartbeats,
+    validation failures and a missing namespace.
+  - Every status matched. The bodies matched apart from identifiers, the
+    cutoff time and `timeline_entries_deleted` (PHP 6 and 3, native 0).
+  - The probe found that PHP 2.5.14 requires `capability_manifest` on
+    registration at protocol 1.19, and native does not. This is recorded as a
+    known gap.

docs/rust-server-port.md

@@ -59,8 +59,8 @@ CLI and Waterline use the first group.
 - **Used by the CLI or Waterline.**
   - Task queue build IDs.
   - Namespace update, delete and external storage.
-  - `system` repair, retention and activity-timeout passes, metrics, and
-    the operator dashboard filtered by workflow type.
+  - `system` repair and activity-timeout passes, metrics, and the operator
+    dashboard filtered by workflow type.
   - Activities (standalone).
 - **Worker protocol features.**
   - Worker sessions.
@@ -144,11 +144,14 @@ contract fixture before its correction.
   workflow task guard does not check registration freshness yet, and the
   local activity and cancellation scope checkpoints still use the older fence.
   Add a fixture for each before changing them.
-- **History retention.** Published PHP's worker heartbeat runs an inline
-  history retention pass, and the default namespace retention is bounded at
-  30 days, so PHP prunes old closed runs. Native has no retention yet. Its
-  heartbeat omits PHP's `retention` report rather than claim a pass. The
-  `system/retention` endpoints are absent too.
+- **Retention of runs with external payloads or message streams.**
+  Published PHP deletes a pruned run's external payload objects and releases
+  its message stream items. Native does not clean either up yet. It skips
+  such a run with `external_payload_cleanup_unavailable` or
+  `message_stream_cleanup_unavailable` and keeps its details.
+- **Registration without a capability manifest.** At protocol 1.19,
+  published PHP 2.5.14 refuses a registration that has no
+  `capability_manifest` (422). Native accepts it.
 - **Python worker shutdown after a late synchronous query handler (not yet
   diagnosed).** In the SQLite query-authority cohort on #515's run
   39431491496, the published Python SDK 2.5.15 client in `late-handler-sync`
@@ -255,7 +258,12 @@ support is not evidence of multi-node safety.
 ## Checkpoint history
 
 Newest checkpoints first. Each entry is exact to the source, tree and runs it
-names, including unsuccessful observations.
+names, including unsuccessful observations. From October 21, each new
+checkpoint is its own file in
+[`rust-server-port-checkpoints/`](rust-server-port-checkpoints/), named by
+date and topic:
+
+- [History retention](rust-server-port-checkpoints/2026-10-21-history-retention.md).
 
 October 21, operator dashboard:
 

rust/src/runtime/execution.rs

@@ -299,6 +299,9 @@ impl Runtime {
     delegate!(operator_metrics(config: &super::operator_metrics::MetricsConfig) -> Result<Value>);
     delegate!(system_health(config: &super::operator_metrics::MetricsConfig) -> Result<(bool, Value)>);
     delegate!(operator_dashboard(config: &super::operator_metrics::MetricsConfig, bounded: bool) -> Result<Value>);
+    delegate!(retention_status(limit: i64) -> Result<Value>);
+    delegate!(retention_pass(limit: i64, run_ids: Vec<String>) -> Result<(Value, bool)>);
+    delegate!(activity_timeout_status(limit: i64) -> Result<Value>);
     delegate!(create_namespace(body: Value) -> Result<Value>);
     delegate!(list_search_attributes() -> Result<Value>);
     delegate!(

rust/src/runtime/http.rs

@@ -198,6 +198,9 @@ pub fn router(runtime: Runtime) -> Router {
             "/api/system/operator-dashboard/bounded/workflow-types",
             get(workflow_type_operator_dashboard),
         )
+        .route("/api/system/retention", get(retention_status))
+        .route("/api/system/retention/pass", post(retention_pass))
+        .route("/api/system/activity-timeouts", get(activity_timeout_status))
         .route("/api/workers", get(list_workers))
         .route("/api/task-queues", get(list_task_queues))
         .route("/api/task-queues/{task_queue}", get(describe_task_queue))
@@ -532,9 +535,16 @@ fn named_namespace_operation(method: &str, path: &str) -> bool {
         || path == "/api/system/operator-dashboard"
         || path == "/api/system/operator-dashboard/bounded"
         || path == "/api/system/operator-dashboard/bounded/workflow-types"
+        // Retention reads and prunes the namespace's own runs; the activity
+        // timeout scan spans namespaces in PHP too.
+        || path == "/api/system/retention"
+        || path == "/api/system/activity-timeouts"
     {
         return method == "GET";
     }
+    if path == "/api/system/retention/pass" {
+        return method == "POST";
+    }
     if let Some(path) = path.strip_prefix("/api/workflows/") {
         let parts: Vec<_> = path.split('/').collect();
         if matches!(
@@ -1155,6 +1165,30 @@ async fn workflow_type_operator_dashboard(
         .into_response())
 }
 
+type QueryMap = Query<std::collections::HashMap<String, String>>;
+
+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?))
+}
+
+/// PHP answers 207 when any run failed to prune.
+async fn retention_pass(Extension(runtime): Extension<Runtime>, body: Option<Json<Value>>) -> Result<Response> {
+    let body = body.map_or(Value::Null, |Json(body)| body);
+    let (limit, run_ids) = super::retention::pass_request(&body)?;
+    let (report, failed) = runtime.retention_pass(limit, run_ids).await?;
+    let status = if failed { StatusCode::MULTI_STATUS } else { StatusCode::OK };
+    Ok((status, Json(report)).into_response())
+}
+
+async fn activity_timeout_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.activity_timeout_status(limit).await?))
+}
+
 /// PHP answers 503 while a check reports an error.
 async fn system_health(Extension(runtime): Extension<Runtime>) -> Result<Response> {
     let (unavailable, body) = runtime.system_health(&metrics_config(&runtime)).await?;

rust/src/runtime/laravel_validation.rs

@@ -229,7 +229,8 @@ fn messages(
     value: Option<&Value>,
     custom: fn(&str, Rule) -> Option<&'static str>,
 ) -> Vec<String> {
-    let attribute = field.replace('_', " ");
+    // Laravel shows a nested attribute such as `run_ids.0` as written.
+    let attribute = if field.contains('.') { field.to_owned() } else { field.replace('_', " ") };
     let nullable = rules.iter().any(|rule| matches!(rule, Nullable));
     let numeric = rules.iter().any(|rule| matches!(rule, Integer));
     let array = rules.iter().any(|rule| matches!(rule, Array));

rust/src/runtime/mod.rs

@@ -45,6 +45,7 @@ pub mod postgres;
 mod previews;
 mod queries;
 mod query_authority;
+mod retention;
 mod run_observation;
 mod run_observation_store;
 mod run_observation_waits;

rust/src/runtime/operator_metrics.rs

@@ -12,6 +12,7 @@
 use super::{
     Result,
     backend::Backend,
+    retention,
     store::{Store, now},
 };
 use chrono::{DateTime, Duration, SecondsFormat, Timelike, Utc};
@@ -1077,6 +1078,184 @@ where
             "identity_policy": {"provider_credentials_exposed": false, "provider_locations_exposed": false, "reusable_reference_identities_exposed": false},
         }))
     }
+
+    /// `HistoryRetentionEnforcer::retentionPolicy` for this namespace.
+    async fn retention_policy(&self) -> Result<retention::Policy> {
+        let row = Self::query("SELECT retention_mode,retention_days FROM workflow_namespaces WHERE name=$1")
+            .bind(self.namespace.as_ref())
+            .fetch_optional(&self.pool)
+            .await?;
+        let (mode, days) = match &row {
+            Some(row) => (DB::optional_string(row, "retention_mode")?, DB::optional_number(row, "retention_days")?),
+            None => (None, None),
+        };
+        Ok(retention::Policy::new(mode.as_deref(), days, now()))
+    }
+
+    /// `expiredRunIds`: closed, unarchived completed or failed runs past the
+    /// cutoff, oldest first.
+    async fn expired_run_ids(&self, policy: &retention::Policy, limit: i64) -> Result<Vec<String>> {
+        let Some(cutoff) = policy.cutoff else {
+            return Ok(Vec::new());
+        };
+        Self::query("SELECT id FROM workflow_run_summaries WHERE namespace=$1 AND status_bucket IN ('completed','failed') AND closed_at IS NOT NULL AND archived_at IS NULL AND closed_at<$2 ORDER BY closed_at,id LIMIT $3")
+            .bind(self.namespace.as_ref())
+            .bind(DB::bind_time(cutoff))
+            .bind(limit)
+            .fetch_all(&self.pool)
+            .await?
+            .iter()
+            .map(|row| DB::string(row, "id"))
+            .collect()
+    }
+
+    /// `GET /api/system/retention`.
+    pub(super) async fn retention_status(&self, limit: i64) -> Result<Value> {
+        let policy = self.retention_policy().await?;
+        let expired = self.expired_run_ids(&policy, limit).await?;
+        let mut fields = policy.fields(self.namespace.as_ref());
+        fields.insert("expired_run_count".into(), json!(expired.len()));
+        fields.insert("scan_pressure".into(), json!(expired.len() as i64 >= limit));
+        fields.insert("expired_run_ids".into(), json!(expired));
+        fields.insert("scan_limit".into(), json!(limit));
+        Ok(Value::Object(fields))
+    }
+
+    /// `POST /api/system/retention/pass`: the report, and whether any run failed.
+    pub(super) async fn retention_pass(&self, limit: i64, run_ids: Vec<String>) -> Result<(Value, bool)> {
+        let policy = self.retention_policy().await?;
+        let report = self.prune_runs(&policy, limit, run_ids).await?;
+        let failed = report["failed"].as_i64().is_some_and(|failed| failed > 0);
+        let mut fields = policy.fields(self.namespace.as_ref());
+        fields.extend(report);
+        Ok((Value::Object(fields), failed))
+    }
+
+    /// `GET /api/system/activity-timeouts`: open activity executions past a
+    /// deadline. Like PHP's `ActivityTimeoutScanner`, the scan spans every
+    /// namespace.
+    pub(super) async fn activity_timeout_status(&self, limit: i64) -> Result<Value> {
+        let at = now();
+        let expired: Vec<String> = Self::query("SELECT id FROM activity_executions WHERE status IN ('pending','running') AND ((status='pending' AND schedule_deadline_at IS NOT NULL AND schedule_deadline_at<=$1) OR (status='running' AND close_deadline_at IS NOT NULL AND close_deadline_at<=$2) OR (schedule_to_close_deadline_at IS NOT NULL AND schedule_to_close_deadline_at<=$3) OR (status='running' AND heartbeat_deadline_at IS NOT NULL AND heartbeat_deadline_at<=$4)) ORDER BY id LIMIT $5")
+            .bind(DB::bind_time(at))
+            .bind(DB::bind_time(at))
+            .bind(DB::bind_time(at))
+            .bind(DB::bind_time(at))
+            .bind(limit)
+            .fetch_all(&self.pool)
+            .await?
+            .iter()
+            .map(|row| DB::string(row, "id"))
+            .collect::<Result<_>>()?;
+        Ok(json!({"expired_count": expired.len(), "scan_pressure": expired.len() as i64 >= limit,
+            "expired_execution_ids": expired, "scan_limit": limit}))
+    }
+
+    /// `runInlinePass`, reported on the worker heartbeat.
+    pub(super) async fn retention_inline_pass(&self) -> Result<Value> {
+        let policy = self.retention_policy().await?;
+        if policy.forever {
+            return Ok(retention::inline_report(false, None));
+        }
+        if !retention::inline_pass_due(self.namespace.as_ref()) {
+            return Ok(retention::inline_report(true, None));
+        }
+        let report = self.prune_runs(&policy, 1, Vec::new()).await?;
+        Ok(retention::inline_report(false, Some(&report)))
+    }
+
+    /// `runPass`: the given runs, or the oldest expired ones.
+    async fn prune_runs(
+        &self,
+        policy: &retention::Policy,
+        limit: i64,
+        run_ids: Vec<String>,
+    ) -> Result<serde_json::Map<String, Value>> {
+        let run_ids = if run_ids.is_empty() { self.expired_run_ids(policy, limit).await? } else { run_ids };
+        let mut outcomes = Vec::with_capacity(run_ids.len());
+        for run_id in run_ids {
+            let outcome = match self.prune_run(policy, &run_id).await {
+                Ok(outcome) => outcome,
+                Err(error) => retention::Outcome::Error(error.to_string()),
+            };
+            outcomes.push((run_id, outcome));
+        }
+        Ok(retention::report(outcomes))
+    }
+
+    /// `pruneRun`: one run of this namespace loses its details in one
+    /// transaction, unless a hold keeps it.
+    async fn prune_run(&self, policy: &retention::Policy, run_id: &str) -> Result<retention::Outcome> {
+        use retention::Outcome::{Error, Pruned, Skipped};
+        if policy.forever {
+            return Ok(Skipped("namespace_retention_forever"));
+        }
+        let Some(summary) = Self::query("SELECT status,archived_at FROM workflow_run_summaries WHERE id=$1 AND namespace=$2")
+            .bind(run_id)
+            .bind(self.namespace.as_ref())
+            .fetch_optional(&self.pool)
+            .await?
+        else {
+            return Ok(Skipped("run_not_found"));
+        };
+        if !retention::terminal(&DB::string(&summary, "status")?) {
+            return Ok(Skipped("run_not_terminal"));
+        }
+        if DB::optional_instant(&summary, "archived_at")?.is_some() {
+            return Ok(Skipped("run_archived"));
+        }
+        if let Some(reason) = self.retention_hold(run_id).await? {
+            return Ok(Skipped(reason));
+        }
+        let mut tx = self.begin().await?;
+        let run = Self::query("SELECT status,closed_at FROM workflow_runs WHERE id=$1")
+            .bind(run_id)
+            .fetch_one(&mut *tx)
+            .await?;
+        // `assertPrunable`.
+        if !retention::terminal(&DB::string(&run, "status")?) {
+            return Ok(Error(format!("Workflow run [{run_id}] is not terminal.")));
+        }
+        if DB::optional_instant(&run, "closed_at")?.is_none() {
+            return Ok(Error(format!("Workflow run [{run_id}] has not been closed.")));
+        }
+        let mut deleted = serde_json::Map::new();
+        deleted.insert("run_deleted".into(), json!(0));
+        // A run with message streams is held above, so none are released.
+        for key in retention::STREAM_COUNTS {
+            deleted.insert(key.into(), json!(0));
+        }
+        for (key, sql, binds) in retention::PRUNED {
+            let mut query = Self::query(sql);
+            for _ in 0..*binds {
+                query = query.bind(run_id);
+            }
+            deleted.insert((*key).into(), json!(DB::affected(query.execute(&mut *tx).await?)));
+        }
+        Self::query("UPDATE workflow_runs SET details_pruned_at=$1 WHERE id=$2 AND details_pruned_at IS NULL")
+            .bind(DB::bind_time(now()))
+            .bind(run_id)
+            .execute(&mut *tx)
+            .await?;
+        tx.commit().await?;
+        Ok(Pruned(deleted))
+    }
+
+    /// What keeps a closed run's details: an activity it abandoned that is
+    /// still open, a message stream, or a stored external payload reference,
+    /// whose cleanup native does not perform yet.
+    async fn retention_hold(&self, run_id: &str) -> Result<Option<&'static str>> {
+        for (sql, binds, reason) in retention::HOLDS {
+            let mut query = Self::scalar(sql);
+            for _ in 0..*binds {
+                query = query.bind(run_id);
+            }
+            if query.fetch_one(&self.pool).await? > 0 {
+                return Ok(Some(reason));
+            }
+        }
+        Ok(None)
+    }
 }
 
 struct UnhealthyTasks {