diff --git a/docs/rust-server-port-checkpoints/2026-10-21-history-retention.md b/docs/rust-server-port-checkpoints/2026-10-21-history-retention.md new file mode 100644 index 000000000..2197157e3 --- /dev/null +++ b/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. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 6e7514267..493d4a25a 100644 --- a/docs/rust-server-port.md +++ b/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: diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 112154fe9..30c2c55a9 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -299,6 +299,9 @@ impl Runtime { delegate!(operator_metrics(config: &super::operator_metrics::MetricsConfig) -> Result); delegate!(system_health(config: &super::operator_metrics::MetricsConfig) -> Result<(bool, Value)>); delegate!(operator_dashboard(config: &super::operator_metrics::MetricsConfig, bounded: bool) -> Result); + delegate!(retention_status(limit: i64) -> Result); + delegate!(retention_pass(limit: i64, run_ids: Vec) -> Result<(Value, bool)>); + delegate!(activity_timeout_status(limit: i64) -> Result); delegate!(create_namespace(body: Value) -> Result); delegate!(list_search_attributes() -> Result); delegate!( diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index cc30326d3..5bc9b3832 100644 --- a/rust/src/runtime/http.rs +++ b/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>; + +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?)) +} + +/// PHP answers 207 when any run failed to prune. +async fn retention_pass(Extension(runtime): Extension, body: Option>) -> Result { + 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, + Query(query): QueryMap, +) -> Result> { + 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) -> Result { let (unavailable, body) = runtime.system_health(&metrics_config(&runtime)).await?; diff --git a/rust/src/runtime/laravel_validation.rs b/rust/src/runtime/laravel_validation.rs index ee3eec6b8..f3623fad6 100644 --- a/rust/src/runtime/laravel_validation.rs +++ b/rust/src/runtime/laravel_validation.rs @@ -229,7 +229,8 @@ fn messages( value: Option<&Value>, custom: fn(&str, Rule) -> Option<&'static str>, ) -> Vec { - 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)); diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index f6d977d9d..d0a7938e8 100644 --- a/rust/src/runtime/mod.rs +++ b/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; diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs index 182e17dd5..90d0be081 100644 --- a/rust/src/runtime/operator_metrics.rs +++ b/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 { + 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> { + 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 { + 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) -> 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 { + let at = now(); + let expired: Vec = 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::>()?; + 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 { + 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, + ) -> Result> { + 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 { + 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> { + 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 { diff --git a/rust/src/runtime/retention.rs b/rust/src/runtime/retention.rs new file mode 100644 index 000000000..f72d7540b --- /dev/null +++ b/rust/src/runtime/retention.rs @@ -0,0 +1,278 @@ +//! `HistoryRetentionEnforcer`: a closed run past its namespace's retention +//! loses its details, as published PHP prunes it. The run row stays, marked +//! with `details_pruned_at`. +//! +//! Native never deletes stored external payload objects. A run that holds a +//! reference to one, an open abandoned activity or a message stream is +//! skipped with its reason, so a prune never strands or loses data PHP would +//! have handled. +use super::laravel_validation::{Rule, failure_messages, rule_errors}; +use chrono::{DateTime, Duration, Utc}; +use serde_json::{Map, Value, json}; +use std::{ + collections::HashMap, + sync::{Mutex, OnceLock}, + time::Instant, +}; + +/// The most runs one status read or pass considers. +pub(super) const MAX_LIMIT: i64 = 100; +/// PHP's inline pass prunes one run per namespace at most once a minute. +const INLINE_THROTTLE_SECONDS: u64 = 60; + +/// The deletes of PHP's `WorkflowRunRetentionCleanup::pruneRun`, in its +/// order: report key, statement, and how many times it binds the run id. +pub(super) const PRUNED: &[(&str, &str, usize)] = &[ + ("activity_attempts_deleted", "DELETE FROM activity_attempts WHERE workflow_run_id=$1", 1), + ("updates_deleted", "DELETE FROM workflow_updates WHERE workflow_run_id=$1 OR requested_workflow_run_id=$2 OR resolved_workflow_run_id=$3", 3), + ("signals_deleted", "DELETE FROM workflow_signal_records WHERE workflow_run_id=$1 OR requested_workflow_run_id=$2 OR resolved_workflow_run_id=$3", 3), + ("messages_deleted", "DELETE FROM workflow_messages WHERE workflow_run_id=$1", 1), + ("child_calls_deleted", "DELETE FROM workflow_child_calls WHERE parent_workflow_run_id=$1", 1), + ("links_deleted", "DELETE FROM workflow_links WHERE parent_workflow_run_id=$1 OR child_workflow_run_id=$2", 2), + ("activity_executions_deleted", "DELETE FROM activity_executions WHERE workflow_run_id=$1", 1), + ("commands_deleted", "DELETE FROM workflow_commands WHERE workflow_run_id=$1 OR requested_workflow_run_id=$2 OR resolved_workflow_run_id=$3", 3), + ("failures_deleted", "DELETE FROM workflow_failures WHERE workflow_run_id=$1", 1), + ("child_projection_repairs_deleted", "DELETE FROM workflow_child_projection_repairs WHERE workflow_run_id=$1", 1), + ("run_waits_deleted", "DELETE FROM workflow_run_waits WHERE workflow_run_id=$1", 1), + ("timeline_entries_deleted", "DELETE FROM workflow_run_timeline_entries WHERE workflow_run_id=$1", 1), + ("run_timer_entries_deleted", "DELETE FROM workflow_run_timer_entries WHERE workflow_run_id=$1", 1), + ("run_lineage_entries_deleted", "DELETE FROM workflow_run_lineage_entries WHERE workflow_run_id=$1", 1), + ("timers_deleted", "DELETE FROM workflow_run_timers WHERE workflow_run_id=$1", 1), + ("memos_deleted", "DELETE FROM workflow_memos WHERE workflow_run_id=$1", 1), + ("search_attributes_deleted", "DELETE FROM workflow_search_attributes WHERE workflow_run_id=$1", 1), + ("history_events_deleted", "DELETE FROM workflow_history_events WHERE workflow_run_id=$1", 1), + ("tasks_deleted", "DELETE FROM workflow_tasks WHERE workflow_run_id=$1", 1), + ("run_summary_deleted", "DELETE FROM workflow_run_summaries WHERE id=$1", 1), +]; + +/// PHP's terminal run statuses. +pub(super) fn terminal(status: &str) -> bool { + matches!(status, "completed" | "failed" | "cancelled" | "terminated") +} + +/// A text column holding a stored external payload reference, PHP's or +/// native's, or an external storage envelope. +macro_rules! reference { + ($column:literal) => { + concat!( + "(", $column, " LIKE '%dw-external-payload:v1:%' OR ", $column, " LIKE '%dw-native-external-payload-v1:%' OR ", + $column, " LIKE '%external_storage%' OR ", $column, " LIKE '%external_payload%')" + ) + }; +} + +/// The run's payload columns that can hold an external reference. History +/// events repeat these payloads, so they are not scanned again. +const REFERENCES_SQL: &str = concat!( + "SELECT (SELECT COUNT(*) FROM workflow_runs WHERE id=$1 AND (", reference!("arguments"), " OR ", reference!("output"), "))", + "+(SELECT COUNT(*) FROM workflow_commands WHERE workflow_run_id=$2 AND ", reference!("payload"), ")", + "+(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$3 AND ", reference!("arguments"), ")", + "+(SELECT COUNT(*) FROM workflow_updates WHERE workflow_run_id=$4 AND (", reference!("arguments"), " OR ", reference!("result"), "))", + "+(SELECT COUNT(*) FROM activity_executions WHERE workflow_run_id=$5 AND (", reference!("arguments"), " OR ", reference!("result"), "))", +); + +/// `MessageStreamRetentionCleanup::apply`'s counts in the prune report. +pub(super) const STREAM_COUNTS: [&str; 3] = + ["message_streams_deleted", "message_stream_items_deleted", "message_stream_items_compacted"]; + +/// Counts that hold a run's details, the run-id binds of each, and the +/// skip reason. Native closes a run only after cancelling its open +/// activities, so an open one is an activity PHP abandoned. +pub(super) const HOLDS: &[(&str, usize, &str)] = &[ + ("SELECT COUNT(*) FROM activity_executions WHERE workflow_run_id=$1 AND status IN ('pending','running')", 1, "detached_activity_still_open"), + ("SELECT (SELECT COUNT(*) FROM workflow_durable_streams WHERE workflow_run_id=$1)+(SELECT COUNT(*) FROM workflow_messages WHERE workflow_run_id=$2)", 2, "message_stream_cleanup_unavailable"), + (REFERENCES_SQL, 5, "external_payload_cleanup_unavailable"), +]; + +/// `server.history.retention_days`. +fn default_days() -> i64 { + ["DW_HISTORY_RETENTION_DAYS", "WORKFLOW_HISTORY_RETENTION_DAYS"] + .iter() + .find_map(|name| std::env::var(name).ok()) + .and_then(|days| days.trim().parse().ok()) + .unwrap_or(30) +} + +/// `HistoryRetentionEnforcer::retentionPolicy`. +pub(super) struct Policy { + pub(super) forever: bool, + pub(super) days: Option, + pub(super) cutoff: Option>, +} + +impl Policy { + pub(super) fn new(mode: Option<&str>, days: Option, now: DateTime) -> Self { + let forever = mode == Some("forever"); + let days = (!forever).then(|| days.unwrap_or_else(default_days)); + Self { forever, days, cutoff: days.map(|days| now - Duration::days(days)) } + } + + /// The policy fields of PHP's status and pass responses. + pub(super) fn fields(&self, namespace: &str) -> Map { + let mut fields = Map::new(); + fields.insert("namespace".into(), json!(namespace)); + fields.insert("retention_mode".into(), json!(if self.forever { "forever" } else { "bounded" })); + fields.insert("retention_days".into(), json!(self.days)); + fields.insert( + "cutoff".into(), + json!(self.cutoff.map(|cutoff| cutoff.format("%Y-%m-%dT%H:%M:%S+00:00").to_string())), + ); + fields + } +} + +/// `min(100, limit ?? 100)` after Laravel's `nullable|integer|min:1`. +pub(super) fn limit(value: Option) -> i64 { + value.unwrap_or(MAX_LIMIT).min(MAX_LIMIT) +} + +fn defaults(_: &str, _: Rule) -> Option<&'static str> { + None +} + +fn checked(errors: Vec<(String, Vec)>) -> super::Result<()> { + match failure_messages(errors) { + Some(error) => Err(error), + None => Ok(()), + } +} + +/// The `limit` rule of the status endpoints: `nullable|integer|min:1`. +pub(super) fn status_limit(value: Option<&str>) -> super::Result { + let value = super::laravel_validation::normalized(value.map(|text| json!(text)).as_ref()).unwrap_or(Value::Null); + let body = json!({"limit": value}); + checked(rule_errors(&body, &[("limit", &[Rule::Nullable, Rule::Integer, Rule::Min(1)])], defaults))?; + Ok(limit(super::laravel_validation::integer(&body["limit"]))) +} + +/// `retentionEnforcePass`'s rules: the limit and up to 100 trimmed run ids. +pub(super) fn pass_request(body: &Value) -> super::Result<(i64, Vec)> { + let body = &super::start_request::normalized(body.clone()); + let rules: &[(&str, &[Rule])] = &[ + ("limit", &[Rule::Nullable, Rule::Integer, Rule::Min(1)]), + ("run_ids", &[Rule::Nullable, Rule::Array, Rule::Max(MAX_LIMIT)]), + ]; + let mut errors = rule_errors(body, rules, defaults); + // Laravel names each item by its key, which is the index of a list. + 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(), + }; + let item_rules: &[Rule] = &[Rule::String, Rule::Min(1), Rule::Max(128)]; + for (key, item) in &items { + let field = format!("run_ids.{key}"); + let mut value = Map::new(); + value.insert(field.clone(), item.clone()); + let nested = rule_errors(&Value::Object(value), &[(field.as_str(), item_rules)], defaults); + errors.extend(nested); + } + checked(errors)?; + let run_ids = items.iter().filter_map(|(_, id)| id.as_str()).map(|id| id.trim().to_owned()).collect(); + Ok((limit(super::laravel_validation::integer(&body["limit"])), run_ids)) +} + +/// What a prune attempt did to one run. +pub(super) enum Outcome { + Pruned(Map), + Skipped(&'static str), + Error(String), +} + +/// `runPass`'s report over each run's outcome. +pub(super) fn report(outcomes: Vec<(String, Outcome)>) -> Map { + let (mut pruned, mut skipped, mut failed) = (0, 0, 0); + let processed = outcomes.len(); + let results: Vec = outcomes + .into_iter() + .map(|(run_id, outcome)| match outcome { + Outcome::Pruned(deleted) => { + pruned += 1; + json!({"run_id": run_id, "outcome": "pruned", "history_events_deleted": deleted["history_events_deleted"], + "tasks_deleted": deleted["tasks_deleted"], "external_payloads_deleted": 0, "deleted": deleted}) + } + Outcome::Skipped(reason) => { + skipped += 1; + json!({"run_id": run_id, "outcome": "skipped", "reason": reason}) + } + Outcome::Error(reason) => { + failed += 1; + json!({"run_id": run_id, "outcome": "error", "reason": reason}) + } + }) + .collect(); + let mut report = Map::new(); + report.insert("processed".into(), json!(processed)); + report.insert("pruned".into(), json!(pruned)); + report.insert("skipped".into(), json!(skipped)); + report.insert("failed".into(), json!(failed)); + report.insert("results".into(), json!(results)); + report +} + +/// `runInlinePass`'s throttle: at most one inline pass per namespace a +/// minute in this process. +pub(super) fn inline_pass_due(namespace: &str) -> bool { + static LAST: OnceLock>> = OnceLock::new(); + let mut last = LAST.get_or_init(Default::default).lock().unwrap_or_else(std::sync::PoisonError::into_inner); + let now = Instant::now(); + match last.get(namespace) { + Some(at) if now.duration_since(*at).as_secs() < INLINE_THROTTLE_SECONDS => false, + _ => { + last.insert(namespace.to_owned(), now); + true + } + } +} + +/// `runInlinePass`'s heartbeat report. +pub(super) fn inline_report(throttled: bool, report: Option<&Map>) -> Value { + let count = |key: &str| report.and_then(|report| report.get(key)).cloned().unwrap_or(json!(0)); + json!({"throttled": throttled, "processed": count("processed"), "pruned": count("pruned"), + "skipped": count("skipped"), "failed": count("failed")}) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_delete_binds_the_run_once_per_reference() { + for (_, sql, binds) in PRUNED { + assert_eq!(sql.matches('$').count(), *binds, "{sql}"); + } + assert_eq!(PRUNED.last().unwrap().0, "run_summary_deleted"); + for (sql, binds, _) in HOLDS { + assert_eq!((1..=*binds).filter(|index| sql.contains(&format!("${index}"))).count(), *binds, "{sql}"); + assert!(!sql.contains(&format!("${}", binds + 1)), "{sql}"); + } + } + + #[test] + fn policies_follow_php() { + let now = DateTime::parse_from_rfc3339("2026-10-21T12:00:00Z").unwrap().with_timezone(&Utc); + let bounded = Policy::new(Some("bounded"), Some(7), now); + assert_eq!(bounded.fields("default")["cutoff"], "2026-10-14T12:00:00+00:00"); + let forever = Policy::new(Some("forever"), Some(7), now); + assert_eq!((forever.days, forever.cutoff), (None, None)); + assert_eq!(forever.fields("default")["retention_mode"], "forever"); + assert_eq!(limit(Some(500)), 100); + assert_eq!(limit(None), 100); + } + + #[test] + fn reports_count_each_outcome() { + let mut deleted = Map::new(); + deleted.insert("history_events_deleted".into(), json!(3)); + deleted.insert("tasks_deleted".into(), json!(1)); + let report = report(vec![ + ("a".into(), Outcome::Pruned(deleted)), + ("b".into(), Outcome::Skipped("run_not_terminal")), + ("c".into(), Outcome::Error("boom".into())), + ]); + assert_eq!((report["processed"].clone(), report["pruned"].clone(), report["skipped"].clone(), report["failed"].clone()), + (json!(3), json!(1), json!(1), json!(1))); + assert_eq!(report["results"][0]["history_events_deleted"], 3); + assert_eq!(inline_report(true, None), json!({"throttled":true,"processed":0,"pruned":0,"skipped":0,"failed":0})); + } +} diff --git a/rust/src/runtime/worker_management.rs b/rust/src/runtime/worker_management.rs index 9aa709ac6..a80d77aa7 100644 --- a/rust/src/runtime/worker_management.rs +++ b/rust/src/runtime/worker_management.rs @@ -480,11 +480,11 @@ where Self::set_count(&mut tx, &INTERVAL, interval, worker_id, namespace).await?; } tx.commit().await?; - // PHP also reports an inline history retention pass here. Native has - // no retention yet, so it does not claim one. + // Like PHP, each heartbeat may prune one expired run of its namespace. + let retention = self.retention_inline_pass().await?; Ok(json!({"worker_id": worker_id, "acknowledged": true, "heartbeat_interval_seconds": config.worker_heartbeat_interval_seconds, - "stale_after_seconds": config.worker_stale_after_seconds})) + "stale_after_seconds": config.worker_stale_after_seconds, "retention": retention})) } fn worker_summary(row: &DB::Row, stale_after: i64, at: DateTime) -> Result { diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index f617471cb..dc61fc76d 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -85,6 +85,8 @@ mod read_snapshots; mod query_responses; #[path = "execution/history_export.rs"] mod history_export; +#[path = "execution/history_retention.rs"] +mod history_retention; #[path = "execution/start_request.rs"] mod start_request; #[path = "execution/command_requests.rs"] diff --git a/rust/tests/execution/history_retention.rs b/rust/tests/execution/history_retention.rs new file mode 100644 index 000000000..b06440979 --- /dev/null +++ b/rust/tests/execution/history_retention.rs @@ -0,0 +1,111 @@ +use super::*; + +const STATUS: &str = "/api/system/retention"; +const PASS: &str = "/api/system/retention/pass"; + +async fn completed_run(app: &Router, workflow: &str) -> String { + let started = start(app, workflow).await; + let task = poll(app, "worker", "workflow").await; + let complete = json!({"type":"complete_workflow","result":envelope(Payload::Long(1))}); + assert_eq!(finish_task(app, &task, json!([complete])).await.0, StatusCode::OK); + started["run_id"].as_str().unwrap().to_owned() +} + +fn messages(body: &Value, field: &str) -> Value { + body["errors"][field].clone() +} + +// Published PHP keeps closed runs for the namespace's retention days, then +// prunes their details: one run per heartbeat, or a batch on request. +#[tokio::test] +async fn retention_prunes_only_expired_closed_runs_like_php() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + register(&app, "worker", json!(["echo"]), json!([])).await; + let expired = completed_run(&app, "expired").await; + let kept = completed_run(&app, "kept").await; + let open = start(&app, "open").await["run_id"].as_str().unwrap().to_owned(); + + let (status, fresh) = request(&app, "GET", STATUS, Value::Null).await; + assert_eq!(status, StatusCode::OK, "{fresh}"); + assert_eq!( + [&fresh["namespace"], &fresh["retention_mode"], &fresh["retention_days"], &fresh["expired_run_count"]], + [&json!("default"), &json!("bounded"), &json!(30), &json!(0)] + ); + assert!(fresh["cutoff"].as_str().unwrap().ends_with("+00:00"), "{fresh}"); + assert_eq!((fresh["scan_limit"].clone(), fresh["scan_pressure"].clone()), (json!(100), json!(false))); + + // The expired run closed long before the 30-day cutoff. + for table in ["workflow_runs", "workflow_run_summaries"] { + database + .execute(sqlx::AssertSqlSafe(format!( + "UPDATE {table} SET closed_at='2020-01-01 00:00:00.000000' WHERE id='{expired}'" + ))) + .await; + } + let (status, pressured) = request(&app, "GET", &format!("{STATUS}?limit=1"), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{pressured}"); + assert_eq!(pressured["expired_run_ids"], json!([expired])); + assert_eq!((pressured["scan_limit"].clone(), pressured["scan_pressure"].clone()), (json!(1), json!(true))); + + // Laravel's validation messages. + let (status, invalid) = request(&app, "GET", &format!("{STATUS}?limit=0"), Value::Null).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{invalid}"); + assert_eq!(messages(&invalid, "limit"), json!(["The limit field must be at least 1."])); + let (status, invalid) = request(&app, "POST", PASS, json!({"run_ids":[1, "x"]})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{invalid}"); + assert_eq!(messages(&invalid, "run_ids.0"), json!(["The run_ids.0 field must be a string."])); + let (status, invalid) = request(&app, "POST", PASS, json!({"run_ids":"x"})).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{invalid}"); + assert_eq!(messages(&invalid, "run_ids"), json!(["The run ids field must be an array."])); + + // Named runs that are missing or still open are skipped with a reason. + let (status, skipped) = request(&app, "POST", PASS, json!({"run_ids":[" missing ", open]})).await; + assert_eq!(status, StatusCode::OK, "{skipped}"); + assert_eq!((skipped["processed"].clone(), skipped["skipped"].clone(), skipped["pruned"].clone()), (json!(2), json!(2), json!(0))); + assert_eq!(skipped["results"][0], json!({"run_id":"missing","outcome":"skipped","reason":"run_not_found"})); + assert_eq!(skipped["results"][1]["reason"], "run_not_terminal"); + + // Without run ids, the pass prunes the expired run in one transaction. + let (status, pruned) = request(&app, "POST", PASS, json!([])).await; + assert_eq!(status, StatusCode::OK, "{pruned}"); + assert_eq!((pruned["processed"].clone(), pruned["pruned"].clone(), pruned["failed"].clone()), (json!(1), json!(1), json!(0))); + let result = &pruned["results"][0]; + assert_eq!((result["run_id"].clone(), result["outcome"].clone()), (json!(expired), json!("pruned"))); + assert!(result["history_events_deleted"].as_i64().unwrap() > 0, "{result}"); + assert_eq!(result["deleted"]["run_summary_deleted"], 1, "{result}"); + assert_eq!(result["deleted"]["run_deleted"], 0, "{result}"); + assert_eq!(result["deleted"]["message_stream_items_compacted"], 0, "{result}"); + assert_eq!(result["external_payloads_deleted"], 0, "{result}"); + + let (_, after) = request(&app, "GET", STATUS, Value::Null).await; + assert_eq!(after["expired_run_count"], 0, "{after}"); + let (status, again) = request(&app, "POST", PASS, json!({"run_ids":[expired]})).await; + assert_eq!(status, StatusCode::OK, "{again}"); + assert_eq!(again["results"][0]["reason"], "run_not_found", "{again}"); + // The run that has not expired keeps its history. + let (status, export) = request(&app, "GET", &format!("/api/workflows/kept/runs/{kept}/history/export"), Value::Null).await; + assert_eq!(status, StatusCode::OK, "{export}"); + assert!(!export["history_events"].as_array().unwrap().is_empty()); + + // A heartbeat reports its namespace's inline pass, at most once a minute. + let (status, heartbeat) = request(&app, "POST", "/api/worker/heartbeat", json!({"worker_id":"worker"})).await; + assert_eq!(status, StatusCode::OK, "{heartbeat}"); + let retention = &heartbeat["retention"]; + assert_eq!(retention["failed"], 0, "{heartbeat}"); + for field in ["throttled", "processed", "pruned", "skipped"] { + assert!(!retention[field].is_null(), "{field}: {heartbeat}"); + } +} + +#[tokio::test] +async fn activity_timeout_status_reports_no_expired_executions_like_php() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let (status, body) = request(&app, "GET", "/api/system/activity-timeouts?limit=5", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body, json!({"expired_count":0,"expired_execution_ids":[],"scan_limit":5,"scan_pressure":false})); + let (status, invalid) = request(&app, "GET", "/api/system/activity-timeouts?limit=abc", Value::Null).await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{invalid}"); + assert_eq!(messages(&invalid, "limit"), json!(["The limit field must be an integer."])); +}