Delete a namespace and everything it owns, like PHP

.github/workflows/server-parity.yml

@@ -34,6 +34,7 @@ on:
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityOperatorMetricsContractTest.mjs
+      - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
       - tests/Unit/ServerParityPublishedPatchContractTest.mjs
@@ -81,6 +82,7 @@ on:
       - tests/Unit/ServerParityWorkerListingContractTest.mjs
       - tests/Unit/ServerParityWorkerReplacementContractTest.mjs
       - tests/Unit/ServerParityOperatorMetricsContractTest.mjs
+      - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
       - tests/Unit/ServerParityPublishedPatchContractTest.mjs
@@ -1470,7 +1472,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 > 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/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

rust/src/runtime/execution.rs

@@ -359,6 +359,7 @@ impl Runtime {
     delegate!(delete_search_attribute(name: &str) -> Result<Value>);
     delegate!(set_external_policy(name: &str, body: Value) -> Result<Value>);
     delegate!(update_namespace(name: &str, body: Value) -> Result<Value>);
+    delegate!(delete_namespace(name: &str) -> Result<Value>);
     delegate!(
         upload_external_local(
             local: crate::external_payload::LocalStorage,

rust/src/runtime/http.rs

@@ -32,7 +32,7 @@ pub fn router(runtime: Runtime) -> Router {
         )
         .route(
             "/api/namespaces/{namespace}",
-            get(describe_namespace).put(update_namespace),
+            get(describe_namespace).put(update_namespace).delete(delete_namespace),
         )
         .route(
             "/api/namespaces/{namespace}/external-storage",
@@ -411,8 +411,8 @@ async fn guard(State(runtime): State<Runtime>, mut request: Request, next: Next)
     } else if path.starts_with("/api/workers/") && request.method() != Method::GET
         || path == "/api/namespaces" && request.method() == Method::POST
         || path == "/api/storage/test"
-        // Updating a namespace or its external storage needs an admin.
-        || path.starts_with("/api/namespaces/") && request.method() == Method::PUT
+        // Updating or deleting a namespace, or its external storage, needs an admin.
+        || path.starts_with("/api/namespaces/") && matches!(*request.method(), Method::PUT | Method::DELETE)
         // PHP's system routes are admin-only.
         || path.starts_with("/api/system/")
         || path == "/api/runtime-credentials"
@@ -721,6 +721,12 @@ async fn update_namespace(
 ) -> Result<Json<Value>> {
     Ok(Json(runtime.update_namespace(&namespace, body).await?))
 }
+async fn delete_namespace(
+    Extension(runtime): Extension<Runtime>,
+    Path(namespace): Path<String>,
+) -> Result<Json<Value>> {
+    Ok(Json(runtime.delete_namespace(&namespace).await?))
+}
 async fn create_namespace(
     Extension(runtime): Extension<Runtime>,
     Json(body): Json<Value>,

rust/src/runtime/mod.rs

@@ -2,6 +2,27 @@
 //! arguments use the official codec on a bounded blocking worker.
 //! Typed database adapters share transitions; existing PHP data requires qualification.
 
+/// Store methods for every backend: the bounds the shared SQL helpers need,
+/// written once rather than in each module.
+macro_rules! store_methods {
+    ($($body:tt)*) => {
+        impl<DB: $crate::runtime::backend::Backend> $crate::runtime::store::Store<DB>
+        where
+            for<'c> &'c mut DB::Connection: ::sqlx::Executor<'c, Database = DB>,
+            DB::Arguments: ::sqlx::IntoArguments<DB>,
+            for<'q> &'q str: ::sqlx::Encode<'q, DB> + ::sqlx::Type<DB>,
+            for<'q> Option<&'q str>: ::sqlx::Encode<'q, DB>,
+            for<'q> Option<DB::EncodedDocument>: ::sqlx::Encode<'q, DB>,
+            for<'q> String: ::sqlx::Encode<'q, DB> + ::sqlx::Type<DB>,
+            for<'q> i64: ::sqlx::Encode<'q, DB> + ::sqlx::Type<DB>,
+            for<'r> i64: ::sqlx::Decode<'r, DB>,
+            usize: ::sqlx::ColumnIndex<DB::Row>,
+        {
+            $($body)*
+        }
+    };
+}
+
 mod active_summary;
 mod activity_completion_uploads;
 mod activity_deadlines;
@@ -42,6 +63,7 @@ mod memo_json;
 mod memo_limits;
 mod memos;
 pub mod mysql;
+mod namespace_cleanup;
 mod namespaces;
 mod operator_dashboard;
 mod operator_metrics;

rust/src/runtime/namespace_cleanup.rs

@@ -0,0 +1,215 @@
+//! `DELETE /api/namespaces/{namespace}`: PHP's NamespaceController::destroy
+//! over NamespaceLifecycleCleanup. One transaction deletes every row the
+//! namespace owns in PHP's order and reports the non-zero counts under PHP's
+//! keys. PHP checks each table's columns first; the statements here use the
+//! columns the verified schema has, so a table without a run or instance
+//! column is matched by the other, as PHP's column check does.
+//!
+//! Native does not delete external payload objects yet. A namespace that
+//! still references any refuses before changing anything, with PHP's answer
+//! for unavailable external payload storage.
+use super::retention::reference;
+use super::{Result, RuntimeError};
+use axum::http::StatusCode;
+use serde_json::{Map, Value, json};
+
+/// The namespace's runs, instances and tasks, as subqueries. Each takes the
+/// namespace in its own numbered placeholder.
+macro_rules! runs {
+    ($n:literal) => {
+        concat!("SELECT id FROM workflow_runs WHERE namespace=", $n)
+    };
+}
+macro_rules! instances {
+    ($a:literal, $b:literal) => {
+        concat!(
+            "SELECT id FROM workflow_instances WHERE namespace=", $a,
+            " UNION SELECT workflow_instance_id FROM workflow_runs WHERE namespace=", $b
+        )
+    };
+}
+macro_rules! tasks {
+    ($a:literal, $b:literal) => {
+        concat!(
+            "SELECT id FROM workflow_tasks WHERE namespace=", $a,
+            " UNION SELECT id FROM workflow_tasks WHERE workflow_run_id IN (", runs!($b), ")"
+        )
+    };
+}
+/// A table matched by its run and instance columns.
+macro_rules! by_run_or_instance {
+    ($table:literal) => {
+        (
+            $table,
+            concat!(
+                "DELETE FROM ", $table, " WHERE workflow_run_id IN (", runs!("$1"),
+                ") OR workflow_instance_id IN (", instances!("$2", "$3"), ")"
+            ),
+            3,
+        )
+    };
+}
+/// A table with only a run column.
+macro_rules! by_run {
+    ($table:literal) => {
+        ($table, concat!("DELETE FROM ", $table, " WHERE workflow_run_id IN (", runs!("$1"), ")"), 1)
+    };
+}
+macro_rules! by_instance_or_run {
+    ($table:literal, $instance:literal, $run:literal) => {
+        (
+            $table,
+            concat!(
+                "DELETE FROM ", $table, " WHERE ", $instance, " IN (", instances!("$1", "$2"),
+                ") OR ", $run, " IN (", runs!("$3"), ")"
+            ),
+            3,
+        )
+    };
+}
+macro_rules! by_namespace {
+    ($table:literal) => {
+        ($table, concat!("DELETE FROM ", $table, " WHERE namespace=$1"), 1)
+    };
+}
+
+/// PHP's report key, statement and namespace binds, in PHP's order.
+const CLEANUP: &[(&str, &str, usize)] = &[
+    by_namespace!("runtime_external_payload_cleanup_stats"),
+    (
+        "activity_attempts",
+        concat!(
+            "DELETE FROM activity_attempts WHERE workflow_run_id IN (", runs!("$1"),
+            ") OR workflow_task_id IN (", tasks!("$2", "$3"), ")"
+        ),
+        3,
+    ),
+    by_instance_or_run!("workflow_updates", "workflow_instance_id", "workflow_run_id"),
+    by_instance_or_run!("workflow_signal_records", "workflow_instance_id", "workflow_run_id"),
+    by_instance_or_run!("workflow_messages", "workflow_instance_id", "workflow_run_id"),
+    by_instance_or_run!("workflow_child_calls", "parent_workflow_instance_id", "parent_workflow_run_id"),
+    by_instance_or_run!("workflow_links", "parent_workflow_instance_id", "parent_workflow_run_id"),
+    (
+        "workflow_service_calls",
+        "DELETE FROM workflow_service_calls WHERE namespace=$1 OR target_namespace=$2",
+        2,
+    ),
+    // A caller namespace's context is cleared from the calls that remain, so
+    // a namespace recreated with the same name does not inherit them.
+    (
+        "workflow_service_call_caller_contexts",
+        concat!(
+            "UPDATE workflow_service_calls SET caller_namespace=NULL,caller_workflow_instance_id=NULL,",
+            "caller_workflow_run_id=NULL,caller_principal_subject=NULL,caller_principal_method=NULL,",
+            "caller_principal_roles=NULL,caller_principal_tenant=NULL,caller_principal_claims=NULL ",
+            "WHERE caller_namespace=$1 OR caller_workflow_instance_id IN (", instances!("$2", "$3"),
+            ") OR caller_workflow_run_id IN (", runs!("$4"), ")"
+        ),
+        4,
+    ),
+    by_run!("activity_executions"),
+    by_run_or_instance!("workflow_commands"),
+    by_run!("workflow_failures"),
+    by_run!("workflow_history_events"),
+    by_run_or_instance!("workflow_memos"),
+    by_run_or_instance!("workflow_run_lineage_entries"),
+    by_run_or_instance!("workflow_run_timer_entries"),
+    by_run_or_instance!("workflow_run_timeline_entries"),
+    by_run_or_instance!("workflow_run_waits"),
+    by_run!("workflow_run_timers"),
+    by_run_or_instance!("workflow_search_attributes"),
+    by_run!("workflow_tasks"),
+    by_namespace!("workflow_schedule_history_events"),
+    by_namespace!("workflow_schedules"),
+    by_namespace!("workflow_service_operations"),
+    by_namespace!("workflow_services"),
+    by_namespace!("workflow_service_endpoints"),
+    by_namespace!("workflow_worker_compatibility_heartbeats"),
+    by_namespace!("workflow_worker_sessions"),
+    by_namespace!("workflow_worker_build_id_rollouts"),
+    by_namespace!("workflow_worker_registrations"),
+    by_namespace!("search_attribute_definitions"),
+    by_namespace!("workflow_durable_stream_items"),
+    by_namespace!("workflow_durable_streams"),
+    by_namespace!("workflow_inbound_stream_items"),
+    by_namespace!("workflow_inbound_streams"),
+    // Run summaries have no run column; PHP matches them by instance.
+    (
+        "workflow_run_summaries",
+        concat!("DELETE FROM workflow_run_summaries WHERE workflow_instance_id IN (", instances!("$1", "$2"), ")"),
+        2,
+    ),
+    by_namespace!("workflow_runs"),
+    by_namespace!("workflow_instances"),
+];
+
+/// Native's own rows for the namespace. PHP has no such tables, so they are
+/// not reported. Task-keyed receipts cascade with their tasks.
+const NATIVE_CLEANUP: &[&str] = &[
+    "DELETE FROM dw_poll_receipts WHERE namespace=$1",
+    "DELETE FROM dw_worker_registration_incarnations WHERE namespace=$1",
+];
+
+/// Registered external payloads and payload columns holding a reference.
+const EXTERNAL_PAYLOADS: &str = concat!(
+    "SELECT (SELECT COUNT(*) FROM runtime_external_payloads WHERE namespace=$1)",
+    "+(SELECT COUNT(*) FROM workflow_runs WHERE namespace=$2 AND (", reference!("arguments"), " OR ", reference!("output"), "))",
+    "+(SELECT COUNT(*) FROM workflow_commands WHERE workflow_run_id IN (", runs!("$3"), ") AND ", reference!("payload"), ")",
+    "+(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id IN (", runs!("$4"), ") AND ", reference!("arguments"), ")",
+    "+(SELECT COUNT(*) FROM workflow_updates WHERE workflow_run_id IN (", runs!("$5"), ") AND (", reference!("arguments"), " OR ", reference!("result"), "))",
+    "+(SELECT COUNT(*) FROM activity_executions WHERE workflow_run_id IN (", runs!("$6"), ") AND (", reference!("arguments"), " OR ", reference!("result"), "))",
+);
+
+pub(super) fn not_found(name: &str) -> RuntimeError {
+    RuntimeError::Protocol {
+        status: StatusCode::NOT_FOUND,
+        response: json!({"message": format!("Namespace [{name}] not found."), "reason": "namespace_not_found", "namespace": name}),
+    }
+}
+
+store_methods! {
+    pub(super) async fn delete_namespace(&self, name: &str) -> Result<Value> {
+        let name = name.to_ascii_lowercase();
+        let mut tx = self.begin().await?;
+        if Self::scalar("SELECT COUNT(*) FROM workflow_namespaces WHERE name=$1")
+            .bind(&name)
+            .fetch_one(&mut *tx)
+            .await?
+            == 0
+        {
+            return Err(not_found(&name));
+        }
+        let mut external = Self::scalar(EXTERNAL_PAYLOADS);
+        for _ in 0..6 {
+            external = external.bind(&name);
+        }
+        if external.fetch_one(&mut *tx).await? > 0 {
+            return Err(RuntimeError::Protocol {
+                status: StatusCode::SERVICE_UNAVAILABLE,
+                response: json!({
+                    "message": "Namespace cleanup requires deleting external payloads, but external payload storage is unavailable.",
+                    "reason": "external_payload_storage_driver_unavailable",
+                    "namespace": name,
+                    "error": "The native server does not delete external payload objects yet.",
+                }),
+            });
+        }
+        let mut deleted = Map::new();
+        for (key, sql, binds) in CLEANUP {
+            let mut query = Self::query(sql);
+            for _ in 0..*binds {
+                query = query.bind(&name);
+            }
+            let count = DB::affected(query.execute(&mut *tx).await?);
+            if count > 0 {
+                deleted.insert((*key).into(), json!(count));
+            }
+        }
+        for sql in NATIVE_CLEANUP {
+            Self::query(sql).bind(&name).execute(&mut *tx).await?;
+        }
+        Self::query("DELETE FROM workflow_namespaces WHERE name=$1").bind(&name).execute(&mut *tx).await?;
+        tx.commit().await?;
+        Ok(json!({"name": name, "status": "deleted", "deleted": deleted}))
+    }
+}

rust/src/runtime/retention.rs

@@ -61,6 +61,8 @@ macro_rules! reference {
     };
 }
 
+pub(super) use reference;
+
 /// 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!(