From a5dc8af52cbc9f9a4b9a0f654522a6d9f037c97b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 23:28:21 +0000 Subject: [PATCH 1/3] Serve external payload cleanup and its pass like published PHP 2.11 GET /api/system/external-payload-cleanup reports the namespace's cleanup metrics. POST .../pass renews expired completion-upload allowances whose activity can still complete and drops the rest, then reclaims the namespace's expired, unclaimed references under their object locks: a shared object keeps its file, a sole reference deletes it. Each pass is recorded in the namespace's totals; a blocked reference answers 207. Native deletes through the local driver only, honours a backup hold that PHP left active, and treats a missing hold row as no hold. --- .github/workflows/server-parity.yml | 4 +- .../2026-10-22-payload-cleanup.md | 43 ++ docs/rust-server-port.md | 7 +- .../runtime/activity_completion_uploads.rs | 4 + rust/src/runtime/activity_lease_renewal.rs | 29 ++ rust/src/runtime/execution.rs | 12 + rust/src/runtime/http.rs | 20 +- rust/src/runtime/mod.rs | 1 + rust/src/runtime/operator_metrics.rs | 2 +- rust/src/runtime/payload_cleanup.rs | 317 ++++++++++++++ rust/tests/execution/external_payloads.rs | 8 +- .../execution/external_payloads/cleanup.rs | 135 ++++++ .../external_payloads/completion_uploads.rs | 27 ++ .../conformance/server-parity/contract.mjs | 3 + .../payload-cleanup-contract.mjs | 32 ++ .../server-parity/payload-cleanup-probe.php | 35 ++ scripts/conformance/server-parity/probe.php | 6 + .../ServerParity/payload-cleanup.json | 391 ++++++++++++++++++ ...ServerParityPayloadCleanupContractTest.mjs | 59 +++ 19 files changed, 1128 insertions(+), 7 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md create mode 100644 rust/src/runtime/payload_cleanup.rs create mode 100644 rust/tests/execution/external_payloads/cleanup.rs create mode 100644 scripts/conformance/server-parity/payload-cleanup-contract.mjs create mode 100644 scripts/conformance/server-parity/payload-cleanup-probe.php create mode 100644 tests/Fixtures/ServerParity/payload-cleanup.json create mode 100644 tests/Unit/ServerParityPayloadCleanupContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 8365e6a7c..5ba5b0809 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -38,6 +38,7 @@ on: - tests/Unit/ServerParityServiceCatalogContractTest.mjs - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs + - tests/Unit/ServerParityPayloadCleanupContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs @@ -90,6 +91,7 @@ on: - tests/Unit/ServerParityServiceCatalogContractTest.mjs - tests/Unit/ServerParitySystemMetricsContractTest.mjs - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs + - tests/Unit/ServerParityPayloadCleanupContractTest.mjs - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs - tests/Unit/ServerParityWaitingHistoryContractTest.mjs - tests/Unit/ServerParityTaskRetryContractTest.mjs @@ -1508,7 +1510,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/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/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-22-payload-cleanup.md b/docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md new file mode 100644 index 000000000..856cbf350 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md @@ -0,0 +1,43 @@ +# October 22, external payload cleanup + +- **Source.** Server 2.11.1 `SystemController::externalPayloadCleanupStatus` + and `externalPayloadCleanupPass`, `RuntimeExternalPayloadCleanup::runPass`, + `RuntimeExternalPayloadCleanupMetrics::snapshot`, + `RuntimePayloadCompletionUploads::cleanup`, `RuntimePayloadCompletionLease:: + mayResume` and `ExternalPayloadBackupHold::deleting`. +- **Served.** Native answers admin-only + `GET /api/system/external-payload-cleanup` and + `POST /api/system/external-payload-cleanup/pass` in the default or a named + namespace. + - The status is the cleanup section native's operator metrics already + report: the expired backlog, batch bounds, totals and the last pass. + - The pass takes `limit` (`nullable|integer|min:1|max:1000`, default 100). + It first visits the namespace's expired completion-upload allowances: one + whose activity can still complete (its task is leased to the same owner + and its attempt can continue) gets another abandoned-upload period, and + any other is dropped. It then reclaims the namespace's expired, unclaimed + references, oldest first, each under its object lock. A reference that + shares its object with another loses only its row; a sole reference also + deletes its object. The report lists processed, deleted, preserved, + skipped and blocked references, the remaining backlog and scan pressure, + and answers 207 when anything was blocked. Each pass is added to the + namespace's totals, with whole-second times as Eloquent stores them. +- **Decisions.** + - Native stores objects only with the local driver. A reference without an + enabled local policy, or whose object lies outside the current local root, + is blocked rather than deleted. PHP would delete through its configured + remote driver or old root. + - Native never takes an online-backup hold. A hold row left active blocks + deletion as in PHP; a missing row means no hold, where PHP would report a + storage failure because its migration seeds the row. + - Native's completion allowances hold activity contexts only. An allowance + whose context native cannot read cannot resume and is dropped once + expired; PHP would fail the pass on a context it cannot parse. +- **Fixture.** `payload-cleanup` checks a fresh namespace's status, a pass, + the recorded totals, limit validation (0, 1001, text and a fraction answer + 422; null and numeric text are accepted) and the final count of five + passes. Deletion needs expired rows and stored objects, which an HTTP probe + cannot make within the abandoned-upload period, so native tests cover it on + SQLite, PostgreSQL and MySQL: shared, live and retained references, the + limit and its backlog, an active backup hold, a disabled policy, and an + allowance renewed while its activity runs and dropped after it completes. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index fbb50e0dc..cb7e5b5a4 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -102,8 +102,8 @@ 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, and external payload cleanup and its pass. (Namespace - delete is served with the `namespace-delete` fixture. The JSON + - 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 @@ -112,7 +112,8 @@ These published PHP routes on this tree have no native handler yet, by family. 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.) + family. External payload cleanup and its pass are served with the + `payload-cleanup` fixture.) - Task queue priority and fairness. - **Worker protocol features.** - Worker sessions. diff --git a/rust/src/runtime/activity_completion_uploads.rs b/rust/src/runtime/activity_completion_uploads.rs index c68b7e4ad..04341c1e4 100644 --- a/rust/src/runtime/activity_completion_uploads.rs +++ b/rust/src/runtime/activity_completion_uploads.rs @@ -129,6 +129,10 @@ impl Context { value, }) } + /// The task, attempt and lease owner this context names. + pub(super) fn lease_identity(&self) -> (&str, &str, &str) { + (&self.task, &self.attempt, &self.owner) + } fn scope(&self, namespace: &str) -> String { identity(&json!([ namespace, diff --git a/rust/src/runtime/activity_lease_renewal.rs b/rust/src/runtime/activity_lease_renewal.rs index 83890c97d..ac1a84f9e 100644 --- a/rust/src/runtime/activity_lease_renewal.rs +++ b/rust/src/runtime/activity_lease_renewal.rs @@ -128,6 +128,35 @@ store_methods! { Ok(status) } + /// `RuntimePayloadCompletionLease::mayResume` for an activity: its task + /// is still leased to the owner and its attempt can continue on that task + /// for that owner. An expired lease alone does not end it, since the + /// activity may still renew before lease recovery. + pub(super) async fn activity_completion_may_resume( + &self, + tx: &mut Transaction<'_, DB>, + task_id: &str, + attempt_id: &str, + owner: &str, + ) -> Result { + let Some(task) = Self::query("SELECT status,lease_owner FROM workflow_tasks WHERE id=$1 AND namespace=$2") + .bind(task_id) + .bind(self.namespace.as_ref()) + .fetch_optional(&mut **tx) + .await? + else { + return Ok(false); + }; + if DB::string(&task, "status")? != "leased" || DB::optional_string(&task, "lease_owner")?.as_deref() != Some(owner) { + return Ok(false); + } + let Some(rows) = Self::attempt_rows(tx, attempt_id).await? else { + return Ok(false); + }; + let status = Self::attempt_status(attempt_id, &rows, None)?; + Ok(status["can_continue"] == true && status["workflow_task_id"] == task_id && status["lease_owner"] == owner) + } + async fn attempt_rows(tx: &mut Transaction<'_, DB>, attempt_id: &str) -> Result>> { let Some(attempt) = Self::query("SELECT * FROM activity_attempts WHERE id=$1") .bind(attempt_id) diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 3eb9ca2bb..3e18a769f 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -341,6 +341,18 @@ impl Runtime { delegate!(repair_status() -> Result); delegate!(system_metrics() -> Result); delegate!(prometheus_metrics() -> Result); + delegate!(payload_cleanup_status(config: &super::operator_metrics::MetricsConfig) -> Result); + /// The namespace's payload cleanup pass, with its local storage when an + /// enabled local policy names it. + pub(crate) async fn payload_cleanup_pass(&self, limit: i64) -> Result { + let local = super::external_transport::local(self).await.ok(); + let retention = self.external.abandoned_seconds; + match &*self.storage { + Storage::Sqlite(store) => store.payload_cleanup_pass(limit, retention, local.as_ref()).await, + Storage::Postgres(store) => store.payload_cleanup_pass(limit, retention, local.as_ref()).await, + Storage::MySql(store) => store.payload_cleanup_pass(limit, retention, local.as_ref()).await, + } + } delegate!(retention_status(limit: i64) -> Result); delegate!(retention_pass(limit: i64, run_ids: Vec) -> Result<(Value, bool)>); delegate!(activity_timeout_status(limit: i64) -> Result); diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index d9b6deecb..8089cea58 100644 --- a/rust/src/runtime/http.rs +++ b/rust/src/runtime/http.rs @@ -205,6 +205,8 @@ pub fn router(runtime: Runtime) -> Router { .route("/api/system/health", get(system_health)) .route("/api/system/metrics", get(system_metrics).post(system_metrics)) .route("/api/system/prometheus-metrics", get(prometheus_metrics)) + .route("/api/system/external-payload-cleanup", get(payload_cleanup_status)) + .route("/api/system/external-payload-cleanup/pass", post(payload_cleanup_pass)) .route("/api/system/operator-dashboard", get(operator_dashboard)) .route("/api/system/operator-dashboard/bounded", get(bounded_operator_dashboard)) .route( @@ -651,9 +653,13 @@ fn named_namespace_operation(method: &str, path: &str) -> bool { if path == "/api/system/metrics" { return matches!(method, "GET" | "POST"); } - if path == "/api/system/prometheus-metrics" { + if path == "/api/system/prometheus-metrics" || path == "/api/system/external-payload-cleanup" { return method == "GET"; } + // The pass reclaims only the namespace's own expired payloads. + if path == "/api/system/external-payload-cleanup/pass" { + return method == "POST"; + } if let Some(path) = path.strip_prefix("/api/workflows/") { let parts: Vec<_> = path.split('/').collect(); if matches!( @@ -1490,6 +1496,18 @@ async fn prometheus_metrics(Extension(runtime): Extension) -> Result) -> Result> { + Ok(Json(runtime.payload_cleanup_status(&metrics_config(&runtime)).await?)) +} + +/// PHP answers 207 when the pass blocked any reference. +async fn payload_cleanup_pass(Extension(runtime): Extension, body: axum::body::Bytes) -> Result { + let limit = super::payload_cleanup::pass_limit(&request_json(&body)?)?; + let report = runtime.payload_cleanup_pass(limit).await?; + let blocked = report["blocked"].as_i64().is_some_and(|blocked| blocked > 0); + Ok(pass_response((json!({"namespace": runtime.namespace(), "cleanup": report}), blocked))) +} + /// 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/mod.rs b/rust/src/runtime/mod.rs index 59f3042c0..e93b05bdc 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -69,6 +69,7 @@ mod namespace_cleanup; mod namespaces; mod operator_dashboard; mod operator_metrics; +mod payload_cleanup; mod payload_limits; mod poll_readiness; pub mod postgres; diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs index 2574ab2f5..0e3fb648c 100644 --- a/rust/src/runtime/operator_metrics.rs +++ b/rust/src/runtime/operator_metrics.rs @@ -1027,7 +1027,7 @@ where /// PHP's `RuntimeExternalPayloadCleanupMetrics`: the expired registrations /// and the namespace's recorded cleanup passes. - async fn external_payload_cleanup_metrics(&self, config: &MetricsConfig) -> Result { + pub(super) async fn external_payload_cleanup_metrics(&self, config: &MetricsConfig) -> Result { let bound = whole_second(now()); let stats = Self::query("SELECT * FROM runtime_external_payload_cleanup_stats WHERE namespace=$1") diff --git a/rust/src/runtime/payload_cleanup.rs b/rust/src/runtime/payload_cleanup.rs new file mode 100644 index 000000000..ed98bb6df --- /dev/null +++ b/rust/src/runtime/payload_cleanup.rs @@ -0,0 +1,317 @@ +//! `GET /api/system/external-payload-cleanup` and +//! `POST /api/system/external-payload-cleanup/pass`, as published Server 2.11 +//! serves them (`RuntimeExternalPayloadCleanup::runPass`). +//! +//! A pass first renews or drops the namespace's expired completion-upload +//! budgets, then reclaims its expired, unclaimed payload references, oldest +//! first. Each reference is reclaimed under its object lock: a reference that +//! shares its object with another only loses its row, and a sole reference +//! also deletes its object. Each pass is recorded in the namespace's cleanup +//! totals. +//! +//! Native stores objects only with the local driver. A reference it cannot +//! map to the namespace's local storage (no enabled local policy, or an object +//! outside the current root) is blocked rather than deleted. Native never +//! takes an online-backup hold; a hold row PHP left active still blocks +//! deletion, and a missing row means no hold. +use super::Result; +use super::activity_completion_uploads::Context; +use super::laravel_validation::{Rule, failure_messages, rule_errors}; +use super::operator_metrics::whole_second; +use super::store::now; +use crate::external_payload::{LocalStorage, REFERENCE_SCHEMA, Reference}; +use chrono::{DateTime, Duration, Utc}; +use serde_json::{Value, json}; + +/// `RuntimeExternalPayloadCleanup::DEFAULT_BATCH_SIZE`. +pub(crate) const DEFAULT_BATCH_SIZE: i64 = 100; +/// `RuntimeExternalPayloadCleanup::MAX_BATCH_SIZE`. +pub(crate) const MAX_BATCH_SIZE: i64 = 1000; + +/// The pass's rule: `limit` is `nullable|integer|min:1|max:1000`, by default 100. +pub(super) fn pass_limit(body: &Value) -> Result { + let body = &super::start_request::normalized(body.clone()); + let rules: &[(&str, &[Rule])] = &[("limit", &[Rule::Nullable, Rule::Integer, Rule::Min(1), Rule::Max(MAX_BATCH_SIZE)])]; + if let Some(error) = failure_messages(rule_errors(body, rules, |_, _| None)) { + return Err(error); + } + Ok(super::laravel_validation::integer(&body["limit"]).unwrap_or(DEFAULT_BATCH_SIZE)) +} + +enum Outcome { + Skipped, + DeletedObject, + DeletedShared, + Blocked, + StorageFailure, +} + +#[derive(Default)] +struct Report { + processed: i64, + deleted_references: i64, + deleted_backing_objects: i64, + shared_objects_preserved: i64, + skipped: i64, + blocked: i64, + storage_driver_failures: i64, +} + +store_methods! { + /// `externalPayloadCleanupStatus`: the namespace's cleanup metrics. + pub(crate) async fn payload_cleanup_status(&self, config: &super::operator_metrics::MetricsConfig) -> Result { + Ok(json!({"namespace": self.namespace.as_ref(), "cleanup": self.external_payload_cleanup_metrics(config).await?})) + } + + /// The pass's report; the caller answers 207 when it blocked anything. + pub(crate) async fn payload_cleanup_pass( + &self, + limit: i64, + retention_seconds: i64, + local: Option<&LocalStorage>, + ) -> Result { + let limit = limit.clamp(1, MAX_BATCH_SIZE); + let cutoff = now(); + self.completion_budget_cleanup(limit, cutoff, retention_seconds.max(1)).await?; + let candidates = Self::query("SELECT id FROM runtime_external_payloads WHERE namespace=$1 AND retained_at IS NULL AND expires_at IS NOT NULL AND expires_at<=$2 ORDER BY expires_at,id LIMIT $3") + .bind(self.namespace.as_ref()) + .bind(DB::bind_time(cutoff)) + .bind(limit) + .fetch_all(&self.pool) + .await?; + let mut report = Report::default(); + for candidate in &candidates { + report.processed += 1; + match self.cleanup_reference(&DB::string(candidate, "id")?, cutoff, local).await? { + Outcome::Skipped => report.skipped += 1, + Outcome::DeletedObject => { + report.deleted_references += 1; + report.deleted_backing_objects += 1; + } + Outcome::DeletedShared => { + report.deleted_references += 1; + report.shared_objects_preserved += 1; + } + Outcome::Blocked => report.blocked += 1, + Outcome::StorageFailure => { + report.blocked += 1; + report.storage_driver_failures += 1; + } + } + } + self.record_cleanup_stats(&report).await?; + let backlog = Self::scalar("SELECT COUNT(*) FROM runtime_external_payloads WHERE namespace=$1 AND retained_at IS NULL AND expires_at IS NOT NULL AND expires_at<=$2") + .bind(self.namespace.as_ref()) + .bind(DB::bind_time(cutoff)) + .fetch_one(&self.pool) + .await?; + Ok(json!({ + "processed": report.processed, "deleted_references": report.deleted_references, + "deleted_backing_objects": report.deleted_backing_objects, + "shared_objects_preserved": report.shared_objects_preserved, "skipped": report.skipped, + "blocked": report.blocked, "storage_driver_failures": report.storage_driver_failures, + "backlog": backlog, "scan_limit": limit, + "scan_pressure": backlog > 0 && candidates.len() as i64 >= limit, + })) + } + + /// `RuntimePayloadCompletionUploads::cleanup`: an expired budget whose + /// activity may still complete gets another retention period; any other + /// expired budget is dropped. + async fn completion_budget_cleanup(&self, limit: i64, cutoff: DateTime, retention_seconds: i64) -> Result<()> { + let candidates = Self::query("SELECT id,context FROM runtime_payload_completion_budgets WHERE expires_at<=$1 AND namespace=$2 ORDER BY expires_at LIMIT $3") + .bind(DB::bind_time(cutoff)) + .bind(self.namespace.as_ref()) + .bind(limit) + .fetch_all(&self.pool) + .await?; + for candidate in &candidates { + let id = DB::string(candidate, "id")?; + // Native holds activity contexts only; any other cannot resume. + let context = Context::parse(&DB::document_row(candidate, "context")?.to_string()).ok(); + let mut tx = self.begin().await?; + // As in PHP, completion ownership is read before the object lock. + let may_resume = match &context { + Some(context) => { + let (task, attempt, owner) = context.lease_identity(); + self.activity_completion_may_resume(&mut tx, task, attempt, owner).await? + } + None => false, + }; + Self::payload_object_lock(&mut tx, &id).await?; + let Some(row) = Self::query("SELECT expires_at FROM runtime_payload_completion_budgets WHERE id=$1") + .bind(&id) + .fetch_optional(&mut *tx) + .await? + else { + tx.commit().await?; + continue; + }; + if DB::instant(&row, "expires_at")? > cutoff { + tx.commit().await?; + continue; + } + if may_resume { + let at = now(); + Self::query("UPDATE runtime_payload_completion_budgets SET expires_at=$1,updated_at=$2 WHERE id=$3") + .bind(DB::bind_time(at + Duration::seconds(retention_seconds))) + .bind(DB::bind_time(at)) + .bind(&id) + .execute(&mut *tx) + .await?; + } else { + Self::query("DELETE FROM runtime_payload_completion_budgets WHERE id=$1") + .bind(&id) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + } + Ok(()) + } + + /// `cleanupReference`, under the reference's object lock. + async fn cleanup_reference(&self, id: &str, cutoff: DateTime, local: Option<&LocalStorage>) -> Result { + let Some(first) = Self::query("SELECT storage_uri FROM runtime_external_payloads WHERE id=$1") + .bind(id) + .fetch_optional(&self.pool) + .await? + else { + return Ok(Outcome::Skipped); + }; + let uri = DB::string(&first, "storage_uri")?; + let mut tx = self.begin().await?; + Self::payload_object_lock(&mut tx, &uri).await?; + let Some(row) = Self::query("SELECT * FROM runtime_external_payloads WHERE id=$1") + .bind(id) + .fetch_optional(&mut *tx) + .await? + else { + tx.commit().await?; + return Ok(Outcome::Skipped); + }; + if DB::optional_instant(&row, "retained_at")?.is_some() + || DB::optional_instant(&row, "expires_at")?.is_none_or(|expires| expires > cutoff) + { + tx.commit().await?; + return Ok(Outcome::Skipped); + } + let uri = DB::string(&row, "storage_uri")?; + let owners = Self::scalar("SELECT COUNT(*) FROM runtime_external_payloads WHERE storage_uri_sha256=$1 AND storage_uri=$2") + .bind(DB::string(&row, "storage_uri_sha256")?) + .bind(&uri) + .fetch_one(&mut *tx) + .await?; + if owners > 1 { + Self::query("DELETE FROM runtime_external_payloads WHERE id=$1").bind(id).execute(&mut *tx).await?; + tx.commit().await?; + return Ok(Outcome::DeletedShared); + } + let reference = Reference { + schema: REFERENCE_SCHEMA, + reference_id: id.to_owned(), + codec: "avro", + size_bytes: DB::number(&row, "size_bytes")?, + sha256: DB::string(&row, "sha256")?, + }; + let Some(local) = local.filter(|local| local.uri_for(&reference).is_ok_and(|current| current == uri)) else { + tx.commit().await?; + return Ok(Outcome::Blocked); + }; + if Self::backup_hold_active(&mut tx).await? { + tx.commit().await?; + return Ok(Outcome::Blocked); + } + if local.delete(&reference).await.is_err() { + tx.commit().await?; + return Ok(Outcome::StorageFailure); + } + Self::query("DELETE FROM runtime_external_payloads WHERE id=$1").bind(id).execute(&mut *tx).await?; + tx.commit().await?; + Ok(Outcome::DeletedObject) + } + + /// `ExternalPayloadBackupHold::deleting`: an owner whose hold has not + /// expired pauses reclamation. + async fn backup_hold_active(tx: &mut super::transaction::Transaction<'_, DB>) -> Result { + // PostgreSQL keeps the owner as a uuid; only its presence matters. + let Some(hold) = Self::query("SELECT CASE WHEN owner IS NULL THEN 0 ELSE 1 END AS held,expires_at FROM runtime_external_payload_backup_hold WHERE id=1") + .fetch_optional(&mut **tx) + .await? + else { + return Ok(false); + }; + Ok(DB::number(&hold, "held")? == 1 + && DB::optional_instant(&hold, "expires_at")?.is_some_and(|expires| expires > now())) + } + + /// `recordStats` for an existing namespace. Times are whole seconds, as + /// Eloquent stores them. + async fn record_cleanup_stats(&self, report: &Report) -> Result<()> { + let namespace = self.namespace.as_ref(); + if Self::scalar("SELECT COUNT(*) FROM workflow_namespaces WHERE name=$1") + .bind(namespace) + .fetch_one(&self.pool) + .await? + == 0 + { + return Ok(()); + } + let at = whole_second(now()); + let mut tx = self.begin().await?; + if Self::scalar("SELECT COUNT(*) FROM runtime_external_payload_cleanup_stats WHERE namespace=$1") + .bind(namespace) + .fetch_one(&mut *tx) + .await? + == 0 + { + Self::query("INSERT INTO runtime_external_payload_cleanup_stats(namespace,created_at,updated_at) VALUES ($1,$2,$3)") + .bind(namespace) + .bind(DB::bind_time(at)) + .bind(DB::bind_time(at)) + .execute(&mut *tx) + .await?; + } + let status = if report.storage_driver_failures > 0 { + "failed" + } else if report.blocked > 0 { + "blocked" + } else { + "completed" + }; + Self::query(concat!( + "UPDATE runtime_external_payload_cleanup_stats SET passes_total=passes_total+1,", + "deleted_references_total=deleted_references_total+$1,deleted_backing_objects_total=deleted_backing_objects_total+$2,", + "shared_objects_preserved_total=shared_objects_preserved_total+$3,blocked_outcomes_total=blocked_outcomes_total+$4,", + "storage_driver_failures_total=storage_driver_failures_total+$5,last_processed=$6,last_deleted_references=$7,", + "last_deleted_backing_objects=$8,last_shared_objects_preserved=$9,last_blocked_outcomes=$10,", + "last_storage_driver_failures=$11,last_pass_status=$12,last_completed_at=$13,updated_at=$14 WHERE namespace=$15" + )) + .bind(report.deleted_references) + .bind(report.deleted_backing_objects) + .bind(report.shared_objects_preserved) + .bind(report.blocked) + .bind(report.storage_driver_failures) + .bind(report.processed) + .bind(report.deleted_references) + .bind(report.deleted_backing_objects) + .bind(report.shared_objects_preserved) + .bind(report.blocked) + .bind(report.storage_driver_failures) + .bind(status) + .bind(DB::bind_time(at)) + .bind(DB::bind_time(at)) + .bind(namespace) + .execute(&mut *tx) + .await?; + if report.storage_driver_failures > 0 { + Self::query("UPDATE runtime_external_payload_cleanup_stats SET last_storage_failure_at=$1 WHERE namespace=$2") + .bind(DB::bind_time(at)) + .bind(namespace) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(()) + } +} diff --git a/rust/tests/execution/external_payloads.rs b/rust/tests/execution/external_payloads.rs index 785bea3c8..040334c00 100644 --- a/rust/tests/execution/external_payloads.rs +++ b/rust/tests/execution/external_payloads.rs @@ -1,4 +1,6 @@ use super::*; +#[path = "external_payloads/cleanup.rs"] +mod cleanup; #[path = "external_payloads/completion_uploads.rs"] mod completion_uploads; use axum::{body::Bytes, http::HeaderMap}; @@ -125,7 +127,11 @@ async fn json_request( } async fn unclaimed(database: &TestDatabase) -> i64 { - let query = "SELECT COUNT(*) FROM runtime_external_payloads WHERE retained_at IS NULL AND expires_at IS NOT NULL"; + count(database, "SELECT COUNT(*) FROM runtime_external_payloads WHERE retained_at IS NULL AND expires_at IS NOT NULL").await +} + +/// One integer the database answers. +async fn count(database: &TestDatabase, query: &'static str) -> i64 { match database { TestDatabase::Sqlite(dir) => { let pool = sqlx::SqlitePool::connect(&format!( diff --git a/rust/tests/execution/external_payloads/cleanup.rs b/rust/tests/execution/external_payloads/cleanup.rs new file mode 100644 index 000000000..4b4e32bb0 --- /dev/null +++ b/rust/tests/execution/external_payloads/cleanup.rs @@ -0,0 +1,135 @@ +use super::*; +use sqlx::AssertSqlSafe; + +async fn upload_bytes(router: &Router, bytes: &[u8]) -> Value { + let (status, _, body) = request(router, "POST", "/api/external-payloads/v1", "worker-token", "default", + Some(&metadata(bytes)), Body::from(bytes.to_vec())).await; + assert_eq!(status, StatusCode::CREATED, "{}", String::from_utf8_lossy(&body)); + serde_json::from_slice::(&body).unwrap()["reference"].clone() +} + +fn id(reference: &Value) -> &str { + reference["reference_id"].as_str().unwrap() +} + +/// The local driver's content-addressed object path. +fn object(root: &std::path::Path, reference: &Value) -> std::path::PathBuf { + let sha = reference["sha256"].as_str().unwrap(); + root.join("avro").join(&sha[..2]).join(sha) +} + +async fn expire(database: &TestDatabase, ids: &[&str]) { + let list = ids.iter().map(|id| format!("'{id}'")).collect::>().join(","); + database.execute(AssertSqlSafe(format!( + "UPDATE runtime_external_payloads SET expires_at='2000-01-01 00:00:00' WHERE id IN ({list})"))).await; +} + +async fn pass(router: &Router, body: Value) -> (StatusCode, Value) { + json_request(router, "POST", "/api/system/external-payload-cleanup/pass", "admin-token", body).await +} + +async fn status(router: &Router) -> Value { + let (status, body) = json_request(router, "GET", "/api/system/external-payload-cleanup", "admin-token", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["namespace"], "default"); + body["cleanup"].clone() +} + +fn report(values: [i64; 7], backlog: i64, limit: i64, pressure: bool) -> Value { + json!({"processed": values[0], "deleted_references": values[1], "deleted_backing_objects": values[2], + "shared_objects_preserved": values[3], "skipped": values[4], "blocked": values[5], + "storage_driver_failures": values[6], "backlog": backlog, "scan_limit": limit, "scan_pressure": pressure}) +} + +#[tokio::test] +async fn cleanup_reclaims_expired_objects_and_keeps_shared_live_and_retained_ones() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("objects"); + let (database, _runtime, router) = setup(ExternalPayloadConfig::default(), &root).await; + let fresh = status(&router).await; + assert_eq!([&fresh["totals"]["passes"], &fresh["last_pass"]["status"]], [&json!(0), &Value::Null]); + let expired = upload_bytes(&router, b"expired object").await; + let shared = upload_bytes(&router, b"shared object").await; + let live = upload_bytes(&router, b"live object").await; + let retained = upload_bytes(&router, b"retained object").await; + // Another namespace's reference to the shared object. + database.execute(AssertSqlSafe(format!("INSERT INTO runtime_external_payloads(id,namespace,storage_uri,storage_uri_sha256,codec,sha256,size_bytes,upload_status,retained_at,expires_at,created_at,updated_at) SELECT 'zz-shared-twin','other',storage_uri,storage_uri_sha256,codec,sha256,size_bytes,upload_status,NULL,expires_at,created_at,updated_at FROM runtime_external_payloads WHERE id='{}'", id(&shared)))).await; + expire(&database, &[id(&expired), id(&shared), "zz-shared-twin", id(&retained)]).await; + database.execute(AssertSqlSafe(format!( + "UPDATE runtime_external_payloads SET retained_at='2026-01-01 00:00:00' WHERE id='{}'", id(&retained)))).await; + + // The pass reclaims only this namespace's references. The shared one + // loses its row and keeps the object the other namespace still holds. + let (code, body) = pass(&router, json!({})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + assert_eq!(body, json!({"namespace": "default", "cleanup": report([2, 2, 1, 1, 0, 0, 0], 0, 100, false)})); + assert!(!object(&root, &expired).exists()); + assert!(object(&root, &shared).exists()); + assert!(object(&root, &live).exists()); + assert!(object(&root, &retained).exists()); + assert_eq!(count(&database, "SELECT COUNT(*) FROM runtime_external_payloads").await, 3); + assert_eq!(count(&database, "SELECT COUNT(*) FROM runtime_external_payloads WHERE id='zz-shared-twin'").await, 1); + + let after = status(&router).await; + assert_eq!( + after["totals"], + json!({"passes": 1, "deleted_references": 2, "deleted_backing_objects": 1, "shared_objects_preserved": 1, + "blocked_outcomes": 0, "storage_driver_failures": 0}) + ); + assert_eq!([&after["last_pass"]["status"], &after["last_pass"]["processed"]], [&json!("completed"), &json!(2)]); + let completed = after["last_pass"]["completed_at"].as_str().unwrap(); + assert!(completed.ends_with(".000000Z"), "{completed}"); + assert_eq!(after["backlog"], json!({"expired_unclaimed": 0, "failed_registration_objects": 0})); +} + +#[tokio::test] +async fn cleanup_respects_its_limit_a_backup_hold_and_a_disabled_policy() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("objects"); + let (database, _runtime, router) = setup(ExternalPayloadConfig::default(), &root).await; + let first = upload_bytes(&router, b"first object").await; + let second = upload_bytes(&router, b"second object").await; + expire(&database, &[id(&first), id(&second)]).await; + assert_eq!(status(&router).await["backlog"]["expired_unclaimed"], 2); + + // A one-reference pass reports the rest as backlog under pressure. + let (code, body) = pass(&router, json!({"limit": "1"})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + assert_eq!(body["cleanup"], report([1, 1, 1, 0, 0, 0, 0], 1, 1, true)); + + // An unexpired online-backup hold pauses deletion: 207, object kept. + database.execute("INSERT INTO runtime_external_payload_backup_hold(id,owner,acquired_at,expires_at) VALUES (1,'7f1c3a1e-0000-4000-8000-000000000001','2026-01-01 00:00:00','2037-01-01 00:00:00')").await; + let (code, body) = pass(&router, json!({})).await; + assert_eq!(code, StatusCode::MULTI_STATUS, "{body}"); + assert_eq!(body["cleanup"], report([1, 0, 0, 0, 0, 1, 0], 1, 100, false)); + let remaining = if object(&root, &first).exists() { &first } else { &second }; + assert!(object(&root, remaining).exists()); + let held = status(&router).await; + assert_eq!([&held["last_pass"]["status"], &held["totals"]["blocked_outcomes"]], [&json!("blocked"), &json!(1)]); + + // With the policy disabled there is no driver to delete through. + database.execute("UPDATE runtime_external_payload_backup_hold SET expires_at='2000-01-01 00:00:00'").await; + assert_eq!(policy(&router, "default", &root, false).await, StatusCode::OK); + let (code, body) = pass(&router, json!({})).await; + assert_eq!(code, StatusCode::MULTI_STATUS, "{body}"); + assert_eq!(body["cleanup"]["blocked"], 1); + + // Once the hold has expired and the policy is back, the object goes. + assert_eq!(policy(&router, "default", &root, true).await, StatusCode::OK); + let (code, body) = pass(&router, json!({"limit": null})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + assert_eq!(body["cleanup"], report([1, 1, 1, 0, 0, 0, 0], 0, 100, false)); + assert!(!object(&root, remaining).exists()); + assert_eq!(status(&router).await["totals"]["passes"], 4); + + for (limit, message) in [ + (json!(0), "The limit field must be at least 1."), + (json!(1001), "The limit field must not be greater than 1000."), + (json!(2.5), "The limit field must be an integer."), + ] { + let (code, body) = pass(&router, json!({"limit": limit})).await; + assert_eq!(code, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert_eq!(body["errors"]["limit"], json!([message])); + } + assert_eq!(status(&router).await["totals"]["passes"], 4); +} diff --git a/rust/tests/execution/external_payloads/completion_uploads.rs b/rust/tests/execution/external_payloads/completion_uploads.rs index 2c4a80878..0a9591d9b 100644 --- a/rust/tests/execution/external_payloads/completion_uploads.rs +++ b/rust/tests/execution/external_payloads/completion_uploads.rs @@ -446,3 +446,30 @@ async fn activity_completion_during_upload_cannot_reopen_its_closed_allowance() runtime.close().await; database.remove().await; } + +#[tokio::test] +async fn cleanup_renews_a_live_completion_budget_and_drops_a_closed_one() { + let (_provider, database, _runtime, app, task, bytes) = original(120).await; + let uploaded = upload(&app, "worker-token", "default", &context(&task), bytes, None).await; + assert_eq!(uploaded.0, StatusCode::CREATED, "{}", uploaded.2); + assert_eq!(count(&database, "SELECT COUNT(*) FROM runtime_payload_completion_budgets").await, 1); + database.execute("UPDATE runtime_payload_completion_budgets SET expires_at='2000-01-01 00:00:00'").await; + let path = "/api/system/external-payload-cleanup/pass"; + let (code, body) = json_request(&app, "POST", path, "admin-token", json!({})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + // The activity can still complete, so its allowance gets another period. + assert_eq!( + count(&database, "SELECT COUNT(*) FROM runtime_payload_completion_budgets WHERE expires_at>'2001-01-01'").await, + 1 + ); + let (code, body) = json_request(&app, "POST", + &format!("/api/worker/activity-tasks/{}/complete", task["task_id"].as_str().unwrap()), "worker-token", + json!({"activity_attempt_id": task["activity_attempt_id"], "lease_owner": task["lease_owner"], + "result": envelope(Payload::Array(vec![]))})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + database.execute("UPDATE runtime_payload_completion_budgets SET expires_at='2000-01-01 00:00:00'").await; + let (code, body) = json_request(&app, "POST", path, "admin-token", json!({})).await; + assert_eq!(code, StatusCode::OK, "{body}"); + // A closed activity cannot resume its upload: the allowance is dropped. + assert_eq!(count(&database, "SELECT COUNT(*) FROM runtime_payload_completion_budgets").await, 0); +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index 03e8203bc..08932d03e 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -16,6 +16,7 @@ import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs'; 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 {checkNamespaceDelete} from './namespace-delete-contract.mjs'; import {checkSchedulePolicies} from './schedule-policies-contract.mjs'; import {checkScheduleCatalog} from './schedule-catalog-contract.mjs'; @@ -150,6 +151,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const serviceCatalog = fixture.service_catalog ? checkServiceCatalog(fixture, observation, workflowId) : null; 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 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; @@ -701,6 +703,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(serviceCatalog ? {service_catalog: serviceCatalog} : {}), ...(systemMetrics ? {system_metrics: systemMetrics} : {}), ...(prometheusMetrics ? {prometheus_metrics: prometheusMetrics} : {}), + ...(payloadCleanup ? {payload_cleanup: payloadCleanup} : {}), ...(namespaceDelete ? {namespace_delete: namespaceDelete} : {}), ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}), ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}), diff --git a/scripts/conformance/server-parity/payload-cleanup-contract.mjs b/scripts/conformance/server-parity/payload-cleanup-contract.mjs new file mode 100644 index 000000000..0240e61b2 --- /dev/null +++ b/scripts/conformance/server-parity/payload-cleanup-contract.mjs @@ -0,0 +1,32 @@ +import assert from 'node:assert/strict'; + +const WHOLE_SECOND = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.000000Z$/; + +// Reviewed external payload cleanup answers (SystemController:: +// externalPayloadCleanupStatus and externalPayloadCleanupPass). Every answer is +// exact except the namespace placeholder and a pass's whole-second time. +export function checkPayloadCleanup(fixture, observation, workflowId) { + const definition = fixture.payload_cleanup; + const state = observation.payload_cleanup; + const projection = {scope: 'payload_cleanup', cases: definition.requests.length}; + assert.ok(state && typeof state === 'object', 'actual payload cleanup observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_system_routes'}, 'embedded cleanup explicitly inapplicable'); + return projection; + } + const namespace = `${workflowId}-cleanup`.toLowerCase(); + assert.deepStrictEqual([state.applicable, state.namespace], [true, namespace], 'cleanup namespace'); + assert.deepStrictEqual(Object.keys(state.requests), definition.requests.map(([label]) => label), 'every request recorded in order'); + for (const [label] of definition.requests) { + const recorded = state.requests[label]; + const [status, body] = definition.expected[label]; + const completed = recorded.response?.cleanup?.last_pass?.completed_at; + if (body.cleanup?.last_pass?.completed_at === '@second') { + assert.match(String(completed), WHOLE_SECOND, `${label}: whole-second completion time`); + } + const expected = JSON.parse(JSON.stringify(body).split('"@own"').join(JSON.stringify(namespace)) + .split('"@second"').join(JSON.stringify(completed ?? null))); + assert.deepStrictEqual([recorded.status, recorded.response], [status, expected], label); + } + return projection; +} diff --git a/scripts/conformance/server-parity/payload-cleanup-probe.php b/scripts/conformance/server-parity/payload-cleanup-probe.php new file mode 100644 index 000000000..4be9fdb51 --- /dev/null +++ b/scripts/conformance/server-parity/payload-cleanup-probe.php @@ -0,0 +1,35 @@ +createNamespace($namespace, 'payload cleanup λ', 30); + $token = getenv('DW_PARITY_TOKEN') ?: null; + $http = new PayloadCleanupHttp(['base_uri' => rtrim($url, '/').'/api/system/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30]); + $raw = static function (string $method, string $path, mixed $body = null) use ($http, $token, $namespace): array { + $response = $http->request($method, $path, ['headers' => array_filter([ + 'Authorization' => $token === null ? null : 'Bearer '.$token, + 'Accept' => 'application/json', 'X-Namespace' => $namespace, + 'X-Durable-Workflow-Control-Plane-Version' => '2', + ]), ...($body === null ? [] : ['json' => $body])]); + + return ['status' => $response->getStatusCode(), + 'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)]; + }; + $state = ['applicable' => true, 'namespace' => $namespace, 'requests' => []]; + foreach ($definition['requests'] as [$label, $method, $path, $body]) { + $state['requests'][$label] = $raw($method, $path, $body === [] ? (object) [] : $body); + } + + return $state; +} diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 95cadcd7d..c5d205f5b 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -434,6 +434,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $serviceCatalog = isset($fixture['service_catalog']) ? finishHttpServiceCatalog($client, $fixture, $workflowId, $url) : null; $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; $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']) @@ -538,6 +539,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'service_catalog' => $serviceCatalog, 'system_metrics' => $systemMetrics, 'prometheus_metrics' => $prometheusMetrics, + 'payload_cleanup' => $payloadCleanup, 'namespace_delete' => $namespaceDelete, 'schedule_policies' => $schedulePolicies, 'schedule_catalog' => $scheduleCatalog, @@ -846,6 +848,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/service-catalog-probe.php'; require __DIR__.'/system-metrics-probe.php'; require __DIR__.'/prometheus-metrics-probe.php'; + require __DIR__.'/payload-cleanup-probe.php'; require __DIR__.'/namespace-delete-probe.php'; require __DIR__.'/schedule-policies-probe.php'; require __DIR__.'/signals-before-worker-probe.php'; @@ -1017,6 +1020,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['prometheus_metrics'])) { $observation['prometheus_metrics'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_metrics']; } + if ($mode === 'embedded' && isset($fixture['payload_cleanup'])) { + $observation['payload_cleanup'] = ['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/tests/Fixtures/ServerParity/payload-cleanup.json b/tests/Fixtures/ServerParity/payload-cleanup.json new file mode 100644 index 000000000..eb6612bd0 --- /dev/null +++ b/tests/Fixtures/ServerParity/payload-cleanup.json @@ -0,0 +1,391 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "payload-cleanup", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "payload cleanup λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "payload cleanup λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "payload_cleanup": { + "requests": [ + [ + "status fresh", + "GET", + "external-payload-cleanup", + null + ], + [ + "pass", + "POST", + "external-payload-cleanup/pass", + [] + ], + [ + "status after pass", + "GET", + "external-payload-cleanup", + null + ], + [ + "pass limit 5", + "POST", + "external-payload-cleanup/pass", + { + "limit": 5 + } + ], + [ + "pass limit 0", + "POST", + "external-payload-cleanup/pass", + { + "limit": 0 + } + ], + [ + "pass limit 1001", + "POST", + "external-payload-cleanup/pass", + { + "limit": 1001 + } + ], + [ + "pass limit text", + "POST", + "external-payload-cleanup/pass", + { + "limit": "x" + } + ], + [ + "pass limit fraction", + "POST", + "external-payload-cleanup/pass", + { + "limit": 2.5 + } + ], + [ + "pass limit null", + "POST", + "external-payload-cleanup/pass", + { + "limit": null + } + ], + [ + "pass limit numeric text", + "POST", + "external-payload-cleanup/pass", + { + "limit": "7" + } + ], + [ + "pass without body", + "POST", + "external-payload-cleanup/pass", + null + ], + [ + "status final", + "GET", + "external-payload-cleanup", + null + ] + ], + "expected": { + "status fresh": [ + 200, + { + "namespace": "@own", + "cleanup": { + "schema": "durable-workflow.v2.runtime-external-payload-cleanup-metrics.v1", + "backlog": { + "expired_unclaimed": 0, + "failed_registration_objects": 0 + }, + "bounds": { + "abandoned_upload_expiry_seconds": 3600, + "default_batch_size": 100, + "maximum_batch_size": 1000 + }, + "totals": { + "passes": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "blocked_outcomes": 0, + "storage_driver_failures": 0 + }, + "last_pass": { + "status": null, + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "blocked_outcomes": 0, + "storage_driver_failures": 0, + "completed_at": null, + "last_storage_failure_at": null + }, + "identity_policy": { + "provider_credentials_exposed": false, + "provider_locations_exposed": false, + "reusable_reference_identities_exposed": false + } + } + } + ], + "pass": [ + 200, + { + "namespace": "@own", + "cleanup": { + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "skipped": 0, + "blocked": 0, + "storage_driver_failures": 0, + "backlog": 0, + "scan_limit": 100, + "scan_pressure": false + } + } + ], + "status after pass": [ + 200, + { + "namespace": "@own", + "cleanup": { + "schema": "durable-workflow.v2.runtime-external-payload-cleanup-metrics.v1", + "backlog": { + "expired_unclaimed": 0, + "failed_registration_objects": 0 + }, + "bounds": { + "abandoned_upload_expiry_seconds": 3600, + "default_batch_size": 100, + "maximum_batch_size": 1000 + }, + "totals": { + "passes": 1, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "blocked_outcomes": 0, + "storage_driver_failures": 0 + }, + "last_pass": { + "status": "completed", + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "blocked_outcomes": 0, + "storage_driver_failures": 0, + "completed_at": "@second", + "last_storage_failure_at": null + }, + "identity_policy": { + "provider_credentials_exposed": false, + "provider_locations_exposed": false, + "reusable_reference_identities_exposed": false + } + } + } + ], + "pass limit 5": [ + 200, + { + "namespace": "@own", + "cleanup": { + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "skipped": 0, + "blocked": 0, + "storage_driver_failures": 0, + "backlog": 0, + "scan_limit": 5, + "scan_pressure": false + } + } + ], + "pass limit 0": [ + 422, + { + "message": "The limit field must be at least 1.", + "reason": "validation_failed", + "errors": { + "limit": [ + "The limit field must be at least 1." + ] + }, + "validation_errors": { + "limit": [ + "The limit field must be at least 1." + ] + } + } + ], + "pass limit 1001": [ + 422, + { + "message": "The limit field must not be greater than 1000.", + "reason": "validation_failed", + "errors": { + "limit": [ + "The limit field must not be greater than 1000." + ] + }, + "validation_errors": { + "limit": [ + "The limit field must not be greater than 1000." + ] + } + } + ], + "pass limit text": [ + 422, + { + "message": "The limit field must be an integer.", + "reason": "validation_failed", + "errors": { + "limit": [ + "The limit field must be an integer." + ] + }, + "validation_errors": { + "limit": [ + "The limit field must be an integer." + ] + } + } + ], + "pass limit fraction": [ + 422, + { + "message": "The limit field must be an integer.", + "reason": "validation_failed", + "errors": { + "limit": [ + "The limit field must be an integer." + ] + }, + "validation_errors": { + "limit": [ + "The limit field must be an integer." + ] + } + } + ], + "pass limit null": [ + 200, + { + "namespace": "@own", + "cleanup": { + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "skipped": 0, + "blocked": 0, + "storage_driver_failures": 0, + "backlog": 0, + "scan_limit": 100, + "scan_pressure": false + } + } + ], + "pass limit numeric text": [ + 200, + { + "namespace": "@own", + "cleanup": { + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "skipped": 0, + "blocked": 0, + "storage_driver_failures": 0, + "backlog": 0, + "scan_limit": 7, + "scan_pressure": false + } + } + ], + "pass without body": [ + 200, + { + "namespace": "@own", + "cleanup": { + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "skipped": 0, + "blocked": 0, + "storage_driver_failures": 0, + "backlog": 0, + "scan_limit": 100, + "scan_pressure": false + } + } + ], + "status final": [ + 200, + { + "namespace": "@own", + "cleanup": { + "schema": "durable-workflow.v2.runtime-external-payload-cleanup-metrics.v1", + "backlog": { + "expired_unclaimed": 0, + "failed_registration_objects": 0 + }, + "bounds": { + "abandoned_upload_expiry_seconds": 3600, + "default_batch_size": 100, + "maximum_batch_size": 1000 + }, + "totals": { + "passes": 5, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "blocked_outcomes": 0, + "storage_driver_failures": 0 + }, + "last_pass": { + "status": "completed", + "processed": 0, + "deleted_references": 0, + "deleted_backing_objects": 0, + "shared_objects_preserved": 0, + "blocked_outcomes": 0, + "storage_driver_failures": 0, + "completed_at": "@second", + "last_storage_failure_at": null + }, + "identity_policy": { + "provider_credentials_exposed": false, + "provider_locations_exposed": false, + "reusable_reference_identities_exposed": false + } + } + } + ] + } + }, + "contract": "GET /api/system/external-payload-cleanup and POST /api/system/external-payload-cleanup/pass (published Server 2.11) report and run the namespace's external payload cleanup. A fresh namespace has no expired unclaimed payloads and no recorded pass. Each valid pass reports what it processed, deleted, preserved, skipped and blocked, with its scan limit (default 100, at most 1000), and records a completed pass in the namespace's totals with its whole-second completion time. A limit that is not an integer from 1 to 1000 answers Laravel's 422 without recording a pass; a null or numeric-text limit is accepted. Embedded applications have no HTTP system routes." +} diff --git a/tests/Unit/ServerParityPayloadCleanupContractTest.mjs b/tests/Unit/ServerParityPayloadCleanupContractTest.mjs new file mode 100644 index 000000000..a5b4ce945 --- /dev/null +++ b/tests/Unit/ServerParityPayloadCleanupContractTest.mjs @@ -0,0 +1,59 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import {checkPayloadCleanup} from '../../scripts/conformance/server-parity/payload-cleanup-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/payload-cleanup.json', import.meta.url))); +const definition = fixture.payload_cleanup; +const id = 'parity-v1-payload-cleanup'; +const own = `${id}-cleanup`; + +const observation = () => { + const requests = {}; + for (const [label] of definition.requests) { + const [status, body] = definition.expected[label]; + requests[label] = {status, response: JSON.parse(JSON.stringify(body).split('"@own"').join(JSON.stringify(own)) + .split('"@second"').join('"2026-10-22T23:08:38.000000Z"'))}; + } + return {mode: 'http', payload_cleanup: {applicable: true, namespace: own, requests}}; +}; + +test('the payload cleanup accepts the published answers', () => { + assert.deepStrictEqual(checkPayloadCleanup(fixture, observation(), id), {scope: 'payload_cleanup', cases: 12}); +}); + +test('the fixture covers status, passes and limit validation', () => { + const statuses = Object.values(definition.expected).map(([status]) => status); + assert.deepStrictEqual([...new Set(statuses)].sort(), [200, 422]); + assert.equal(definition.expected['status final'][1].cleanup.totals.passes, 5); + assert.equal(definition.expected['status fresh'][1].cleanup.last_pass.status, null); +}); + +test('the embedded payload cleanup is explicitly inapplicable', () => { + checkPayloadCleanup(fixture, {mode: 'embedded', payload_cleanup: {applicable: false, reason: 'embedded_has_no_http_system_routes'}}, id); + assert.throws(() => checkPayloadCleanup(fixture, {mode: 'embedded', payload_cleanup: {applicable: true}}, id)); +}); + +const response = (state, label) => state.requests[label].response; +for (const [label, mutate] of [ + ['an unimplemented route', state => { state.requests.pass = {status: 501, response: {reason: 'rust_capability_not_implemented'}}; }], + ['a missing request', state => { delete state.requests['pass limit 5']; }], + ['another namespace', state => { response(state, 'pass').namespace = 'default'; }], + ['a pass that is not recorded', state => { response(state, 'status after pass').cleanup.totals.passes = 0; }], + ['a refused pass counted', state => { response(state, 'status final').cleanup.totals.passes = 9; }], + ['a millisecond completion time', state => { response(state, 'status final').cleanup.last_pass.completed_at = '2026-10-22T23:08:38.123000Z'; }], + ['a missing completion time', state => { response(state, 'status after pass').cleanup.last_pass.completed_at = null; }], + ['an ignored limit', state => { response(state, 'pass limit 5').cleanup.scan_limit = 100; }], + ['an accepted fractional limit', state => { state.requests['pass limit fraction'] = {status: 200, response: {}}; }], + ['a refused null limit', state => { state.requests['pass limit null'].status = 422; }], + ['a min message for the maximum', state => { response(state, 'pass limit 1001').message = 'The limit field must be at least 1.'; }], + ['a pass that reports pressure', state => { response(state, 'pass').cleanup.scan_pressure = true; }], + ['another expiry bound', state => { response(state, 'status fresh').cleanup.bounds.abandoned_upload_expiry_seconds = 60; }], + ['an exposed storage location', state => { response(state, 'status fresh').cleanup.identity_policy.provider_locations_exposed = true; }], +]) { + test(`the payload cleanup rejects ${label}`, () => { + const value = observation(); + mutate(value.payload_cleanup); + assert.throws(() => checkPayloadCleanup(fixture, value, id)); + }); +} From 694656581e8b859cc55568899f571f56a8e4511c Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 23:40:33 +0000 Subject: [PATCH 2/3] Retry the SQLite reference's retryable refusals in three probes Locally, published PHP on SQLite refused a namespace delete with its retryable 503 backend_lock_pressure in two of four parity runs, and a storage diagnostic with a retryable 503 in two. The namespace-delete and payload-cleanup probes now retry lock pressure through the shared middleware, and the storage diagnostic retries a retryable 503 up to ten times. --- .../server-parity/namespace-delete-probe.php | 7 ++++++- .../server-parity/payload-cleanup-probe.php | 5 ++++- .../server-parity/storage-test-probe.php | 15 +++++++++++++-- 3 files changed, 23 insertions(+), 4 deletions(-) diff --git a/scripts/conformance/server-parity/namespace-delete-probe.php b/scripts/conformance/server-parity/namespace-delete-probe.php index e43e212a1..9fe562b7f 100644 --- a/scripts/conformance/server-parity/namespace-delete-probe.php +++ b/scripts/conformance/server-parity/namespace-delete-probe.php @@ -5,6 +5,7 @@ use DurableWorkflow\Client; use DurableWorkflow\Exception\ServerException; use GuzzleHttp\Client as NamespaceDeleteHttp; +use GuzzleHttp\HandlerStack; // Deleting a namespace that holds a search attribute definition and a started, // unclaimed run, through the published SDK; then the namespace's reads, a @@ -16,8 +17,12 @@ function finishHttpNamespaceDelete(Client $admin, array $fixture, string $workfl $namespace = strtolower($workflowId.'-delete'); $doomed = $workflowId.'-doomed'; $token = getenv('DW_PARITY_TOKEN') ?: null; + // The published SQLite reference may refuse a delete with its retryable + // 503 backend_lock_pressure while its maintenance loop holds the file. + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $http = new NamespaceDeleteHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, - 'connect_timeout' => 1, 'timeout' => 30]); + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); $raw = static function (string $method, string $path, string $scope, ?array $body = null) use ($http, $token): array { $response = $http->request($method, $path, ['headers' => array_filter([ 'Authorization' => $token === null ? null : 'Bearer '.$token, diff --git a/scripts/conformance/server-parity/payload-cleanup-probe.php b/scripts/conformance/server-parity/payload-cleanup-probe.php index 4be9fdb51..2a7df99aa 100644 --- a/scripts/conformance/server-parity/payload-cleanup-probe.php +++ b/scripts/conformance/server-parity/payload-cleanup-probe.php @@ -4,6 +4,7 @@ use DurableWorkflow\Client; use GuzzleHttp\Client as PayloadCleanupHttp; +use GuzzleHttp\HandlerStack; // The external payload cleanup status and pass of a fresh namespace over raw // HTTP: the status before and after passes, each pass's report, and the @@ -14,8 +15,10 @@ function finishHttpPayloadCleanup(Client $admin, array $fixture, string $workflo $namespace = strtolower($workflowId.'-cleanup'); $admin->createNamespace($namespace, 'payload cleanup λ', 30); $token = getenv('DW_PARITY_TOKEN') ?: null; + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $http = new PayloadCleanupHttp(['base_uri' => rtrim($url, '/').'/api/system/', 'http_errors' => false, - 'connect_timeout' => 1, 'timeout' => 30]); + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); $raw = static function (string $method, string $path, mixed $body = null) use ($http, $token, $namespace): array { $response = $http->request($method, $path, ['headers' => array_filter([ 'Authorization' => $token === null ? null : 'Bearer '.$token, diff --git a/scripts/conformance/server-parity/storage-test-probe.php b/scripts/conformance/server-parity/storage-test-probe.php index bb61bfcbf..de276b772 100644 --- a/scripts/conformance/server-parity/storage-test-probe.php +++ b/scripts/conformance/server-parity/storage-test-probe.php @@ -42,9 +42,20 @@ function finishHttpStorageTest(Client $admin, array $fixture, string $workflowId foreach ($definition['invalid'] as $body) { $state['invalid'][] = $raw('POST', 'storage/test', $body === [] ? (object) [] : $body); } - $state['passed'] = $raw('POST', 'storage/test', $sizes); + // The published SQLite reference can refuse a round trip with a retryable + // 503 while its maintenance loop holds the file; a later attempt succeeds. + $roundTrip = static function () use ($raw, $sizes): array { + for ($attempt = 1; ; $attempt++) { + $recorded = $raw('POST', 'storage/test', $sizes); + if ($attempt === 10 || $recorded['status'] !== 503 || ($recorded['response']['retryable'] ?? null) !== true) { + return $recorded; + } + usleep(500_000); + } + }; + $state['passed'] = $roundTrip(); // A repeated diagnostic writes and removes the same content-addressed objects. - $state['repeated'] = $raw('POST', 'storage/test', $sizes); + $state['repeated'] = $roundTrip(); return $state; } From 236a5b7575762f59746c5951c037f11c67bc5023 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 23:40:33 +0000 Subject: [PATCH 3/3] Record the payload cleanup evidence --- .../2026-10-22-payload-cleanup.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md b/docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md index 856cbf350..f5c962e8c 100644 --- a/docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md +++ b/docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md @@ -41,3 +41,10 @@ SQLite, PostgreSQL and MySQL: shared, live and retained references, the limit and its backlog, an active backup hold, a disabled policy, and an allowance renewed while its activity runs and dropped after it completes. +- **Evidence.** The new tests pass on SQLite, PostgreSQL and MySQL, and the + native suite with `--all-targets` passes on SQLite (652 tests). A local + copy of the SQLite parity step passed all 54 fixtures on embedded PHP and + native, including `payload-cleanup`. The published PHP fixture answers + were checked against a fresh 2.11.1 server. Its record differed only on + `namespace-delete`, a retryable 503 `backend_lock_pressure` that its probe + did not retry; that probe and the storage diagnostic now retry it.