Serve external payload cleanup and its pass like published PHP 2.11

.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

docs/rust-server-port-checkpoints/2026-10-22-payload-cleanup.md

@@ -0,0 +1,50 @@
+# 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.
+- **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.

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.

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,

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<bool> {
+        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<Option<Rows<DB::Row>>> {
         let Some(attempt) = Self::query("SELECT * FROM activity_attempts WHERE id=$1")
             .bind(attempt_id)

rust/src/runtime/execution.rs

@@ -341,6 +341,18 @@ impl Runtime {
     delegate!(repair_status() -> Result<Value>);
     delegate!(system_metrics() -> Result<Value>);
     delegate!(prometheus_metrics() -> Result<Value>);
+    delegate!(payload_cleanup_status(config: &super::operator_metrics::MetricsConfig) -> Result<Value>);
+    /// 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<Value> {
+        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<Value>);
     delegate!(retention_pass(limit: i64, run_ids: Vec<String>) -> Result<(Value, bool)>);
     delegate!(activity_timeout_status(limit: i64) -> Result<Value>);

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<Runtime>) -> Result<Js
     Ok(Json(runtime.prometheus_metrics().await?))
 }
 
+async fn payload_cleanup_status(Extension(runtime): Extension<Runtime>) -> Result<Json<Value>> {
+    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<Runtime>, body: axum::body::Bytes) -> Result<Response> {
+    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<Runtime>) -> Result<Response> {
     let (unavailable, body) = runtime.system_health(&metrics_config(&runtime)).await?;

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;

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<Value> {
+    pub(super) async fn external_payload_cleanup_metrics(&self, config: &MetricsConfig) -> Result<Value> {
         let bound = whole_second(now());
         let stats =
             Self::query("SELECT * FROM runtime_external_payload_cleanup_stats WHERE namespace=$1")