From 0f0ca8957fa71095f5f59857e2a32756a8f8f574 Mon Sep 17 00:00:00 2001 From: Ravneet Arora Date: Tue, 20 Oct 2026 13:38:09 -0400 Subject: [PATCH] BUZZ-301: Add dummy user deletion workflow Signed-off-by: Ravneet Arora --- crates/buzz-admin/src/main.rs | 7 + crates/buzz-admin/src/user_deletions.rs | 33 ++ crates/buzz-db/src/lib.rs | 8 +- crates/buzz-db/src/runtime/migration.rs | 11 +- crates/buzz-db/src/store/deletion.rs | 6 +- .../src/store/entity_deletion/lifecycle.rs | 218 ++++++++ .../src/store/entity_deletion/manager.rs | 159 ++++++ .../buzz-db/src/store/entity_deletion/mod.rs | 93 ++++ .../src/store/entity_deletion/model.rs | 249 +++++++++ .../store/entity_deletion/postgres_tests.rs | 514 ++++++++++++++++++ .../src/store/entity_deletion/stages.rs | 133 +++++ crates/buzz-db/src/store/mod.rs | 2 + crates/buzz-deletion/src/executor.rs | 406 ++++++++++++++ crates/buzz-deletion/src/lib.rs | 211 +------ crates/buzz-deletion/src/user.rs | 199 +++++++ .../buzz-deletion/src/user_postgres_tests.rs | 216 ++++++++ docs/operator-user-deletion-demo.md | 156 ++++++ migrations/0067_entity_deletion_workflow.sql | 169 ++++++ .../protect_entity_deletion_approval.sql | 16 + .../protect_entity_deletion_request.sql | 19 + .../protect_entity_deletion_stage.sql | 19 + schema/schema.sql | 18 + schema/seeds/entity_deletion_approvals.sql | 2 + .../entity_deletion_executor_heartbeats.sql | 2 + schema/seeds/entity_deletion_requests.sql | 2 + .../seeds/entity_deletion_stage_history.sql | 2 + .../public/entity_deletion_approvals.sql | 8 + .../entity_deletion_executor_heartbeats.sql | 9 + .../public/entity_deletion_requests.sql | 40 ++ .../public/entity_deletion_stage_history.sql | 26 + schema/triggers/entity_deletion_approvals.sql | 3 + schema/triggers/entity_deletion_requests.sql | 3 + .../entity_deletion_stage_history.sql | 3 + scripts/reconcile-schema-after-pgschema.sql | 61 +++ scripts/test-user-deletion-demo.py | 149 +++++ 35 files changed, 2971 insertions(+), 201 deletions(-) create mode 100644 crates/buzz-admin/src/user_deletions.rs create mode 100644 crates/buzz-db/src/store/entity_deletion/lifecycle.rs create mode 100644 crates/buzz-db/src/store/entity_deletion/manager.rs create mode 100644 crates/buzz-db/src/store/entity_deletion/mod.rs create mode 100644 crates/buzz-db/src/store/entity_deletion/model.rs create mode 100644 crates/buzz-db/src/store/entity_deletion/postgres_tests.rs create mode 100644 crates/buzz-db/src/store/entity_deletion/stages.rs create mode 100644 crates/buzz-deletion/src/executor.rs create mode 100644 crates/buzz-deletion/src/user.rs create mode 100644 crates/buzz-deletion/src/user_postgres_tests.rs create mode 100644 docs/operator-user-deletion-demo.md create mode 100644 migrations/0067_entity_deletion_workflow.sql create mode 100644 schema/functions/protect_entity_deletion_approval.sql create mode 100644 schema/functions/protect_entity_deletion_request.sql create mode 100644 schema/functions/protect_entity_deletion_stage.sql create mode 100644 schema/seeds/entity_deletion_approvals.sql create mode 100644 schema/seeds/entity_deletion_executor_heartbeats.sql create mode 100644 schema/seeds/entity_deletion_requests.sql create mode 100644 schema/seeds/entity_deletion_stage_history.sql create mode 100644 schema/tables/public/entity_deletion_approvals.sql create mode 100644 schema/tables/public/entity_deletion_executor_heartbeats.sql create mode 100644 schema/tables/public/entity_deletion_requests.sql create mode 100644 schema/tables/public/entity_deletion_stage_history.sql create mode 100644 schema/triggers/entity_deletion_approvals.sql create mode 100644 schema/triggers/entity_deletion_requests.sql create mode 100644 schema/triggers/entity_deletion_stage_history.sql create mode 100644 scripts/test-user-deletion-demo.py diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 20e4acfa95..87fdf9f065 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -22,6 +22,7 @@ mod communities; mod deletions; mod nip_fi; +mod user_deletions; use std::future::Future; use std::sync::atomic::{AtomicU64, Ordering}; @@ -126,6 +127,11 @@ enum Command { #[command(subcommand)] command: deletions::DeletionsCommand, }, + /// Synthetic multi-community user deletion demonstrations; no real erasure. + UserDeletions { + #[command(subcommand)] + command: user_deletions::UserDeletionsCommand, + }, /// Emit missing kind:39000/39001/39002 channel discovery events, or /// republish only a targeted channel's kind:39002 roster. /// @@ -203,6 +209,7 @@ async fn run(cli: Cli) -> Result { } => cmd_list_product_feedback(limit).await, Command::Communities { command } => communities::run(command).await, Command::Deletions { command } => deletions::run(command).await, + Command::UserDeletions { command } => user_deletions::run(command).await, Command::NipFi { command } => nip_fi::run(command).await, Command::ReconcileChannels { channel, relay_key } => { reconcile_channels(channel, relay_key).await?; diff --git a/crates/buzz-admin/src/user_deletions.rs b/crates/buzz-admin/src/user_deletions.rs new file mode 100644 index 0000000000..db531874ed --- /dev/null +++ b/crates/buzz-admin/src/user_deletions.rs @@ -0,0 +1,33 @@ +//! Thin `buzz-admin user-deletions` adapter. + +pub use buzz_deletion::user::Command as UserDeletionsCommand; + +/// Delegate to the shared deletion engine's PostgreSQL-only demonstration path. +pub async fn run(command: UserDeletionsCommand) -> anyhow::Result { + buzz_deletion::user::run(command).await +} + +#[cfg(test)] +mod tests { + use clap::Parser; + + #[test] + fn user_commands_are_registered_without_a_continuous_worker() { + for args in [ + vec![ + "buzz-admin", + "user-deletions", + "submit", + "--target", + "demo:alice", + "--requested-by", + "operator-a", + ], + vec!["buzz-admin", "user-deletions", "list"], + vec!["buzz-admin", "user-deletions", "drain"], + ] { + assert!(crate::Cli::try_parse_from(args).is_ok()); + } + assert!(crate::Cli::try_parse_from(["buzz-admin", "user-deletions", "worker"]).is_err()); + } +} diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index f495983a28..7aa8607cdf 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -66,10 +66,10 @@ pub(crate) use runtime::{ }; pub use store::{ admin_moderation, allowlist, api_token, archived_identities, artifact, channel, - channel_members, community, deletion, dm, event, feed, git_repo, moderation, nip_fi_binding, - operator_listener, partition, personal_read, product_feedback, push, reaction, read_state, - relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, replaceable, - storage_accounting, thread, thread_window, usage, user, workflow, + channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation, + nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction, + read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, + replaceable, storage_accounting, thread, thread_window, usage, user, workflow, }; pub use allowlist::AllowlistEntry; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 7cb0879ff8..0b6a53a174 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -489,6 +489,10 @@ mod postgres_tests { "storage_taxonomy_sweeps", "community_serving_write_leases", "community_deletion_executor_heartbeats", + "entity_deletion_requests", + "entity_deletion_stage_history", + "entity_deletion_approvals", + "entity_deletion_executor_heartbeats", "relay_operators", "relay_admin_actions", "relay_admin_outbox", @@ -709,7 +713,12 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 66); + assert_eq!(migrations.len(), 67); + assert_eq!(migrations[66].version, 67); + assert!(migrations[66] + .sql + .as_str() + .contains("CREATE TABLE entity_deletion_stage_history")); assert_eq!(migrations[64].version, 65); let retire_ttl = strip_sql_comments(migrations[64].sql.as_str()); for statement in [ diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 2120df5a16..a944f63840 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -702,11 +702,11 @@ pub struct LeaseToken { /// A claimed request with its durable token. #[derive(Debug, Clone)] -pub struct ClaimedDeletion { +pub struct ClaimedDeletion { /// Request snapshot. - pub request: DeletionRequest, + pub request: R, /// Required token. - pub lease: LeaseToken, + pub lease: L, } /// Short-lived durable lease for an external serving side effect. diff --git a/crates/buzz-db/src/store/entity_deletion/lifecycle.rs b/crates/buzz-db/src/store/entity_deletion/lifecycle.rs new file mode 100644 index 0000000000..c3e92b3c48 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/lifecycle.rs @@ -0,0 +1,218 @@ +use std::time::Duration; + +use sqlx::{Postgres, Transaction}; +use uuid::Uuid; + +use crate::Result; + +use super::{ + actor, lock_owned, lock_request, safety, EntityClaim, EntityDeletionStore, EntityLeaseToken, + EntityRequest, +}; + +fn seconds(duration: Duration) -> Result { + i64::try_from(duration.as_secs()) + .ok() + .filter(|v| *v > 0) + .ok_or_else(|| safety("lease/retry duration must be a positive whole number of seconds")) +} + +pub(super) async fn heartbeat_row( + tx: &mut Transaction<'_, Postgres>, + token: &EntityLeaseToken, + mode: &str, +) -> Result<()> { + sqlx::query( + "INSERT INTO entity_deletion_executor_heartbeats (executor_id, mode, request_id) + VALUES ($1, $2, $3) + ON CONFLICT (executor_id) DO UPDATE + SET mode = EXCLUDED.mode, request_id = EXCLUDED.request_id, + heartbeat_at = now(), draining = false, stopped_at = NULL", + ) + .bind(&token.owner) + .bind(mode) + .bind(token.request_id) + .execute(&mut **tx) + .await?; + Ok(()) +} + +impl EntityDeletionStore { + /// Claim one approved due request; expired owners are replaced atomically. + pub async fn claim( + &self, + id: Option, + owner: &str, + duration: Duration, + ) -> Result> { + actor(owner)?; + let duration = seconds(duration)?; + let request: Option = sqlx::query_as( + "WITH candidate AS ( + SELECT r.id FROM entity_deletion_requests r + WHERE ($1::uuid IS NULL OR r.id = $1) + AND r.status IN ('approved', 'running') AND r.blocked_at IS NULL + AND r.next_attempt_at <= clock_timestamp() + AND (r.lease_until IS NULL OR r.lease_until <= clock_timestamp()) + AND EXISTS (SELECT 1 FROM entity_deletion_approvals a + WHERE a.request_id = r.id AND a.inventory_digest = r.inventory_digest) + ORDER BY r.created_at, r.id FOR UPDATE SKIP LOCKED LIMIT 1 + ) + UPDATE entity_deletion_requests r + SET status = 'running', lease_owner = $2, lease_generation = lease_generation + 1, + lease_until = clock_timestamp() + make_interval(secs => $3), + attempts = attempts + 1, updated_at = now() + FROM candidate WHERE r.id = candidate.id RETURNING r.*", + ) + .bind(id) + .bind(owner) + .bind(duration) + .fetch_optional(&self.pool) + .await?; + Ok(request.map(|request| { + let lease = EntityLeaseToken { + request_id: request.id, + owner: owner.into(), + generation: request.lease_generation, + }; + EntityClaim { request, lease } + })) + } + + /// Renew parent ownership and save executor liveness in the same transaction. + pub async fn heartbeat( + &self, + token: &EntityLeaseToken, + mode: &str, + duration: Duration, + ) -> Result<()> { + let duration = seconds(duration)?; + let mut tx = self.pool.begin().await?; + lock_owned(&mut tx, token).await?; + sqlx::query( + "UPDATE entity_deletion_requests + SET lease_until = clock_timestamp() + make_interval(secs => $2), updated_at = now() + WHERE id = $1", + ) + .bind(token.request_id) + .bind(duration) + .execute(&mut *tx) + .await?; + heartbeat_row(&mut tx, token, mode).await?; + tx.commit().await?; + Ok(()) + } + + /// Release only this generation and record executor shutdown. + pub async fn stop_executor(&self, token: Option<&EntityLeaseToken>, owner: &str) -> Result<()> { + let mut tx = self.pool.begin().await?; + if let Some(token) = token { + sqlx::query( + "UPDATE entity_deletion_requests SET lease_owner = NULL, lease_until = NULL, + updated_at = now() WHERE id = $1 AND lease_owner = $2 AND lease_generation = $3", + ).bind(token.request_id).bind(&token.owner).bind(token.generation) + .execute(&mut *tx).await?; + } + sqlx::query( + "UPDATE entity_deletion_executor_heartbeats SET draining = true, stopped_at = now(), + heartbeat_at = now() WHERE executor_id = $1", + ) + .bind(owner) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + /// Persist failure at the current stage and release ownership for a bounded retry. + pub async fn record_failure( + &self, + token: &EntityLeaseToken, + stage_id: Uuid, + error: &str, + permanent: bool, + delay: Duration, + ) -> Result<()> { + let delay = seconds(delay)?; + let message: String = error.chars().take(4096).collect(); + let mut tx = self.pool.begin().await?; + let request = lock_owned(&mut tx, token).await?; + let affected = sqlx::query( + "UPDATE entity_deletion_stage_history + SET status = 'failed', started_at = COALESCE(started_at, now()), + last_error = $3, updated_at = now() + WHERE request_id = $1 AND id = $2 AND is_current", + ) + .bind(token.request_id) + .bind(stage_id) + .bind(&message) + .execute(&mut *tx) + .await? + .rows_affected(); + if affected != 1 { + return Err(safety("failed stage is no longer current")); + } + let blocked = permanent || request.retry_count >= 4; + sqlx::query( + "UPDATE entity_deletion_requests + SET retry_count = retry_count + 1, last_error = $2, last_error_at = now(), + next_attempt_at = clock_timestamp() + make_interval(secs => $3), + blocked_at = CASE WHEN $4 THEN now() ELSE NULL END, + blocked_reason = CASE WHEN $4 THEN $2 ELSE NULL END, + lease_owner = NULL, lease_until = NULL, updated_at = now() + WHERE id = $1", + ) + .bind(token.request_id) + .bind(&message) + .bind(delay) + .bind(blocked) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + /// Resume blocked work without changing completed history or the current step. + pub async fn unblock( + &self, + id: Uuid, + unblocked_by: &str, + reason: &str, + ) -> Result { + actor(unblocked_by)?; + if reason.trim().is_empty() || reason.len() > 4096 { + return Err(safety( + "unblock requires a nonempty reason of at most 4096 bytes", + )); + } + let mut tx = self.pool.begin().await?; + let request = lock_request(&mut tx, id).await?; + if request.blocked_at.is_none() || request.lease_owner.is_some() { + return Err(safety("only an unowned blocked request can be unblocked")); + } + let affected = sqlx::query( + "UPDATE entity_deletion_stage_history SET metadata_json = metadata_json || $2, + updated_at = now() WHERE request_id = $1 AND is_current", + ) + .bind(id) + .bind(serde_json::json!({"last_unblock": { + "by": unblocked_by, "reason": reason, "at": chrono::Utc::now() + }})) + .execute(&mut *tx) + .await? + .rows_affected(); + if affected != 1 { + return Err(safety("blocked request has no current stage")); + } + let request = sqlx::query_as( + "UPDATE entity_deletion_requests SET blocked_at = NULL, blocked_reason = NULL, + retry_count = 0, next_attempt_at = now(), last_error = NULL, + last_error_at = NULL, updated_at = now() WHERE id = $1 RETURNING *", + ) + .bind(id) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(request) + } +} diff --git a/crates/buzz-db/src/store/entity_deletion/manager.rs b/crates/buzz-db/src/store/entity_deletion/manager.rs new file mode 100644 index 0000000000..63b3c52db3 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/manager.rs @@ -0,0 +1,159 @@ +use uuid::Uuid; + +use crate::Result; + +use super::{ + actor, lock_request, safety, DemoInventory, EntityDeletionStore, EntityRequest, Inspection, +}; + +impl EntityDeletionStore { + /// Submit a synthetic target, converging on an identical previous submission. + pub async fn submit( + &self, + target: &str, + requested_by: &str, + reason: Option<&str>, + ) -> Result { + DemoInventory::for_target(target)?; + actor(requested_by)?; + if reason.is_some_and(|r| r.len() > 4096) { + return Err(safety("reason exceeds 4096 bytes")); + } + let mut tx = self.pool.begin().await?; + sqlx::query( + "INSERT INTO entity_deletion_requests (target, requested_by, reason) + VALUES ($1, $2, $3) ON CONFLICT (entity_type, target) DO NOTHING", + ) + .bind(target) + .bind(requested_by) + .bind(reason) + .execute(&mut *tx) + .await?; + let request: EntityRequest = sqlx::query_as( + "SELECT * FROM entity_deletion_requests WHERE entity_type = 'user' AND target = $1", + ) + .bind(target) + .fetch_one(&mut *tx) + .await?; + if request.requested_by != requested_by || request.reason.as_deref() != reason { + return Err(safety(format!( + "target already has request {}; inspect it before resubmission", + request.id + ))); + } + tx.commit().await?; + Ok(request) + } + + /// Freeze the fixed non-owned fixture and its plan in one transaction. + pub async fn prepare_inventory(&self, id: Uuid) -> Result { + let mut tx = self.pool.begin().await?; + let request = lock_request(&mut tx, id).await?; + if request.status != "submitted" { + request.inventory()?; + tx.commit().await?; + return Ok(request); + } + let inventory = DemoInventory::for_target(&request.target)?; + let request = sqlx::query_as( + "UPDATE entity_deletion_requests + SET inventory_manifest = $2, inventory_digest = $3, inventory_frozen_at = now(), + status = 'inventoried', updated_at = now() WHERE id = $1 RETURNING *", + ) + .bind(id) + .bind(serde_json::to_value(&inventory)?) + .bind(inventory.digest()?) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(request) + } + + /// Approve the frozen inventory and create the first pending stage atomically. + pub async fn approve(&self, id: Uuid, approved_by: &str) -> Result { + actor(approved_by)?; + let mut tx = self.pool.begin().await?; + let request = lock_request(&mut tx, id).await?; + if request.requested_by == approved_by { + return Err(safety("approval requires a different operator")); + } + let inventory = request.inventory()?; + let previous: Option = sqlx::query_scalar( + "SELECT approved_by FROM entity_deletion_approvals WHERE request_id = $1", + ) + .bind(id) + .fetch_optional(&mut *tx) + .await?; + if let Some(previous) = previous { + if previous != approved_by { + return Err(safety("approval is already recorded by another operator")); + } + tx.commit().await?; + return Ok(request); + } + if request.status != "inventoried" { + return Err(safety("request must be inventoried before approval")); + } + sqlx::query( + "INSERT INTO entity_deletion_approvals (request_id, inventory_digest, approved_by) + VALUES ($1, $2, $3)", + ) + .bind(id) + .bind(inventory.digest()?) + .bind(approved_by) + .execute(&mut *tx) + .await?; + super::stages::insert_stage(&mut tx, id, &inventory, 1).await?; + let request = sqlx::query_as( + "UPDATE entity_deletion_requests SET status = 'approved', updated_at = now() + WHERE id = $1 RETURNING *", + ) + .bind(id) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(request) + } + + /// Load a request from the authoritative writer database. + pub async fn get(&self, id: Uuid) -> Result { + sqlx::query_as("SELECT * FROM entity_deletion_requests WHERE id = $1") + .bind(id) + .fetch_optional(&self.pool) + .await? + .ok_or_else(|| crate::DbError::NotFound(format!("entity deletion request {id}"))) + } + + /// List bounded operator-global requests, newest first. + pub async fn list(&self, limit: u16) -> Result> { + Ok(sqlx::query_as( + "SELECT * FROM entity_deletion_requests ORDER BY created_at DESC, id LIMIT $1", + ) + .bind(i64::from(limit.clamp(1, 1000))) + .fetch_all(&self.pool) + .await?) + } + + /// Inspect a consistent snapshot while serializing with request mutations. + pub async fn inspect(&self, id: Uuid) -> Result { + let mut tx = self.pool.begin().await?; + let request = lock_request(&mut tx, id).await?; + let approval = + sqlx::query_as("SELECT * FROM entity_deletion_approvals WHERE request_id = $1") + .bind(id) + .fetch_optional(&mut *tx) + .await?; + let stage_history = sqlx::query_as( + "SELECT * FROM entity_deletion_stage_history WHERE request_id = $1 ORDER BY sequence", + ) + .bind(id) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + Ok(Inspection { + request, + approval, + stage_history, + }) + } +} diff --git a/crates/buzz-db/src/store/entity_deletion/mod.rs b/crates/buzz-db/src/store/entity_deletion/mod.rs new file mode 100644 index 0000000000..efd0204a63 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/mod.rs @@ -0,0 +1,93 @@ +//! Operator-global workflow state for synthetic, non-owned user deletion demos. +//! Domain handlers must never infer permission to erase data from these records. + +use sqlx::{PgPool, Postgres, Transaction}; +use uuid::Uuid; + +use crate::{Db, DbError, Result}; + +mod lifecycle; +mod manager; +mod model; +mod stages; + +pub use model::{ + Approval, DemoCommunity, DemoInventory, EntityClaim, EntityLeaseToken, EntityRequest, + Inspection, PlannedStage, StageRecord, +}; + +#[cfg(test)] +mod postgres_tests; + +/// PostgreSQL adapter for operator-driven entity workflow demonstrations. +#[derive(Clone)] +pub struct EntityDeletionStore { + pool: PgPool, +} + +impl Db { + /// Use the writer pool for durable entity deletion workflow state. + pub fn entity_deletion_store(&self) -> EntityDeletionStore { + EntityDeletionStore { + pool: self.pool().clone(), + } + } +} + +fn safety(message: impl Into) -> DbError { + DbError::DeletionSafety(message.into()) +} + +fn stale(id: Uuid) -> DbError { + DbError::AccessDenied(format!("stale entity deletion lease: {id}")) +} + +/// Whether a mutation was rejected because execution ownership is stale. +pub fn is_stale_lease(error: &DbError) -> bool { + matches!(error, DbError::AccessDenied(message) if message.starts_with("stale entity deletion lease:")) +} + +fn actor(value: &str) -> Result<&str> { + if value.is_empty() || value.len() > 256 || value.trim() != value { + return Err(safety( + "operator/executor identity must be 1..256 bytes without surrounding whitespace", + )); + } + Ok(value) +} + +async fn lock_request(tx: &mut Transaction<'_, Postgres>, id: Uuid) -> Result { + sqlx::query_as("SELECT * FROM entity_deletion_requests WHERE id = $1 FOR UPDATE") + .bind(id) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("entity deletion request {id}"))) +} + +async fn lock_owned( + tx: &mut Transaction<'_, Postgres>, + token: &EntityLeaseToken, +) -> Result { + // Keep the row lock until the child mutation commits. A concurrent claim + // cannot change the generation between ownership verification and progress. + let request = lock_request(tx, token.request_id).await?; + let valid: bool = sqlx::query_scalar( + "SELECT EXISTS ( + SELECT 1 FROM entity_deletion_requests r + JOIN entity_deletion_approvals a + ON a.request_id = r.id AND a.inventory_digest = r.inventory_digest + WHERE r.id = $1 AND r.lease_owner = $2 AND r.lease_generation = $3 + AND r.lease_until > clock_timestamp() + AND r.status = 'running' AND r.blocked_at IS NULL + )", + ) + .bind(token.request_id) + .bind(&token.owner) + .bind(token.generation) + .fetch_one(&mut **tx) + .await?; + if !valid { + return Err(stale(token.request_id)); + } + Ok(request) +} diff --git a/crates/buzz-db/src/store/entity_deletion/model.rs b/crates/buzz-db/src/store/entity_deletion/model.rs new file mode 100644 index 0000000000..bd0d80f463 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/model.rs @@ -0,0 +1,249 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use uuid::Uuid; + +use crate::{deletion::ClaimedDeletion, Result}; + +use super::safety; + +/// A participant community in the fixed synthetic inventory. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DemoCommunity { + /// Synthetic provenance; no real community row is required. + pub id: Uuid, + /// Reserved example hostname. + pub host: String, + /// Only participant relationships are supported by this demonstration. + pub relationship: String, +} + +/// An ordered action in the frozen, approved demonstration plan. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PlannedStage { + /// None identifies a request-wide step. + pub target_community_id: Option, + /// Explicit action selected by the dummy handler. + pub stage: String, +} + +/// Fixed three-community fixture; never a discovery result for a real user. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DemoInventory { + /// Fixture format version. + pub version: u32, + /// Synthetic target, prefixed with `demo:`. + pub target: String, + /// Three non-owned participant communities. + pub communities: Vec, + /// Bounded deterministic execution order included in the approval digest. + pub stages: Vec, +} + +impl DemoInventory { + /// Build the only supported fixture, without reading user or membership data. + pub fn for_target(target: &str) -> Result { + let slug = target.strip_prefix("demo:").unwrap_or_default(); + if slug.is_empty() + || slug.len() > 64 + || !slug.as_bytes()[0].is_ascii_alphanumeric() + || !slug + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-') + { + return Err(safety( + "target must be demo:<1..64 lowercase letters, digits, underscores or hyphens>", + )); + } + let communities: Vec<_> = (1..=3) + .map(|n| DemoCommunity { + id: Uuid::from_u128(n), + host: format!("participant-{n}.invalid"), + relationship: "participant".into(), + }) + .collect(); + let mut stages = vec![PlannedStage { + target_community_id: None, + stage: "gate_user".into(), + }]; + for community in &communities { + stages.push(PlannedStage { + target_community_id: Some(community.id), + stage: "gate_community".into(), + }); + } + for community in &communities { + for stage in ["drain", "delete_data", "verify_community"] { + stages.push(PlannedStage { + target_community_id: Some(community.id), + stage: stage.into(), + }); + } + } + for stage in ["verify_user", "post_delete"] { + stages.push(PlannedStage { + target_community_id: None, + stage: stage.into(), + }); + } + Ok(Self { + version: 1, + target: target.into(), + communities, + stages, + }) + } + + /// Reject edited fixtures, including owned communities or arbitrary actions. + pub fn validate(&self) -> Result<()> { + if *self != Self::for_target(&self.target)? { + return Err(safety( + "only the fixed non-owned participant demo inventory is supported", + )); + } + Ok(()) + } + + /// Digest the validated typed manifest so JSON object ordering is irrelevant. + pub fn digest(&self) -> Result> { + self.validate()?; + Ok(Sha256::digest(serde_json::to_vec(self)?).to_vec()) + } +} + +/// Overall lifecycle and the sole ownership record for a demonstration. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct EntityRequest { + /// Request identifier. + pub id: Uuid, + /// Currently always user. + pub entity_type: String, + /// Synthetic user label. + pub target: String, + /// Overall lifecycle; granular progress is in stage history. + pub status: String, + /// Submitting operator. + pub requested_by: String, + /// Operator's optional reason. + pub reason: Option, + /// Frozen synthetic inventory and stage plan. + pub inventory_manifest: Option, + /// Digest covered by approval. + pub inventory_digest: Option>, + /// Inventory freeze time. + pub inventory_frozen_at: Option>, + /// Executor holding the parent lease. + pub lease_owner: Option, + /// Monotonic ownership generation. + pub lease_generation: i64, + /// Ownership deadline, checked using database time. + pub lease_until: Option>, + /// Total claims. + pub attempts: i32, + /// Consecutive failures of the current stage. + pub retry_count: i32, + /// Earliest automatic retry. + pub next_attempt_at: DateTime, + /// Latest workflow error. + pub last_error: Option, + /// Latest error time. + pub last_error_at: Option>, + /// Manual intervention marker. + pub blocked_at: Option>, + /// Why automatic attempts stopped. + pub blocked_reason: Option, + /// Submission time. + pub created_at: DateTime, + /// Last lifecycle change. + pub updated_at: DateTime, + /// Demonstration completion time. + pub completed_at: Option>, +} + +impl EntityRequest { + /// Decode and validate the approved fixture before dispatching any stage. + pub fn inventory(&self) -> Result { + let inventory: DemoInventory = serde_json::from_value( + self.inventory_manifest + .clone() + .ok_or_else(|| safety("inventory is not prepared"))?, + )?; + if inventory.target != self.target + || self.inventory_digest.as_deref() != Some(inventory.digest()?.as_slice()) + { + return Err(safety("inventory does not match request target/digest")); + } + Ok(inventory) + } +} + +/// Ownership proof carried by user workflow operations; no community-wide fence. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EntityLeaseToken { + /// Parent request. + pub request_id: Uuid, + /// Executor identity. + pub owner: String, + /// Claim generation. + pub generation: i64, +} + +/// The same claim container used by community deletion, with user record types. +pub type EntityClaim = ClaimedDeletion; + +/// Current or completed granular stage, with one row reused across retries. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct StageRecord { + /// Stage record identifier. + pub id: Uuid, + /// Parent request. + pub request_id: Uuid, + /// One-based position in the approved plan. + pub sequence: i32, + /// Community provenance, or None for request-wide work. + pub target_community_id: Option, + /// Action name. + pub stage: String, + /// pending, running, failed, or completed. + pub status: String, + /// The request's one current unfinished stage. + pub is_current: bool, + /// First attempt start time. + pub started_at: Option>, + /// Successful completion time. + pub finished_at: Option>, + /// Stage details and result. + pub metadata_json: Value, + /// Last stage failure, retained for diagnosis after retry. + pub last_error: Option, + /// Latest stage record update. + pub updated_at: DateTime, +} + +/// Independent approval of the exact frozen inventory. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct Approval { + /// Parent request. + pub request_id: Uuid, + /// Approved manifest digest. + pub inventory_digest: Vec, + /// Approving operator. + pub approved_by: String, + /// Approval time. + pub approved_at: DateTime, +} + +/// Consistent request, approval and stage-history snapshot for inspection. +#[derive(Debug, Serialize)] +pub struct Inspection { + /// Overall lifecycle and ownership. + pub request: EntityRequest, + /// Approval, if recorded. + pub approval: Option, + /// Reached stages in plan order. + pub stage_history: Vec, +} diff --git a/crates/buzz-db/src/store/entity_deletion/postgres_tests.rs b/crates/buzz-db/src/store/entity_deletion/postgres_tests.rs new file mode 100644 index 0000000000..38e0c8bb71 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/postgres_tests.rs @@ -0,0 +1,514 @@ +use std::time::Duration; + +use serde_json::json; +use sqlx::PgPool; + +use super::*; + +const LEASE: Duration = Duration::from_secs(120); + +async fn store() -> EntityDeletionStore { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect isolated DB"); + EntityDeletionStore { pool } +} + +async fn approved(store: &EntityDeletionStore) -> EntityRequest { + let target = format!("demo:{}", Uuid::new_v4().simple()); + let request = store + .submit(&target, "operator-a", None) + .await + .expect("submit"); + store + .prepare_inventory(request.id) + .await + .expect("inventory"); + store + .approve(request.id, "operator-b") + .await + .expect("approve") +} + +async fn claim(store: &EntityDeletionStore, id: Uuid, owner: &str) -> EntityClaim { + store + .claim(Some(id), owner, LEASE) + .await + .expect("claim") + .expect("eligible") +} + +async fn due(store: &EntityDeletionStore, id: Uuid) { + sqlx::query("UPDATE entity_deletion_requests SET next_attempt_at = now() WHERE id = $1") + .bind(id) + .execute(&store.pool) + .await + .expect("make retry due"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn inventory_approval_and_completed_history_are_immutable() { + let store = store().await; + assert!(store.submit("real-user", "operator-a", None).await.is_err()); + let request = store + .submit("demo:alice", "operator-a", None) + .await + .expect("submit"); + assert_eq!( + store + .submit("demo:alice", "operator-a", None) + .await + .expect("repeat") + .id, + request.id + ); + assert!(store + .submit("demo:alice", "someone-else", None) + .await + .is_err()); + assert!(store + .claim(Some(request.id), "worker", LEASE) + .await + .expect("unapproved") + .is_none()); + assert!(store.approve(request.id, "operator-b").await.is_err()); + store + .prepare_inventory(request.id) + .await + .expect("inventory"); + assert!(sqlx::query( + "INSERT INTO entity_deletion_approvals (request_id, inventory_digest, approved_by) + SELECT id, inventory_digest, requested_by FROM entity_deletion_requests WHERE id = $1", + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + assert!(sqlx::query( + "UPDATE entity_deletion_requests SET target = 'demo:retargeted' WHERE id = $1" + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + assert!(sqlx::query( + "UPDATE entity_deletion_requests SET requested_by = 'replacement' WHERE id = $1" + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + assert!(store.approve(request.id, "operator-a").await.is_err()); + store + .approve(request.id, "operator-b") + .await + .expect("approve"); + store + .approve(request.id, "operator-b") + .await + .expect("idempotent approve"); + assert_eq!( + store + .inspect(request.id) + .await + .expect("inspect") + .stage_history + .len(), + 1 + ); + assert!(sqlx::query( + "UPDATE entity_deletion_requests SET inventory_manifest = '{}' WHERE id = $1" + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + assert!( + sqlx::query("DELETE FROM entity_deletion_approvals WHERE request_id = $1") + .bind(request.id) + .execute(&store.pool) + .await + .is_err() + ); + assert!(sqlx::query("INSERT INTO entity_deletion_stage_history (request_id, sequence, stage) VALUES ($1, 2, 'drain')") + .bind(request.id).execute(&store.pool).await.is_err()); + let claim = claim(&store, request.id, "worker").await; + let stage = store.start_stage(&claim.lease).await.expect("start"); + assert!(sqlx::query( + "UPDATE entity_deletion_stage_history SET stage = 'delete_data' WHERE id = $1" + ) + .bind(stage.id) + .execute(&store.pool) + .await + .is_err()); + assert!(sqlx::query("UPDATE entity_deletion_stage_history SET started_at = now() + interval '1 second' WHERE id = $1") + .bind(stage.id).execute(&store.pool).await.is_err()); + store + .complete_stage(&claim.lease, stage.id, &json!({"simulated": true})) + .await + .expect("complete"); + assert!(sqlx::query( + "UPDATE entity_deletion_stage_history SET metadata_json = '{}' WHERE id = $1" + ) + .bind(stage.id) + .execute(&store.pool) + .await + .is_err()); + assert!( + sqlx::query("DELETE FROM entity_deletion_stage_history WHERE id = $1") + .bind(stage.id) + .execute(&store.pool) + .await + .is_err() + ); + let mut inventory = claim.request.inventory().expect("fixture"); + inventory.communities[0].relationship = "owner".into(); + assert!( + inventory.validate().is_err(), + "owned-community fixtures are unsupported" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn claim_generation_guards_children_heartbeat_and_release() { + let store = store().await; + let request = approved(&store).await; + let (a, b) = tokio::join!( + store.claim(Some(request.id), "a", LEASE), + store.claim(Some(request.id), "b", LEASE), + ); + let a = a.expect("claim a"); + let b = b.expect("claim b"); + assert_ne!(a.is_some(), b.is_some()); + let first = a.or(b).expect("one claim"); + let stage = store.start_stage(&first.lease).await.expect("start"); + store + .heartbeat(&first.lease, "run", LEASE) + .await + .expect("heartbeat"); + let heartbeat: i64 = sqlx::query_scalar( + "SELECT count(*) FROM entity_deletion_executor_heartbeats WHERE executor_id = $1", + ) + .bind(&first.lease.owner) + .fetch_one(&store.pool) + .await + .expect("heartbeat row"); + assert_eq!(heartbeat, 1); + assert_eq!( + store + .get(request.id) + .await + .expect("request") + .lease_generation, + first.lease.generation + ); + sqlx::query("UPDATE entity_deletion_requests SET lease_until = now() - interval '1 second' WHERE id = $1") + .bind(request.id).execute(&store.pool).await.expect("expire"); + assert!(is_stale_lease( + &store + .heartbeat(&first.lease, "run", LEASE) + .await + .expect_err("expired") + )); + let second = claim(&store, request.id, "successor").await; + assert_eq!(second.lease.generation, first.lease.generation + 1); + assert!(is_stale_lease( + &store + .complete_stage(&first.lease, stage.id, &json!({})) + .await + .expect_err("stale child write") + )); + assert!(is_stale_lease( + &store + .start_stage(&first.lease) + .await + .expect_err("stale start") + )); + store + .stop_executor(Some(&first.lease), &first.lease.owner) + .await + .expect("stale release"); + assert_eq!( + store + .get(request.id) + .await + .expect("request") + .lease_owner + .as_deref(), + Some("successor") + ); + let resumed = store.start_stage(&second.lease).await.expect("resume"); + assert_eq!(resumed.id, stage.id); + assert_eq!(resumed.started_at, stage.started_at); + store + .complete_stage(&second.lease, stage.id, &json!({})) + .await + .expect("successor advances"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn stage_and_successor_creation_roll_back_together() { + let store = store().await; + let request = approved(&store).await; + let claim = claim(&store, request.id, "worker").await; + let stage = store.start_stage(&claim.lease).await.expect("start"); + sqlx::raw_sql( + "CREATE FUNCTION reject_demo_successor() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'injected successor insert failure'; END $$; + CREATE TRIGGER reject_demo_successor BEFORE INSERT ON entity_deletion_stage_history + FOR EACH ROW EXECUTE FUNCTION reject_demo_successor();", + ) + .execute(&store.pool) + .await + .expect("inject next-row failure"); + assert!(store + .complete_stage(&claim.lease, stage.id, &json!({"result": "done"})) + .await + .is_err()); + let inspection = store.inspect(request.id).await.expect("inspect rollback"); + assert_eq!(inspection.stage_history.len(), 1); + let current = &inspection.stage_history[0]; + assert_eq!(current.status, "running"); + assert!(current.is_current); + assert!(current.finished_at.is_none()); + assert_eq!(current.metadata_json, json!({})); + sqlx::raw_sql("DROP TRIGGER reject_demo_successor ON entity_deletion_stage_history; DROP FUNCTION reject_demo_successor();") + .execute(&store.pool).await.expect("remove fault"); + store + .complete_stage(&claim.lease, stage.id, &json!({"result": "done"})) + .await + .expect("advance"); + let inspection = store.inspect(request.id).await.expect("inspect"); + assert_eq!(inspection.stage_history.len(), 2); + assert_eq!( + inspection + .stage_history + .iter() + .filter(|s| s.is_current) + .count(), + 1 + ); + assert_eq!(inspection.stage_history[1].sequence, 2); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn community_b_failure_resumes_without_repeating_a_and_retries_are_bounded() { + let store = store().await; + let request = approved(&store).await; + let first = claim(&store, request.id, "worker-1").await; + for sequence in 1..9 { + let stage = store.start_stage(&first.lease).await.expect("start"); + assert_eq!(stage.sequence, sequence); + store + .complete_stage(&first.lease, stage.id, &json!({"sequence": sequence})) + .await + .expect("advance"); + } + let failed = store.start_stage(&first.lease).await.expect("B delete"); + assert_eq!(failed.target_community_id, Some(Uuid::from_u128(2))); + assert_eq!(failed.stage, "delete_data"); + store + .record_failure( + &first.lease, + failed.id, + "temporary", + false, + Duration::from_secs(30), + ) + .await + .expect("fail"); + assert!(store + .claim(Some(request.id), "too-early", LEASE) + .await + .expect("claim delayed") + .is_none()); + let snapshot = store.inspect(request.id).await.expect("inspect failed"); + assert_eq!( + snapshot + .stage_history + .iter() + .filter(|s| s.status == "completed") + .count(), + 8 + ); + assert_eq!( + snapshot.stage_history.last().expect("B stage").status, + "failed" + ); + assert!(snapshot.request.lease_owner.is_none()); + for attempt in 2..=5 { + due(&store, request.id).await; + let retry = claim(&store, request.id, &format!("worker-{attempt}")).await; + let stage = store + .start_stage(&retry.lease) + .await + .expect("retry same step"); + assert_eq!(stage.id, failed.id); + assert_eq!(stage.started_at, failed.started_at); + store + .record_failure( + &retry.lease, + stage.id, + "temporary", + false, + Duration::from_secs(30), + ) + .await + .expect("fail retry"); + } + assert!(store + .get(request.id) + .await + .expect("blocked") + .blocked_at + .is_some()); + due(&store, request.id).await; + assert!(store + .claim(Some(request.id), "blocked", LEASE) + .await + .expect("blocked claim") + .is_none()); + store + .unblock(request.id, "operator-c", "dependency repaired") + .await + .expect("unblock"); + let retry = claim(&store, request.id, "recovered").await; + for sequence in 9..=15 { + let stage = store.start_stage(&retry.lease).await.expect("resume"); + assert_eq!(stage.sequence, sequence); + store + .complete_stage(&retry.lease, stage.id, &json!({"simulated": true})) + .await + .expect("complete"); + } + let finished = store.inspect(request.id).await.expect("finished"); + assert_eq!(finished.request.status, "demo_completed"); + assert_eq!(finished.request.retry_count, 0); + assert!(finished.request.lease_owner.is_none()); + assert!(finished + .stage_history + .iter() + .all(|s| !s.is_current && s.status == "completed")); + assert_eq!(finished.stage_history.len(), 15); + for (before, after) in snapshot.stage_history[..8] + .iter() + .zip(&finished.stage_history[..8]) + { + assert_eq!( + serde_json::to_value(before).expect("before"), + serde_json::to_value(after).expect("after") + ); + } + assert_eq!( + finished.stage_history[8].last_error.as_deref(), + Some("temporary") + ); + assert_eq!( + finished.stage_history[8].metadata_json["last_unblock"]["by"], + "operator-c" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_schema_classifies_workflow_tables_without_community_fences() { + let store = store().await; + let names = [ + "entity_deletion_requests", + "entity_deletion_stage_history", + "entity_deletion_approvals", + "entity_deletion_executor_heartbeats", + ]; + let classified: i64 = sqlx::query_scalar( + "SELECT count(*) FROM _operator_global_tables WHERE table_name = ANY($1)", + ) + .bind(names.as_slice()) + .fetch_one(&store.pool) + .await + .expect("classifications"); + assert_eq!(classified, 4); + let fences: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_trigger t JOIN pg_class c ON c.oid = t.tgrelid + WHERE c.relname = ANY($1) AND t.tgname LIKE 'community_write_fence_%'", + ) + .bind(names.as_slice()) + .fetch_one(&store.pool) + .await + .expect("fences"); + assert_eq!(fences, 0); + let checks: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_constraint WHERE conname IN + ('entity_deletion_request_shape', 'entity_deletion_stage_shape') AND convalidated", + ) + .fetch_one(&store.pool) + .await + .expect("shape checks"); + assert_eq!(checks, 2); + let request = approved(&store).await; + assert!(sqlx::query( + "UPDATE entity_deletion_requests SET lease_owner = 'partial-lease' WHERE id = $1" + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + assert!(sqlx::query( + "UPDATE entity_deletion_stage_history SET status = 'completed' WHERE request_id = $1" + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + // Adding operator state must not break the existing destructive catalog. + crate::Db::from_pool(store.pool.clone()) + .validate_deletion_catalog() + .await + .expect("community catalog"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_entity_workflow_upgrade_preserves_existing_data() { + let store = store().await; + crate::migration::run_migrations_through(&store.pool, 66) + .await + .expect("previous schema"); + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'upgrade-fixture.invalid')") + .bind(id) + .execute(&store.pool) + .await + .expect("existing community"); + crate::migration::run_migrations(&store.pool) + .await + .expect("upgrade"); + let request = approved(&store).await; + crate::migration::run_migrations(&store.pool) + .await + .expect("migration rerun"); + assert_eq!( + store + .inspect(request.id) + .await + .expect("preserved request") + .stage_history + .len(), + 1 + ); + let host: String = sqlx::query_scalar("SELECT host FROM communities WHERE id = $1") + .bind(id) + .fetch_one(&store.pool) + .await + .expect("preserved community"); + assert_eq!(host, "upgrade-fixture.invalid"); + let claimed = claim(&store, request.id, "migration-worker").await; + let stage = store.start_stage(&claimed.lease).await.expect("start"); + store + .complete_stage(&claimed.lease, stage.id, &json!({})) + .await + .expect("advance"); +} diff --git a/crates/buzz-db/src/store/entity_deletion/stages.rs b/crates/buzz-db/src/store/entity_deletion/stages.rs new file mode 100644 index 0000000000..5f49796a74 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/stages.rs @@ -0,0 +1,133 @@ +use serde_json::Value; +use sqlx::{Postgres, Transaction}; +use uuid::Uuid; + +use crate::Result; + +use super::{ + lock_owned, safety, DemoInventory, EntityDeletionStore, EntityLeaseToken, EntityRequest, + StageRecord, +}; + +pub(super) async fn insert_stage( + tx: &mut Transaction<'_, Postgres>, + id: Uuid, + inventory: &DemoInventory, + sequence: i32, +) -> Result<()> { + let index = usize::try_from(sequence - 1).map_err(|_| safety("invalid stage sequence"))?; + let stage = inventory + .stages + .get(index) + .ok_or_else(|| safety("stage is outside the approved plan"))?; + sqlx::query( + "INSERT INTO entity_deletion_stage_history (request_id, sequence, target_community_id, stage) + VALUES ($1, $2, $3, $4)", + ).bind(id).bind(sequence).bind(stage.target_community_id).bind(&stage.stage) + .execute(&mut **tx).await?; + Ok(()) +} + +async fn current( + tx: &mut Transaction<'_, Postgres>, + request: &EntityRequest, +) -> Result { + let stage: StageRecord = sqlx::query_as( + "SELECT * FROM entity_deletion_stage_history WHERE request_id = $1 AND is_current FOR UPDATE", + ).bind(request.id).fetch_optional(&mut **tx).await? + .ok_or_else(|| safety("runnable request has no current stage"))?; + let inventory = request.inventory()?; + let index = + usize::try_from(stage.sequence - 1).map_err(|_| safety("invalid stage sequence"))?; + let planned = inventory + .stages + .get(index) + .ok_or_else(|| safety("stage is outside the approved plan"))?; + if stage.stage != planned.stage || stage.target_community_id != planned.target_community_id { + return Err(safety("current stage differs from the approved plan")); + } + Ok(stage) +} + +impl EntityDeletionStore { + /// Read the current stage for dispatch; starting it still validates ownership. + pub async fn current_stage(&self, id: Uuid) -> Result { + sqlx::query_as( + "SELECT * FROM entity_deletion_stage_history WHERE request_id = $1 AND is_current", + ) + .bind(id) + .fetch_optional(&self.pool) + .await? + .ok_or_else(|| safety("runnable request has no current stage")) + } + + /// Start or resume the current stage under the parent ownership lock. + pub async fn start_stage(&self, token: &EntityLeaseToken) -> Result { + let mut tx = self.pool.begin().await?; + let request = lock_owned(&mut tx, token).await?; + let stage = current(&mut tx, &request).await?; + let stage = sqlx::query_as( + "UPDATE entity_deletion_stage_history + SET status = 'running', started_at = COALESCE(started_at, clock_timestamp()), updated_at = now() + WHERE id = $1 RETURNING *", + ).bind(stage.id).fetch_one(&mut *tx).await?; + tx.commit().await?; + Ok(stage) + } + + /// Commit a stage result and its successor (or final completion) atomically. + pub async fn complete_stage( + &self, + token: &EntityLeaseToken, + stage_id: Uuid, + metadata: &Value, + ) -> Result { + if !metadata.is_object() || serde_json::to_vec(metadata)?.len() > 16_384 { + return Err(safety( + "stage metadata must be an object of at most 16384 bytes", + )); + } + let mut tx = self.pool.begin().await?; + let request = lock_owned(&mut tx, token).await?; + let stage = current(&mut tx, &request).await?; + if stage.id != stage_id || stage.status != "running" { + return Err(safety("only the current running stage can complete")); + } + let inventory = request.inventory()?; + let completed: i64 = sqlx::query_scalar( + "SELECT count(*) FROM entity_deletion_stage_history + WHERE request_id = $1 AND sequence < $2 AND status = 'completed'", + ) + .bind(request.id) + .bind(stage.sequence) + .fetch_one(&mut *tx) + .await?; + if completed != i64::from(stage.sequence - 1) { + return Err(safety("earlier stages have not all completed")); + } + sqlx::query( + "UPDATE entity_deletion_stage_history SET status = 'completed', is_current = false, + finished_at = clock_timestamp(), metadata_json = metadata_json || $2, updated_at = now() + WHERE id = $1", + ).bind(stage.id).bind(metadata).execute(&mut *tx).await?; + let finished = usize::try_from(stage.sequence).ok() == Some(inventory.stages.len()); + if !finished { + insert_stage(&mut tx, request.id, &inventory, stage.sequence + 1).await?; + } + let request = sqlx::query_as( + "UPDATE entity_deletion_requests + SET status = CASE WHEN $2 THEN 'demo_completed' ELSE status END, + completed_at = CASE WHEN $2 THEN clock_timestamp() ELSE NULL END, + lease_owner = CASE WHEN $2 THEN NULL ELSE lease_owner END, + lease_until = CASE WHEN $2 THEN NULL ELSE lease_until END, + retry_count = 0, last_error = NULL, last_error_at = NULL, updated_at = now() + WHERE id = $1 RETURNING *", + ) + .bind(request.id) + .bind(finished) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(request) + } +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 6850b625cf..9311f34927 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -18,6 +18,8 @@ pub mod community; pub mod deletion; /// Direct message channel persistence. pub mod dm; +/// Operator-global synthetic user deletion workflow persistence. +pub mod entity_deletion; /// Event storage and retrieval. pub mod event; /// App-owned push enqueue and channel TTL refresh for inserted events. diff --git a/crates/buzz-deletion/src/executor.rs b/crates/buzz-deletion/src/executor.rs new file mode 100644 index 0000000000..4ff1c63c02 --- /dev/null +++ b/crates/buzz-deletion/src/executor.rs @@ -0,0 +1,406 @@ +//! Shared request selection, execution, heartbeats, cancellation and recovery. +//! Dispatch is explicit; each store owns its transaction and safety semantics. + +use super::*; +use buzz_db::entity_deletion::{EntityClaim, EntityRequest, StageRecord}; + +#[derive(Clone)] +pub(super) enum WorkflowServices { + Community(Services), + User(user::Services), +} + +impl From for WorkflowServices { + fn from(services: Services) -> Self { + Self::Community(services) + } +} + +impl From for WorkflowServices { + fn from(services: user::Services) -> Self { + Self::User(services) + } +} + +#[derive(Clone)] +enum Execution { + Community { + services: Services, + claim: ClaimedDeletion, + }, + User { + services: user::Services, + claim: EntityClaim, + stage: StageRecord, + }, +} + +enum Selection { + Ready(Box), + Empty, + Finished(i32), +} + +#[cfg(test)] +pub(super) async fn execute_claim( + services: &Services, + mode: LoopMode, + claim: ClaimedDeletion, + shutdown: &CancellationToken, +) -> Result { + let mut execution = Execution::Community { + services: services.clone(), + claim, + }; + match execution.run(mode, shutdown).await? { + Output::Community(output) => Ok(output), + Output::User { .. } => anyhow::bail!("community execution returned a user result"), + } +} + +#[derive(Serialize)] +#[serde(untagged)] +enum Output { + Community(RunOutput), + User { + request: Box, + demonstration: &'static str, + }, +} + +impl Output { + fn failed(&self) -> bool { + match self { + Self::Community(output) => { + output.last_error.is_some() || output.blocked_reason.is_some() + } + Self::User { request, .. } => { + request.last_error.is_some() || request.blocked_reason.is_some() + } + } + } +} + +pub(super) async fn run_loop( + services: impl Into, + mode: LoopMode, + request_id: Option, + executor_id: String, +) -> Result { + let services = services.into(); + let shutdown = shutdown_token(); + loop { + if shutdown.is_cancelled() { + services.stop(&executor_id).await?; + return Ok(0); + } + let mut execution = match services + .claim(mode, request_id, &executor_id, &shutdown) + .await? + { + Selection::Ready(execution) => execution, + Selection::Finished(code) => return Ok(code), + Selection::Empty => { + if mode == LoopMode::Run { + anyhow::bail!("deletion request is not runnable, is blocked, or is leased by another executor"); + } + return Ok(0); + } + }; + let output = execution.run(mode, &shutdown).await?; + print_json(&output)?; + let failed = output.failed(); + let stopped = execution.stop_requested(); + if mode == LoopMode::Run || shutdown.is_cancelled() || failed || stopped { + return Ok(i32::from(failed)); + } + } +} + +impl WorkflowServices { + async fn stop(&self, owner: &str) -> Result<()> { + match self { + Self::Community(s) => s.store.stop_executor(None, owner).await?, + Self::User(s) => s.store.stop_executor(None, owner).await?, + } + Ok(()) + } + + async fn claim( + &self, + mode: LoopMode, + id: Option, + owner: &str, + shutdown: &CancellationToken, + ) -> Result { + match self { + Self::User(services) => { + let Some(claim) = services + .store + .claim(id, owner, DEFAULT_LEASE_DURATION) + .await? + else { + return Ok(Selection::Empty); + }; + let stage = services.store.current_stage(claim.request.id).await?; + Ok(Selection::Ready(Box::new(Execution::User { + services: services.clone(), + claim, + stage, + }))) + } + Self::Community(services) => { + let claim = match id { + Some(id) => { + services + .store + .claim_specific(id, owner, DEFAULT_LEASE_DURATION) + .await? + } + None => { + services + .store + .claim_next(owner, DEFAULT_LEASE_DURATION) + .await? + } + }; + let claim = if claim.is_none() && mode == LoopMode::Drain && id.is_none() { + services + .store + .claim_next_owner_submission(owner, DEFAULT_LEASE_DURATION) + .await? + } else { + claim + }; + let Some(mut claim) = claim else { + return Ok(Selection::Empty); + }; + if claim.request.stage == DeletionStage::Submitted { + let request = claim.request.clone(); + match prepare_owner_claim_with( + services, mode, claim, shutdown, HEARTBEAT_INTERVAL, + |lost| async move { build_inventory(services, &request, Some(&lost)).await }, + ).await? { + OwnerPreparationOutcome::Prepared(prepared) => claim = *prepared, + OwnerPreparationOutcome::Stopped => return Ok(Selection::Finished(0)), + OwnerPreparationOutcome::Failed(output) => { + // Preserve the original drain's nonzero exit and JSON evidence. + print_json(&output)?; + return Ok(Selection::Finished(1)); + } + } + } + Ok(Selection::Ready(Box::new(Execution::Community { + services: services.clone(), + claim, + }))) + } + } + } +} + +impl Execution { + async fn run(&mut self, mode: LoopMode, shutdown: &CancellationToken) -> Result { + let mut completed = 0u16; + loop { + if shutdown.is_cancelled() { + self.stop(mode).await?; + return self.output().await; + } + self.heartbeat(mode).await?; + match self.run_stage(mode, shutdown).await { + StageOutcome::Completed => {} + StageOutcome::Shutdown => { + self.stop(mode).await?; + return self.output().await; + } + StageOutcome::Failed(error) => { + self.record_failure(&error).await?; + return self.output().await; + } + } + completed += 1; + if self.refresh().await? { + if let Self::User { + services, claim, .. + } = self + { + services + .store + .stop_executor(Some(&claim.lease), &claim.lease.owner) + .await?; + } + return self.output().await; + } + if let Self::User { services, .. } = self { + if services + .controls + .stop_after + .is_some_and(|limit| completed >= limit) + { + self.stop(mode).await?; + return self.output().await; + } + } + } + } + + fn stop_requested(&self) -> bool { + matches!(self, Self::User { services, .. } if services.controls.stop_after.is_some()) + } + + async fn heartbeat(&self, mode: LoopMode) -> Result<()> { + match self { + Self::Community { services, claim } => { + services + .store + .heartbeat(&claim.lease, mode.as_str(), DEFAULT_LEASE_DURATION, false) + .await?; + } + Self::User { + services, claim, .. + } => { + services + .store + .heartbeat(&claim.lease, mode.as_str(), DEFAULT_LEASE_DURATION) + .await?; + } + } + Ok(()) + } + + async fn stop(&self, mode: LoopMode) -> Result<()> { + match self { + Self::Community { services, claim } => { + stop_claim_executor(services, mode, &claim.lease).await? + } + Self::User { + services, claim, .. + } => { + services + .store + .stop_executor(Some(&claim.lease), &claim.lease.owner) + .await?; + } + } + Ok(()) + } + + async fn output(&self) -> Result { + match self { + Self::Community { services, claim } => Ok(Output::Community(run_output( + services.store.get(claim.request.id).await?, + ))), + Self::User { + services, claim, .. + } => Ok(Output::User { + request: Box::new(services.store.get(claim.request.id).await?), + demonstration: "Synthetic workflow only; no user data was deleted.", + }), + } + } + + async fn refresh(&mut self) -> Result { + match self { + Self::Community { services, claim } => { + let request = services.store.get(claim.request.id).await?; + let terminal = request.stage == DeletionStage::RetentionPending + || request.blocked_reason.is_some(); + claim.lease.fence_generation = request.fence_generation; + claim.request = request; + Ok(terminal) + } + Self::User { + services, + claim, + stage, + } => { + claim.request = services.store.get(claim.request.id).await?; + let terminal = + claim.request.status == "demo_completed" || claim.request.blocked_at.is_some(); + if !terminal { + *stage = services.store.current_stage(claim.request.id).await?; + } + Ok(terminal) + } + } + } + + async fn record_failure(&self, error: &anyhow::Error) -> Result<()> { + match self { + Self::Community { services, claim } => { + let request = services.store.get(claim.request.id).await?; + if request.lease_owner.as_deref() == Some(&claim.lease.owner) + && request.lease_generation == claim.lease.generation + { + record_stage_failure(services, &claim.lease, claim.request.stage, error) + .await?; + } + } + Self::User { + services, + claim, + stage, + } => { + match services + .store + .record_failure( + &claim.lease, + stage.id, + &format!("{error:#}"), + is_permanent_error(error), + RETRY_DELAY, + ) + .await + { + Ok(()) => {} + Err(error) if buzz_db::entity_deletion::is_stale_lease(&error) => {} + Err(error) => return Err(error.into()), + } + } + } + Ok(()) + } + + async fn run_stage(&self, mode: LoopMode, shutdown: &CancellationToken) -> StageOutcome { + let heartbeat_execution = self.clone(); + let cancel = CancellationToken::new(); + let heartbeat_cancel = cancel.clone(); + let lost = CancellationToken::new(); + let heartbeat_lost = lost.clone(); + let heartbeat = tokio::spawn(async move { + let mut interval = tokio::time::interval(heartbeat_interval()); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + loop { + tokio::select! { + _ = heartbeat_cancel.cancelled() => return, + _ = interval.tick() => { + if heartbeat_execution.heartbeat(mode).await.is_err() { + heartbeat_lost.cancel(); + return; + } + } + } + } + }); + let operation = async { + match self { + Self::Community { services, claim } => execute_stage(services, claim, &lost).await, + Self::User { + services, + claim, + stage, + } => user::execute_stage(services, claim, stage, &lost).await, + } + }; + let result = await_stage(operation, shutdown, &lost).await; + cancel.cancel(); + match heartbeat.await { + Ok(()) => result, + Err(error) => { + StageOutcome::Failed(anyhow::anyhow!("deletion heartbeat task failed: {error}")) + } + } + } +} diff --git a/crates/buzz-deletion/src/lib.rs b/crates/buzz-deletion/src/lib.rs index c3fb598523..508fa4e723 100644 --- a/crates/buzz-deletion/src/lib.rs +++ b/crates/buzz-deletion/src/lib.rs @@ -1,6 +1,18 @@ #![deny(unsafe_code)] #![warn(missing_docs)] -//! Shared durable whole-community deletion engine and store adapters. +//! Shared durable deletion engine and store adapters. + +mod executor; +/// Synthetic multi-community user deletion commands. +pub mod user; + +use executor::run_loop; + +#[cfg(test)] +use executor::execute_claim; + +#[cfg(test)] +mod user_postgres_tests; use std::future::Future; #[cfg(test)] @@ -546,6 +558,10 @@ fn resolve_submit_host(host: Option<&str>, relay_url: Option<&str>) -> Result Result { + Ok(store(&connect_database().await?)) +} + +async fn connect_database() -> Result { let database_url = required_env("DATABASE_URL")?; // The scheduled drain starts in a fresh pod whose first connection can // outlast the relay's three-second budget. @@ -559,7 +575,7 @@ async fn connect_store() -> Result { "deletion", ) .await?; - Ok(store(&db)) + Ok(db) } fn resolve_s3_region(buzz_region: Option, aws_region: Option) -> String { @@ -962,91 +978,6 @@ async fn freeze_destructive_manifest( Ok(manifest) } -async fn run_loop( - services: Services, - mode: LoopMode, - request_id: Option, - executor_id: String, -) -> Result { - let shutdown = shutdown_token(); - let mut ran = false; - loop { - if shutdown.is_cancelled() { - services - .store - .stop_executor(None, &executor_id) - .await - .context("record executor drain")?; - return Ok(0); - } - let claim = match request_id { - Some(id) => { - services - .store - .claim_specific(id, &executor_id, DEFAULT_LEASE_DURATION) - .await? - } - None => { - services - .store - .claim_next(&executor_id, DEFAULT_LEASE_DURATION) - .await? - } - }; - let claim = if claim.is_none() && mode == LoopMode::Drain && request_id.is_none() { - services - .store - .claim_next_owner_submission(&executor_id, DEFAULT_LEASE_DURATION) - .await? - } else { - claim - }; - let Some(mut claim) = claim else { - if mode == LoopMode::Run && !ran { - anyhow::bail!( - "deletion request is not runnable, is blocked, or is leased by another executor" - ); - } - return Ok(0); - }; - ran = true; - if claim.request.stage == DeletionStage::Submitted { - let preparation_request = claim.request.clone(); - let preparation_services = &services; - match prepare_owner_claim_with( - &services, - mode, - claim, - &shutdown, - HEARTBEAT_INTERVAL, - |heartbeat_lost| async move { - build_inventory( - preparation_services, - &preparation_request, - Some(&heartbeat_lost), - ) - .await - }, - ) - .await? - { - OwnerPreparationOutcome::Prepared(prepared) => claim = *prepared, - OwnerPreparationOutcome::Stopped => return Ok(0), - OwnerPreparationOutcome::Failed(output) => { - print_json(&output)?; - return Ok(1); - } - } - } - let output = execute_claim(&services, mode, claim, &shutdown).await?; - print_json(&output)?; - let failed = output.last_error.is_some() || output.blocked_reason.is_some(); - if mode == LoopMode::Run || shutdown.is_cancelled() || failed { - return Ok(i32::from(failed)); - } - } -} - enum OwnerPreparationOutcome { Prepared(Box), Stopped, @@ -1254,57 +1185,6 @@ async fn record_stage_failure( } } -async fn execute_claim( - services: &Services, - mode: LoopMode, - mut claim: ClaimedDeletion, - shutdown: &CancellationToken, -) -> Result { - let token = claim.lease.clone(); - loop { - if shutdown.is_cancelled() { - stop_claim_executor(services, mode, &token).await?; - let request = services.store.get(token.request_id).await?; - return Ok(run_output(request)); - } - services - .store - .heartbeat(&token, mode.as_str(), DEFAULT_LEASE_DURATION, false) - .await?; - let stage_result = run_stage_with_heartbeat(services, mode, &claim, shutdown).await; - match stage_result { - StageOutcome::Completed => {} - StageOutcome::Shutdown => { - stop_claim_executor(services, mode, &token).await?; - let request = services.store.get(token.request_id).await?; - return Ok(run_output(request)); - } - StageOutcome::Failed(error) => { - let request = services.store.get(token.request_id).await?; - if request.lease_owner.as_deref() != Some(&token.owner) - || request.lease_generation != token.generation - { - return Ok(run_output(request)); - } - if !record_stage_failure(services, &token, claim.request.stage, &error).await? { - // Ownership expired between the stage failure and durable - // error recording. A successor now owns retry/block policy. - let request = services.store.get(token.request_id).await?; - return Ok(run_output(request)); - } - let request = services.store.get(token.request_id).await?; - return Ok(run_output(request)); - } - } - let request = services.store.get(token.request_id).await?; - if request.stage == DeletionStage::RetentionPending || request.blocked_reason.is_some() { - return Ok(run_output(request)); - } - claim.request = request.clone(); - claim.lease.fence_generation = request.fence_generation; - } -} - enum StageOutcome { Completed, Shutdown, @@ -1330,61 +1210,6 @@ where } } -async fn run_stage_with_heartbeat( - services: &Services, - mode: LoopMode, - claim: &ClaimedDeletion, - shutdown: &CancellationToken, -) -> StageOutcome { - let heartbeat_services = services.clone(); - let heartbeat_token = claim.lease.clone(); - let heartbeat_mode = mode.as_str(); - let heartbeat_shutdown = CancellationToken::new(); - let heartbeat_cancel = heartbeat_shutdown.clone(); - let heartbeat_error = CancellationToken::new(); - let heartbeat_error_signal = heartbeat_error.clone(); - let heartbeat = tokio::spawn(async move { - let mut interval = tokio::time::interval(heartbeat_interval()); - interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - interval.tick().await; - loop { - tokio::select! { - _ = heartbeat_cancel.cancelled() => return, - _ = interval.tick() => { - if heartbeat_services - .store - .heartbeat( - &heartbeat_token, - heartbeat_mode, - DEFAULT_LEASE_DURATION, - false, - ) - .await - .is_err() - { - heartbeat_error_signal.cancel(); - return; - } - } - } - } - }); - - let stage = await_stage( - execute_stage(services, claim, &heartbeat_error), - shutdown, - &heartbeat_error, - ) - .await; - heartbeat_shutdown.cancel(); - match heartbeat.await { - Ok(()) => stage, - Err(error) => { - StageOutcome::Failed(anyhow::anyhow!("deletion heartbeat task failed: {error}")) - } - } -} - async fn run_guarded_external_step( services: &Services, token: &LeaseToken, diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs new file mode 100644 index 0000000000..6db23eb73d --- /dev/null +++ b/crates/buzz-deletion/src/user.rs @@ -0,0 +1,199 @@ +//! PostgreSQL-only operator commands and dummy actions for user deletion. + +use anyhow::Result; +use buzz_db::entity_deletion::{EntityClaim, EntityDeletionStore, StageRecord}; +use clap::{Args, Subcommand}; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +use crate::{default_executor_id, print_json, LoopMode}; + +/// Commands for synthetic user-deletion demonstrations; no real erasure occurs. +#[derive(Debug, Subcommand)] +pub enum Command { + /// Save a synthetic request. Repeating the same submission returns its row. + Submit { + /// Synthetic user label, for example demo:alice. + #[arg(long)] + target: String, + /// Submitting operator identity. + #[arg(long)] + requested_by: String, + /// Optional operator reason. + #[arg(long)] + reason: Option, + }, + /// Prepare and freeze the three-participant-community demonstration plan. + Inventory { + /// Request UUID. + id: Uuid, + }, + /// List demonstration requests. + List { + /// Maximum number of rows. + #[arg(long, default_value_t = 100, value_parser = clap::value_parser!(u16).range(1..=1000))] + limit: u16, + }, + /// Show inventory, approval, ownership, and all reached stages. + Inspect { + /// Request UUID. + id: Uuid, + }, + /// Independently approve the frozen demonstration inventory. + Approve { + /// Request UUID. + id: Uuid, + /// Must differ from the submitting operator. + #[arg(long)] + approved_by: String, + }, + /// Run an approved demonstration through its remaining dummy stages. + Run { + /// Request UUID. + id: Uuid, + /// Executor identity, otherwise a unique process identity. + #[arg(long)] + executor_id: Option, + /// Controls for demonstrating failure and recovery. + #[command(flatten)] + controls: DemoControls, + }, + /// Drain eligible synthetic requests; exits when none are runnable. + Drain { + /// Executor identity. + #[arg(long)] + executor_id: Option, + }, + /// Resume blocked work after operator remediation. + Unblock { + /// Request UUID. + id: Uuid, + /// Operator identity recorded in current-stage metadata. + #[arg(long)] + unblocked_by: String, + /// Explanation of the remediation. + #[arg(long)] + reason: String, + }, +} + +/// Bounded, explicit failure controls for an artificial workflow. +#[derive(Debug, Clone, Default, Args)] +pub struct DemoControls { + /// Fail at this one-based plan sequence on each invocation using this flag. + #[arg(long, value_parser = clap::value_parser!(u16).range(1..=15))] + pub fail_at: Option, + /// Release ownership after this many successful steps in this invocation. + #[arg(long, value_parser = clap::value_parser!(u16).range(1..=15))] + pub stop_after: Option, + /// Delay each dummy action to exercise heartbeats or process interruption. + #[arg(long, default_value_t = 0, value_parser = clap::value_parser!(u64).range(0..=60000))] + pub step_delay_ms: u64, +} + +#[derive(Clone)] +pub(crate) struct Services { + pub store: EntityDeletionStore, + pub controls: DemoControls, +} + +/// Execute one user-deletion command with a PostgreSQL-only connection. +pub async fn run(command: Command) -> Result { + let db = crate::connect_database().await?; + run_with_store(command, db.entity_deletion_store()).await +} + +pub(crate) async fn run_with_store(command: Command, store: EntityDeletionStore) -> Result { + match command { + Command::Submit { + target, + requested_by, + reason, + } => { + print_json( + &store + .submit(&target, &requested_by, reason.as_deref()) + .await?, + )?; + } + Command::Inventory { id } => print_json(&store.prepare_inventory(id).await?)?, + Command::List { limit } => print_json(&store.list(limit).await?)?, + Command::Inspect { id } => print_json(&store.inspect(id).await?)?, + Command::Approve { id, approved_by } => { + print_json(&store.approve(id, &approved_by).await?)? + } + Command::Unblock { + id, + unblocked_by, + reason, + } => { + print_json(&store.unblock(id, &unblocked_by, &reason).await?)?; + } + Command::Run { + id, + executor_id, + controls, + } => { + return crate::executor::run_loop( + Services { store, controls }, + LoopMode::Run, + Some(id), + executor_id.unwrap_or_else(default_executor_id), + ) + .await; + } + Command::Drain { executor_id } => { + return crate::executor::run_loop( + Services { + store, + controls: DemoControls::default(), + }, + LoopMode::Drain, + None, + executor_id.unwrap_or_else(default_executor_id), + ) + .await; + } + } + Ok(0) +} + +pub(crate) async fn execute_stage( + services: &Services, + claim: &EntityClaim, + expected: &StageRecord, + heartbeat_lost: &CancellationToken, +) -> Result<()> { + let stage = services.store.start_stage(&claim.lease).await?; + if stage.id != expected.id { + anyhow::bail!("current user stage changed before dispatch"); + } + let scope = stage + .target_community_id + .map_or_else(|| "request".into(), |id| id.to_string()); + let message = format!( + "DEMO ONLY: would execute {} for {} in {}; no user data is deleted", + stage.stage, claim.request.target, scope, + ); + eprintln!("{message}"); + tokio::select! { + biased; + _ = heartbeat_lost.cancelled() => return Err(crate::DeletionLeaseLost.into()), + _ = tokio::time::sleep(std::time::Duration::from_millis(services.controls.step_delay_ms)) => {} + } + if services.controls.fail_at.map(i32::from) == Some(stage.sequence) { + return Err(crate::transient(format!( + "injected demo failure at stage {}", + stage.sequence + ))); + } + services + .store + .complete_stage( + &claim.lease, + stage.id, + &serde_json::json!({"message": message, "simulated": true}), + ) + .await?; + Ok(()) +} diff --git a/crates/buzz-deletion/src/user_postgres_tests.rs b/crates/buzz-deletion/src/user_postgres_tests.rs new file mode 100644 index 0000000000..aa7de4b44a --- /dev/null +++ b/crates/buzz-deletion/src/user_postgres_tests.rs @@ -0,0 +1,216 @@ +use super::*; +use user::{Command, DemoControls}; + +async fn fixture() -> (Db, buzz_db::entity_deletion::EntityRequest) { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .expect("isolated PostgreSQL URL"); + let pool = sqlx::PgPool::connect(&url).await.expect("connect"); + let db = Db::from_pool(pool); + let store = db.entity_deletion_store(); + let target = format!("demo:{}", Uuid::new_v4().simple()); + user::run_with_store( + Command::Submit { + target: target.clone(), + requested_by: "submitter".into(), + reason: None, + }, + store.clone(), + ) + .await + .expect("submit command"); + let request = store + .list(100) + .await + .expect("list") + .into_iter() + .find(|r| r.target == target) + .expect("submitted request"); + user::run_with_store(Command::Inventory { id: request.id }, store.clone()) + .await + .expect("inventory command"); + assert!(user::run_with_store( + Command::Run { + id: request.id, + executor_id: Some("unapproved".into()), + controls: DemoControls::default() + }, + store.clone(), + ) + .await + .is_err()); + user::run_with_store( + Command::Approve { + id: request.id, + approved_by: "approver".into(), + }, + store.clone(), + ) + .await + .expect("approve command"); + (db, store.get(request.id).await.expect("request")) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn shared_executor_runs_three_communities_and_resumes_b() { + let (db, request) = fixture().await; + let store = db.entity_deletion_store(); + // Canary data at the synthetic community IDs makes accidental real cleanup + // observable, including updates that would preserve row counts. + for n in 1..=3 { + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(Uuid::from_u128(n)) + .bind(format!("canary-{n}.invalid")) + .execute(db.pool()) + .await + .expect("community canary"); + sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)") + .bind(Uuid::from_u128(n)) + .bind(vec![7u8; 32]) + .execute(db.pool()) + .await + .expect("user canary"); + sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')", + ) + .bind(Uuid::from_u128(n)) + .bind(hex::encode([7u8; 32])) + .execute(db.pool()) + .await + .expect("membership canary"); + } + let before: serde_json::Value = sqlx::query_scalar( + "SELECT jsonb_build_object( + 'communities', (SELECT jsonb_agg(to_jsonb(c) ORDER BY id) FROM communities c), + 'users', (SELECT jsonb_agg(to_jsonb(u) ORDER BY community_id, pubkey) FROM users u), + 'members', (SELECT jsonb_agg(to_jsonb(m) ORDER BY community_id, pubkey) FROM relay_members m), + 'events', (SELECT count(*) FROM events), + 'community_requests', (SELECT count(*) FROM community_deletion_requests))", + ) + .fetch_one(db.pool()) + .await + .expect("domain snapshot"); + let result = user::run_with_store( + Command::Run { + id: request.id, + executor_id: Some("first-process".into()), + controls: DemoControls { + fail_at: Some(9), + ..Default::default() + }, + }, + store.clone(), + ) + .await + .expect("failed demonstration"); + assert_eq!(result, 1); + let snapshot = store.inspect(request.id).await.expect("history"); + assert_eq!(snapshot.stage_history.len(), 9); + assert_eq!(snapshot.stage_history[8].stage, "delete_data"); + assert_eq!( + snapshot.stage_history[8].target_community_id, + Some(Uuid::from_u128(2)) + ); + assert_eq!(snapshot.stage_history[8].status, "failed"); + assert!(snapshot.request.lease_owner.is_none()); + assert_eq!( + snapshot.stage_history[..4] + .iter() + .map(|s| s.stage.as_str()) + .collect::>(), + [ + "gate_user", + "gate_community", + "gate_community", + "gate_community" + ] + ); + assert!(snapshot.stage_history[4..7] + .iter() + .all(|s| s.target_community_id == Some(Uuid::from_u128(1)) && s.status == "completed")); + sqlx::query("UPDATE entity_deletion_requests SET next_attempt_at = now() WHERE id = $1") + .bind(request.id) + .execute(db.pool()) + .await + .expect("make retry due"); + assert_eq!( + user::run_with_store( + Command::Drain { + executor_id: Some("replacement-process".into()), + }, + store.clone() + ) + .await + .expect("drain retry"), + 0 + ); + let finished = store.inspect(request.id).await.expect("finished"); + assert_eq!(finished.request.status, "demo_completed"); + assert_eq!(finished.request.lease_generation, 2); + assert_eq!(finished.stage_history.len(), 15); + assert!(finished + .stage_history + .iter() + .all(|s| !s.is_current && s.metadata_json["simulated"] == true)); + for (a, b) in snapshot.stage_history[..8] + .iter() + .zip(&finished.stage_history[..8]) + { + assert_eq!( + serde_json::to_value(a).expect("before"), + serde_json::to_value(b).expect("after") + ); + } + let after: serde_json::Value = sqlx::query_scalar( + "SELECT jsonb_build_object( + 'communities', (SELECT jsonb_agg(to_jsonb(c) ORDER BY id) FROM communities c), + 'users', (SELECT jsonb_agg(to_jsonb(u) ORDER BY community_id, pubkey) FROM users u), + 'members', (SELECT jsonb_agg(to_jsonb(m) ORDER BY community_id, pubkey) FROM relay_members m), + 'events', (SELECT count(*) FROM events), + 'community_requests', (SELECT count(*) FROM community_deletion_requests))", + ) + .fetch_one(db.pool()) + .await + .expect("domain snapshot"); + assert_eq!(before, after); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn voluntary_stop_retains_one_pending_stage_for_next_executor() { + let (db, request) = fixture().await; + let store = db.entity_deletion_store(); + user::run_with_store( + Command::Run { + id: request.id, + executor_id: Some("stopping-process".into()), + controls: DemoControls { + stop_after: Some(7), + ..Default::default() + }, + }, + store.clone(), + ) + .await + .expect("stop after A"); + let stopped = store.inspect(request.id).await.expect("stopped"); + assert_eq!(stopped.request.status, "running"); + assert!(stopped.request.lease_owner.is_none()); + assert_eq!(stopped.stage_history.len(), 8); + assert_eq!(stopped.stage_history[7].status, "pending"); + assert!(stopped.stage_history[7].is_current); + user::run_with_store( + Command::Run { + id: request.id, + executor_id: Some("next-process".into()), + controls: DemoControls::default(), + }, + store.clone(), + ) + .await + .expect("resume"); + let finished = store.inspect(request.id).await.expect("finished"); + assert_eq!(finished.request.status, "demo_completed"); + assert_eq!(finished.stage_history[7].id, stopped.stage_history[7].id); +} diff --git a/docs/operator-user-deletion-demo.md b/docs/operator-user-deletion-demo.md new file mode 100644 index 0000000000..153c69d29a --- /dev/null +++ b/docs/operator-user-deletion-demo.md @@ -0,0 +1,156 @@ +# User deletion workflow demonstration + +This exercises real PostgreSQL requests, approvals, leases, stage history and +recovery for a synthetic user in three participant communities. Every action +prints `DEMO ONLY`. It does not discover or delete real users, change memberships, +gate access, contact storage, or delete owned communities. + +## Prepare an isolated development database + +Activate Hermit and start a separate PostgreSQL instance. Use a different +container name and port from a running Buzz development environment. + +```bash +. ./bin/activate-hermit +docker run --detach --name buzz-user-deletion-demo \ + --publish 127.0.0.1:55432:5432 \ + --env POSTGRES_USER=buzz \ + --env POSTGRES_PASSWORD=buzz-local-demo \ + --env POSTGRES_DB=buzz postgres:17-alpine +export DATABASE_URL=postgres://buzz:buzz-local-demo@127.0.0.1:55432/buzz +cargo build -p buzz-admin +./target/debug/buzz-admin migrate +``` + +These are disposable local credentials. If the port is occupied, choose another +one in both the Docker command and URL. No Redis or media-storage configuration +is required by `user-deletions`. + +## Submit, prepare and approve + +An operator manually verifies the incoming request in the real operational +process. This demonstration uses `demo:alice`; it is not an account lookup. +Only `demo:` labels are accepted, and inventory is always the fixed participant +fixture. A different label creates an independent demonstration. + +```bash +./target/debug/buzz-admin user-deletions list +./target/debug/buzz-admin user-deletions submit \ + --target demo:alice --requested-by operator-a +``` + +Copy the returned `id` into `REQUEST_ID`. Submission alone creates a `submitted` +request. Repeating the identical target/operator/reason returns that same +request without resetting its progress. + +```bash +REQUEST_ID= +./target/debug/buzz-admin user-deletions inventory "$REQUEST_ID" +./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID" +./target/debug/buzz-admin user-deletions approve "$REQUEST_ID" \ + --approved-by operator-b +``` + +Inventory saves the three targets and ordered plan with a digest. Approval must +come from a different operator and covers the saved digest. Approval also creates +the first pending/current stage. A request cannot run before approval. + +## Execute and inspect + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" +./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID" +``` + +The 15 stages are: + +| Sequence | Target | Action | +| --- | --- | --- | +| 1 | Request | `gate_user` | +| 2–4 | Communities A, B, C | `gate_community` for each | +| 5–7 | Community A | `drain`, `delete_data`, `verify_community` | +| 8–10 | Community B | `drain`, `delete_data`, `verify_community` | +| 11–13 | Community C | `drain`, `delete_data`, `verify_community` | +| 14 | Request | `verify_user` | +| 15 | Request | `post_delete` | + +All gates are simulated. They do not enforce access restrictions. + +`entity_deletion_requests` holds overall status, inventory and worker ownership. +`entity_deletion_stage_history` holds each reached stage's target, status, +`is_current`, `started_at`, `finished_at`, `metadata_json` and latest error. +The final parent status is `demo_completed`, with no current stage or live lease. +Completion output explicitly says no user data was deleted. + +## Demonstrate recovery + +Use a fresh synthetic target for each scenario and repeat submission, inventory +and approval first. + +To stop after A finishes: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" --stop-after 7 +./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID" +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" +``` + +The stop releases ownership and leaves B's `drain` row pending/current. +The next invocation claims a new generation and continues that row. + +To fail during B's data-removal stage: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" --fail-at 9 +./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID" +# Wait until the saved next_attempt_at (normally 30 seconds after failure). +./target/debug/buzz-admin user-deletions drain +``` + +The failed command exits nonzero. A's completed rows remain unchanged, B's current +row records the error, and ownership is released. The next invocation without +`--fail-at` resumes B. The flag injects a failure on every invocation where it is +specified; it is not a persistent request setting. + +Five consecutive failures at the same stage block automatic claims. After +remediation, unblock and retry: + +```bash +./target/debug/buzz-admin user-deletions unblock "$REQUEST_ID" \ + --unblocked-by operator-c --reason "Removed the injected failure" +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" +``` + +`--step-delay-ms 15000` allows observing periodic heartbeats or interrupting a +worker. SIGTERM/Ctrl-C releases its matching lease. After an abrupt process exit, +a later worker can claim the request once `lease_until` expires. An old generation +cannot update stage progress. A printed action whose transition did not commit +can print again when retried. + +## Repeatable checks + +The PostgreSQL lane uses an isolated database per test. Follow +`crates/buzz-db/TESTING.md` for its environment and run: + +```bash +scripts/test-postgres-test-discovery.sh +scripts/postgres-test-run.sh -p buzz-db -p buzz-deletion --lib \ + -E 'test(entity_deletion) | test(user_postgres_tests) | (package(buzz-deletion) & test(postgres_tests))' +``` + +Run the actual binary demonstration against a disposable, migrated database: + +```bash +BUZZ_USER_DELETION_TEST_DATABASE_URL="$DATABASE_URL" \ + python3 scripts/test-user-deletion-demo.py --binary ./target/debug/buzz-admin +``` + +That script creates unique synthetic requests and verifies approval rejection, +completion, stopping/resumption, failure in B followed by a delayed retry, +and heartbeat renewal followed by SIGTERM and resumption. +It leaves the demonstration rows for inspection. + +Schema migrations may roll out normally while these tables are unused. +Introducing real execution requires a separate change and cleanup of +demonstration records. Owned-community policy, real identity discovery, real +write gates and real cleanup handlers remain future work. diff --git a/migrations/0067_entity_deletion_workflow.sql b/migrations/0067_entity_deletion_workflow.sql new file mode 100644 index 0000000000..626d1e5781 --- /dev/null +++ b/migrations/0067_entity_deletion_workflow.sql @@ -0,0 +1,169 @@ +-- Durable synthetic user-deletion workflows; existing community state is unchanged. + +CREATE FUNCTION protect_entity_deletion_request() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF (NEW.id, NEW.entity_type, NEW.target, NEW.requested_by, NEW.reason, NEW.created_at) + IS DISTINCT FROM + (OLD.id, OLD.entity_type, OLD.target, OLD.requested_by, OLD.reason, OLD.created_at) + THEN + RAISE EXCEPTION 'entity deletion target and provenance are immutable'; + END IF; + IF OLD.inventory_frozen_at IS NOT NULL AND + (NEW.inventory_manifest, NEW.inventory_digest, NEW.inventory_frozen_at) + IS DISTINCT FROM + (OLD.inventory_manifest, OLD.inventory_digest, OLD.inventory_frozen_at) + THEN + RAISE EXCEPTION 'entity deletion inventory is immutable'; + END IF; + RETURN NEW; +END; +$$; + +CREATE FUNCTION protect_entity_deletion_approval() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP <> 'INSERT' THEN + RAISE EXCEPTION 'entity deletion approval evidence is immutable'; + END IF; + IF NOT EXISTS ( + SELECT 1 FROM entity_deletion_requests + WHERE id = NEW.request_id AND inventory_digest = NEW.inventory_digest + AND status = 'inventoried' AND requested_by <> NEW.approved_by + ) THEN + RAISE EXCEPTION 'approval requires a frozen inventory and a different operator'; + END IF; + RETURN NEW; +END; +$$; + +CREATE FUNCTION protect_entity_deletion_stage() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + RAISE EXCEPTION 'entity deletion stage history cannot be removed'; + END IF; + IF OLD.status = 'completed' OR + (NEW.id, NEW.request_id, NEW.sequence, NEW.target_community_id, NEW.stage) + IS DISTINCT FROM + (OLD.id, OLD.request_id, OLD.sequence, OLD.target_community_id, OLD.stage) + THEN + RAISE EXCEPTION 'completed history and stage identity are immutable'; + END IF; + IF OLD.started_at IS NOT NULL AND NEW.started_at IS DISTINCT FROM OLD.started_at THEN + RAISE EXCEPTION 'first stage start time is immutable'; + END IF; + RETURN NEW; +END; +$$; + +-- Operator-global workflow state. Initially used only by synthetic user demos. +CREATE TABLE entity_deletion_requests ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + entity_type TEXT NOT NULL DEFAULT 'user' CHECK (entity_type = 'user'), + target TEXT NOT NULL CHECK (target ~ '^demo:[a-z0-9][a-z0-9_-]{0,63}$'), + status TEXT NOT NULL DEFAULT 'submitted' + CHECK (status IN ('submitted', 'inventoried', 'approved', 'running', 'demo_completed')), + requested_by TEXT NOT NULL CHECK (length(trim(requested_by)) BETWEEN 1 AND 256), + reason TEXT, + inventory_manifest JSONB, + inventory_digest BYTEA CHECK (inventory_digest IS NULL OR length(inventory_digest) = 32), + inventory_frozen_at TIMESTAMPTZ, + lease_owner TEXT, + lease_generation BIGINT NOT NULL DEFAULT 0 CHECK (lease_generation >= 0), + lease_until TIMESTAMPTZ, + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + retry_count INTEGER NOT NULL DEFAULT 0 CHECK (retry_count >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + last_error_at TIMESTAMPTZ, + blocked_at TIMESTAMPTZ, + blocked_reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ, + CONSTRAINT entity_deletion_request_shape CHECK ( + ((lease_owner IS NULL) = (lease_until IS NULL)) + AND ((blocked_at IS NULL) = (blocked_reason IS NULL)) + AND ((status = 'demo_completed') = (completed_at IS NOT NULL)) + AND (status <> 'demo_completed' OR lease_owner IS NULL) + AND ((status = 'submitted') = (inventory_manifest IS NULL)) + AND ((inventory_manifest IS NULL) = (inventory_digest IS NULL)) + AND ((inventory_manifest IS NULL) = (inventory_frozen_at IS NULL)) + ), + UNIQUE (entity_type, target), + UNIQUE (id, inventory_digest) +); +CREATE INDEX entity_deletion_requests_runnable + ON entity_deletion_requests (next_attempt_at, created_at, id) + WHERE status IN ('approved', 'running') AND blocked_at IS NULL; + +-- target_community_id is operator provenance, including synthetic references. +-- It is not tenant data and intentionally has no FK to the communities table. +CREATE TABLE entity_deletion_stage_history ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + request_id UUID NOT NULL REFERENCES entity_deletion_requests(id), + sequence INTEGER NOT NULL CHECK (sequence > 0), + target_community_id UUID, + stage TEXT NOT NULL CHECK (length(stage) BETWEEN 1 AND 64), + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'running', 'failed', 'completed')), + is_current BOOLEAN NOT NULL DEFAULT true, + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + metadata_json JSONB NOT NULL DEFAULT '{}' CHECK (jsonb_typeof(metadata_json) = 'object'), + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT entity_deletion_stage_shape CHECK ( + (is_current = (status <> 'completed')) + AND ((status = 'completed') = (finished_at IS NOT NULL)) + AND ((status = 'pending') = (started_at IS NULL)) + AND (finished_at IS NULL OR finished_at >= started_at) + ), + UNIQUE (request_id, sequence) +); +CREATE UNIQUE INDEX entity_deletion_stage_history_current + ON entity_deletion_stage_history (request_id) WHERE is_current; + +CREATE TABLE entity_deletion_approvals ( + request_id UUID PRIMARY KEY, + inventory_digest BYTEA NOT NULL CHECK (length(inventory_digest) = 32), + approved_by TEXT NOT NULL CHECK (length(trim(approved_by)) BETWEEN 1 AND 256), + approved_at TIMESTAMPTZ NOT NULL DEFAULT now(), + FOREIGN KEY (request_id, inventory_digest) + REFERENCES entity_deletion_requests(id, inventory_digest) +); + +CREATE TABLE entity_deletion_executor_heartbeats ( + executor_id TEXT PRIMARY KEY, + mode TEXT NOT NULL CHECK (mode IN ('run', 'drain')), + request_id UUID REFERENCES entity_deletion_requests(id), + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), + draining BOOLEAN NOT NULL DEFAULT false, + stopped_at TIMESTAMPTZ +); + +CREATE TRIGGER protect_entity_deletion_request + BEFORE UPDATE ON entity_deletion_requests + FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_request(); + +CREATE TRIGGER protect_entity_deletion_stage + BEFORE UPDATE OR DELETE ON entity_deletion_stage_history + FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_stage(); + +CREATE TRIGGER protect_entity_deletion_approval + BEFORE INSERT OR UPDATE OR DELETE ON entity_deletion_approvals + FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_approval(); + +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_requests', 'deployment entity deletion workflow and synthetic demo state'); + +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_stage_history', 'deployment entity deletion workflow and synthetic demo state'); + +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_approvals', 'deployment entity deletion workflow and synthetic demo state'); + +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_executor_heartbeats', 'deployment entity deletion workflow and synthetic demo state'); diff --git a/schema/functions/protect_entity_deletion_approval.sql b/schema/functions/protect_entity_deletion_approval.sql new file mode 100644 index 0000000000..5e3b87ab15 --- /dev/null +++ b/schema/functions/protect_entity_deletion_approval.sql @@ -0,0 +1,16 @@ +CREATE FUNCTION protect_entity_deletion_approval() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP <> 'INSERT' THEN + RAISE EXCEPTION 'entity deletion approval evidence is immutable'; + END IF; + IF NOT EXISTS ( + SELECT 1 FROM entity_deletion_requests + WHERE id = NEW.request_id AND inventory_digest = NEW.inventory_digest + AND status = 'inventoried' AND requested_by <> NEW.approved_by + ) THEN + RAISE EXCEPTION 'approval requires a frozen inventory and a different operator'; + END IF; + RETURN NEW; +END; +$$; diff --git a/schema/functions/protect_entity_deletion_request.sql b/schema/functions/protect_entity_deletion_request.sql new file mode 100644 index 0000000000..67d4a9b183 --- /dev/null +++ b/schema/functions/protect_entity_deletion_request.sql @@ -0,0 +1,19 @@ +CREATE FUNCTION protect_entity_deletion_request() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF (NEW.id, NEW.entity_type, NEW.target, NEW.requested_by, NEW.reason, NEW.created_at) + IS DISTINCT FROM + (OLD.id, OLD.entity_type, OLD.target, OLD.requested_by, OLD.reason, OLD.created_at) + THEN + RAISE EXCEPTION 'entity deletion target and provenance are immutable'; + END IF; + IF OLD.inventory_frozen_at IS NOT NULL AND + (NEW.inventory_manifest, NEW.inventory_digest, NEW.inventory_frozen_at) + IS DISTINCT FROM + (OLD.inventory_manifest, OLD.inventory_digest, OLD.inventory_frozen_at) + THEN + RAISE EXCEPTION 'entity deletion inventory is immutable'; + END IF; + RETURN NEW; +END; +$$; diff --git a/schema/functions/protect_entity_deletion_stage.sql b/schema/functions/protect_entity_deletion_stage.sql new file mode 100644 index 0000000000..97b20459e2 --- /dev/null +++ b/schema/functions/protect_entity_deletion_stage.sql @@ -0,0 +1,19 @@ +CREATE FUNCTION protect_entity_deletion_stage() +RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + RAISE EXCEPTION 'entity deletion stage history cannot be removed'; + END IF; + IF OLD.status = 'completed' OR + (NEW.id, NEW.request_id, NEW.sequence, NEW.target_community_id, NEW.stage) + IS DISTINCT FROM + (OLD.id, OLD.request_id, OLD.sequence, OLD.target_community_id, OLD.stage) + THEN + RAISE EXCEPTION 'completed history and stage identity are immutable'; + END IF; + IF OLD.started_at IS NOT NULL AND NEW.started_at IS DISTINCT FROM OLD.started_at THEN + RAISE EXCEPTION 'first stage start time is immutable'; + END IF; + RETURN NEW; +END; +$$; diff --git a/schema/schema.sql b/schema/schema.sql index d252cbdb7e..cddf2adf3d 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -78,6 +78,10 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i functions/attach_community_write_fence.sql \i functions/profile_name_fields.sql +\i functions/protect_entity_deletion_request.sql +\i functions/protect_entity_deletion_approval.sql +\i functions/protect_entity_deletion_stage.sql + -- tables \i tables/public/communities.sql \i tables/public/channels.sql @@ -138,11 +142,19 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/artifact_heads.sql \i tables/public/artifact_revisions.sql +\i tables/public/entity_deletion_requests.sql +\i tables/public/entity_deletion_stage_history.sql +\i tables/public/entity_deletion_approvals.sql +\i tables/public/entity_deletion_executor_heartbeats.sql + -- partitions \i partitions/events.sql \i partitions/delivery_log.sql -- triggers +\i triggers/entity_deletion_requests.sql +\i triggers/entity_deletion_stage_history.sql +\i triggers/entity_deletion_approvals.sql \i triggers/channels.sql \i triggers/events.sql \i triggers/community_deletion_requests.sql @@ -210,3 +222,9 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; -- Fence any community-scoped table that lacks an explicit attach. \i community_write_fence.sql + +-- Operator-global entity deletion state. +\i seeds/entity_deletion_requests.sql +\i seeds/entity_deletion_stage_history.sql +\i seeds/entity_deletion_approvals.sql +\i seeds/entity_deletion_executor_heartbeats.sql diff --git a/schema/seeds/entity_deletion_approvals.sql b/schema/seeds/entity_deletion_approvals.sql new file mode 100644 index 0000000000..ed1fa75fc2 --- /dev/null +++ b/schema/seeds/entity_deletion_approvals.sql @@ -0,0 +1,2 @@ +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_approvals', 'deployment entity deletion workflow and synthetic demo state'); diff --git a/schema/seeds/entity_deletion_executor_heartbeats.sql b/schema/seeds/entity_deletion_executor_heartbeats.sql new file mode 100644 index 0000000000..acea0a3578 --- /dev/null +++ b/schema/seeds/entity_deletion_executor_heartbeats.sql @@ -0,0 +1,2 @@ +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_executor_heartbeats', 'deployment entity deletion workflow and synthetic demo state'); diff --git a/schema/seeds/entity_deletion_requests.sql b/schema/seeds/entity_deletion_requests.sql new file mode 100644 index 0000000000..e995fd0c61 --- /dev/null +++ b/schema/seeds/entity_deletion_requests.sql @@ -0,0 +1,2 @@ +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_requests', 'deployment entity deletion workflow and synthetic demo state'); diff --git a/schema/seeds/entity_deletion_stage_history.sql b/schema/seeds/entity_deletion_stage_history.sql new file mode 100644 index 0000000000..3b1d4b2086 --- /dev/null +++ b/schema/seeds/entity_deletion_stage_history.sql @@ -0,0 +1,2 @@ +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_stage_history', 'deployment entity deletion workflow and synthetic demo state'); diff --git a/schema/tables/public/entity_deletion_approvals.sql b/schema/tables/public/entity_deletion_approvals.sql new file mode 100644 index 0000000000..042e634053 --- /dev/null +++ b/schema/tables/public/entity_deletion_approvals.sql @@ -0,0 +1,8 @@ +CREATE TABLE entity_deletion_approvals ( + request_id UUID PRIMARY KEY, + inventory_digest BYTEA NOT NULL CHECK (length(inventory_digest) = 32), + approved_by TEXT NOT NULL CHECK (length(trim(approved_by)) BETWEEN 1 AND 256), + approved_at TIMESTAMPTZ NOT NULL DEFAULT now(), + FOREIGN KEY (request_id, inventory_digest) + REFERENCES entity_deletion_requests(id, inventory_digest) +); diff --git a/schema/tables/public/entity_deletion_executor_heartbeats.sql b/schema/tables/public/entity_deletion_executor_heartbeats.sql new file mode 100644 index 0000000000..195aa0d037 --- /dev/null +++ b/schema/tables/public/entity_deletion_executor_heartbeats.sql @@ -0,0 +1,9 @@ +CREATE TABLE entity_deletion_executor_heartbeats ( + executor_id TEXT PRIMARY KEY, + mode TEXT NOT NULL CHECK (mode IN ('run', 'drain')), + request_id UUID REFERENCES entity_deletion_requests(id), + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + heartbeat_at TIMESTAMPTZ NOT NULL DEFAULT now(), + draining BOOLEAN NOT NULL DEFAULT false, + stopped_at TIMESTAMPTZ +); diff --git a/schema/tables/public/entity_deletion_requests.sql b/schema/tables/public/entity_deletion_requests.sql new file mode 100644 index 0000000000..706d22d51a --- /dev/null +++ b/schema/tables/public/entity_deletion_requests.sql @@ -0,0 +1,40 @@ +-- Operator-global workflow state. Initially used only by synthetic user demos. +CREATE TABLE entity_deletion_requests ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + entity_type TEXT NOT NULL DEFAULT 'user' CHECK (entity_type = 'user'), + target TEXT NOT NULL CHECK (target ~ '^demo:[a-z0-9][a-z0-9_-]{0,63}$'), + status TEXT NOT NULL DEFAULT 'submitted' + CHECK (status IN ('submitted', 'inventoried', 'approved', 'running', 'demo_completed')), + requested_by TEXT NOT NULL CHECK (length(trim(requested_by)) BETWEEN 1 AND 256), + reason TEXT, + inventory_manifest JSONB, + inventory_digest BYTEA CHECK (inventory_digest IS NULL OR length(inventory_digest) = 32), + inventory_frozen_at TIMESTAMPTZ, + lease_owner TEXT, + lease_generation BIGINT NOT NULL DEFAULT 0 CHECK (lease_generation >= 0), + lease_until TIMESTAMPTZ, + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + retry_count INTEGER NOT NULL DEFAULT 0 CHECK (retry_count >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_error TEXT, + last_error_at TIMESTAMPTZ, + blocked_at TIMESTAMPTZ, + blocked_reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ, + CONSTRAINT entity_deletion_request_shape CHECK ( + ((lease_owner IS NULL) = (lease_until IS NULL)) + AND ((blocked_at IS NULL) = (blocked_reason IS NULL)) + AND ((status = 'demo_completed') = (completed_at IS NOT NULL)) + AND (status <> 'demo_completed' OR lease_owner IS NULL) + AND ((status = 'submitted') = (inventory_manifest IS NULL)) + AND ((inventory_manifest IS NULL) = (inventory_digest IS NULL)) + AND ((inventory_manifest IS NULL) = (inventory_frozen_at IS NULL)) + ), + UNIQUE (entity_type, target), + UNIQUE (id, inventory_digest) +); +CREATE INDEX entity_deletion_requests_runnable + ON entity_deletion_requests (next_attempt_at, created_at, id) + WHERE status IN ('approved', 'running') AND blocked_at IS NULL; diff --git a/schema/tables/public/entity_deletion_stage_history.sql b/schema/tables/public/entity_deletion_stage_history.sql new file mode 100644 index 0000000000..3b560af7e8 --- /dev/null +++ b/schema/tables/public/entity_deletion_stage_history.sql @@ -0,0 +1,26 @@ +-- target_community_id is operator provenance, including synthetic references. +-- It is not tenant data and intentionally has no FK to the communities table. +CREATE TABLE entity_deletion_stage_history ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + request_id UUID NOT NULL REFERENCES entity_deletion_requests(id), + sequence INTEGER NOT NULL CHECK (sequence > 0), + target_community_id UUID, + stage TEXT NOT NULL CHECK (length(stage) BETWEEN 1 AND 64), + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'running', 'failed', 'completed')), + is_current BOOLEAN NOT NULL DEFAULT true, + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + metadata_json JSONB NOT NULL DEFAULT '{}' CHECK (jsonb_typeof(metadata_json) = 'object'), + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT entity_deletion_stage_shape CHECK ( + (is_current = (status <> 'completed')) + AND ((status = 'completed') = (finished_at IS NOT NULL)) + AND ((status = 'pending') = (started_at IS NULL)) + AND (finished_at IS NULL OR finished_at >= started_at) + ), + UNIQUE (request_id, sequence) +); +CREATE UNIQUE INDEX entity_deletion_stage_history_current + ON entity_deletion_stage_history (request_id) WHERE is_current; diff --git a/schema/triggers/entity_deletion_approvals.sql b/schema/triggers/entity_deletion_approvals.sql new file mode 100644 index 0000000000..eb5480d6af --- /dev/null +++ b/schema/triggers/entity_deletion_approvals.sql @@ -0,0 +1,3 @@ +CREATE TRIGGER protect_entity_deletion_approval + BEFORE INSERT OR UPDATE OR DELETE ON entity_deletion_approvals + FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_approval(); diff --git a/schema/triggers/entity_deletion_requests.sql b/schema/triggers/entity_deletion_requests.sql new file mode 100644 index 0000000000..fc9bc66c97 --- /dev/null +++ b/schema/triggers/entity_deletion_requests.sql @@ -0,0 +1,3 @@ +CREATE TRIGGER protect_entity_deletion_request + BEFORE UPDATE ON entity_deletion_requests + FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_request(); diff --git a/schema/triggers/entity_deletion_stage_history.sql b/schema/triggers/entity_deletion_stage_history.sql new file mode 100644 index 0000000000..5b07b57239 --- /dev/null +++ b/schema/triggers/entity_deletion_stage_history.sql @@ -0,0 +1,3 @@ +CREATE TRIGGER protect_entity_deletion_stage + BEFORE UPDATE OR DELETE ON entity_deletion_stage_history + FOR EACH ROW EXECUTE FUNCTION protect_entity_deletion_stage(); diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 06d6258643..92d67dfdbd 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -232,3 +232,64 @@ BEGIN RAISE EXCEPTION 'replica_heartbeat must contain its singleton row after pgschema apply'; END IF; END $$; + +-- Entity deletion is operator-global, even when a stage targets a community. +INSERT INTO _operator_global_tables (table_name, reason) VALUES + ('entity_deletion_requests', 'deployment entity deletion workflow and synthetic demo state'), + ('entity_deletion_stage_history', 'deployment entity deletion workflow and synthetic demo state'), + ('entity_deletion_approvals', 'deployment entity deletion workflow and synthetic demo state'), + ('entity_deletion_executor_heartbeats', 'deployment entity deletion workflow and synthetic demo state') +ON CONFLICT (table_name) DO NOTHING; + +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'entity_deletion_requests'::regclass + AND conname = 'entity_deletion_request_shape') THEN + ALTER TABLE entity_deletion_requests ADD CONSTRAINT entity_deletion_request_shape CHECK ( + ((lease_owner IS NULL) = (lease_until IS NULL)) + AND ((blocked_at IS NULL) = (blocked_reason IS NULL)) + AND ((status = 'demo_completed') = (completed_at IS NOT NULL)) + AND (status <> 'demo_completed' OR lease_owner IS NULL) + AND ((status = 'submitted') = (inventory_manifest IS NULL)) + AND ((inventory_manifest IS NULL) = (inventory_digest IS NULL)) + AND ((inventory_manifest IS NULL) = (inventory_frozen_at IS NULL)) + ); + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'entity_deletion_requests'::regclass + AND conname = 'entity_deletion_request_shape' AND convalidated) THEN + RAISE EXCEPTION 'entity_deletion_request_shape must be validated'; + END IF; +END $$; + +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'entity_deletion_stage_history'::regclass + AND conname = 'entity_deletion_stage_shape') THEN + ALTER TABLE entity_deletion_stage_history ADD CONSTRAINT entity_deletion_stage_shape CHECK ( + (is_current = (status <> 'completed')) + AND ((status = 'completed') = (finished_at IS NOT NULL)) + AND ((status = 'pending') = (started_at IS NULL)) + AND (finished_at IS NULL OR finished_at >= started_at) + ); + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_constraint + WHERE conrelid = 'entity_deletion_stage_history'::regclass + AND conname = 'entity_deletion_stage_shape' AND convalidated) THEN + RAISE EXCEPTION 'entity_deletion_stage_shape must be validated'; + END IF; +END $$; + +DO $$ +BEGIN + IF (SELECT count(*) FROM _operator_global_tables WHERE table_name IN ( + 'entity_deletion_requests', + 'entity_deletion_stage_history', + 'entity_deletion_approvals', + 'entity_deletion_executor_heartbeats' + )) <> 4 THEN + RAISE EXCEPTION 'entity deletion tables must be operator-global'; + END IF; +END $$; diff --git a/scripts/test-user-deletion-demo.py b/scripts/test-user-deletion-demo.py new file mode 100644 index 0000000000..20c395337a --- /dev/null +++ b/scripts/test-user-deletion-demo.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Exercise the actual user-deletion CLI against an explicit disposable database.""" + +import argparse +import json +import os +import subprocess +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--binary", type=Path, required=True) + args = parser.parse_args() + database_url = os.environ.get("BUZZ_USER_DELETION_TEST_DATABASE_URL") + if not database_url: + parser.error("set BUZZ_USER_DELETION_TEST_DATABASE_URL to an isolated, migrated database") + environment = dict(os.environ, DATABASE_URL=database_url) + binary = str(args.binary.resolve()) + + def command(*parts, expected=0): + result = subprocess.run( + [binary, "user-deletions", *parts], + env=environment, + capture_output=True, + text=True, + timeout=90, + check=False, + ) + if result.returncode != expected: + raise AssertionError( + f"{parts}: expected exit {expected}, got {result.returncode}\n" + f"{result.stdout[-8000:]}\n{result.stderr[-8000:]}" + ) + if expected != 0: + return None + return json.loads(result.stdout) + + def prepare(): + target = "demo:cli-" + uuid.uuid4().hex + request = command("submit", "--target", target, "--requested-by", "operator-a") + request_id = request["id"] + repeated = command("submit", "--target", target, "--requested-by", "operator-a") + assert repeated["id"] == request_id + command("run", request_id, expected=5) + inventory = command("inventory", request_id) + assert len(inventory["inventory_manifest"]["communities"]) == 3 + assert all(c["relationship"] == "participant" + for c in inventory["inventory_manifest"]["communities"]) + command("approve", request_id, "--approved-by", "operator-a", expected=5) + command("approve", request_id, "--approved-by", "operator-b") + return request_id + + def finished(request_id): + result = command("inspect", request_id) + assert result["request"]["status"] == "demo_completed" + assert result["request"]["lease_owner"] is None + history = result["stage_history"] + assert len(history) == 15 + assert all(not s["is_current"] and s["status"] == "completed" for s in history) + assert all(s["metadata_json"]["simulated"] for s in history) + assert [s["stage"] for s in history[:4]] == [ + "gate_user", "gate_community", "gate_community", "gate_community" + ] + for start, community in [(4, 1), (7, 2), (10, 3)]: + assert all(s["target_community_id"] == str(uuid.UUID(int=community)) + for s in history[start:start + 3]) + return result + + stopped_id = prepare() + command("run", stopped_id, "--stop-after", "7") + stopped = command("inspect", stopped_id) + assert stopped["request"]["lease_owner"] is None + assert len(stopped["stage_history"]) == 8 + assert stopped["stage_history"][-1]["status"] == "pending" + command("run", stopped_id) + completed = finished(stopped_id) + assert completed["stage_history"][:7] == stopped["stage_history"][:7] + print(f"PASS CLI stop/resume: {stopped_id}", flush=True) + + failed_id = prepare() + command("run", failed_id, "--fail-at", "9", expected=1) + failed = command("inspect", failed_id) + assert len(failed["stage_history"]) == 9 + assert failed["stage_history"][-1]["status"] == "failed" + assert failed["request"]["lease_owner"] is None + retry_at = datetime.fromisoformat(failed["request"]["next_attempt_at"].replace("Z", "+00:00")) + delay = max(0, (retry_at - datetime.now(timezone.utc)).total_seconds()) + 0.2 + assert delay < 35, "unexpected retry delay" + time.sleep(delay) + command("run", failed_id) + recovered = finished(failed_id) + assert recovered["stage_history"][:8] == failed["stage_history"][:8] + assert recovered["stage_history"][8]["id"] == failed["stage_history"][8]["id"] + assert recovered["request"]["lease_generation"] == 2 + print(f"PASS CLI failure in B/retry: {failed_id}", flush=True) + + interrupted_id = prepare() + process = subprocess.Popen( + [binary, "user-deletions", "run", interrupted_id, "--step-delay-ms", "30000"], + env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + try: + # Cold database connection setup has its own retry budget. Once the + # stage starts, require renewal within two heartbeat intervals. + deadline = time.monotonic() + 45 + initial = None + while time.monotonic() < deadline: + if process.poll() is not None: + stdout, stderr = process.communicate(timeout=5) + raise AssertionError( + f"worker exited before heartbeat: {process.returncode}\n" + f"{stdout[-8000:]}\n{stderr[-8000:]}" + ) + current = command("inspect", interrupted_id) + if current["stage_history"][0]["status"] == "running": + if initial is None: + initial = current + deadline = time.monotonic() + 20 + elif current["request"]["lease_until"] != initial["request"]["lease_until"]: + break + time.sleep(0.25) + else: + raise AssertionError("worker did not renew its lease while a dummy action ran") + command("run", interrupted_id, expected=5) + process.terminate() + stdout, stderr = process.communicate(timeout=15) + assert process.returncode == 0, (stdout[-8000:], stderr[-8000:]) + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=10) + interrupted = command("inspect", interrupted_id) + assert interrupted["request"]["lease_owner"] is None + assert interrupted["stage_history"][0]["is_current"] + command("run", interrupted_id) + resumed = finished(interrupted_id) + assert resumed["request"]["lease_generation"] == 2 + assert resumed["stage_history"][0]["id"] == interrupted["stage_history"][0]["id"] + assert resumed["stage_history"][0]["started_at"] == interrupted["stage_history"][0]["started_at"] + print(f"PASS CLI heartbeat/SIGTERM/resume: {interrupted_id}", flush=True) + print("PASS: all actions were simulated; no owned-community path was exercised.") + + +if __name__ == "__main__": + main()