From efb13cb09a36feeb5ff28a10243ba91524bb8184 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 20:16:34 +0000 Subject: [PATCH 1/2] Delete a namespace and everything it owns, like PHP DELETE /api/namespaces/{namespace} (dw namespace:delete, the SDK's deleteNamespace) follows PHP's NamespaceController::destroy over NamespaceLifecycleCleanup. One transaction deletes the namespace's rows in PHP's order, matching each table by the run, instance or namespace columns the verified schema has, and reports each table's non-zero count under PHP's keys. Native also clears its own poll receipts and worker incarnations for the namespace. A missing namespace answers PHP's 404 namespace_not_found; deleting needs an admin, like updating. Native does not delete external payload objects yet, so a namespace that still references one refuses with PHP's 503 external_payload_storage_driver_unavailable before changing anything. The new reviewed namespace-delete fixture runs through the unchanged published PHP SDK: one search attribute definition and one started run, the exact counts published PHP 2.5.14 reports, the 404s afterwards, the SDK's typed refusal of a second delete, an untouched run elsewhere and the freed workflow ID. PHP's asynchronously filled timeline, timer and lineage projections may be counted. Embedded has no HTTP namespaces. --- .github/workflows/server-parity.yml | 4 +- rust/src/runtime/execution.rs | 1 + rust/src/runtime/http.rs | 12 +- rust/src/runtime/mod.rs | 1 + rust/src/runtime/namespace_cleanup.rs | 227 ++++++++++++++++++ rust/src/runtime/retention.rs | 2 + rust/tests/execution/namespaces.rs | 63 +++++ .../conformance/server-parity/contract.mjs | 3 + .../namespace-delete-contract.mjs | 42 ++++ .../server-parity/namespace-delete-probe.php | 53 ++++ scripts/conformance/server-parity/probe.php | 6 + .../ServerParity/namespace-delete.json | 29 +++ ...erverParityNamespaceDeleteContractTest.mjs | 52 ++++ 13 files changed, 491 insertions(+), 4 deletions(-) create mode 100644 rust/src/runtime/namespace_cleanup.rs create mode 100644 scripts/conformance/server-parity/namespace-delete-contract.mjs create mode 100644 scripts/conformance/server-parity/namespace-delete-probe.php create mode 100644 tests/Fixtures/ServerParity/namespace-delete.json create mode 100644 tests/Unit/ServerParityNamespaceDeleteContractTest.mjs diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 1c6e559c1..7b440a302 100644 --- a/.github/workflows/server-parity.yml +++ b/.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 diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs index 7fd36531f..c13573c2d 100644 --- a/rust/src/runtime/execution.rs +++ b/rust/src/runtime/execution.rs @@ -359,6 +359,7 @@ impl Runtime { delegate!(delete_search_attribute(name: &str) -> Result); delegate!(set_external_policy(name: &str, body: Value) -> Result); delegate!(update_namespace(name: &str, body: Value) -> Result); + delegate!(delete_namespace(name: &str) -> Result); delegate!( upload_external_local( local: crate::external_payload::LocalStorage, diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs index b6bb73eef..1c105139d 100644 --- a/rust/src/runtime/http.rs +++ b/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, 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> { Ok(Json(runtime.update_namespace(&namespace, body).await?)) } +async fn delete_namespace( + Extension(runtime): Extension, + Path(namespace): Path, +) -> Result> { + Ok(Json(runtime.delete_namespace(&namespace).await?)) +} async fn create_namespace( Extension(runtime): Extension, Json(body): Json, diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index 8723ac1ff..bfe81d629 100644 --- a/rust/src/runtime/mod.rs +++ b/rust/src/runtime/mod.rs @@ -42,6 +42,7 @@ mod memo_json; mod memo_limits; mod memos; pub mod mysql; +mod namespace_cleanup; mod namespaces; mod operator_dashboard; mod operator_metrics; diff --git a/rust/src/runtime/namespace_cleanup.rs b/rust/src/runtime/namespace_cleanup.rs new file mode 100644 index 000000000..ec8939747 --- /dev/null +++ b/rust/src/runtime/namespace_cleanup.rs @@ -0,0 +1,227 @@ +//! `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, backend::Backend, store::Store}; +use axum::http::StatusCode; +use serde_json::{Map, Value, json}; +use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; + +/// 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}), + } +} + +impl Store +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, + for<'q> &'q str: Encode<'q, DB> + Type, + for<'q> Option<&'q str>: Encode<'q, DB>, + for<'q> Option: Encode<'q, DB>, + for<'q> String: Encode<'q, DB> + Type, + for<'q> i64: Encode<'q, DB> + Type, + for<'r> i64: Decode<'r, DB>, + usize: ColumnIndex, +{ + pub(super) async fn delete_namespace(&self, name: &str) -> Result { + 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})) + } +} diff --git a/rust/src/runtime/retention.rs b/rust/src/runtime/retention.rs index d331102c1..9e888fde5 100644 --- a/rust/src/runtime/retention.rs +++ b/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!( diff --git a/rust/tests/execution/namespaces.rs b/rust/tests/execution/namespaces.rs index c63637d7b..933665b6a 100644 --- a/rust/tests/execution/namespaces.rs +++ b/rust/tests/execution/namespaces.rs @@ -749,3 +749,66 @@ async fn a_named_namespace_child_runs_and_resolves_in_its_parents_namespace() { let (_, described) = scoped(&app, "alpha", "GET", "/api/workflows/named-parent", Value::Null).await; assert_eq!(described["status"], "completed", "{described}"); } + +// `DELETE /api/namespaces/{namespace}` follows PHP's NamespaceController::destroy: +// the namespace's rows go in one transaction, the non-zero counts are reported, +// other namespaces keep theirs, and a missing namespace answers 404. +#[tokio::test] +async fn namespace_delete_removes_only_what_the_namespace_owns() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + create(&app, "doomed").await; + let (status, body) = scoped(&app, "doomed", "POST", "/api/search-attributes", + json!({"name":"CustomerTier","type":"keyword"})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + start_scoped(&app, "doomed", "namespace-delete-run").await; + start_scoped(&app, "default", "namespace-delete-peer").await; + let (status, body) = scoped(&app, "default", "DELETE", "/api/namespaces/DOOMED", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + // What published PHP 2.5.14 reports for one started run and one search + // attribute definition. + assert_eq!(body, json!({"name":"doomed","status":"deleted","deleted":{ + "workflow_commands":1,"workflow_history_events":2,"workflow_tasks":1, + "search_attribute_definitions":1,"workflow_run_summaries":1,"workflow_runs":1,"workflow_instances":1}})); + for (method, uri) in [("GET", "/api/namespaces/doomed"), ("DELETE", "/api/namespaces/doomed")] { + let (status, body) = scoped(&app, "default", method, uri, Value::Null).await; + assert_eq!(status, StatusCode::NOT_FOUND, "{method} {uri}: {body}"); + assert_eq!(body, json!({"message":"Namespace [doomed] not found.","reason":"namespace_not_found","namespace":"doomed"})); + } + let (status, body) = scoped(&app, "default", "GET", "/api/workflows/namespace-delete-peer", Value::Null).await; + assert_eq!(status, StatusCode::OK, "another namespace keeps its run: {body}"); + // The deleted run's workflow id is free again. + let restarted = start_scoped(&app, "default", "namespace-delete-run").await; + assert_eq!(restarted["outcome"], "started_new", "{restarted}"); + runtime.close().await; + database.remove().await; +} + +// Native does not delete external payload objects yet, so a namespace that +// still references one refuses, as PHP does when its storage is unavailable, +// and nothing is deleted. +#[tokio::test] +async fn namespace_delete_refuses_while_external_payloads_remain() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + create(&app, "held").await; + start_scoped(&app, "held", "namespace-delete-held").await; + database + .execute("UPDATE workflow_runs SET output='{\"reference\":\"dw-external-payload:v1:held\"}' WHERE namespace='held'") + .await; + let (status, body) = scoped(&app, "default", "DELETE", "/api/namespaces/held", Value::Null).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); + assert_eq!(body["reason"], "external_payload_storage_driver_unavailable"); + assert_eq!(body["namespace"], "held"); + let (status, body) = scoped(&app, "default", "GET", "/api/namespaces/held", Value::Null).await; + assert_eq!(status, StatusCode::OK, "the refused namespace remains: {body}"); + // The planted reference makes describe resolve an external payload, so + // the listing shows the run is still there. + let (status, body) = scoped(&app, "held", "GET", "/api/workflows", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert!(body.to_string().contains("namespace-delete-held"), "its run remains: {body}"); + runtime.close().await; + database.remove().await; +} diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index a181ab10d..fbc2d4fa0 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -12,6 +12,7 @@ import {checkWorkerDeregistration} from './worker-deregistration-contract.mjs'; import {checkLateWorkflowCompletion} from './late-workflow-completion-contract.mjs'; import {checkSearchAttributes} from './search-attribute-contract.mjs'; import {checkStorageTest} from './storage-test-contract.mjs'; +import {checkNamespaceDelete} from './namespace-delete-contract.mjs'; import {checkSchedulePolicies} from './schedule-policies-contract.mjs'; import {checkScheduleCatalog} from './schedule-catalog-contract.mjs'; import {checkSignalsBeforeWorker} from './signals-before-worker-contract.mjs'; @@ -141,6 +142,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const lateWorkflowCompletion = fixture.late_workflow_completion ? checkLateWorkflowCompletion(fixture, observation, workflowId) : null; const searchAttributes = fixture.search_attributes ? checkSearchAttributes(fixture, observation, workflowId) : null; const storageTest = fixture.storage_test ? checkStorageTest(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; const signalsBeforeWorker = fixture.signals_before_worker ? checkSignalsBeforeWorker(fixture, observation, workflowId) : null; @@ -687,6 +689,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { ...(lateWorkflowCompletion ? {late_workflow_completion: lateWorkflowCompletion} : {}), ...(searchAttributes ? {search_attributes: searchAttributes} : {}), ...(storageTest ? {storage_test: storageTest} : {}), + ...(namespaceDelete ? {namespace_delete: namespaceDelete} : {}), ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}), ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}), ...(signalsBeforeWorker ? {signals_before_worker: signalsBeforeWorker} : {}), diff --git a/scripts/conformance/server-parity/namespace-delete-contract.mjs b/scripts/conformance/server-parity/namespace-delete-contract.mjs new file mode 100644 index 000000000..b7cb741e1 --- /dev/null +++ b/scripts/conformance/server-parity/namespace-delete-contract.mjs @@ -0,0 +1,42 @@ +import assert from 'node:assert/strict'; + +// Reviewed namespace deletion (NamespaceController::destroy over +// NamespaceLifecycleCleanup). The namespace holds one search attribute +// definition and one started, unclaimed run. The deleted counts are exact, +// except PHP's timeline, timer and lineage projections, which PHP fills +// asynchronously: they may be absent or positive. Embedded applications have +// no HTTP namespaces, so the projection is the definition alone. +export function checkNamespaceDelete(fixture, observation, workflowId) { + const definition = fixture.namespace_delete; + const state = observation.namespace_delete; + const projection = {scope: 'namespace_delete', deleted: definition.deleted}; + assert.ok(state && typeof state === 'object', 'actual namespace delete observation'); + if (observation.mode === 'embedded') { + assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_namespaces'}, + 'embedded namespace delete explicitly inapplicable'); + return projection; + } + const equal = (actual, expected, label) => assert.deepStrictEqual(actual, expected, label); + const namespace = `${workflowId}-delete`.toLowerCase(); + equal([state.applicable, state.namespace, state.created?.name, state.created?.status], [true, namespace, namespace, 'active'], + 'namespace created'); + equal([state.search_attribute, state.started], [201, 'pending'], 'definition and run in the namespace'); + + equal(state.deleted.status, 200, 'namespace deleted'); + const {deleted, ...response} = state.deleted.response; + equal(response, {name: namespace, status: 'deleted'}, 'delete response'); + const exact = Object.fromEntries(Object.entries(deleted).filter(([key]) => !definition.asynchronous_projections.includes(key))); + equal(exact, definition.deleted, 'each owned table deleted with its count'); + for (const key of definition.asynchronous_projections) { + assert.ok(!(key in deleted) || (Number.isInteger(deleted[key]) && deleted[key] > 0), `${key} absent or positive`); + } + + const gone = {status: 404, response: {message: `Namespace [${namespace}] not found.`, reason: 'namespace_not_found', namespace}}; + equal(state.described_after, gone, 'a deleted namespace reads as not found'); + equal(state.deleted_again, gone, 'a deleted namespace cannot be deleted again'); + equal(state.sdk_delete_again, {class: 'DurableWorkflow\\Exception\\ServerException', status: 404, reason: 'namespace_not_found'}, + 'the published SDK refuses a second delete'); + equal(state.own_run, workflowId, 'a run in another namespace remains'); + equal(state.restarted, `${workflowId}-doomed`, "the deleted run's workflow ID starts a new run"); + return projection; +} diff --git a/scripts/conformance/server-parity/namespace-delete-probe.php b/scripts/conformance/server-parity/namespace-delete-probe.php new file mode 100644 index 000000000..e43e212a1 --- /dev/null +++ b/scripts/conformance/server-parity/namespace-delete-probe.php @@ -0,0 +1,53 @@ + rtrim($url, '/').'/api/', 'http_errors' => false, + 'connect_timeout' => 1, 'timeout' => 30]); + $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, + 'Accept' => 'application/json', 'X-Namespace' => $scope, + '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]; + $state['created'] = $admin->createNamespace($namespace, $definition['description'], 30)->raw; + $state['search_attribute'] = $raw('POST', 'search-attributes', $namespace, $definition['search_attribute'])['status']; + $scoped = parityClient($url, namespace: $namespace, token: $token); + $state['started'] = $scoped->startWorkflow($fixture['workflow_type'], $doomed, $queue.'-delete', [$fixture['input']]) + ->describeSelectedRun()->raw['status'] ?? null; + // The delete itself over HTTP, so its status is recorded too. + $state['deleted'] = $raw('DELETE', 'namespaces/'.$namespace, 'default'); + $state['described_after'] = $raw('GET', 'namespaces/'.$namespace, 'default'); + $state['deleted_again'] = $raw('DELETE', 'namespaces/'.$namespace, 'default'); + // The published SDK's typed refusal for a namespace that is gone. + try { + $admin->deleteNamespace($namespace); + $state['sdk_delete_again'] = ['accepted' => true]; + } catch (ServerException $error) { + $state['sdk_delete_again'] = ['class' => get_class($error), 'status' => $error->status, 'reason' => $error->reason]; + } + $state['own_run'] = $admin->describeWorkflow($workflowId)->raw['workflow_id'] ?? null; + $state['restarted'] = $admin->startWorkflow($fixture['workflow_type'], $doomed, $queue.'-delete', [$fixture['input']]) + ->describeSelectedRun()->raw['workflow_id'] ?? null; + + return $state; +} diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php index 0b40d67a2..e20bbb5c4 100644 --- a/scripts/conformance/server-parity/probe.php +++ b/scripts/conformance/server-parity/probe.php @@ -430,6 +430,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, $lateWorkflowCompletion = isset($fixture['late_workflow_completion']) ? finishHttpLateWorkflowCompletion($client, $fixture, $workflowId) : null; $searchAttributes = isset($fixture['search_attributes']) ? finishHttpSearchAttributes($client, $fixture, $workflowId, $url) : null; $storageTest = isset($fixture['storage_test']) ? finishHttpStorageTest($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']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url, 'schedule_catalog') @@ -529,6 +530,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace, 'late_workflow_completion' => $lateWorkflowCompletion, 'search_attributes' => $searchAttributes, 'storage_test' => $storageTest, + 'namespace_delete' => $namespaceDelete, 'schedule_policies' => $schedulePolicies, 'schedule_catalog' => $scheduleCatalog, 'signals_before_worker' => $signalsBeforeWorker, @@ -832,6 +834,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa require __DIR__.'/namespace-probe.php'; require __DIR__.'/search-attribute-probe.php'; require __DIR__.'/storage-test-probe.php'; + require __DIR__.'/namespace-delete-probe.php'; require __DIR__.'/schedule-policies-probe.php'; require __DIR__.'/signals-before-worker-probe.php'; require __DIR__.'/worker-deregistration-probe.php'; @@ -990,6 +993,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa if ($mode === 'embedded' && isset($fixture['storage_test'])) { $observation['storage_test'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_storage_diagnostic']; } + if ($mode === 'embedded' && isset($fixture['namespace_delete'])) { + $observation['namespace_delete'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_namespaces']; + } if ($mode === 'embedded' && isset($fixture['schedule_policies'])) { $observation['schedule_policies'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_schedules']; } diff --git a/tests/Fixtures/ServerParity/namespace-delete.json b/tests/Fixtures/ServerParity/namespace-delete.json new file mode 100644 index 000000000..37fdd283f --- /dev/null +++ b/tests/Fixtures/ServerParity/namespace-delete.json @@ -0,0 +1,29 @@ +{ + "schema": "durable-workflow.server-parity-fixture/v1", + "id": "namespace-delete", + "review_state": "reviewed", + "workflow_type": "parity.v1.echo", + "input": {"message": "namespace delete λ", "count": 9007199254740993, "optional": null}, + "typed_value": {"type": "map", "value": { + "message": {"type": "string", "value": "namespace delete λ"}, + "count": {"type": "int64", "value": "9007199254740993"}, + "optional": {"type": "null", "value": null} + }}, + "activity": false, + "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"], + "namespace_delete": { + "description": "delete me λ", + "search_attribute": {"name": "CustomerTier", "type": "keyword"}, + "deleted": { + "workflow_commands": 1, + "workflow_history_events": 2, + "workflow_tasks": 1, + "search_attribute_definitions": 1, + "workflow_run_summaries": 1, + "workflow_runs": 1, + "workflow_instances": 1 + }, + "asynchronous_projections": ["workflow_run_timeline_entries", "workflow_run_timer_entries", "workflow_run_lineage_entries"] + }, + "contract": "DELETE /api/namespaces/{namespace} (published SDK deleteNamespace) removes a namespace and every row it owns in one transaction. Here the namespace holds one search attribute definition and one started, unclaimed run, and the response is 200 {name, status: deleted, deleted} with each table's non-zero count: one command, two history events, one task, one definition, one run summary, one run and one instance. PHP fills its timeline, timer and lineage projections asynchronously, so those counts may be absent or positive. Afterwards the namespace reads and deletes as 404 namespace_not_found with the message Namespace [name] not found. A run in another namespace is untouched, and the deleted run's workflow ID starts a new run. Embedded applications have no HTTP namespaces." +} diff --git a/tests/Unit/ServerParityNamespaceDeleteContractTest.mjs b/tests/Unit/ServerParityNamespaceDeleteContractTest.mjs new file mode 100644 index 000000000..d8dc16523 --- /dev/null +++ b/tests/Unit/ServerParityNamespaceDeleteContractTest.mjs @@ -0,0 +1,52 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import {checkNamespaceDelete} from '../../scripts/conformance/server-parity/namespace-delete-contract.mjs'; + +const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/namespace-delete.json', import.meta.url))); +const namespace = 'test-delete'; +const gone = {status: 404, response: {message: `Namespace [${namespace}] not found.`, reason: 'namespace_not_found', namespace}}; +// What published PHP 2.5.14 answered for this fixture. +const model = () => structuredClone({mode: 'http', namespace_delete: {applicable: true, namespace, + created: {name: namespace, description: 'delete me λ', retention_mode: 'bounded', retention_days: 30, status: 'active'}, + search_attribute: 201, started: 'pending', + deleted: {status: 200, response: {name: namespace, status: 'deleted', deleted: {workflow_commands: 1, + workflow_history_events: 2, workflow_tasks: 1, search_attribute_definitions: 1, workflow_run_summaries: 1, + workflow_runs: 1, workflow_instances: 1}}}, + described_after: gone, deleted_again: gone, + sdk_delete_again: {class: 'DurableWorkflow\\Exception\\ServerException', status: 404, reason: 'namespace_not_found'}, + own_run: 'TEST', restarted: 'TEST-doomed'}}); +const projection = {scope: 'namespace_delete', deleted: fixture.namespace_delete.deleted}; + +test('published PHP deletion passes', () => assert.deepStrictEqual(checkNamespaceDelete(fixture, model(), 'TEST'), projection)); +test("PHP's asynchronous projections may be counted", () => { + const observation = model(); + Object.assign(observation.namespace_delete.deleted.response.deleted, {workflow_run_timeline_entries: 2, + workflow_run_lineage_entries: 1}); + checkNamespaceDelete(fixture, observation, 'TEST'); +}); +test('embedded explicitly declares HTTP namespaces inapplicable', () => assert.deepStrictEqual(checkNamespaceDelete(fixture, + {mode: 'embedded', namespace_delete: {applicable: false, reason: 'embedded_has_no_http_namespaces'}}, 'test'), projection)); +for (const [label, mutate] of [ + ['a missing observation', s => { for (const key of Object.keys(s)) delete s[key]; }], + ['another namespace', s => s.namespace = 'default'], + ['a refused delete', s => s.deleted = {status: 503, response: {reason: 'external_payload_storage_driver_unavailable'}}], + ['another status word', s => s.deleted.response.status = 'removed'], + ['an owned table left behind', s => delete s.deleted.response.deleted.workflow_tasks], + ['a count off by one', s => s.deleted.response.deleted.workflow_history_events = 3], + ['a zero count reported', s => s.deleted.response.deleted.workflow_memos = 0], + ['an extra table deleted', s => s.deleted.response.deleted.workflow_schedules = 1], + ['a zero asynchronous projection reported', s => s.deleted.response.deleted.workflow_run_timeline_entries = 0], + ['a deleted namespace still described', s => s.described_after = {status: 200, response: s.created}], + ['a second delete accepted', s => s.deleted_again = s.deleted], + ['another not-found message', s => s.deleted_again.response.message = 'Namespace not found.'], + ['the SDK accepting a second delete', s => s.sdk_delete_again = {accepted: true}], + ["another namespace's run removed", s => s.own_run = null], + ['the workflow ID still reserved', s => s.restarted = null], +]) { + test(`rejects ${label}`, () => { + const observation = model(); + mutate(observation.namespace_delete); + assert.throws(() => checkNamespaceDelete(fixture, observation, 'TEST')); + }); +} From de738c663e937c9c8246445594b906196272b402 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 21:28:20 +0000 Subject: [PATCH 2/2] Write the store's backend bounds once in a macro Every runtime module repeated the same fifteen-line bound list on its Store impl, which Sonar counts as duplicated new code. store_methods! holds the list; namespace cleanup uses it. --- rust/src/runtime/mod.rs | 21 +++++++++++++++++++++ rust/src/runtime/namespace_cleanup.rs | 16 ++-------------- 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs index bfe81d629..009170809 100644 --- a/rust/src/runtime/mod.rs +++ b/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 $crate::runtime::store::Store + where + for<'c> &'c mut DB::Connection: ::sqlx::Executor<'c, Database = DB>, + DB::Arguments: ::sqlx::IntoArguments, + for<'q> &'q str: ::sqlx::Encode<'q, DB> + ::sqlx::Type, + for<'q> Option<&'q str>: ::sqlx::Encode<'q, DB>, + for<'q> Option: ::sqlx::Encode<'q, DB>, + for<'q> String: ::sqlx::Encode<'q, DB> + ::sqlx::Type, + for<'q> i64: ::sqlx::Encode<'q, DB> + ::sqlx::Type, + for<'r> i64: ::sqlx::Decode<'r, DB>, + usize: ::sqlx::ColumnIndex, + { + $($body)* + } + }; +} + mod active_summary; mod activity_completion_uploads; mod activity_deadlines; diff --git a/rust/src/runtime/namespace_cleanup.rs b/rust/src/runtime/namespace_cleanup.rs index ec8939747..3fbd980f1 100644 --- a/rust/src/runtime/namespace_cleanup.rs +++ b/rust/src/runtime/namespace_cleanup.rs @@ -9,10 +9,9 @@ //! still references any refuses before changing anything, with PHP's answer //! for unavailable external payload storage. use super::retention::reference; -use super::{Result, RuntimeError, backend::Backend, store::Store}; +use super::{Result, RuntimeError}; use axum::http::StatusCode; use serde_json::{Map, Value, json}; -use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type}; /// The namespace's runs, instances and tasks, as subqueries. Each takes the /// namespace in its own numbered placeholder. @@ -168,18 +167,7 @@ pub(super) fn not_found(name: &str) -> RuntimeError { } } -impl Store -where - for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, - DB::Arguments: IntoArguments, - for<'q> &'q str: Encode<'q, DB> + Type, - for<'q> Option<&'q str>: Encode<'q, DB>, - for<'q> Option: Encode<'q, DB>, - for<'q> String: Encode<'q, DB> + Type, - for<'q> i64: Encode<'q, DB> + Type, - for<'r> i64: Decode<'r, DB>, - usize: ColumnIndex, -{ +store_methods! { pub(super) async fn delete_namespace(&self, name: &str) -> Result { let name = name.to_ascii_lowercase(); let mut tx = self.begin().await?;