diff --git a/crates/buzz-admin/src/user_deletions.rs b/crates/buzz-admin/src/user_deletions.rs index c8db6027fb..6508df8c6e 100644 --- a/crates/buzz-admin/src/user_deletions.rs +++ b/crates/buzz-admin/src/user_deletions.rs @@ -12,7 +12,7 @@ mod tests { use clap::Parser; #[test] - fn user_intake_commands_are_registered_without_execution() { + fn user_workflow_commands_are_registered() { for args in [ vec![ "buzz-admin", @@ -23,6 +23,31 @@ mod tests { "--requested-by", "operator-a", ], + vec![ + "buzz-admin", + "user-deletions", + "approve", + "00000000-0000-0000-0000-000000000001", + "--approved-by", + "operator-b", + "--inventory-digest", + "digest-from-inspect", + ], + vec![ + "buzz-admin", + "user-deletions", + "run", + "00000000-0000-0000-0000-000000000001", + "--next-step", + "--executor-id", + "demo-worker", + "--fail-stage", + "purge_db", + "--stage-delay-ms", + "1000", + "--demo-no-agents-in", + "00000000-0000-0000-0000-00000000000c", + ], vec!["buzz-admin", "user-deletions", "list"], vec![ "buzz-admin", @@ -33,7 +58,39 @@ mod tests { ] { assert!(crate::Cli::try_parse_from(args).is_ok()); } - for command in ["inventory", "approve", "run", "drain", "worker"] { + assert!(crate::Cli::try_parse_from([ + "buzz-admin", + "user-deletions", + "approve", + "00000000-0000-0000-0000-000000000001", + "--approved-by", + "operator-b", + ]) + .is_err()); + for (flag, value) in [ + ("--max-stages", "1"), + ("--stage-delay-ms", "60001"), + ("--fail-stage", "real_delete"), + ] { + assert!(crate::Cli::try_parse_from([ + "buzz-admin", + "user-deletions", + "run", + "00000000-0000-0000-0000-000000000001", + flag, + value + ]) + .is_err()); + } + assert!(crate::Cli::try_parse_from([ + "buzz-admin", + "user-deletions", + "run", + "00000000-0000-0000-0000-000000000001", + "--pause-after-handler" + ]) + .is_err()); + for command in ["inventory", "run", "drain", "worker"] { assert!(crate::Cli::try_parse_from(["buzz-admin", "user-deletions", command]).is_err()); } } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 7aa8607cdf..c85b7dc6b9 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -69,7 +69,7 @@ pub use store::{ 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, + replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion, workflow, }; pub use allowlist::AllowlistEntry; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index b59455b502..f1c804cf4f 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -543,7 +543,9 @@ mod postgres_tests { "product_feedback", "replica_heartbeat", "community_deletion_requests", + "entity_deletion_approvals", "entity_deletion_requests", + "entity_deletion_stage_history", "community_deletion_approvals", "community_deletion_checkpoints", "community_deletion_manifest_keys", @@ -770,7 +772,9 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 68); + assert_eq!(migrations.len(), 70); + assert_eq!(migrations[69].version, 70); + assert_eq!(migrations[68].version, 69); assert_eq!(migrations[67].version, 68); assert!(migrations[67] .sql @@ -2887,7 +2891,9 @@ mod postgres_tests { "relay_admin_actions", "relay_admin_outbox", "relay_operator_audit", + "entity_deletion_approvals", "entity_deletion_requests", + "entity_deletion_stage_history", ] { assert_eq!( columns(&desired, table).await, @@ -2912,11 +2918,17 @@ mod postgres_tests { scripts/reconcile-schema-after-pgschema.sql and scripts/repair-events-pkey.sql" ); - assert_eq!( - constraints(&desired, "entity_deletion_requests").await, - constraints(&migrated, "entity_deletion_requests").await, - "constraint parity mismatch for entity_deletion_requests" - ); + for table in [ + "entity_deletion_requests", + "entity_deletion_stage_history", + "entity_deletion_approvals", + ] { + assert_eq!( + constraints(&desired, table).await, + constraints(&migrated, table).await, + "constraint parity mismatch for {table}" + ); + } desired.close().await; migrated.close().await; diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index c5d6245127..3ef4fdfb33 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -8507,6 +8507,37 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .as_uuid() .to_owned(); seed_relay_admin_rows(&db, bystander).await; + // Operator workflow history is retained even when its provenance + // community is purged; it is not a tenant-owned serving table. + let entity = db + .user_deletion_store() + .submit("demo:history-retention", "operator", None) + .await + .unwrap(); + db.entity_deletion_store() + .approve( + entity.id, + "approver", + entity.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let approvals_before = db + .entity_deletion_store() + .inspect(entity.id) + .await + .unwrap() + .approvals; + let history_id: Uuid = sqlx::query_scalar( + "INSERT INTO entity_deletion_stage_history + (request_id, target_community_id, stage, is_current) + VALUES ($1, $2, 'drain', false) RETURNING id", + ) + .bind(entity.id) + .bind(community) + .fetch_one(&db.pool) + .await + .unwrap(); let live = store .inventory_schema(request.community_id) .await @@ -8540,6 +8571,25 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); .expect("bindings"); let deleted = store.purge_postgres(&token).await.expect("purge postgres"); + let retained: Uuid = sqlx::query_scalar( + "SELECT target_community_id FROM entity_deletion_stage_history WHERE id = $1", + ) + .bind(history_id) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(retained, community); + let approvals_after = db + .entity_deletion_store() + .inspect(entity.id) + .await + .unwrap() + .approvals; + assert_eq!( + serde_json::to_value(approvals_before).unwrap(), + serde_json::to_value(approvals_after).unwrap() + ); + assert_eq!(deleted.get("relay_admin_outbox"), Some(&2)); assert_eq!(deleted.get("relay_admin_actions"), Some(&2)); assert_eq!(relay_admin_row_counts(&db, community).await, (0, 0)); diff --git a/crates/buzz-db/src/store/entity_deletion.rs b/crates/buzz-db/src/store/entity_deletion.rs deleted file mode 100644 index c2d5a07b69..0000000000 --- a/crates/buzz-db/src/store/entity_deletion.rs +++ /dev/null @@ -1,130 +0,0 @@ -//! Operator-global intake for synthetic user-deletion requests. - -use chrono::{DateTime, Utc}; -use serde::Serialize; -use sqlx::PgPool; -use uuid::Uuid; - -use crate::{Db, DbError, Result}; - -/// A submitted demonstration request; no cleanup is authorized by this record. -#[derive(Debug, Serialize, sqlx::FromRow)] -pub struct EntityRequest { - /// Stable request identifier. - pub id: Uuid, - /// Currently always `user`. - pub entity_type: String, - /// Synthetic `demo:` label, not a resolved account identity. - pub target: String, - /// Currently always `submitted`. - pub status: String, - /// Operator identity supplied at submission. - pub requested_by: String, - /// Optional explanation supplied at submission. - pub reason: Option, - /// Time of the original submission. - pub created_at: DateTime, - /// Time of the last change; identical resubmissions do not change it. - pub updated_at: DateTime, -} - -/// Writer-database access to synthetic deletion requests. -#[derive(Clone)] -pub struct EntityDeletionStore { - pool: PgPool, -} - -impl Db { - /// Use the writer pool for operator-global request intake and inspection. - pub fn entity_deletion_store(&self) -> EntityDeletionStore { - EntityDeletionStore { - pool: self.pool().clone(), - } - } -} - -impl EntityDeletionStore { - /// Save a request or return an identical previous submission. - /// - /// A conflicting operator or reason returns an error without changing the - /// existing request. Concurrent identical submissions converge on one row. - pub async fn submit( - &self, - target: &str, - requested_by: &str, - reason: Option<&str>, - ) -> Result { - validate_submission(target, requested_by, reason)?; - 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(DbError::DeletionSafety(format!( - "target already has request {}; inspect it before resubmission", - request.id - ))); - } - 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(|| DbError::NotFound(format!("entity deletion request {id}"))) - } - - /// List up to 1,000 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?) - } -} - -fn validate_submission(target: &str, requested_by: &str, reason: Option<&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(DbError::DeletionSafety( - "target must be demo:<1..64 lowercase letters, digits, underscores or hyphens>".into(), - )); - } - if requested_by.is_empty() || requested_by.len() > 256 || requested_by.trim() != requested_by { - return Err(DbError::DeletionSafety( - "operator identity must be 1..256 bytes without surrounding whitespace".into(), - )); - } - if reason.is_some_and(|r| r.len() > 4096) { - return Err(DbError::DeletionSafety("reason exceeds 4096 bytes".into())); - } - Ok(()) -} - -#[cfg(test)] -#[path = "entity_deletion_postgres_tests.rs"] -mod postgres_tests; diff --git a/crates/buzz-db/src/store/entity_deletion/approvals.rs b/crates/buzz-db/src/store/entity_deletion/approvals.rs new file mode 100644 index 0000000000..a507b58648 --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/approvals.rs @@ -0,0 +1,128 @@ +//! Approval evidence bound to the saved inventory fingerprint. +use super::requests::REQUEST_COLUMNS; +use super::*; + +/// One retained approval of a particular request inventory. +#[derive(Debug, Serialize, sqlx::FromRow)] +pub struct EntityApproval { + /// Stable identifier; identical approval retries retain it. + pub id: Uuid, + /// Parent request, checked and locked by the approval transaction. + pub request_id: Uuid, + /// SHA-256 of the inventory this operator approved. + pub inventory_digest: String, + /// Self-asserted demo identity, not authenticated real-deletion authorization. + pub approved_by: String, + /// Time this approval was first recorded. + pub approved_at: DateTime, +} + +impl EntityDeletionStore { + /// Record approval of the saved inventory without creating or executing stages. + /// + /// An identical retry is read-only. Another operator adds retained evidence + /// without restarting the workflow. The operator identity is self-asserted and + /// this demo-only boundary does not authorize real user deletion. + pub async fn approve( + &self, + id: Uuid, + approved_by: &str, + expected_inventory_digest: &str, + ) -> Result { + self.approve_validated(id, approved_by, expected_inventory_digest, |_| Ok(())) + .await + } + + pub(crate) async fn approve_validated( + &self, + id: Uuid, + approved_by: &str, + expected_inventory_digest: &str, + validate: F, + ) -> Result + where + F: FnOnce(&EntityRequest) -> Result<()> + Send, + { + validate_operator(approved_by)?; + let mut tx = self.pool.begin().await?; + let request: EntityRequest = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + 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}")))?; + validate(&request)?; + let digest = validated_inventory_digest(&request)?; + if expected_inventory_digest != digest { + return Err(DbError::DeletionSafety( + "approval does not match saved inventory".into(), + )); + } + let has_history: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM entity_deletion_stage_history WHERE request_id = $1)", + ) + .bind(id) + .fetch_one(&mut *tx) + .await?; + if has_history { + return Err(DbError::DeletionSafety( + "approval cannot change a started workflow".into(), + )); + } + if request.status == "approved" { + let has_approval: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM entity_deletion_approvals + WHERE request_id = $1 AND inventory_digest = $2)", + ) + .bind(id) + .bind(&digest) + .fetch_one(&mut *tx) + .await?; + if !has_approval { + return Err(DbError::DeletionSafety( + "approved request has no approval for its saved inventory".into(), + )); + } + save_approval(&mut tx, id, approved_by, &digest).await?; + tx.commit().await?; + return Ok(request); + } + if request.status != "inventoried" { + return Err(DbError::DeletionSafety("request is not inventoried".into())); + } + let request = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "UPDATE entity_deletion_requests + SET status = 'approved', updated_at = now() + WHERE id = $1 + RETURNING {REQUEST_COLUMNS}", + ))) + .bind(id) + .fetch_one(&mut *tx) + .await?; + save_approval(&mut tx, id, approved_by, &digest).await?; + tx.commit().await?; + Ok(request) + } +} + +// The caller holds the parent row lock and has validated inventory and lifecycle. +async fn save_approval( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + request_id: Uuid, + approved_by: &str, + inventory_digest: &str, +) -> Result<()> { + sqlx::query( + "INSERT INTO entity_deletion_approvals (request_id, inventory_digest, approved_by) + VALUES ($1, $2, $3) + ON CONFLICT (request_id, inventory_digest, approved_by) DO NOTHING", + ) + .bind(request_id) + .bind(inventory_digest) + .bind(approved_by) + .execute(&mut **tx) + .await?; + Ok(()) +} diff --git a/crates/buzz-db/src/store/entity_deletion/leases.rs b/crates/buzz-db/src/store/entity_deletion/leases.rs new file mode 100644 index 0000000000..a8b88ef9ae --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/leases.rs @@ -0,0 +1,206 @@ +//! Parent-request ownership and stage attempts for the synthetic workflow. + +use std::time::Duration; + +use super::requests::REQUEST_COLUMNS; +use super::*; + +pub(super) type Tx<'a> = sqlx::Transaction<'a, sqlx::Postgres>; + +/// Ownership token; this generation fences executor writes, not serving traffic. +#[derive(Debug, Clone)] +pub struct EntityLease { + /// Claimed parent request. + pub request_id: Uuid, + /// Worker identity recorded on the parent. + pub owner: String, + /// Incremented whenever a worker claims or reclaims the request. + pub generation: i64, +} + +/// Execution fields stored on the parent, exposed by inspection. +#[derive(Debug, Serialize, sqlx::FromRow)] +pub struct EntityExecution { + /// Worker that currently owns the request, if any. + pub lease_owner: Option, + /// Monotonic claim generation. + pub lease_generation: i64, + /// Database-clock deadline for ownership. + pub lease_expires_at: Option>, + /// Most recent claim or heartbeat time. + pub heartbeat_at: Option>, + /// Recorded failures of the current stage; reset on advancement. + pub retry_count: i32, + /// Earliest time another run may retry a failed stage. + pub next_attempt_at: Option>, + /// Last recorded stage failure. + pub last_error: Option, + /// Time all synthetic stages completed. + pub completed_at: Option>, +} + +pub(super) async fn read_execution(tx: &mut Tx<'_>, id: Uuid) -> Result { + Ok(sqlx::query_as( + "SELECT lease_owner, lease_generation, lease_expires_at, heartbeat_at, + retry_count, next_attempt_at, last_error, completed_at + FROM entity_deletion_requests WHERE id = $1", + ) + .bind(id) + .fetch_one(&mut **tx) + .await?) +} + +impl EntityDeletionStore { + /// Claim one approved, due, unleased request; concurrent claimers cannot both win. + /// + /// Expired running requests resume their current row with a new generation. + /// A completed, unapproved, not-yet-due, or already-owned request returns `None`. + pub async fn claim( + &self, + id: Uuid, + owner: &str, + lease_duration: Duration, + ) -> Result> { + validate_operator(owner)?; + let millis = duration_millis(lease_duration)?; + let mut tx = self.pool.begin().await?; + let request: Option = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + FROM entity_deletion_requests + WHERE id = $1 AND status IN ('approved', 'running', 'retry_wait') + AND (lease_expires_at IS NULL OR lease_expires_at <= clock_timestamp()) + AND (next_attempt_at IS NULL OR next_attempt_at <= clock_timestamp()) + FOR UPDATE SKIP LOCKED", + ))) + .bind(id) + .fetch_optional(&mut *tx) + .await?; + let Some(request) = request else { + return Ok(None); + }; + validate_approved_inventory(&mut tx, &request).await?; + let generation = sqlx::query_scalar( + "UPDATE entity_deletion_requests + SET lease_owner = $2, + lease_generation = lease_generation + 1, + lease_expires_at = clock_timestamp() + $3 * interval '1 millisecond', + heartbeat_at = clock_timestamp(), + updated_at = clock_timestamp() + WHERE id = $1 RETURNING lease_generation", + ) + .bind(id) + .bind(owner) + .bind(millis) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(Some(EntityLease { + request_id: id, + owner: owner.to_owned(), + generation, + })) + } + + /// Renew an unexpired lease. An expired or superseded worker cannot revive it. + pub async fn heartbeat(&self, lease: &EntityLease, duration: Duration) -> Result<()> { + let millis = duration_millis(duration)?; + let mut tx = self.pool.begin().await?; + lock_owned(&mut tx, lease).await?; + sqlx::query( + "UPDATE entity_deletion_requests + SET lease_expires_at = clock_timestamp() + $2 * interval '1 millisecond', + heartbeat_at = clock_timestamp() + WHERE id = $1", + ) + .bind(lease.request_id) + .bind(millis) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + /// Release this live claim after a pause or shutdown, without advancing its stage. + /// + /// Returns false after ownership loss; never clears a successor's lease. + pub async fn release(&self, lease: &EntityLease) -> Result { + Ok(sqlx::query( + "UPDATE entity_deletion_requests SET lease_owner = NULL, lease_expires_at = NULL, + updated_at = clock_timestamp() + WHERE id = $1 AND lease_owner = $2 AND lease_generation = $3 + AND lease_expires_at > clock_timestamp()", + ) + .bind(lease.request_id) + .bind(&lease.owner) + .bind(lease.generation) + .execute(&self.pool) + .await? + .rows_affected() + == 1) + } +} + +pub(super) async fn lock_owned(tx: &mut Tx<'_>, lease: &EntityLease) -> Result { + // Lock by id first, then check expiry after any wait for that lock. An old + // snapshot/transaction start time must not extend a worker's authority. + let request: EntityRequest = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + FROM entity_deletion_requests WHERE id = $1 FOR UPDATE", + ))) + .bind(lease.request_id) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| stale_lease(lease))?; + let owned: bool = sqlx::query_scalar( + "SELECT COALESCE(status IN ('approved', 'running', 'retry_wait', 'completed') AND lease_owner = $2 AND lease_generation = $3 + AND lease_expires_at > clock_timestamp(), false) + FROM entity_deletion_requests WHERE id = $1", + ) + .bind(lease.request_id) + .bind(&lease.owner) + .bind(lease.generation) + .fetch_one(&mut **tx) + .await?; + if !owned { + return Err(stale_lease(lease)); + } + Ok(request) +} + +pub(super) async fn validate_approved_inventory( + tx: &mut Tx<'_>, + request: &EntityRequest, +) -> Result<()> { + let digest = validated_inventory_digest(request)?; + let approved: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM entity_deletion_approvals + WHERE request_id = $1 AND inventory_digest = $2)", + ) + .bind(request.id) + .bind(digest) + .fetch_one(&mut **tx) + .await?; + if !approved { + return Err(DbError::DeletionSafety( + "saved inventory has no approval".into(), + )); + } + Ok(()) +} + +pub(super) fn duration_millis(duration: Duration) -> Result { + if duration.is_zero() || duration > Duration::from_secs(3600) || duration.as_millis() == 0 { + return Err(DbError::InvalidData( + "lease/retry duration must be 1ms..1h".into(), + )); + } + Ok(duration.as_millis() as i64) +} + +fn stale_lease(lease: &EntityLease) -> DbError { + // Use the existing deletion lease classifier for executor control flow. + DbError::AccessDenied(format!( + "stale deletion lease {} owner {:?} generation {}", + lease.request_id, lease.owner, lease.generation + )) +} 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..2090b0de7e --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/mod.rs @@ -0,0 +1,35 @@ +//! Shared deletion-workflow persistence. User policy lives in buzz-deletion. +use crate::{Db, DbError, Result}; +use chrono::{DateTime, Utc}; +use serde::Serialize; +use sqlx::PgPool; +use uuid::Uuid; + +/// Writer-database access to synthetic deletion requests. +#[derive(Clone)] +pub struct EntityDeletionStore { + pool: PgPool, +} + +impl Db { + /// Use the writer pool for operator-global request intake and inspection. + pub fn entity_deletion_store(&self) -> EntityDeletionStore { + EntityDeletionStore { + pool: self.pool().clone(), + } + } +} + +mod approvals; +mod leases; +pub(crate) mod requests; +mod stages; +pub use approvals::EntityApproval; +pub use leases::{EntityExecution, EntityLease}; +pub use requests::{inventory_digest, EntityInspection, EntityRequest}; +use requests::{validate_operator, validated_inventory_digest}; +pub use stages::{EntityStage, NewEntityStage, NextStage, StageTransition}; + +#[cfg(test)] +#[path = "../entity_deletion_postgres_tests.rs"] +mod postgres_tests; diff --git a/crates/buzz-db/src/store/entity_deletion/requests.rs b/crates/buzz-db/src/store/entity_deletion/requests.rs new file mode 100644 index 0000000000..b8a59a254b --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/requests.rs @@ -0,0 +1,142 @@ +//! Shared request reads, inventory fingerprints, and consistent inspection. +use super::*; +use sha2::{Digest, Sha256}; + +// Shared by SELECT and RETURNING queries that decode EntityRequest. +// This constant is the only SQL interpolation; all request values stay bound. +pub(crate) const REQUEST_COLUMNS: &str = + "id, entity_type, target, status, requested_by, reason, created_at, + updated_at, inventory_manifest, inventory_digest"; + +/// A demonstration request; approval authorizes only the synthetic workflow. +#[derive(Debug, Serialize, sqlx::FromRow)] +pub struct EntityRequest { + /// Stable request identifier. + pub id: Uuid, + /// Currently always `user`. + pub entity_type: String, + /// Synthetic `demo:` label, not a resolved account identity. + pub target: String, + /// Intake, approval, execution (`running`/`retry_wait`), or `completed`. + pub status: String, + /// Operator identity supplied at submission. + pub requested_by: String, + /// Optional explanation supplied at submission. + pub reason: Option, + /// Time of the original submission. + pub created_at: DateTime, + /// Time of the last change; identical resubmissions do not change it. + pub updated_at: DateTime, + /// Saved synthetic communities in processing order; absent on legacy intake. + pub inventory_manifest: Option, + /// SHA-256 of the saved inventory. + pub inventory_digest: Option, +} + +/// A consistent snapshot of a request, its approvals, and reached stages. +#[derive(Debug, Serialize)] +pub struct EntityInspection { + /// Request fields remain at the top level of CLI JSON. + #[serde(flatten)] + pub request: EntityRequest, + /// Parent ownership, heartbeat, and retry state. + pub execution: EntityExecution, + /// Retained approvals, ordered by approval time and identifier. + pub approvals: Vec, + /// Reached stages, including the current resume position after execution starts. + pub stage_history: Vec, +} + +impl EntityDeletionStore { + /// Inspect the parent and its history from one database snapshot. + pub async fn inspect(&self, id: Uuid) -> Result { + let mut tx = self.pool.begin().await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY") + .execute(&mut *tx) + .await?; + let request = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + FROM entity_deletion_requests WHERE id = $1", + ))) + .bind(id) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| DbError::NotFound(format!("entity deletion request {id}")))?; + let execution = leases::read_execution(&mut tx, id).await?; + let approvals = sqlx::query_as( + "SELECT id, request_id, inventory_digest, approved_by, approved_at + FROM entity_deletion_approvals WHERE request_id = $1 + ORDER BY approved_at, id", + ) + .bind(id) + .fetch_all(&mut *tx) + .await?; + let stage_history = sqlx::query_as( + "SELECT id, request_id, target_community_id, stage, status, is_current, + started_at, finished_at, metadata_json, last_error, updated_at + FROM entity_deletion_stage_history WHERE request_id = $1 + ORDER BY started_at NULLS LAST, id", + ) + .bind(id) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + Ok(EntityInspection { + request, + execution, + approvals, + stage_history, + }) + } + + /// Load a request from the authoritative writer database. + pub async fn get(&self, id: Uuid) -> Result { + sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + FROM entity_deletion_requests WHERE id = $1", + ))) + .bind(id) + .fetch_optional(&self.pool) + .await? + .ok_or_else(|| DbError::NotFound(format!("entity deletion request {id}"))) + } + + /// List up to 1,000 requests, newest first. + pub async fn list(&self, limit: u16) -> Result> { + Ok(sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + FROM entity_deletion_requests ORDER BY created_at DESC, id LIMIT $1", + ))) + .bind(i64::from(limit.clamp(1, 1000))) + .fetch_all(&self.pool) + .await?) + } +} +pub(crate) fn validate_operator(operator: &str) -> Result<()> { + if operator.is_empty() || operator.len() > 256 || operator.trim() != operator { + return Err(DbError::DeletionSafety( + "operator identity must be 1..256 bytes without surrounding whitespace".into(), + )); + } + Ok(()) +} + +/// Compute a stable fingerprint of saved inventory, preserving array order. +pub fn inventory_digest(inventory: &serde_json::Value) -> Result { + // JSONB does not preserve object-key order. Sorting preserves the digest + // across a database round trip while retaining the community array order. + let mut canonical = inventory.clone(); + canonical.sort_all_objects(); + Ok(hex::encode(Sha256::digest(serde_json::to_vec(&canonical)?))) +} +pub(super) fn validated_inventory_digest(request: &EntityRequest) -> Result { + let inventory = request + .inventory_manifest + .as_ref() + .ok_or_else(|| super::stages::invalid("missing inventory"))?; + let digest = inventory_digest(inventory)?; + if request.inventory_digest.as_deref() != Some(&digest) { + return Err(super::stages::invalid("saved inventory digest changed")); + } + Ok(digest) +} 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..d31be5d12b --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/stages.rs @@ -0,0 +1,360 @@ +//! Stage records, handler outcomes, and atomic transition persistence. +use super::leases::*; +use super::*; +use std::time::Duration; + +/// A reached stage. Retries will update the same row, not append attempts. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct EntityStage { + /// Stable stage-history row identifier. + pub id: Uuid, + /// Parent workflow; application transactions enforce this relationship. + pub request_id: Uuid, + /// Community provenance, or null for a request-wide stage. + pub target_community_id: Option, + /// Action to dispatch. + pub stage: String, + /// `pending`, `running`, `failed`, or `completed`. + pub status: String, + /// Whether this is the workflow's resume position. + pub is_current: bool, + /// First execution time; pending stages have not started. + pub started_at: Option>, + /// Successful completion time. + pub finished_at: Option>, + /// Stage-specific saved progress. + pub metadata_json: serde_json::Value, + /// Most recent failure, if any. + pub last_error: Option, + /// Last stage update. + pub updated_at: DateTime, +} + +impl EntityDeletionStore { + /// Start the supplied current stage, preserving its first start time on retry. + pub async fn start_stage(&self, lease: &EntityLease, stage_id: Uuid) -> Result { + let mut tx = self.pool.begin().await?; + let request = lock_owned(&mut tx, lease).await?; + validate_approved_inventory(&mut tx, &request).await?; + let stage = current_stage(&mut tx, lease.request_id).await?; + if stage.id != stage_id + || !matches!(stage.status.as_str(), "pending" | "running" | "failed") + { + return Err(DbError::DeletionSafety( + "stage is no longer executable".into(), + )); + } + let due: bool = sqlx::query_scalar( + "SELECT next_attempt_at IS NULL OR next_attempt_at <= clock_timestamp() + FROM entity_deletion_requests WHERE id = $1", + ) + .bind(lease.request_id) + .fetch_one(&mut *tx) + .await?; + if !due { + return Err(DbError::DeletionSafety("stage retry is not due".into())); + } + sqlx::query("UPDATE entity_deletion_requests SET status = 'running', next_attempt_at = NULL WHERE id = $1") + .bind(lease.request_id).execute(&mut *tx).await?; + let stage = sqlx::query_as( + "UPDATE entity_deletion_stage_history + SET status = 'running', started_at = COALESCE(started_at, clock_timestamp()), + updated_at = clock_timestamp() + WHERE id = $1 + RETURNING id, request_id, target_community_id, stage, status, is_current, + started_at, finished_at, metadata_json, last_error, updated_at", + ) + .bind(stage.id) + .fetch_one(&mut *tx) + .await?; + tx.commit().await?; + Ok(stage) + } + + /// Record failure and its retry deadline; get_next_stage returns WaitUntil. + /// Ownership is released by the worker after it observes that outcome. + pub async fn fail_stage( + &self, + lease: &EntityLease, + stage_id: Uuid, + message: &str, + retry_delay: Duration, + ) -> Result<()> { + let millis = duration_millis(retry_delay)?; + let message: String = message.chars().take(2048).collect(); + let mut tx = self.pool.begin().await?; + lock_owned(&mut tx, lease).await?; + require_running_stage(&mut tx, lease, stage_id).await?; + sqlx::query( + "UPDATE entity_deletion_stage_history + SET status = 'failed', last_error = $2, updated_at = clock_timestamp() + WHERE id = $1", + ) + .bind(stage_id) + .bind(&message) + .execute(&mut *tx) + .await?; + sqlx::query( + "UPDATE entity_deletion_requests + SET status = 'retry_wait', retry_count = retry_count + 1, last_error = $2, + next_attempt_at = clock_timestamp() + $3 * interval '1 millisecond', + updated_at = clock_timestamp() + WHERE id = $1", + ) + .bind(lease.request_id) + .bind(message) + .bind(millis) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + /// Save a handler's result without switching stages. A completed current row + /// is the durable journal for a transition that has not committed yet. + pub async fn save_stage_success( + &self, + lease: &EntityLease, + stage_id: Uuid, + result: serde_json::Value, + ) -> Result<()> { + let mut tx = self.pool.begin().await?; + let request = lock_owned(&mut tx, lease).await?; + validate_approved_inventory(&mut tx, &request).await?; + require_running_stage(&mut tx, lease, stage_id).await?; + sqlx::query( + "UPDATE entity_deletion_stage_history SET status = 'completed', + finished_at = clock_timestamp(), last_error = NULL, + metadata_json = jsonb_set(metadata_json, '{result}', $2), + updated_at = clock_timestamp() WHERE id = $1", + ) + .bind(stage_id) + .bind(result) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } +} +pub(super) async fn current_stage(tx: &mut Tx<'_>, id: Uuid) -> Result { + let stage: EntityStage = sqlx::query_as( + "SELECT id, request_id, target_community_id, stage, status, is_current, + started_at, finished_at, metadata_json, last_error, updated_at + FROM entity_deletion_stage_history WHERE request_id = $1 AND is_current", + ) + .bind(id) + .fetch_optional(&mut **tx) + .await? + .ok_or_else(|| DbError::DeletionSafety("unfinished request has no current stage".into()))?; + Ok(stage) +} + +pub(super) async fn require_running_stage( + tx: &mut Tx<'_>, + lease: &EntityLease, + stage_id: Uuid, +) -> Result { + let stage = current_stage(tx, lease.request_id).await?; + if stage.id != stage_id || stage.status != "running" { + return Err(DbError::DeletionSafety( + "stage is no longer current and running".into(), + )); + } + Ok(stage) +} + +/// Inputs for a stage chosen by the caller's workflow policy. +pub struct NewEntityStage { + /// Handler name; interpreted by the deletion worker, not the store. + pub stage: String, + /// Community provenance, absent for request-wide work. + pub community: Option, + /// Saved handler arguments. + pub inputs: serde_json::Value, +} + +/// A workflow decision applied while the request's database lock is held. +pub enum StageTransition { + /// Create the first stage of an approved request. + Initialize(NewEntityStage), + /// Return the current stage, or its retry deadline if not yet due. + Resume(Uuid), + /// Retire a completed current stage and create its successor, or finish. + Advance { + /// The completed stage whose result was interpreted. + previous: Uuid, + /// Workflow-specific decision retained for recovery and audit. + decision: serde_json::Value, + /// None records successful request completion. + successor: Option, + }, + /// Return an already completed request without changing it. + Finished, +} + +/// The saved position returned to a deletion worker. +#[derive(Debug)] +pub enum NextStage { + /// Full saved stage row, including its inputs. + Stage(EntityStage), + /// Request completion has been saved. + Finished, + /// Saved retry deadline has not arrived. + WaitUntil(DateTime), +} + +pub(super) fn invalid(message: &str) -> DbError { + DbError::DeletionSafety(message.into()) +} + +impl EntityDeletionStore { + /// Lock the request, let the caller decide from its saved state, and persist + /// that decision atomically. The synchronous callback must only compute a + /// decision; it must not perform external work. No stage policy lives here. + pub async fn transition_stage( + &self, + lease: &EntityLease, + previous_stage_id: Option, + decide: F, + ) -> Result + where + F: FnOnce(&EntityRequest, &[EntityStage]) -> Result + Send, + { + let mut tx = self.pool.begin().await?; + let request = lock_owned(&mut tx, lease).await?; + validate_approved_inventory(&mut tx, &request).await?; + let history: Vec = sqlx::query_as( + "SELECT id, request_id, target_community_id, stage, status, is_current, + started_at, finished_at, metadata_json, last_error, updated_at + FROM entity_deletion_stage_history WHERE request_id = $1 LIMIT 10001", + ) + .bind(request.id) + .fetch_all(&mut *tx) + .await?; + if history.len() > 10000 { + return Err(invalid("workflow exceeds 10000 reached stages")); + } + if previous_stage_id.is_some_and(|id| !history.iter().any(|row| row.id == id)) { + return Err(invalid("previous stage does not belong to this request")); + } + let current = history.iter().find(|row| row.is_current); + let next = match decide(&request, &history)? { + StageTransition::Finished => { + if request.status != "completed" || current.is_some() { + return Err(invalid("request is not completed")); + } + return Ok(NextStage::Finished); + } + StageTransition::Initialize(stage) => { + if request.status != "approved" || !history.is_empty() { + return Err(invalid( + "only an approved request without history can initialize", + )); + } + let row = insert_stage(&mut tx, request.id, None, stage).await?; + sqlx::query( + "UPDATE entity_deletion_requests + SET status = 'running', updated_at = clock_timestamp() WHERE id = $1", + ) + .bind(request.id) + .execute(&mut *tx) + .await?; + NextStage::Stage(row) + } + StageTransition::Resume(id) => { + let row = current + .filter(|row| row.id == id) + .ok_or_else(|| invalid("resume stage is not current"))?; + match row.status.as_str() { + "pending" | "running" if request.status == "running" => { + return Ok(NextStage::Stage(row.clone())); + } + "failed" => { + let retry: Option<(DateTime, bool)> = sqlx::query_as( + "SELECT next_attempt_at, next_attempt_at > clock_timestamp() + FROM entity_deletion_requests + WHERE id = $1 AND status = 'retry_wait' AND next_attempt_at IS NOT NULL", + ) + .bind(request.id) + .fetch_optional(&mut *tx) + .await?; + let (deadline, waiting) = + retry.ok_or_else(|| invalid("failed stage has no retry record"))?; + return Ok(if waiting { + NextStage::WaitUntil(deadline) + } else { + NextStage::Stage(row.clone()) + }); + } + _ => return Err(invalid("current stage is not executable")), + } + } + StageTransition::Advance { + previous, + decision, + successor, + } => { + let row = current + .filter(|row| row.id == previous && row.status == "completed") + .ok_or_else(|| invalid("only a completed current stage can advance"))?; + if request.status != "running" { + return Err(invalid("completed stage has inconsistent parent status")); + } + sqlx::query( + "UPDATE entity_deletion_stage_history SET is_current = false, + metadata_json = jsonb_set(metadata_json, '{transition}', $2), + updated_at = clock_timestamp() WHERE id = $1", + ) + .bind(row.id) + .bind(decision) + .execute(&mut *tx) + .await?; + let next = match successor { + Some(stage) => NextStage::Stage( + insert_stage(&mut tx, request.id, Some(row.id), stage).await?, + ), + None => NextStage::Finished, + }; + sqlx::query( + "UPDATE entity_deletion_requests + SET status = CASE WHEN $2 THEN 'completed' ELSE 'running' END, + completed_at = CASE WHEN $2 THEN clock_timestamp() ELSE NULL END, + retry_count = 0, next_attempt_at = NULL, last_error = NULL, + updated_at = clock_timestamp() WHERE id = $1", + ) + .bind(request.id) + .bind(matches!(next, NextStage::Finished)) + .execute(&mut *tx) + .await?; + next + } + }; + tx.commit().await?; + Ok(next) + } +} + +async fn insert_stage( + tx: &mut Tx<'_>, + request_id: Uuid, + previous: Option, + stage: NewEntityStage, +) -> Result { + let metadata = serde_json::json!({"inputs": stage.inputs, "previous_stage_id": previous}); + Ok(sqlx::query_as( + "INSERT INTO entity_deletion_stage_history (request_id, target_community_id, stage, metadata_json) + VALUES ($1, $2, $3, $4) + RETURNING id, request_id, target_community_id, stage, status, is_current, + started_at, finished_at, metadata_json, last_error, updated_at", + ) + .bind(request_id) + .bind(stage.community) + .bind(stage.stage) + .bind(metadata) + .fetch_one(&mut **tx) + .await?) +} + +#[cfg(test)] +#[path = "stages_postgres_tests.rs"] +mod postgres_tests; diff --git a/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs b/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs new file mode 100644 index 0000000000..b2c19b7bfa --- /dev/null +++ b/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs @@ -0,0 +1,83 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn stage_storage_accepts_caller_policy_and_rolls_back_invalid_decisions() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool); + let request = db + .user_deletion_store() + .submit("demo:generic-storage", "operator", None) + .await + .unwrap(); + let store = db.entity_deletion_store(); + store + .approve( + request.id, + "approver", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = store + .claim(request.id, "worker", Duration::from_secs(60)) + .await + .unwrap() + .unwrap(); + let before = serde_json::to_value(store.inspect(request.id).await.unwrap()).unwrap(); + assert!(store + .transition_stage(&lease, None, |_, _| Err(invalid("policy refused"))) + .await + .is_err()); + assert_eq!( + serde_json::to_value(store.inspect(request.id).await.unwrap()).unwrap(), + before + ); + let NextStage::Stage(row) = store + .transition_stage(&lease, None, |request, history| { + assert!(history.is_empty()); + assert_eq!(request.status, "approved"); + Ok(StageTransition::Initialize(NewEntityStage { + stage: "arbitrary_policy_stage".into(), + community: None, + inputs: serde_json::json!({"example": true}), + })) + }) + .await + .unwrap() + else { + panic!("expected stage"); + }; + assert_eq!(row.stage, "arbitrary_policy_stage"); + assert!(store + .transition_stage(&lease, Some(row.id), |_, _| Ok(StageTransition::Advance { + previous: row.id, + decision: serde_json::json!({"kind": "done"}), + successor: None, + })) + .await + .is_err()); // pending work cannot be skipped by an invalid decision + store.start_stage(&lease, row.id).await.unwrap(); + store + .save_stage_success(&lease, row.id, serde_json::json!({"done": true})) + .await + .unwrap(); + assert!(matches!( + store + .transition_stage(&lease, Some(row.id), |_, history| { + assert_eq!(history[0].metadata_json["result"]["done"], true); + Ok(StageTransition::Advance { + previous: row.id, + decision: serde_json::json!({"kind": "done"}), + successor: None, + }) + }) + .await + .unwrap(), + NextStage::Finished + )); + assert_eq!(store.get(request.id).await.unwrap().status, "completed"); + assert!(store.release(&lease).await.unwrap()); +} diff --git a/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs b/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs index 32a15cec24..51a98f7421 100644 --- a/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs +++ b/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs @@ -11,7 +11,9 @@ async fn store() -> EntityDeletionStore { async fn assert_plain_schema(pool: &PgPool) { let fences: i64 = sqlx::query_scalar( "SELECT count(*) FROM pg_trigger - WHERE tgrelid = 'entity_deletion_requests'::regclass", + WHERE tgrelid IN ('entity_deletion_requests'::regclass, + 'entity_deletion_stage_history'::regclass, + 'entity_deletion_approvals'::regclass)", ) .fetch_one(pool) .await @@ -19,7 +21,9 @@ async fn assert_plain_schema(pool: &PgPool) { assert_eq!(fences, 0); let foreign_keys: i64 = sqlx::query_scalar( "SELECT count(*) FROM pg_constraint - WHERE conrelid = 'entity_deletion_requests'::regclass AND contype = 'f'", + WHERE conrelid IN ('entity_deletion_requests'::regclass, + 'entity_deletion_stage_history'::regclass, + 'entity_deletion_approvals'::regclass) AND contype = 'f'", ) .fetch_one(pool) .await @@ -31,22 +35,23 @@ async fn assert_plain_schema(pool: &PgPool) { #[ignore = "requires Postgres"] async fn concurrent_submissions_converge_and_conflicts_preserve_request() { let store = store().await; + let submissions = users(&store); let (first, second) = tokio::join!( - store.submit("demo:alice", "operator-a", Some("manual test")), - store.submit("demo:alice", "operator-a", Some("manual test")), + submissions.submit("demo:alice", "operator-a", Some("manual test")), + submissions.submit("demo:alice", "operator-a", Some("manual test")), ); let first = first.expect("first submission"); let second = second.expect("concurrent submission"); assert_eq!(first.id, second.id); - assert_eq!(first.status, "submitted"); + assert_eq!(first.status, "inventoried"); assert_eq!(first.entity_type, "user"); assert_eq!(first.created_at, second.created_at); assert_eq!(first.updated_at, second.updated_at); - assert!(store + assert!(users(&store) .submit("demo:alice", "operator-b", Some("manual test")) .await .is_err()); - assert!(store + assert!(users(&store) .submit("demo:alice", "operator-a", None) .await .is_err()); @@ -56,7 +61,7 @@ async fn concurrent_submissions_converge_and_conflicts_preserve_request() { serde_json::to_value(first).unwrap() ); assert_eq!(store.list(100).await.expect("list").len(), 1); - store + users(&store) .submit("demo:bob", "operator-a", None) .await .expect("second target"); @@ -81,22 +86,28 @@ async fn invalid_submissions_do_not_create_rows() { "demo:alice@example.com", ] { assert!( - store.submit(target, "operator-a", None).await.is_err(), + users(&store) + .submit(target, "operator-a", None) + .await + .is_err(), "{target}" ); } - assert!(store + assert!(users(&store) .submit(&format!("demo:{}", "a".repeat(65)), "operator-a", None) .await .is_err()); for actor in ["", " operator-a", "operator-a "] { - assert!(store.submit("demo:alice", actor, None).await.is_err()); + assert!(users(&store) + .submit("demo:alice", actor, None) + .await + .is_err()); } - assert!(store + assert!(users(&store) .submit("demo:alice", &"a".repeat(257), None) .await .is_err()); - assert!(store + assert!(users(&store) .submit("demo:alice", "operator-a", Some(&"a".repeat(4097))) .await .is_err()); @@ -121,12 +132,12 @@ async fn desired_schema_has_plain_intake_without_database_guards() { .is_err()); assert!(sqlx::query( "INSERT INTO entity_deletion_requests (target, requested_by, status) - VALUES ('demo:alice', 'operator-a', 'running')", + VALUES ('demo:alice', 'operator-a', 'unsupported')", ) .execute(&store.pool) .await .is_err()); - store + users(&store) .submit("demo:alice", "operator-a", None) .await .expect("valid submission"); @@ -157,7 +168,7 @@ async fn migration_schema_entity_intake_preserves_existing_data() { .await .expect("registry"); assert!(registered); - let request = store + let request = users(&store) .submit("demo:alice", "operator-a", None) .await .expect("submit"); @@ -183,3 +194,258 @@ async fn migration_schema_entity_intake_preserves_existing_data() { .await .expect("community catalog"); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn approval_is_atomic_idempotent_and_bound_to_inventory() { + let store = store().await; + let request = users(&store) + .submit("demo:approval", "operator-a", None) + .await + .unwrap(); + let digest = request.inventory_digest.as_deref().unwrap(); + let before = store.inspect(request.id).await.unwrap(); + assert!(before.stage_history.is_empty()); + assert!(before.approvals.is_empty()); + let inventory = before.request.inventory_manifest.as_ref().unwrap(); + assert_eq!(inventory["communities"][0]["relationship"], "participant"); + assert_eq!( + inventory["communities"][1]["relationship"], + "former_participant" + ); + for (actor, hash) in [("operator-b", "wrong"), (" operator-b", digest)] { + assert!(store.approve(request.id, actor, hash).await.is_err()); + assert_eq!( + serde_json::to_value(store.inspect(request.id).await.unwrap()).unwrap(), + serde_json::to_value(&before).unwrap() + ); + } + let (a, b) = tokio::join!( + store.approve(request.id, "operator-b", digest), + store.approve(request.id, "operator-b", digest), + ); + let approved = a.unwrap(); + assert_eq!(approved.status, "approved"); + assert_eq!(approved.updated_at, b.unwrap().updated_at); + let after = store.inspect(request.id).await.unwrap(); + assert_eq!(after.approvals.len(), 1); + assert_eq!(after.approvals[0].request_id, request.id); + assert_eq!(after.approvals[0].approved_by, "operator-b"); + assert_eq!(after.approvals[0].inventory_digest, digest); + assert!(after.stage_history.is_empty()); + store + .approve(request.id, "operator-b", digest) + .await + .unwrap(); + users(&store) + .submit("demo:approval", "operator-a", None) + .await + .unwrap(); + assert_eq!( + serde_json::to_value(store.inspect(request.id).await.unwrap()).unwrap(), + serde_json::to_value(&after).unwrap() + ); + // A different operator adds evidence without replacing the first approval + // or recreating the starting stage. Concurrent retries still add one row. + let (a, b) = tokio::join!( + store.approve(request.id, "operator-c", digest), + store.approve(request.id, "operator-c", digest), + ); + a.unwrap(); + b.unwrap(); + let multiple = store.inspect(request.id).await.unwrap(); + assert_eq!(multiple.approvals.len(), 2); + assert_eq!(multiple.approvals[1].approved_by, "operator-c"); + assert_ne!(multiple.approvals[0].id, multiple.approvals[1].id); + assert!(multiple.approvals[1].approved_at > multiple.approvals[0].approved_at); + assert_eq!( + serde_json::to_value(&multiple.approvals[0]).unwrap(), + serde_json::to_value(&after.approvals[0]).unwrap() + ); + assert_eq!( + serde_json::to_value(&multiple.stage_history).unwrap(), + serde_json::to_value(&after.stage_history).unwrap() + ); + assert_eq!(multiple.request.updated_at, after.request.updated_at); + assert!(store + .approve(request.id, "operator-d", "wrong") + .await + .is_err()); + assert_eq!(store.inspect(request.id).await.unwrap().approvals.len(), 2); + // The invariant is enforced by the live partial unique index. + sqlx::query( + "INSERT INTO entity_deletion_stage_history (request_id, stage) VALUES ($1, 'first')", + ) + .bind(request.id) + .execute(&store.pool) + .await + .unwrap(); + assert!(sqlx::query( + "INSERT INTO entity_deletion_stage_history (request_id, stage) VALUES ($1, 'drain')" + ) + .bind(request.id) + .execute(&store.pool) + .await + .is_err()); + // A missing parent is rejected by application code, without a foreign key. + assert!(matches!( + store.approve(Uuid::new_v4(), "operator-b", digest).await, + Err(DbError::NotFound(_)) + )); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn approval_rolls_back_when_evidence_cannot_be_saved() { + let store = store().await; + let request = users(&store) + .submit("demo:rollback", "operator-a", None) + .await + .unwrap(); + // Force approval evidence to fail after the parent status update. + sqlx::query("ALTER TABLE entity_deletion_approvals ADD CONSTRAINT reject_approval CHECK (approved_by <> 'operator-b') NOT VALID") + .execute(&store.pool).await.unwrap(); + let before = store.inspect(request.id).await.unwrap(); + assert!(store + .approve( + request.id, + "operator-b", + request.inventory_digest.as_deref().unwrap() + ) + .await + .is_err()); + assert_eq!( + serde_json::to_value(store.inspect(request.id).await.unwrap()).unwrap(), + serde_json::to_value(before).unwrap() + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn approval_refuses_changed_inventory_and_unexpected_stage_history() { + let store = store().await; + let request = users(&store) + .submit("demo:changed", "operator-a", None) + .await + .unwrap(); + let digest = request.inventory_digest.as_deref().unwrap(); + let mut inventory = request.inventory_manifest.clone().unwrap(); + inventory["communities"].as_array_mut().unwrap().reverse(); + sqlx::query("UPDATE entity_deletion_requests SET inventory_manifest = $2 WHERE id = $1") + .bind(request.id) + .bind(&inventory) + .execute(&store.pool) + .await + .unwrap(); + assert!(store + .approve(request.id, "operator-b", digest) + .await + .is_err()); + assert_eq!(store.get(request.id).await.unwrap().status, "inventoried"); + sqlx::query("UPDATE entity_deletion_requests SET inventory_manifest = $2 WHERE id = $1") + .bind(request.id) + .bind(&request.inventory_manifest) + .execute(&store.pool) + .await + .unwrap(); + store + .approve(request.id, "operator-b", digest) + .await + .unwrap(); + sqlx::query( + "INSERT INTO entity_deletion_stage_history (request_id, stage) VALUES ($1, 'unexpected')", + ) + .bind(request.id) + .execute(&store.pool) + .await + .unwrap(); + assert!(store + .approve(request.id, "operator-b", digest) + .await + .is_err()); + assert_eq!( + store.inspect(request.id).await.unwrap().stage_history.len(), + 1 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_approval_preserves_legacy_intake_and_resubmission_prepares_it() { + let store = store().await; + crate::migration::run_migrations_through(&store.pool, 68) + .await + .unwrap(); + let id: Uuid = sqlx::query_scalar( + "INSERT INTO entity_deletion_requests (target, requested_by) VALUES ('demo:legacy', 'operator-a') RETURNING id") + .fetch_one(&store.pool).await.unwrap(); + crate::migration::run_migrations(&store.pool).await.unwrap(); + let before = store.inspect(id).await.unwrap(); + assert_eq!(before.request.status, "submitted"); + assert!(before.request.inventory_manifest.is_none()); + assert!(before.stage_history.is_empty()); + assert!(before.approvals.is_empty()); + assert!(store.approve(id, "operator-b", "unknown").await.is_err()); + let prepared = users(&store) + .submit("demo:legacy", "operator-a", None) + .await + .unwrap(); + assert_eq!(prepared.id, id); + assert_eq!(prepared.created_at, before.request.created_at); + store + .approve( + id, + "operator-b", + prepared.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + assert_plain_schema(&store.pool).await; + let registered: i64 = sqlx::query_scalar( + "SELECT count(*) FROM _operator_global_tables WHERE table_name IN + ('entity_deletion_stage_history', 'entity_deletion_approvals')", + ) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(registered, 2); + let approved = store.inspect(id).await.unwrap(); + assert_eq!(approved.approvals.len(), 1); + assert_eq!(approved.approvals[0].approved_by, "operator-b"); + // The previous CLI's INSERT remains valid after the upgrade. + sqlx::query("INSERT INTO entity_deletion_requests (target, requested_by) VALUES ('demo:old-cli', 'operator-a')") + .execute(&store.pool).await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn approval_refuses_to_repair_missing_evidence() { + let store = store().await; + let request = users(&store) + .submit("demo:missing-evidence", "operator-a", None) + .await + .unwrap(); + let digest = request.inventory_digest.as_deref().unwrap(); + store + .approve(request.id, "operator-b", digest) + .await + .unwrap(); + sqlx::query("DELETE FROM entity_deletion_approvals WHERE request_id = $1") + .bind(request.id) + .execute(&store.pool) + .await + .unwrap(); + let before = store.inspect(request.id).await.unwrap(); + assert!(store + .approve(request.id, "operator-c", digest) + .await + .is_err()); + assert_eq!( + serde_json::to_value(store.inspect(request.id).await.unwrap()).unwrap(), + serde_json::to_value(before).unwrap() + ); +} + +fn users(store: &EntityDeletionStore) -> crate::user_deletion::UserDeletionStore { + Db::from_pool(store.pool.clone()).user_deletion_store() +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 198bb6f407..da1bdba444 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -18,7 +18,7 @@ pub mod community; pub mod deletion; /// Direct message channel persistence. pub mod dm; -/// Operator-global synthetic user-deletion request intake. +/// Shared operator-global deletion workflow persistence. pub mod entity_deletion; /// Event storage and retrieval. pub mod event; @@ -26,6 +26,8 @@ pub mod event; pub(crate) mod event_follow_up; /// Per-community NIP-FI issuer bindings. pub mod nip_fi_binding; +/// User-specific deletion submission and inventory preparation. +pub mod user_deletion; #[cfg(test)] mod event_follow_up_postgres_tests; diff --git a/crates/buzz-db/src/store/user_deletion/inventory.rs b/crates/buzz-db/src/store/user_deletion/inventory.rs new file mode 100644 index 0000000000..3c565db326 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/inventory.rs @@ -0,0 +1,121 @@ +//! User target validation and synthetic inventory generation. +use crate::entity_deletion::EntityRequest; +use crate::store::entity_deletion::requests::validate_operator; +use crate::{DbError, Result}; +use serde_json::{json, Value}; +use std::collections::HashSet; +use uuid::Uuid; + +fn invalid(message: &str) -> DbError { + DbError::DeletionSafety(message.into()) +} +pub(super) fn validate_submission( + target: &str, + requested_by: &str, + reason: Option<&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(DbError::DeletionSafety( + "target must be demo:<1..64 lowercase letters, digits, underscores or hyphens>".into(), + )); + } + validate_operator(requested_by)?; + if reason.is_some_and(|r| r.len() > 4096) { + return Err(DbError::DeletionSafety("reason exceeds 4096 bytes".into())); + } + Ok(()) +} + +pub(super) fn demo_inventory(target: &str) -> serde_json::Value { + serde_json::json!({ + "version": 1, + "target": target, + "communities": [ + { + "community_id": "00000000-0000-0000-0000-00000000000b", + "label": "demo:community-b", + "relationship": "participant", + "action": "delete_subject_data" + }, + { + "community_id": "00000000-0000-0000-0000-00000000000c", + "label": "demo:community-c", + "relationship": "former_participant", + "action": "delete_subject_data" + } + ] + }) +} + +/// Validated user-deletion inventory, shared by preparation and workflow policy. +pub struct UserDeletionInventory { + /// Communities in approved processing order, with their saved attributes. + pub communities: Vec<(Uuid, Value)>, +} + +impl UserDeletionInventory { + /// Validate and read a saved user inventory. + pub fn read(request: &EntityRequest) -> Result { + if request.entity_type != "user" { + return Err(invalid("user workflow requires entity_type user")); + } + validate_submission( + &request.target, + &request.requested_by, + request.reason.as_deref(), + )?; + let value = request + .inventory_manifest + .as_ref() + .ok_or_else(|| invalid("missing inventory"))?; + if value["version"] != 1 || value["target"] != request.target { + return Err(invalid("unsupported inventory version or target")); + } + let rows = value["communities"] + .as_array() + .ok_or_else(|| invalid("missing communities"))?; + if rows.len() > 1000 { + return Err(invalid("inventory exceeds 1000 communities")); + } + let mut seen = HashSet::new(); + let mut communities = Vec::new(); + for row in rows { + let id = row["community_id"] + .as_str() + .and_then(|s| Uuid::parse_str(s).ok()) + .ok_or_else(|| invalid("invalid inventory community id"))?; + if !seen.insert(id) + || row["action"] != "delete_subject_data" + || !matches!( + row["relationship"].as_str(), + Some("participant" | "former_participant") + ) + { + return Err(invalid("duplicate community or unsupported deletion scope")); + } + communities.push((id, row.clone())); + } + Ok(Self { communities }) + } + + /// Build handler inputs for a request-wide or community-scoped stage. + pub fn inputs(&self, target: &str, community: Option) -> Result { + let community = match community { + None => Value::Null, + Some(id) => self + .communities + .iter() + .find(|(key, _)| *key == id) + .map(|(_, value)| value.clone()) + .ok_or_else(|| invalid("stage community is outside inventory"))?, + }; + Ok(json!({"target": target, "community": community})) + } +} diff --git a/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs new file mode 100644 index 0000000000..a3181e3cbe --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs @@ -0,0 +1,32 @@ +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn user_approval_validates_inventory_shape_before_saving_evidence() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool.clone()); + let users = db.user_deletion_store(); + let request = users + .submit("demo:invalid-scope", "operator", None) + .await + .unwrap(); + let mut inventory = request.inventory_manifest.unwrap(); + inventory["communities"][0]["relationship"] = "owner".into(); + let digest = crate::entity_deletion::inventory_digest(&inventory).unwrap(); + sqlx::query("UPDATE entity_deletion_requests SET inventory_manifest = $2, inventory_digest = $3 WHERE id = $1") + .bind(request.id).bind(inventory).bind(&digest).execute(&pool).await.unwrap(); + assert!(users + .approve(request.id, "approver", &digest) + .await + .is_err()); + let after = db + .entity_deletion_store() + .inspect(request.id) + .await + .unwrap(); + assert_eq!(after.request.status, "inventoried"); + assert!(after.approvals.is_empty()); + assert!(after.stage_history.is_empty()); +} diff --git a/crates/buzz-db/src/store/user_deletion/mod.rs b/crates/buzz-db/src/store/user_deletion/mod.rs new file mode 100644 index 0000000000..25ad3b32dc --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/mod.rs @@ -0,0 +1,24 @@ +//! User-specific submission and inventory preparation over the shared deletion tables. +use crate::Db; +use sqlx::PgPool; +mod inventory; +mod requests; +pub use inventory::UserDeletionInventory; + +/// Database operations that prepare user-deletion requests. +#[derive(Clone)] +pub struct UserDeletionStore { + pool: PgPool, +} +impl Db { + /// Access user-specific deletion request preparation using the writer pool. + pub fn user_deletion_store(&self) -> UserDeletionStore { + UserDeletionStore { + pool: self.pool().clone(), + } + } +} + +#[cfg(test)] +#[path = "inventory_postgres_tests.rs"] +mod postgres_tests; diff --git a/crates/buzz-db/src/store/user_deletion/requests.rs b/crates/buzz-db/src/store/user_deletion/requests.rs new file mode 100644 index 0000000000..913ed1d99e --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/requests.rs @@ -0,0 +1,81 @@ +//! Submit a user request and its dummy inventory in one transaction. +use super::{ + inventory::{demo_inventory, validate_submission}, + UserDeletionStore, +}; +use crate::entity_deletion::EntityRequest; +use crate::store::entity_deletion::requests::{inventory_digest, REQUEST_COLUMNS}; +use crate::{DbError, Result}; +impl UserDeletionStore { + /// Save a request or return an identical previous submission. + /// + /// A conflicting operator or reason returns an error without changing the + /// existing request. Concurrent identical submissions converge on one row. + /// Legacy submitted rows acquire their demo inventory on matching resubmission. + pub async fn submit( + &self, + target: &str, + requested_by: &str, + reason: Option<&str>, + ) -> Result { + validate_submission(target, requested_by, reason)?; + 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 mut request: EntityRequest = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "SELECT {REQUEST_COLUMNS} + FROM entity_deletion_requests WHERE entity_type = 'user' AND target = $1 FOR UPDATE", + ))) + .bind(target) + .fetch_one(&mut *tx) + .await?; + if request.requested_by != requested_by || request.reason.as_deref() != reason { + return Err(DbError::DeletionSafety(format!( + "target already has request {}; inspect it before resubmission", + request.id + ))); + } + if request.status == "submitted" { + let inventory = demo_inventory(target); + let digest = inventory_digest(&inventory)?; + request = sqlx::query_as(sqlx::AssertSqlSafe(format!( + "UPDATE entity_deletion_requests + SET status = 'inventoried', inventory_manifest = $2, inventory_digest = $3, + updated_at = now() + WHERE id = $1 + RETURNING {REQUEST_COLUMNS}", + ))) + .bind(request.id) + .bind(inventory) + .bind(digest) + .fetch_one(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(request) + } +} + +impl UserDeletionStore { + /// Validate the user inventory and save approval in the shared approval transaction. + pub async fn approve( + &self, + id: uuid::Uuid, + approved_by: &str, + digest: &str, + ) -> Result { + crate::Db::from_pool(self.pool.clone()) + .entity_deletion_store() + .approve_validated(id, approved_by, digest, |request| { + super::UserDeletionInventory::read(request).map(|_| ()) + }) + .await + } +} diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs index 798c14f71f..d63f74cc7a 100644 --- a/crates/buzz-deletion/src/user.rs +++ b/crates/buzz-deletion/src/user.rs @@ -1,10 +1,10 @@ -//! PostgreSQL-only intake commands for synthetic user-deletion requests. +//! PostgreSQL-only intake, approval, and execution of synthetic user-deletion requests. use anyhow::Result; use clap::Subcommand; use uuid::Uuid; -/// Submit and inspect synthetic requests; these commands never execute deletion. +/// Prepare and execute synthetic requests; these commands never delete real data. #[derive(Debug, Subcommand)] pub enum Command { /// Save a request. An identical resubmission returns the existing row. @@ -19,6 +19,37 @@ pub enum Command { #[arg(long)] reason: Option, }, + /// Approve the saved inventory; execution initializes the first stage. + Approve { + /// Request UUID returned by submit. + id: Uuid, + /// Approving operator identity (self-asserted for this demo). + #[arg(long)] + approved_by: String, + /// Inventory digest from inspect; approval must match this exact inventory. + #[arg(long)] + inventory_digest: String, + }, + /// Run dummy stages for an approved request; no real user data is changed. + Run { + /// Approved request UUID. + id: Uuid, + /// Worker identity; defaults to hostname and process id. + #[arg(long)] + executor_id: Option, + /// Execute one handler, transition to the next stage, and release ownership. + #[arg(long)] + next_step: bool, + /// Inject a retryable failure when this dummy stage is reached. + #[arg(long, value_parser = ["gate_user", "purge_db", "purge_agents"])] + fail_stage: Option, + /// Delay each dummy action for inspection and shutdown testing (up to 60s). + #[arg(long, default_value_t = 0, value_parser = clap::value_parser!(u64).range(0..=60_000))] + stage_delay_ms: u64, + /// Simulate no agents in this community; may be repeated. Saves a runtime branch result. + #[arg(long)] + demo_no_agents_in: Vec, + }, /// List synthetic requests, newest first. List { /// Maximum number of rows. @@ -32,7 +63,7 @@ pub enum Command { }, } -/// Run intake commands using the existing deletion database connection. +/// Run synthetic workflow commands using the existing deletion database connection. pub async fn run(command: Command) -> Result { let db = crate::connect_database().await?; let store = db.entity_deletion_store(); @@ -42,12 +73,51 @@ pub async fn run(command: Command) -> Result { requested_by, reason, } => crate::print_json( - &store + &db.user_deletion_store() .submit(&target, &requested_by, reason.as_deref()) .await?, )?, + Command::Approve { + id, + approved_by, + inventory_digest, + } => { + db.user_deletion_store() + .approve(id, &approved_by, &inventory_digest) + .await?; + crate::print_json(&store.inspect(id).await?)?; + } + Command::Run { + id, + executor_id, + next_step, + fail_stage, + stage_delay_ms, + demo_no_agents_in, + } => { + let code = worker::run( + &store, + id, + &executor_id.unwrap_or_else(crate::default_executor_id), + &worker::RunOptions { + next_step, + fail_stage, + stage_delay: std::time::Duration::from_millis(stage_delay_ms), + no_agents_in: demo_no_agents_in, + }, + &crate::shutdown_token(), + ) + .await?; + crate::print_json(&store.inspect(id).await?)?; + return Ok(code); + } Command::List { limit } => crate::print_json(&store.list(limit).await?)?, - Command::Inspect { id } => crate::print_json(&store.get(id).await?)?, + Command::Inspect { id } => crate::print_json(&store.inspect(id).await?)?, } Ok(0) } + +mod handlers; +mod transition; +mod worker; +mod workflow; diff --git a/crates/buzz-deletion/src/user/handlers.rs b/crates/buzz-deletion/src/user/handlers.rs new file mode 100644 index 0000000000..9849c2d0d1 --- /dev/null +++ b/crates/buzz-deletion/src/user/handlers.rs @@ -0,0 +1,104 @@ +//! Dummy handlers read full saved rows and persist outcomes; they never route. + +use super::worker::RunOptions; +use crate::RETRY_DELAY; +use anyhow::{bail, Result}; +use buzz_db::entity_deletion::{EntityDeletionStore, EntityLease, EntityStage}; +use uuid::Uuid; + +pub(super) async fn dispatch( + store: &EntityDeletionStore, + lease: &EntityLease, + row: &EntityStage, + options: &RunOptions, +) -> Result<()> { + match row.stage.as_str() { + "gate_user" => gate_user(store, lease, row, options).await, + "purge_db" => purge_db(store, lease, row, options).await, + "purge_agents" => purge_agents(store, lease, row, options).await, + name => bail!("unknown dummy stage {name}"), + } +} + +async fn gate_user( + store: &EntityDeletionStore, + lease: &EntityLease, + row: &EntityStage, + options: &RunOptions, +) -> Result<()> { + dummy_action(store, lease, row, options, "would gate user").await +} + +async fn purge_db( + store: &EntityDeletionStore, + lease: &EntityLease, + row: &EntityStage, + options: &RunOptions, +) -> Result<()> { + dummy_action(store, lease, row, options, "would purge user database rows").await +} + +async fn purge_agents( + store: &EntityDeletionStore, + lease: &EntityLease, + row: &EntityStage, + options: &RunOptions, +) -> Result<()> { + dummy_action(store, lease, row, options, "would purge user agents").await +} + +async fn dummy_action( + store: &EntityDeletionStore, + lease: &EntityLease, + row: &EntityStage, + options: &RunOptions, + message: &str, +) -> Result<()> { + store.start_stage(lease, row.id).await?; + let result = perform(row, options, message).await; + match result { + Ok(result) => store.save_stage_success(lease, row.id, result).await?, + Err(error) => { + eprintln!("dummy stage {} failed: {error:#}", row.stage); + // Ok means the outcome is durable, not that the action succeeded. + // The transition function will reload this row and return WaitUntil. + store + .fail_stage(lease, row.id, &format!("{error:#}"), RETRY_DELAY) + .await?; + } + } + Ok(()) +} + +async fn perform( + row: &EntityStage, + options: &RunOptions, + message: &str, +) -> Result { + let inputs = &row.metadata_json["inputs"]; + let target = inputs["target"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing target input"))?; + eprintln!( + "DUMMY request={} stage={} community={:?} target={target}: {message}", + row.request_id, row.stage, row.target_community_id + ); + tokio::time::sleep(options.stage_delay).await; + if options.fail_stage.as_deref() == Some(&row.stage) { + bail!("injected dummy failure for {}", row.stage); + } + let mut result = serde_json::json!({"dummy": true, "message": message}); + if row.stage == "purge_db" { + let community = inputs["community"]["community_id"] + .as_str() + .ok_or_else(|| anyhow::anyhow!("missing community input"))? + .parse::()?; + if Some(community) != row.target_community_id { + bail!("community input does not match stage"); + } + // A simulated observation, persisted before routing. Restart uses this + // result even if the next invocation supplies different demo options. + result["has_agents"] = (!options.no_agents_in.contains(&community)).into(); + } + Ok(result) +} diff --git a/crates/buzz-deletion/src/user/transition.rs b/crates/buzz-deletion/src/user/transition.rs new file mode 100644 index 0000000000..1ef6902b00 --- /dev/null +++ b/crates/buzz-deletion/src/user/transition.rs @@ -0,0 +1,94 @@ +//! User-deletion routing. The store owns the lock and atomic writes; this module +//! interprets the saved user inventory, history, and handler results. +use super::workflow::{decision, invalid, Progress, Step, INITIAL_STAGE}; +use buzz_db::entity_deletion::{ + EntityDeletionStore, EntityLease, EntityRequest, EntityStage, NewEntityStage, NextStage, + StageTransition, +}; +use buzz_db::user_deletion::UserDeletionInventory as Inventory; +use buzz_db::Result; +use uuid::Uuid; + +pub(super) struct UserDeletionTransition<'a> { + store: &'a EntityDeletionStore, +} + +impl<'a> UserDeletionTransition<'a> { + pub fn new(store: &'a EntityDeletionStore) -> Self { + Self { store } + } + + /// On restart, None recovers the current position. After a handler returns, + /// pass its ID; all decisions still use freshly loaded, locked database rows. + pub async fn get_next_stage( + &self, + lease: &EntityLease, + previous: Option, + ) -> Result { + self.store + .transition_stage(lease, previous, decide_next) + .await + } +} + +fn decide_next(request: &EntityRequest, history: &[EntityStage]) -> Result { + let inventory = Inventory::read(request)?; + let mut progress = Progress::rebuild(request, &inventory, history)?; + if request.status == "completed" { + if !progress.terminal || progress.current.is_some() { + return Err(invalid("completed request has inconsistent stage history")); + } + return Ok(StageTransition::Finished); + } + let Some(current) = progress.current.take() else { + if request.status != "approved" || !history.is_empty() { + return Err(invalid( + "started request has no current stage or terminal outcome", + )); + } + return Ok(StageTransition::Initialize(saved_stage( + request, + &inventory, + Step { + name: INITIAL_STAGE, + community: None, + }, + )?)); + }; + match current.status.as_str() { + "pending" | "running" | "failed" => Ok(StageTransition::Resume(current.id)), + "completed" => { + let outcome = decision(¤t)?; + let successor = progress.apply(&inventory, ¤t, outcome)?; + if successor.is_none() + && progress.completed_communities.len() != inventory.communities.len() + { + return Err(invalid("terminal transition leaves uncovered communities")); + } + Ok(StageTransition::Advance { + previous: current.id, + decision: serde_json::to_value(outcome)?, + successor: successor + .map(|step| saved_stage(request, &inventory, step)) + .transpose()?, + }) + } + _ => Err(invalid("unknown current stage status")), + } +} + +fn saved_stage( + request: &EntityRequest, + inventory: &Inventory, + step: Step, +) -> Result { + Ok(NewEntityStage { + stage: step.name.to_owned(), + community: step.community, + inputs: inventory.inputs(&request.target, step.community)?, + }) +} + +#[cfg(test)] +#[path = "transition_postgres_tests.rs"] +mod postgres_tests; diff --git a/crates/buzz-deletion/src/user/transition_postgres_tests.rs b/crates/buzz-deletion/src/user/transition_postgres_tests.rs new file mode 100644 index 0000000000..830c12c038 --- /dev/null +++ b/crates/buzz-deletion/src/user/transition_postgres_tests.rs @@ -0,0 +1,480 @@ +use super::*; +use buzz_db::Db; +use serde_json::Value; +use sqlx::PgPool; +use std::time::Duration; + +// This fixture exposes the real stores and transition component. It contains +// no routing rules; each helper delegates to the production API. +struct TestStore { + store: EntityDeletionStore, + pool: PgPool, +} +impl std::ops::Deref for TestStore { + type Target = EntityDeletionStore; + fn deref(&self) -> &Self::Target { + &self.store + } +} +impl TestStore { + async fn submit( + &self, + target: &str, + operator: &str, + reason: Option<&str>, + ) -> Result { + Db::from_pool(self.pool.clone()) + .user_deletion_store() + .submit(target, operator, reason) + .await + } + async fn get_next_stage( + &self, + lease: &EntityLease, + previous: Option, + ) -> Result { + UserDeletionTransition::new(&self.store) + .get_next_stage(lease, previous) + .await + } +} +fn inventory_digest(value: &Value) -> Result { + // Use the same public fingerprint operation as request preparation. + buzz_db::entity_deletion::inventory_digest(value) +} + +async fn fixture() -> (TestStore, Uuid) { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap(); + let pool = PgPool::connect(&url).await.unwrap(); + let store = TestStore { + store: Db::from_pool(pool.clone()).entity_deletion_store(), + pool, + }; + let request = store + .submit("demo:executor", "submitter", None) + .await + .unwrap(); + store + .approve( + request.id, + "approver", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + (store, request.id) +} + +async fn claim(store: &TestStore, id: Uuid, owner: &str) -> EntityLease { + store + .claim(id, owner, Duration::from_secs(60)) + .await + .unwrap() + .unwrap() +} + +async fn snapshot(store: &TestStore, id: Uuid) -> serde_json::Value { + serde_json::to_value(store.inspect(id).await.unwrap()).unwrap() +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn concurrent_claimers_have_one_owner_and_unapproved_requests_are_not_claimed() { + let (store, id) = fixture().await; + let unapproved = store + .submit("demo:unapproved", "submitter", None) + .await + .unwrap(); + assert!(store + .claim(unapproved.id, "worker", Duration::from_secs(60)) + .await + .unwrap() + .is_none()); + let (a, b) = tokio::join!( + store.claim(id, "worker-a", Duration::from_secs(60)), + store.claim(id, "worker-b", Duration::from_secs(60)), + ); + let claims: Vec<_> = [a.unwrap(), b.unwrap()].into_iter().flatten().collect(); + assert_eq!(claims.len(), 1); + let lease = &claims[0]; + let inspection = store.inspect(id).await.unwrap(); + assert_eq!(inspection.request.status, "approved"); + assert!(inspection.stage_history.is_empty()); + assert_eq!( + inspection.execution.lease_owner.as_deref(), + Some(lease.owner.as_str()) + ); + assert_eq!(inspection.execution.lease_generation, 1); + assert!(store + .claim(id, &lease.owner, Duration::from_secs(60)) + .await + .unwrap() + .is_none()); + let before = snapshot(&store, id).await; + assert!(store.claim(id, "worker", Duration::ZERO).await.is_err()); + assert_eq!(snapshot(&store, id).await, before); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn expired_and_superseded_workers_cannot_change_any_execution_state() { + let (store, id) = fixture().await; + let old = claim(&store, id, "worker-a").await; + let stage = start(&store, &old).await.unwrap(); + sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at = clock_timestamp() - interval '1 second' WHERE id = $1") + .bind(id).execute(&store.pool).await.unwrap(); + // Expiration alone rejects renewal, even before another worker claims. + assert!(buzz_db::deletion::is_stale_deletion_lease( + &store + .heartbeat(&old, Duration::from_secs(60)) + .await + .unwrap_err() + )); + assert!(!store.release(&old).await.unwrap()); + // Reusing the same executor ID isolates the generation check from owner checks. + let new = claim(&store, id, "worker-a").await; + assert_eq!(new.owner, old.owner); + assert_eq!(new.generation, old.generation + 1); + let resumed = start(&store, &new).await.unwrap(); + assert_eq!(resumed.id, stage.id); + assert_eq!(resumed.started_at, stage.started_at); + let before = snapshot(&store, id).await; + for error in [ + store.get_next_stage(&old, None).await.unwrap_err(), + start(&store, &old).await.unwrap_err(), + store + .heartbeat(&old, Duration::from_secs(60)) + .await + .unwrap_err(), + store + .fail_stage(&old, stage.id, "stale failure", Duration::from_secs(30)) + .await + .unwrap_err(), + store + .save_stage_success(&old, stage.id, serde_json::json!({})) + .await + .unwrap_err(), + ] { + assert!( + buzz_db::deletion::is_stale_deletion_lease(&error), + "{error}" + ); + } + assert!(!store.release(&old).await.unwrap()); + assert_eq!(snapshot(&store, id).await, before); + store + .save_stage_success(&new, stage.id, serde_json::json!({"dummy": true})) + .await + .unwrap(); + let next = next(&store, &new, Some(stage.id)).await; + assert_eq!(next.stage, "purge_db"); + assert_eq!(store.inspect(id).await.unwrap().stage_history.len(), 2); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn heartbeat_checks_authority_after_waiting_for_parent_lock() { + let (store, id) = fixture().await; + let lease = claim(&store, id, "worker").await; + let before = store.inspect(id).await.unwrap().execution; + store + .heartbeat(&lease, Duration::from_secs(120)) + .await + .unwrap(); + let renewed = store.inspect(id).await.unwrap().execution; + assert!(renewed.lease_expires_at > before.lease_expires_at); + assert!(renewed.heartbeat_at > before.heartbeat_at); + + let mut blocker = store.pool.begin().await.unwrap(); + sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at = clock_timestamp() - interval '1 second' WHERE id = $1") + .bind(id).execute(&mut *blocker).await.unwrap(); + let renewal = store.heartbeat(&lease, Duration::from_secs(60)); + tokio::pin!(renewal); + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut renewal) + .await + .is_err() + ); + blocker.commit().await.unwrap(); + assert!(buzz_db::deletion::is_stale_deletion_lease( + &renewal.await.unwrap_err() + )); +} + +async fn next(store: &TestStore, lease: &EntityLease, previous: Option) -> EntityStage { + match store.get_next_stage(lease, previous).await.unwrap() { + NextStage::Stage(row) => row, + other => panic!("expected a stage, got {other:?}"), + } +} + +async fn start(store: &TestStore, lease: &EntityLease) -> Result { + match store.get_next_stage(lease, None).await? { + NextStage::Stage(row) => store.start_stage(lease, row.id).await, + _ => panic!("expected current stage"), + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn transition_recovers_on_both_sides_of_commit_and_repeated_previous_id_is_read_only() { + let (store, id) = fixture().await; + let initial_claim = claim(&store, id, "worker-before-initialization").await; + store.release(&initial_claim).await.unwrap(); + let first = claim(&store, id, "worker-a").await; + sqlx::query("ALTER TABLE entity_deletion_stage_history ADD CONSTRAINT reject_initial CHECK (stage <> 'gate_user') NOT VALID") + .execute(&store.pool).await.unwrap(); + let before = snapshot(&store, id).await; + assert!(store.get_next_stage(&first, None).await.is_err()); + assert_eq!(snapshot(&store, id).await, before); + assert_eq!(store.get(id).await.unwrap().status, "approved"); + sqlx::query("ALTER TABLE entity_deletion_stage_history DROP CONSTRAINT reject_initial") + .execute(&store.pool) + .await + .unwrap(); + let gate = start(&store, &first).await.unwrap(); + store + .save_stage_success(&first, gate.id, serde_json::json!({"dummy": true})) + .await + .unwrap(); + let saved = store.inspect(id).await.unwrap(); + assert_eq!(saved.stage_history.len(), 1); + assert_eq!(saved.stage_history[0].status, "completed"); + assert!(saved.stage_history[0].is_current); + let result_time = saved.stage_history[0].finished_at; + store.release(&first).await.unwrap(); + + // Restart before transition: None finds the completed current row and advances. + let second = claim(&store, id, "worker-b").await; + let db_stage = next(&store, &second, None).await; + assert_eq!(db_stage.stage, "purge_db"); + assert_eq!( + db_stage.metadata_json["previous_stage_id"], + gate.id.to_string() + ); + assert_eq!(db_stage.metadata_json["inputs"]["target"], "demo:executor"); + let before = snapshot(&store, id).await; + assert_eq!(next(&store, &second, Some(gate.id)).await.id, db_stage.id); + assert_eq!(snapshot(&store, id).await, before); + store.release(&second).await.unwrap(); + + // Restart after transition: None returns the saved successor, without replay. + let third = claim(&store, id, "worker-c").await; + assert_eq!(next(&store, &third, None).await.id, db_stage.id); + let after = store.inspect(id).await.unwrap(); + assert_eq!(after.stage_history.len(), 2); + assert_eq!(after.stage_history[0].finished_at, result_time); + let before = snapshot(&store, id).await; + assert!(store + .get_next_stage(&third, Some(Uuid::new_v4())) + .await + .is_err()); + assert_eq!(snapshot(&store, id).await, before); + assert!(store + .save_stage_success(&third, gate.id, serde_json::json!({})) + .await + .is_err()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn failed_transition_keeps_completed_current_row_and_retry_creates_only_one_successor() { + let (store, id) = fixture().await; + let lease = claim(&store, id, "worker").await; + let gate = start(&store, &lease).await.unwrap(); + store + .save_stage_success(&lease, gate.id, serde_json::json!({"dummy":true})) + .await + .unwrap(); + sqlx::query("ALTER TABLE entity_deletion_stage_history ADD CONSTRAINT reject_successor CHECK (stage <> 'purge_db') NOT VALID") + .execute(&store.pool).await.unwrap(); + let before = snapshot(&store, id).await; + assert!(store.get_next_stage(&lease, Some(gate.id)).await.is_err()); + assert_eq!(snapshot(&store, id).await, before); + sqlx::query("ALTER TABLE entity_deletion_stage_history DROP CONSTRAINT reject_successor") + .execute(&store.pool) + .await + .unwrap(); + let successor = next(&store, &lease, Some(gate.id)).await; + assert_eq!(next(&store, &lease, Some(gate.id)).await.id, successor.id); + assert_eq!(store.inspect(id).await.unwrap().stage_history.len(), 2); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn failed_stage_waits_then_resumes_same_row_without_advancing() { + let (store, id) = fixture().await; + let lease = claim(&store, id, "worker-a").await; + let stage = start(&store, &lease).await.unwrap(); + store + .fail_stage(&lease, stage.id, "injected", Duration::from_secs(30)) + .await + .unwrap(); + assert!(matches!( + store.get_next_stage(&lease, Some(stage.id)).await.unwrap(), + NextStage::WaitUntil(_) + )); + let failed = store.inspect(id).await.unwrap(); + assert_eq!(failed.request.status, "retry_wait"); + assert_eq!(failed.execution.retry_count, 1); + assert_eq!(failed.stage_history.len(), 1); + assert!(failed.stage_history[0].is_current); + assert_eq!(failed.stage_history[0].status, "failed"); + assert!(store.start_stage(&lease, stage.id).await.is_err()); + store.release(&lease).await.unwrap(); + assert!(store + .claim(id, "worker-b", Duration::from_secs(60)) + .await + .unwrap() + .is_none()); + sqlx::query("UPDATE entity_deletion_requests SET next_attempt_at = clock_timestamp() - interval '1 second' WHERE id = $1") + .bind(id).execute(&store.pool).await.unwrap(); + let retry = claim(&store, id, "worker-b").await; + let resumed = start(&store, &retry).await.unwrap(); + assert_eq!(resumed.id, stage.id); + assert_eq!(resumed.started_at, stage.started_at); + store + .save_stage_success(&retry, stage.id, serde_json::json!({})) + .await + .unwrap(); + next(&store, &retry, Some(stage.id)).await; + let after = store.inspect(id).await.unwrap(); + assert_eq!(after.execution.retry_count, 0); + assert!(after.execution.last_error.is_none()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn routing_rebuilds_variable_community_progress_and_obeys_saved_branch_results() { + let (store, _) = fixture().await; + for count in [0, 1, 3] { + let mut request = store + .submit(&format!("demo:communities-{count}"), "submitter", None) + .await + .unwrap(); + let mut inventory = request.inventory_manifest.take().unwrap(); + let template = inventory["communities"][0].clone(); + inventory["communities"] = (0..count) + .map(|i| { + let mut row = template.clone(); + row["community_id"] = Uuid::from_u128(100 + i).to_string().into(); + row + }) + .collect::>() + .into(); + let digest = inventory_digest(&inventory).unwrap(); + sqlx::query("UPDATE entity_deletion_requests SET inventory_manifest = $2, inventory_digest = $3 WHERE id = $1") + .bind(request.id).bind(&inventory).bind(&digest).execute(&store.pool).await.unwrap(); + store + .approve(request.id, "approver", &digest) + .await + .unwrap(); + let mut lease = claim(&store, request.id, "worker").await; + let mut stages = Vec::new(); + let mut previous = None; + for _ in 0..10 { + let row = match store.get_next_stage(&lease, previous).await.unwrap() { + NextStage::Finished => break, + NextStage::Stage(row) => row, + other => panic!("{other:?}"), + }; + stages.push((row.stage.clone(), row.target_community_id)); + store.start_stage(&lease, row.id).await.unwrap(); + // Alternate paths: first community skips agents, the next executes them. + let has_agents = row + .target_community_id + .is_some_and(|id| id.as_u128() % 2 == 1); + store + .save_stage_success(&lease, row.id, serde_json::json!({"has_agents":has_agents})) + .await + .unwrap(); + // Rebuild from PostgreSQL after every successful action, with no previous id. + store.release(&lease).await.unwrap(); + lease = claim(&store, request.id, "replacement").await; + previous = None; + } + assert_eq!(store.get(request.id).await.unwrap().status, "completed"); + assert_eq!(stages.len(), 1 + count as usize + (count / 2) as usize); + assert_eq!( + stages.iter().filter(|(s, _)| s == "purge_agents").count(), + (count / 2) as usize + ); + assert!(matches!( + store.get_next_stage(&lease, None).await.unwrap(), + NextStage::Finished + )); + assert!(!store + .inspect(request.id) + .await + .unwrap() + .stage_history + .iter() + .any(|s| s.is_current)); + store.release(&lease).await.unwrap(); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn corrupt_inventory_history_and_results_fail_without_inventing_progress() { + let (store, id) = fixture().await; + let lease = claim(&store, id, "worker").await; + let gate = start(&store, &lease).await.unwrap(); + let manifest = store.get(id).await.unwrap().inventory_manifest.unwrap(); + sqlx::query("UPDATE entity_deletion_requests SET inventory_manifest = '{}' WHERE id = $1") + .bind(id) + .execute(&store.pool) + .await + .unwrap(); + assert!(store.get_next_stage(&lease, None).await.is_err()); + sqlx::query("UPDATE entity_deletion_requests SET inventory_manifest = $2 WHERE id = $1") + .bind(id) + .bind(manifest) + .execute(&store.pool) + .await + .unwrap(); + store + .save_stage_success(&lease, gate.id, serde_json::json!({})) + .await + .unwrap(); + let db = next(&store, &lease, Some(gate.id)).await; + let saved_transition = + store.inspect(id).await.unwrap().stage_history[0].metadata_json["transition"].clone(); + sqlx::query("UPDATE entity_deletion_stage_history SET metadata_json = jsonb_set(metadata_json, '{transition}', $2) WHERE id = $1") + .bind(gate.id).bind(serde_json::json!({"kind": "community_complete"})) + .execute(&store.pool).await.unwrap(); + let corrupted = snapshot(&store, id).await; + let error = store + .get_next_stage(&lease, Some(gate.id)) + .await + .unwrap_err(); + assert!(error + .to_string() + .contains("saved transition disagrees with handler result")); + assert_eq!(snapshot(&store, id).await, corrupted); + sqlx::query("UPDATE entity_deletion_stage_history SET metadata_json = jsonb_set(metadata_json, '{transition}', $2) WHERE id = $1") + .bind(gate.id).bind(saved_transition).execute(&store.pool).await.unwrap(); + store.start_stage(&lease, db.id).await.unwrap(); + store + .save_stage_success(&lease, db.id, serde_json::json!({})) + .await + .unwrap(); + let before = snapshot(&store, id).await; + assert!(store.get_next_stage(&lease, Some(db.id)).await.is_err()); + assert_eq!(snapshot(&store, id).await, before); // missing has_agents cannot skip work + sqlx::query("DELETE FROM entity_deletion_stage_history WHERE id = $1") + .bind(db.id) + .execute(&store.pool) + .await + .unwrap(); + assert!(store.get_next_stage(&lease, None).await.is_err()); // never reinitialize gate + assert_eq!(store.inspect(id).await.unwrap().stage_history.len(), 1); + sqlx::query("DELETE FROM entity_deletion_approvals WHERE request_id = $1") + .bind(id) + .execute(&store.pool) + .await + .unwrap(); + assert!(store.get_next_stage(&lease, None).await.is_err()); +} diff --git a/crates/buzz-deletion/src/user/worker.rs b/crates/buzz-deletion/src/user/worker.rs new file mode 100644 index 0000000000..7d3cb54b7b --- /dev/null +++ b/crates/buzz-deletion/src/user/worker.rs @@ -0,0 +1,169 @@ +//! Ownership and dispatch only; get_next_stage owns all workflow routing. + +use std::time::Duration; + +use anyhow::{bail, Result}; +use buzz_db::entity_deletion::{EntityDeletionStore, EntityLease, EntityStage, NextStage}; +use tokio_util::sync::CancellationToken; +use uuid::Uuid; + +use super::{handlers, transition::UserDeletionTransition}; +use crate::{StageOutcome, DEFAULT_LEASE_DURATION, HEARTBEAT_INTERVAL}; + +/// Controls for a synthetic run. Delays and failures never invoke real deletion. +pub(super) struct RunOptions { + pub next_step: bool, + pub fail_stage: Option, + pub stage_delay: Duration, + pub no_agents_in: Vec, +} + +pub(super) async fn run( + store: &EntityDeletionStore, + id: Uuid, + owner: &str, + options: &RunOptions, + shutdown: &CancellationToken, +) -> Result { + let Some(lease) = store.claim(id, owner, DEFAULT_LEASE_DURATION).await? else { + if store.get(id).await?.status == "completed" { + return Ok(0); + } + bail!("request is unapproved, not yet due for retry, or leased by another worker"); + }; + let result = execute_claim(store, &lease, options, shutdown).await; + // This generation-checked release is harmless after completion, a recorded + // failure, or takeover. Never leave a background heartbeat running on return. + let released = store.release(&lease).await; + finish_claim(result, released) +} + +fn finish_claim(result: Result, released: buzz_db::Result) -> Result { + match (result, released) { + (Err(error), Err(release_error)) => { + // buzz-admin prints Display, so the outer message must include both. + let message = + format!("{error:#}; also failed to release deletion lease: {release_error}"); + Err(error.context(message)) + } + (Err(error), Ok(_)) => Err(error), + (Ok(_), Err(error)) => Err(error.into()), + (Ok(code), Ok(_)) => Ok(code), + } +} + +async fn execute_claim( + store: &EntityDeletionStore, + lease: &EntityLease, + options: &RunOptions, + shutdown: &CancellationToken, +) -> Result { + let transition = UserDeletionTransition::new(store); + let mut previous_stage_id = None; + let mut executed = 0; + // Keep ordinary runs bounded; stepping uses the same transition boundary. + let handler_budget = if options.next_step { 1 } else { 100 }; + loop { + if shutdown.is_cancelled() { + return Ok(0); + } + match transition.get_next_stage(lease, previous_stage_id).await? { + NextStage::Finished => return Ok(0), + NextStage::WaitUntil(deadline) => { + eprintln!("request {} waits until {deadline}", lease.request_id); + return Ok(1); + } + NextStage::Stage(row) => { + if executed >= handler_budget { + return Ok(0); + } + match run_stage_with_heartbeat( + store, + lease, + &row, + options, + shutdown, + HEARTBEAT_INTERVAL, + ) + .await + { + StageOutcome::Completed => {} + StageOutcome::Shutdown => return Ok(0), + StageOutcome::Failed(error) => return Err(error), + } + previous_stage_id = Some(row.id); + executed += 1; + } + } + } +} + +async fn run_stage_with_heartbeat( + store: &EntityDeletionStore, + lease: &EntityLease, + stage: &EntityStage, + options: &RunOptions, + shutdown: &CancellationToken, + heartbeat_period: Duration, +) -> StageOutcome { + let handler = handlers::dispatch(store, lease, stage, options); + tokio::pin!(handler); + let mut interval = tokio::time::interval(heartbeat_period); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Poll renewal alongside the handler. Awaiting renewal inside a selected + // timer branch would stop polling a handler that may hold the parent lock. + // Both futures live here, so cancellation leaves no detached heartbeat task. + let renewal = async { + loop { + interval.tick().await; + if let Err(error) = store.heartbeat(lease, DEFAULT_LEASE_DURATION).await { + return error; + } + } + }; + tokio::select! { + biased; + _ = shutdown.cancelled() => StageOutcome::Shutdown, + result = &mut handler => match result { + Ok(()) => StageOutcome::Completed, + Err(error) => StageOutcome::Failed(error), + }, + error = renewal => StageOutcome::Failed(error.into()), + } +} + +#[cfg(test)] +#[path = "worker_postgres_tests.rs"] +mod postgres_tests; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn release_failure_preserves_execution_error_and_reports_both() { + let original = std::io::Error::new(std::io::ErrorKind::BrokenPipe, "handler failed"); + let error = finish_claim( + Err(original.into()), + Err(buzz_db::DbError::InvalidData("release failed".into())), + ) + .unwrap_err(); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::BrokenPipe + ); + let report = error.to_string(); + assert!(report.contains("handler failed")); + assert!(report.contains("release failed")); + } + + #[test] + fn release_failure_after_success_is_not_swallowed() { + assert!(finish_claim( + Ok(0), + Err(buzz_db::DbError::InvalidData("release failed".into())) + ) + .is_err()); + assert_eq!(finish_claim(Ok(1), Ok(false)).unwrap(), 1); + } +} diff --git a/crates/buzz-deletion/src/user/worker_postgres_tests.rs b/crates/buzz-deletion/src/user/worker_postgres_tests.rs new file mode 100644 index 0000000000..8259f1ad8c --- /dev/null +++ b/crates/buzz-deletion/src/user/worker_postgres_tests.rs @@ -0,0 +1,448 @@ +use super::*; +use buzz_db::Db; + +async fn fixture() -> (EntityDeletionStore, Uuid, sqlx::PgPool) { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap(); + let pool = sqlx::PgPool::connect(&url).await.unwrap(); + let db = Db::from_pool(pool.clone()); + if std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() != Ok("desired") { + db.migrate().await.unwrap(); + } + let store = db.entity_deletion_store(); + let request = db + .user_deletion_store() + .submit("demo:runner", "submitter", None) + .await + .unwrap(); + store + .approve( + request.id, + "approver", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + ( + store, + request.id, + sqlx::PgPool::connect(&url).await.unwrap(), + ) +} + +fn options() -> RunOptions { + RunOptions { + next_step: false, + fail_stage: None, + stage_delay: Duration::ZERO, + no_agents_in: Vec::new(), + } +} + +async fn assert_completed(store: &EntityDeletionStore, id: Uuid) { + let inspection = store.inspect(id).await.unwrap(); + assert_eq!(inspection.request.status, "completed"); + assert!(inspection.execution.completed_at.is_some()); + assert!(inspection.execution.lease_owner.is_none()); + assert_eq!(inspection.approvals.len(), 1); + let actual: Vec<_> = inspection + .stage_history + .iter() + .map(|s| { + ( + s.stage.as_str(), + s.target_community_id.map(|id| id.to_string()), + ) + }) + .collect(); + let b = Some("00000000-0000-0000-0000-00000000000b".to_owned()); + let c = Some("00000000-0000-0000-0000-00000000000c".to_owned()); + assert_eq!( + actual, + vec![ + ("gate_user", None), + ("purge_db", b.clone()), + ("purge_agents", b), + ("purge_db", c.clone()), + ("purge_agents", c) + ] + ); + for stage in &inspection.stage_history { + assert_eq!(stage.status, "completed"); + assert!(!stage.is_current); + assert!(stage.started_at.is_some()); + assert!(stage.finished_at >= stage.started_at); + assert_eq!(stage.metadata_json["result"]["dummy"], true); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dummy_executor_completes_two_communities_and_completed_rerun_is_read_only() { + let (store, id, _) = fixture().await; + assert_eq!( + tokio::time::timeout( + Duration::from_secs(15), + run(&store, id, "worker", &options(), &CancellationToken::new()) + ) + .await + .unwrap() + .unwrap(), + 0 + ); + assert_completed(&store, id).await; + let before = serde_json::to_value(store.inspect(id).await.unwrap()).unwrap(); + assert_eq!( + run(&store, id, "worker", &options(), &CancellationToken::new()) + .await + .unwrap(), + 0 + ); + assert_eq!( + serde_json::to_value(store.inspect(id).await.unwrap()).unwrap(), + before + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn next_step_advances_one_handler_at_a_time_through_completion() { + let (store, id, _) = fixture().await; + let one = RunOptions { + next_step: true, + ..options() + }; + run(&store, id, "worker-a", &one, &CancellationToken::new()) + .await + .unwrap(); + let paused = store.inspect(id).await.unwrap(); + assert_eq!(paused.request.status, "running"); + assert!(paused.execution.lease_owner.is_none()); + assert_eq!(paused.stage_history.len(), 2); + assert_eq!(paused.stage_history[1].stage, "purge_db"); + assert_eq!(paused.stage_history[1].status, "pending"); + let gate = serde_json::to_value(&paused.stage_history[0]).unwrap(); + for completed_handlers in 2..=5 { + run(&store, id, "worker-b", &one, &CancellationToken::new()) + .await + .unwrap(); + let after = store.inspect(id).await.unwrap(); + assert!(after.execution.lease_owner.is_none()); + assert_eq!( + after + .stage_history + .iter() + .filter(|s| s.status == "completed") + .count(), + completed_handlers + ); + if completed_handlers < 5 { + assert_eq!(after.stage_history.len(), completed_handlers + 1); + let current = after.stage_history.iter().find(|s| s.is_current).unwrap(); + assert_eq!(current.status, "pending"); + assert!(current.started_at.is_none()); + } else { + assert_eq!(after.stage_history.len(), 5); + assert!(!after.stage_history.iter().any(|s| s.is_current)); + } + } + assert_completed(&store, id).await; + assert_eq!( + serde_json::to_value(&store.inspect(id).await.unwrap().stage_history[0]).unwrap(), + gate + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dummy_executor_failure_retries_same_community_stage() { + let (store, id, pool) = fixture().await; + let fail = RunOptions { + fail_stage: Some("purge_db".to_owned()), + ..options() + }; + assert_eq!( + run(&store, id, "worker-a", &fail, &CancellationToken::new()) + .await + .unwrap(), + 1 + ); + let failed = store.inspect(id).await.unwrap(); + assert_eq!(failed.request.status, "retry_wait"); + assert_eq!(failed.execution.retry_count, 1); + assert!(failed + .execution + .last_error + .as_deref() + .unwrap() + .contains("injected dummy failure")); + assert_eq!(failed.stage_history.len(), 2); + let stage_id = failed.stage_history[1].id; + let started = failed.stage_history[1].started_at; + assert_eq!(failed.stage_history[1].status, "failed"); + assert!(run( + &store, + id, + "worker-b", + &options(), + &CancellationToken::new() + ) + .await + .is_err()); + sqlx::query("UPDATE entity_deletion_requests SET next_attempt_at = clock_timestamp() - interval '1 second' WHERE id = $1") + .bind(id).execute(&pool).await.unwrap(); + run( + &store, + id, + "worker-b", + &options(), + &CancellationToken::new(), + ) + .await + .unwrap(); + assert_completed(&store, id).await; + let after = store.inspect(id).await.unwrap(); + assert_eq!(after.stage_history[1].id, stage_id); + assert_eq!(after.stage_history[1].started_at, started); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dummy_executor_shutdown_keeps_running_stage_for_next_worker() { + let (store, id, _) = fixture().await; + let shutdown = CancellationToken::new(); + let delayed = RunOptions { + stage_delay: Duration::from_secs(60), + ..options() + }; + let work = run(&store, id, "worker-a", &delayed, &shutdown); + tokio::pin!(work); + // Poll the production executor until its durable running row is visible. + tokio::time::timeout(Duration::from_secs(5), async { + loop { + tokio::select! { + result = &mut work => panic!("run finished before cancellation: {result:?}"), + _ = tokio::time::sleep(Duration::from_millis(10)) => { + if store.inspect(id).await.unwrap().stage_history.first().is_some_and(|s| s.status == "running") { break; } + } + } + } + }).await.unwrap(); + shutdown.cancel(); + assert_eq!(work.await.unwrap(), 0); + let paused = store.inspect(id).await.unwrap(); + assert_eq!(paused.stage_history[0].status, "running"); + assert!(paused.execution.lease_owner.is_none()); + let stage_id = paused.stage_history[0].id; + run( + &store, + id, + "worker-b", + &options(), + &CancellationToken::new(), + ) + .await + .unwrap(); + assert_completed(&store, id).await; + assert_eq!( + store.inspect(id).await.unwrap().stage_history[0].id, + stage_id + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dummy_handler_is_heartbeated_and_lease_loss_prevents_completion() { + let (store, id, pool) = fixture().await; + let lease = store + .claim(id, "worker", Duration::from_secs(1)) + .await + .unwrap() + .unwrap(); + let stage = match UserDeletionTransition::new(&store) + .get_next_stage(&lease, None) + .await + .unwrap() + { + NextStage::Stage(row) => row, + other => panic!("{other:?}"), + }; + let before = store.inspect(id).await.unwrap().execution; + let delayed = RunOptions { + stage_delay: Duration::from_millis(1100), + ..options() + }; + assert!(matches!( + run_stage_with_heartbeat( + &store, + &lease, + &stage, + &delayed, + &CancellationToken::new(), + Duration::from_millis(10) + ) + .await, + StageOutcome::Completed + )); + let after = store.inspect(id).await.unwrap(); + assert!( + after.execution.heartbeat_at.unwrap() + > before.heartbeat_at.unwrap() + chrono::Duration::seconds(1) + ); + assert!(after.execution.lease_expires_at > before.lease_expires_at); + assert_eq!(after.stage_history.len(), 1); + assert!(after.stage_history[0].is_current); + assert_eq!(after.stage_history[0].status, "completed"); + + let stage = match UserDeletionTransition::new(&store) + .get_next_stage(&lease, None) + .await + .unwrap() + { + NextStage::Stage(row) => row, + other => panic!("{other:?}"), + }; + let delayed = RunOptions { + stage_delay: Duration::from_secs(60), + ..options() + }; + let shutdown = CancellationToken::new(); + let work = run_stage_with_heartbeat( + &store, + &lease, + &stage, + &delayed, + &shutdown, + Duration::from_millis(10), + ); + tokio::pin!(work); + // Lose the lease while the handler is already running, not before start_stage. + tokio::time::timeout(Duration::from_secs(5), async { + loop { + tokio::select! { + _ = &mut work => panic!("handler finished before takeover"), + _ = tokio::time::sleep(Duration::from_millis(10)) => { + if store.inspect(id).await.unwrap().stage_history[1].status == "running" { + break; + } + } + } + } + }) + .await + .unwrap(); + let takeover = async { + sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at = clock_timestamp() - interval '1 second' WHERE id = $1") + .bind(id).execute(&pool).await.unwrap(); + // Claim uses SKIP LOCKED; an in-flight heartbeat may still own the row. + // The enclosing timeout bounds this retry, while join keeps work polled. + loop { + if let Some(replacement) = store + .claim(id, "worker", DEFAULT_LEASE_DURATION) + .await + .unwrap() + { + break replacement; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }; + // Keep polling renewal while takeover waits for the parent lock; an + // in-flight heartbeat must be allowed to commit and release that lock. + let (outcome, replacement) = tokio::time::timeout(Duration::from_secs(5), async { + tokio::join!(&mut work, takeover) + }) + .await + .unwrap(); + assert_eq!(replacement.generation, lease.generation + 1); + assert!(matches!(outcome, StageOutcome::Failed(_))); + let after = store.inspect(id).await.unwrap(); + assert_eq!(after.stage_history.len(), 2); + assert_eq!(after.stage_history[1].status, "running"); + assert!(after.stage_history[1].finished_at.is_none()); + assert_eq!(after.execution.lease_generation, replacement.generation); + assert_eq!(after.execution.lease_owner.as_deref(), Some("worker")); + assert!(store.release(&replacement).await.unwrap()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_dummy_executor_completes_two_communities() { + let (store, id, _) = fixture().await; + run(&store, id, "worker", &options(), &CancellationToken::new()) + .await + .unwrap(); + assert_completed(&store, id).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn restart_after_handler_resumes_saved_branch_without_rerunning_action() { + let (store, id, _) = fixture().await; + assert!(store.inspect(id).await.unwrap().stage_history.is_empty()); + execute_without_transition(&store, id, &options()).await; + let after_gate = store.inspect(id).await.unwrap(); + assert_eq!(after_gate.stage_history.len(), 1); + assert_eq!(after_gate.stage_history[0].status, "completed"); + assert!(after_gate.stage_history[0].is_current); + let gate_finished = after_gate.stage_history[0].finished_at; + + let branch = RunOptions { + no_agents_in: vec![Uuid::from_u128(11)], + ..options() + }; + execute_without_transition(&store, id, &branch).await; + let after_db = store.inspect(id).await.unwrap(); + assert_eq!(after_db.stage_history.len(), 2); + assert_eq!( + after_db.stage_history[1].metadata_json["result"]["has_agents"], + false + ); + let db_finished = after_db.stage_history[1].finished_at; + // No skip flag on restart: routing must use the persisted result, not new options. + run( + &store, + id, + "worker-c", + &options(), + &CancellationToken::new(), + ) + .await + .unwrap(); + let completed = store.inspect(id).await.unwrap(); + assert_eq!(completed.request.status, "completed"); + assert_eq!(completed.stage_history.len(), 4); + assert_eq!(completed.stage_history[0].finished_at, gate_finished); + assert_eq!(completed.stage_history[1].finished_at, db_finished); + assert_eq!( + completed.stage_history[1].metadata_json["transition"]["kind"], + "community_complete" + ); + assert!(!completed + .stage_history + .iter() + .any(|row| row.target_community_id == Some(Uuid::from_u128(11)) + && row.stage == "purge_agents")); +} + +// Simulate process exit after the real handler commits but before the worker +// calls get_next_stage again. No operator-facing pause flag is needed. +async fn execute_without_transition(store: &EntityDeletionStore, id: Uuid, options: &RunOptions) { + let lease = store + .claim(id, "interrupted-worker", DEFAULT_LEASE_DURATION) + .await + .unwrap() + .unwrap(); + let NextStage::Stage(row) = UserDeletionTransition::new(store) + .get_next_stage(&lease, None) + .await + .unwrap() + else { + panic!("expected an executable stage"); + }; + handlers::dispatch(store, &lease, &row, options) + .await + .unwrap(); + store.release(&lease).await.unwrap(); +} diff --git a/crates/buzz-deletion/src/user/workflow.rs b/crates/buzz-deletion/src/user/workflow.rs new file mode 100644 index 0000000000..ce8e77a35a --- /dev/null +++ b/crates/buzz-deletion/src/user/workflow.rs @@ -0,0 +1,213 @@ +//! User stage rules and recovery of community progress; no SQL. +use buzz_db::entity_deletion::{EntityRequest, EntityStage}; +use buzz_db::user_deletion::UserDeletionInventory as Inventory; +use buzz_db::{DbError, Result}; +use serde::Serialize; +use serde_json::Value; +use std::collections::{HashMap, HashSet}; +use uuid::Uuid; +pub(super) fn invalid(message: &str) -> DbError { + DbError::DeletionSafety(message.into()) +} +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(tag = "kind", content = "stage", rename_all = "snake_case")] +pub(super) enum Decision { + Stage(&'static str), + EnterCommunities, + CommunityComplete, + RequestComplete, +} + +pub(super) struct Rule { + pub name: &'static str, + pub community_scope: bool, + pub allowed: &'static [Decision], + pub decide: fn(&Value) -> Result, +} + +pub(super) const INITIAL_STAGE: &str = "gate_user"; +const COMMUNITY_INITIAL_STAGE: &str = "purge_db"; +const RULES: &[Rule] = &[ + Rule { + name: "gate_user", + community_scope: false, + allowed: &[Decision::EnterCommunities], + decide: |_| Ok(Decision::EnterCommunities), + }, + Rule { + name: "purge_db", + community_scope: true, + allowed: &[Decision::Stage("purge_agents"), Decision::CommunityComplete], + decide: after_purge_db, + }, + Rule { + name: "purge_agents", + community_scope: true, + allowed: &[Decision::CommunityComplete], + decide: |_| Ok(Decision::CommunityComplete), + }, +]; + +fn after_purge_db(result: &Value) -> Result { + match result["has_agents"].as_bool() { + Some(true) => Ok(Decision::Stage("purge_agents")), + Some(false) => Ok(Decision::CommunityComplete), + None => Err(invalid("purge_db result must contain has_agents")), + } +} + +pub(super) fn rule(name: &str) -> Result<&'static Rule> { + RULES + .iter() + .find(|rule| rule.name == name) + .ok_or_else(|| invalid("unknown workflow stage")) +} + +pub(super) fn decision(stage: &EntityStage) -> Result { + let rule = rule(&stage.stage)?; + let result = stage + .metadata_json + .get("result") + .ok_or_else(|| invalid("completed stage has no saved result"))?; + let next = (rule.decide)(result)?; + if !rule.allowed.contains(&next) { + return Err(invalid("stage selected a disallowed transition")); + } + Ok(next) +} + +#[derive(Debug, PartialEq)] +pub(super) struct Step { + pub name: &'static str, + pub community: Option, +} + +pub(super) struct Progress { + pub current: Option, + pub completed_communities: HashSet, + pub terminal: bool, +} + +impl Progress { + /// Replay saved decisions through the persisted predecessor chain. This + /// validates chronology without relying on timestamps or UUID ordering. + pub fn rebuild( + request: &EntityRequest, + inventory: &Inventory, + history: &[EntityStage], + ) -> Result { + let mut progress = Self { + current: None, + completed_communities: HashSet::new(), + terminal: false, + }; + let mut successors = HashMap::new(); + for row in history { + let predecessor: Option = serde_json::from_value( + row.metadata_json + .get("previous_stage_id") + .cloned() + .ok_or_else(|| invalid("stage has no predecessor metadata"))?, + ) + .map_err(|_| invalid("invalid predecessor id"))?; + if successors.insert(predecessor, row).is_some() { + return Err(invalid( + "stage history branches into multiple current positions", + )); + } + } + let mut previous = None; + let mut expected = Some(Step { + name: INITIAL_STAGE, + community: None, + }); + let mut visited = HashSet::new(); + while let Some(row) = successors.get(&previous) { + if !visited.insert(row.id) { + return Err(invalid("cycle in stage history")); + } + let step = expected + .take() + .ok_or_else(|| invalid("stage exists after terminal outcome"))?; + if row.stage != step.name + || row.target_community_id != step.community + || rule(&row.stage)?.community_scope != row.target_community_id.is_some() + || row.metadata_json["inputs"] + != inventory.inputs(&request.target, step.community)? + { + return Err(invalid( + "stage does not match saved workflow position or inputs", + )); + } + previous = Some(row.id); + if row.is_current { + if row.metadata_json.get("transition").is_some() + || successors.contains_key(&previous) + { + return Err(invalid("current stage already has a committed transition")); + } + progress.current = Some((*row).clone()); + break; + } + if row.status != "completed" { + return Err(invalid("non-current stage is not completed")); + } + let outcome = decision(row)?; + if row.metadata_json["transition"] != serde_json::to_value(outcome)? { + return Err(invalid("saved transition disagrees with handler result")); + } + expected = progress.apply(inventory, row, outcome)?; + progress.terminal = expected.is_none(); + } + if visited.len() != history.len() { + return Err(invalid("disconnected stage history")); + } + Ok(progress) + } + + pub fn apply( + &mut self, + inventory: &Inventory, + row: &EntityStage, + outcome: Decision, + ) -> Result> { + match outcome { + Decision::Stage(name) => { + if rule(name)?.community_scope != row.target_community_id.is_some() { + return Err(invalid("transition crosses incompatible stage scopes")); + } + Ok(Some(Step { + name, + community: row.target_community_id, + })) + } + Decision::EnterCommunities | Decision::CommunityComplete => { + if outcome == Decision::CommunityComplete { + let id = row + .target_community_id + .ok_or_else(|| invalid("community outcome on request stage"))?; + if !self.completed_communities.insert(id) { + return Err(invalid("community completed twice")); + } + } + match inventory + .communities + .iter() + .find(|(id, _)| !self.completed_communities.contains(id)) + { + Some((id, _)) => Ok(Some(Step { + name: COMMUNITY_INITIAL_STAGE, + community: Some(*id), + })), + None => self.apply(inventory, row, Decision::RequestComplete), + } + } + Decision::RequestComplete => { + if self.completed_communities.len() != inventory.communities.len() { + return Err(invalid("request cannot finish with uncovered communities")); + } + Ok(None) + } + } + } +} diff --git a/docs/operator-user-deletion-executor.md b/docs/operator-user-deletion-executor.md new file mode 100644 index 0000000000..592ec378c4 --- /dev/null +++ b/docs/operator-user-deletion-executor.md @@ -0,0 +1,304 @@ +# Run and recover a dummy user-deletion workflow + +This workflow tests orchestration using real PostgreSQL rows and dummy actions. +It never gates a real account or deletes user data. Submit still creates an +artificial inventory: B (participant), then C (former participant). Only `demo:` +targets are accepted. Owned communities and production fencing are future work. + +## Code map: worker, policy, and persistence + +The worker and user-specific routing live in `buzz-deletion`. PostgreSQL +transactions live in `buzz-db`. Both are libraries in the same CLI process; +there is no additional service or database-side workflow function. + +| File | Responsibility | +| --- | --- | +| `buzz-deletion/src/user.rs` | Parse command arguments, call the user/shared stores or worker, print results | +| `buzz-deletion/src/user/worker.rs` | Claim, heartbeat, call transition, dispatch handler, repeat, release | +| `buzz-deletion/src/user/handlers.rs` | Dummy actions; save start/result/failure through the shared store | +| `buzz-deletion/src/user/transition.rs` | `UserDeletionTransition.get_next_stage`: recover or choose the next saved position | +| `buzz-deletion/src/user/workflow.rs` | Allowed user stages, result-based branching, community-progress reconstruction | +| `buzz-db/src/store/entity_deletion/mod.rs` | Shared store definition and exports; no stage policy | +| `buzz-db/src/store/entity_deletion/requests.rs` | Read/list/inspect requests and calculate/verify inventory fingerprints | +| `buzz-db/src/store/entity_deletion/approvals.rs` | Persist approval evidence and parent status together | +| `buzz-db/src/store/entity_deletion/leases.rs` | Claim, heartbeat, release, and owner/generation checks | +| `buzz-db/src/store/entity_deletion/stages.rs` | Save handler outcomes and commit stage transitions atomically | +| `buzz-db/src/store/user_deletion/mod.rs` | User-specific store definition and exports | +| `buzz-db/src/store/user_deletion/requests.rs` | Submit user requests and validate user inventory during approval | +| `buzz-db/src/store/user_deletion/inventory.rs` | Demo target checks, B/C inventory generation, and user inventory validation | + +The common store does not know the names `gate_user`, `purge_db`, or +`purge_agents`. `UserDeletionTransition` supplies a synchronous decision function +to `EntityDeletionStore.transition_stage`. The store locks the parent and loads +history, invokes that function while still holding the lock, checks the returned +operation against the current lifecycle, applies it, and commits. The decision +function performs no SQL or external work. Returning an error rolls back the +transaction. This avoids a gap between reading state and saving a decision. + +The operations are initialize, resume, advance, and already finished. An advance +includes the expected completed stage ID, the decision to retain in history, +and the successor's name, community, and inputs (or no successor for completion). +The store enforces ownership and current-stage preconditions; the user policy +ensures that the chosen destination follows the user workflow rules. + +User submission and inventory interpretation are deliberately separate from +common workflow storage, even though they use the same tables. User approval +runs its inventory validator inside the common approval transaction. Future +real PostgreSQL deletion operations belong in the user-specific database folder; +the current handlers still perform no real deletion. Existing community deletion +continues using its existing implementation. + +## The execution contract + +```text +approve: save approval, create no stages +run: claim parent request + get_next_stage(lease, None) → full stage row + handler(lease, row) → persist status and result + get_next_stage(lease, Some(row.id)) → full next row, Finished, or WaitUntil + repeat; release ownership on return +``` + +The worker owns claiming, heartbeats, shutdown, dispatch, and a bounded execution +budget. It does not inspect handler results, reconstruct community progress, or +choose branches. `get_next_stage` owns all those decisions. Specific rules live +in `user/workflow.rs`; the database transition transaction contains no +stage-name comparisons or two-community requirement. + +The handler gets the whole row. `metadata_json.inputs` holds the target and +community input. The handler marks the row running and either saves success +under `metadata_json.result` or records a failure and retry deadline. Returning +success from the handler means its outcome was saved, including a retryable +failure. Failure to save propagates out of the worker. + +### Durable state + +| Record | State | +| --- | --- | +| `entity_deletion_requests` | Inventory/digest, overall lifecycle, lease owner/generation/expiry, heartbeat, retry deadline/error/count, completion time | +| `entity_deletion_approvals` | Operator, time, approved inventory digest; retained separately | +| `entity_deletion_stage_history` | Stage/community, current position, execution status, timestamps, inputs, result, transition decision, predecessor ID | + +`is_current` means the workflow's durable position. It does not mean a handler +is presently running. In particular, **completed + current is valid**: the +handler finished, but the transition still needs to be committed. + +The handler's success transaction sets `completed`, saves its result and finish +time, and leaves `is_current = true`. The next transition transaction checks the +lease, reloads that result, chooses an allowed branch, clears the old current +flag, saves its decision, and inserts the successor with its inputs. If there +is no successor, it records request completion instead. Those transition writes +commit together; an insertion failure leaves the completed old row current. + +Every inserted stage stores `metadata_json.previous_stage_id` (null for the +initial stage). Reconstruction follows this chain, not timestamps or UUID sort +order. It checks inputs against the approved inventory, validates saved +transition decisions, and builds the set of completed communities. A persisted +`community_complete` decision covers a community even when a runtime branch +skipped a stage. Unknown stages, disconnected history, and missing results fail +without inventing progress. Reads are bounded to 10,000 reached stages and +inventories to 1,000 communities; these are resource limits, not a demo shape. + +### Starting, branching, and finishing + +Claiming saves ownership while preserving the lifecycle status. Thus a crash +between claim and initialization leaves an `approved` request with no stages. +The first `get_next_stage` creates the configured initial stage and sets the +parent to `running` in one transaction. A previously started request with no +current row and no terminal outcome is an error, not permission to create a +second gate. + +The version-1 workflow rules are: + +| Stage | Allowed outcome | Runtime choice | +| --- | --- | --- | +| `gate_user` | Enter communities | Select the first uncovered community | +| `purge_db` | `purge_agents` or community complete | Read saved boolean `result.has_agents` | +| `purge_agents` | Community complete | Mark this community covered | + +After a community completes, select the first uncovered community in inventory +order. After all are covered, finish the request. Missing `has_agents` is an +error, not an implicit skip. The inventory's version selects the currently +supported rules; future versions must retain the rules needed by active requests. +No full execution plan is created upfront. + +`previous_stage_id` is bookkeeping for uninterrupted execution. It must belong +to this request. After restart pass `None`; the database current row is the +source of truth. An old non-current stage is never transitioned twice. + +| Restart point | Recovery | +| --- | --- | +| Before the first transition | Initialize an approved request with empty history | +| During a handler | Resume the unfinished current row after lease takeover | +| After a saved failure | Wait for the deadline, then retry the same row | +| After saved success, before transition | Process the completed current row's saved result; do not rerun its handler | +| After transition commit | Return the saved successor row | +| After terminal commit | A rerun returns success without changing saved state | + +A lease lasts 60 seconds and handlers renew it every ten seconds. Handler and +renewal futures are polled independently in the same task, so a renewal waiting +for the handler's parent lock cannot stop the handler from committing. No +detached renewal task survives cancellation. Generation checks reject stale +workers' status, result, transition, heartbeat, and release writes. + +This is one request at a time through an explicit CLI command. There is no relay +scheduler or automatic retry daemon. Real handlers will need idempotent or +reconciled external actions: a crash after an external action but before saving +its result may repeat that handler. + +## Prepare a local test + +Follow [the intake guide](operator-user-deletion-intake.md) to build the binary +and migrate a fresh disposable database. The commands below use its container +`buzz-user-deletion-intake` and database `buzz`. Keep all commands in one shell. +These commands need PostgreSQL only, not a relay, Redis, or storage credentials. + +```bash +TARGET="demo:workflow-$(date +%s)" +REQUEST_JSON=$(./target/debug/buzz-admin user-deletions submit \ + --target "$TARGET" --requested-by operator-a) +REQUEST_ID=$(printf '%s\n' "$REQUEST_JSON" | jq -r '.id') +INVENTORY_DIGEST=$(printf '%s\n' "$REQUEST_JSON" | jq -r '.inventory_digest') +./target/debug/buzz-admin user-deletions approve "$REQUEST_ID" \ + --approved-by operator-b --inventory-digest "$INVENTORY_DIGEST" +``` + +Expected after approval: parent `approved`, one approval, **zero stage rows**. + +Define a helper to inspect the actual tables after each command: + +```bash +show_rows() { + docker exec buzz-user-deletion-intake psql -X -U buzz -d buzz -c " + SELECT target, status, lease_owner, lease_generation, lease_expires_at, + heartbeat_at, retry_count, next_attempt_at, last_error, completed_at + FROM entity_deletion_requests WHERE id = '$REQUEST_ID'; + SELECT id, stage, target_community_id, status, is_current, + started_at, finished_at, metadata_json, last_error + FROM entity_deletion_stage_history WHERE request_id = '$REQUEST_ID' + ORDER BY started_at NULLS LAST, id; + SELECT approved_by, approved_at, inventory_digest + FROM entity_deletion_approvals WHERE request_id = '$REQUEST_ID' + ORDER BY approved_at, id;" +} +show_rows +``` + +## Advance one stage at a time + +Run the same command each time, then inspect the tables: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" --next-step +show_rows +``` + +Each invocation executes one handler, processes its saved result through +`get_next_stage`, and releases ownership. It leaves the successor pending and +current. The final invocation marks the request completed instead. + +Starting with a newly approved request, expect: + +| Invocation | Handler completed | Current stage after the command | +| --- | --- | --- | +| 1 | `gate_user` | B `purge_db`, pending | +| 2 | B `purge_db` | B `purge_agents`, pending | +| 3 | B `purge_agents` | C `purge_db`, pending | +| 4 | C `purge_db` | C `purge_agents`, pending | +| 5 | C `purge_agents` | None; request completed | + +All earlier rows remain completed and non-current. Every command releases the +worker lease. Re-running a completed request is read-only. If a handler fails, +the same row remains failed and current; the command exits 1 with a saved retry +deadline instead of advancing past the failure. + +To finish the remaining workflow in one command, omit `--next-step`: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" +show_rows +``` + +Ordinary runs have an internal budget of 100 handler calls per invocation. +There are no `--max-stages` or `--pause-after-handler` options. A crash can still +leave a completed current row before its transition; the next run processes +that saved result before executing the next unfinished handler. + +## Test runtime branching + +Submit and approve a new request using the first block. Then: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" \ + --executor-id worker-a \ + --demo-no-agents-in 00000000-0000-0000-0000-00000000000c +show_rows +``` + +Expected: four completed rows: gate, B DB, B agents, C DB. C's DB result contains +`has_agents: false`; its saved transition is `community_complete`. There is no +C agents row. The flag simulates a runtime observation in the dummy handler; it +does not change the saved inventory or transition map. It may be repeated. + +To inspect branching step by step, submit and approve another request. Run once +with `--next-step` to finish the gate, then run with +`--next-step --demo-no-agents-in 00000000-0000-0000-0000-00000000000b`. +B DB completes and transitions directly to C DB. Resume without the flag: the +saved B decision remains unchanged, and no B agents row appears. + +## Test failure and retry + +On a new approved request: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" \ + --executor-id worker-a --fail-stage purge_db +echo "Exit code: $?" +show_rows +``` + +Expected exit 1. Gate has transitioned; B DB is failed and current. The parent +is `retry_wait`, with error, retry count 1, and `next_attempt_at` 30 seconds later. +The handler saved this outcome; `get_next_stage` returned `WaitUntil`, and the +worker released ownership. An immediate retry is refused. + +After the saved deadline: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" --executor-id worker-b +show_rows +``` + +The same B DB row resumes with its original first start time. The request +finishes. Successful transition clears the parent retry state. + +## Observe heartbeat and process interruption + +On a new approved request: + +```bash +./target/debug/buzz-admin user-deletions run "$REQUEST_ID" \ + --executor-id worker-a --stage-delay-ms 60000 +``` + +Inspect from another terminal using the same database/request. Heartbeat and +expiry advance every ten seconds. A competing worker is refused. Ctrl-C stops +execution and releases ownership; the interrupted row remains current. After +an abrupt process kill, wait for the saved lease expiry before running another +worker. It resumes the same unfinished row with a higher generation. + +## Rollout, recovery, and retained demo data + +Migrations 0069–0070 are additive. The serving relay does not schedule these +requests; only an operator explicitly running the demo CLI executes them. + +The Helm chart defaults `migrate.autoMigrate=true`. After these migrations are +applied, rolling back to a binary that does not contain them fails at startup +with sqlx `VersionMissing`. For this rollout, recover by rolling forward to a +corrected binary that retains migrations 0069–0070. Do not remove migration +records or change their checksums to force an older binary to start. An older +binary deployed with auto-migration disabled requires a separately verified +schema-compatibility plan; it is not the recovery procedure assumed here. + +Demo request rows, approval evidence, and stage history are deliberately retained +for inspection. This scaffold has no cleanup command or automatic retention job. diff --git a/docs/operator-user-deletion-intake.md b/docs/operator-user-deletion-intake.md index 6900661e08..44801d5c75 100644 --- a/docs/operator-user-deletion-intake.md +++ b/docs/operator-user-deletion-intake.md @@ -1,8 +1,11 @@ -# Submit a synthetic user-deletion request +# Prepare and approve a synthetic user-deletion request -This first slice saves and reads requests. It does not discover users, build -inventory, approve requests, start workers, or delete data. Only synthetic -`demo:` labels are accepted. +Submit saves a request and a fixed dummy inventory: community B (participant), +then community C (former participant). These are artificial UUIDs, not discovered +communities. Approve saves authorization without creating stages. The +`run` command initializes execution through `get_next_stage`, dispatches dummy +handlers, and asks the transition function for the next saved row. No user or +community data is deleted. Only `demo:` labels are accepted. ## Prepare a local database @@ -40,11 +43,13 @@ The command prints the saved row as JSON. It contains: | `id` | New UUID; copy this for inspection | | `entity_type` | `user` | | `target` | `demo:alice` | -| `status` | `submitted` | +| `status` | `inventoried` | | `requested_by` | `operator-a` | | `reason` | `null`, unless supplied with `--reason` | | `created_at` | Submission time | | `updated_at` | Same time; no execution has started | +| `inventory_manifest` | Ordered dummy communities and their actions | +| `inventory_digest` | SHA-256 identifying the saved inventory | List recent requests, then inspect the UUID returned above: @@ -67,6 +72,97 @@ target, operator, and reason returns that row without changing its timestamps. Reusing the target with a different operator or reason fails and preserves the original row. Use a new `demo:` label for a new request. -This slice stops at `submitted`. Inventory, approval, stage history, worker -ownership, and execution will be added in later changes. Their fields and tables -can be added through normal additive migrations. +## Approve without starting execution + +Inspect the inventory before approving. Copy its `inventory_digest` into the +command so approval is bound to what you inspected: + +```bash +INVENTORY_DIGEST= +./target/debug/buzz-admin user-deletions approve "$REQUEST_ID" \ + --approved-by operator-b --inventory-digest "$INVENTORY_DIGEST" +./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID" +``` + +The parent now has `status: approved`. Approval evidence lives in the separate +`entity_deletion_approvals` table. Each row has `id`, `request_id`, +`inventory_digest`, `approved_by`, and `approved_at`. Inspection returns these +rows in an `approvals` array (empty before approval), ordered by time and ID. +The parent keeps the inventory; approval rows bind each operator's decision to +its digest without overwriting earlier decisions. + +`stage_history` is empty after approval. The first `run` creates the initial +stage. Inspection reads parent, approvals, and history from one consistent +snapshot. No worker is started by approval. + +```bash +docker exec buzz-user-deletion-intake psql -X -U buzz -d buzz -c \ + "SELECT id, status, inventory_digest + FROM entity_deletion_requests ORDER BY created_at DESC;" +docker exec buzz-user-deletion-intake psql -X -U buzz -d buzz -c \ + "SELECT id, request_id, approved_by, approved_at, inventory_digest + FROM entity_deletion_approvals ORDER BY approved_at, id;" +docker exec buzz-user-deletion-intake psql -X -U buzz -d buzz -c \ + "SELECT request_id, stage, status, is_current, target_community_id, + started_at, finished_at, metadata_json + FROM entity_deletion_stage_history;" +``` + +The first approval inserts an approval row and changes the parent status in +one transaction. A failure rolls both back. +Repeating the same operator and digest preserves the approval ID and timestamp. +Another operator adds a new approval row, with its own time, without creating +another stage or changing the earlier approval: + +```bash +./target/debug/buzz-admin user-deletions approve "$REQUEST_ID" \ + --approved-by operator-c --inventory-digest "$INVENTORY_DIGEST" +./target/debug/buzz-admin user-deletions inspect "$REQUEST_ID" +``` + +There are now two approvals and still no stage rows. The unique +key `(request_id, inventory_digest, approved_by)` makes retries idempotent. +A digest that does not match the saved inventory is rejected. This slice has no +inventory replacement or later approval gates yet; those policies can be added +without replacing these historical rows. A partial unique index prevents two +current stages for one request. Executor retries update the same stage row. + +Operator names are self-asserted demo input. This is not authenticated approval +for real deletion. Owned communities and actual deletion handlers belong to +later changes. Continue with [the executor walkthrough](operator-user-deletion-executor.md) +to run the five dummy stages, inspect transitions, and try failure recovery. + +## Upgrade and rollout + +Apply after the intake schema from PR #8496. Migrations 0069 and 0070 add +inventory and execution fields, widen the lifecycle constraint, and create the +operator-global stage-history and approval tables. Constraint validation uses a +separate transaction so it does not retain the schema-change transaction's +write-blocking lock. Both +migrations set a five-second lock timeout and a thirty-second statement timeout. +There are no new triggers, functions, foreign keys, or data backfills. + +Existing `submitted` requests remain unchanged by migration. Resubmit with the +same target, operator, and reason to prepare their dummy inventory; the request +ID and original creation time stay unchanged. The previous CLI can still insert +and read requests against the expanded schema. The serving relay does not +automatically execute these requests; an operator runs the CLI worker explicitly. +Stage history and approvals intentionally survive +community deletion as operator recovery/audit evidence. + +Verify on a test environment with the submit, inspect, approve, and SQL commands +above using a new `demo:` target. Rollback has the usual sqlx limitation: an older +binary running auto-migration refuses unknown migrations (`VersionMissing`). +Do not roll back to an older auto-migrating relay without the deployment's +reviewed rollback procedure; the additive schema itself can remain in place. + +The unmerged draft migration 0069 changed when approvals moved to their own +table, and execution fields were added to requests. If you tested an earlier +draft, use a fresh disposable local database; +do not edit its sqlx checksums or erase its saved demo requests. + +This executor revision does not change SQL schema. Earlier draft requests whose +stage metadata predates saved inputs/results/predecessor IDs are not a supported +resume format. Keep their records intact and submit a new `demo:` label to test +this version. Approval no longer creates a gate row; it is initialized by the +first `get_next_stage` call under worker ownership. diff --git a/migrations/0069_entity_deletion_approval.sql b/migrations/0069_entity_deletion_approval.sql new file mode 100644 index 0000000000..a1b4de50c0 --- /dev/null +++ b/migrations/0069_entity_deletion_approval.sql @@ -0,0 +1,60 @@ +SET LOCAL lock_timeout = '5s'; +SET LOCAL statement_timeout = '30s'; + +-- Expand the demo intake. Existing submitted rows remain valid and unchanged. +-- A matching resubmission through the new CLI prepares their dummy inventory. +ALTER TABLE entity_deletion_requests + DROP CONSTRAINT entity_deletion_requests_status_check, + ADD CONSTRAINT entity_deletion_requests_status_check + CHECK (status IN ('submitted', 'inventoried', 'approved', 'running', 'retry_wait', 'completed')) NOT VALID, + ADD COLUMN inventory_manifest JSONB, + ADD COLUMN inventory_digest TEXT, + ADD COLUMN lease_owner TEXT, + ADD COLUMN lease_generation BIGINT NOT NULL DEFAULT 0, + ADD COLUMN lease_expires_at TIMESTAMPTZ, + ADD COLUMN heartbeat_at TIMESTAMPTZ, + ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0, + ADD COLUMN next_attempt_at TIMESTAMPTZ, + ADD COLUMN last_error TEXT, + ADD COLUMN completed_at TIMESTAMPTZ; +-- Validate in 0070 so the scan does not retain this transaction's DDL lock. + +-- Operator-global workflow history. Retained after community deletion as +-- orchestration evidence; target_community_id is provenance, not tenant ownership. +CREATE TABLE entity_deletion_stage_history ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + request_id UUID NOT NULL, + target_community_id UUID, + stage TEXT NOT NULL, + 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 '{}'::jsonb, + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX entity_deletion_one_current_stage + ON entity_deletion_stage_history (request_id) WHERE is_current = true; +CREATE INDEX entity_deletion_stage_history_request_idx + ON entity_deletion_stage_history (request_id); + +INSERT INTO _operator_global_tables (table_name, reason) +VALUES ('entity_deletion_stage_history', 'Operator-global deletion workflow history retained for recovery and audit') +ON CONFLICT (table_name) DO NOTHING; + +-- Approval evidence is retained after community deletion for operator audit. +-- Each operator approves an inventory once; retries preserve the original row. +CREATE TABLE entity_deletion_approvals ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + request_id UUID NOT NULL, + inventory_digest TEXT NOT NULL CHECK (inventory_digest ~ '^[0-9a-f]{64}$'), + approved_by TEXT NOT NULL CHECK (approved_by = btrim(approved_by) AND octet_length(approved_by) >= 1 AND octet_length(approved_by) <= 256), + approved_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (request_id, inventory_digest, approved_by) +); + +INSERT INTO _operator_global_tables (table_name, reason) +VALUES ('entity_deletion_approvals', 'Operator-global deletion approval history retained for audit') +ON CONFLICT (table_name) DO NOTHING; diff --git a/migrations/0070_validate_entity_deletion_status.sql b/migrations/0070_validate_entity_deletion_status.sql new file mode 100644 index 0000000000..b8226de360 --- /dev/null +++ b/migrations/0070_validate_entity_deletion_status.sql @@ -0,0 +1,6 @@ +SET LOCAL lock_timeout = '5s'; +SET LOCAL statement_timeout = '30s'; + +-- Separate from the DDL transaction so validation does not block writers. +ALTER TABLE entity_deletion_requests + VALIDATE CONSTRAINT entity_deletion_requests_status_check; diff --git a/schema/schema.sql b/schema/schema.sql index 6e85a9a455..15376d9eb2 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -122,7 +122,9 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i tables/public/push_gateway_delivery_request_replays.sql \i tables/public/replica_heartbeat.sql \i tables/public/community_deletion_requests.sql +\i tables/public/entity_deletion_approvals.sql \i tables/public/entity_deletion_requests.sql +\i tables/public/entity_deletion_stage_history.sql \i tables/public/community_deletion_approvals.sql \i tables/public/community_deletion_checkpoints.sql \i tables/public/community_deletion_manifest_keys.sql @@ -195,7 +197,9 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i seeds/push_gateway_delivery_request_replays.sql \i seeds/replica_heartbeat.sql \i seeds/community_deletion_requests.sql +\i seeds/entity_deletion_approvals.sql \i seeds/entity_deletion_requests.sql +\i seeds/entity_deletion_stage_history.sql \i seeds/community_deletion_approvals.sql \i seeds/community_deletion_checkpoints.sql \i seeds/community_deletion_manifest_keys.sql diff --git a/schema/seeds/entity_deletion_approvals.sql b/schema/seeds/entity_deletion_approvals.sql new file mode 100644 index 0000000000..1686edc158 --- /dev/null +++ b/schema/seeds/entity_deletion_approvals.sql @@ -0,0 +1,3 @@ +INSERT INTO _operator_global_tables (table_name, reason) +VALUES ('entity_deletion_approvals', 'Operator-global deletion approval history retained for audit') +ON CONFLICT (table_name) DO NOTHING; diff --git a/schema/seeds/entity_deletion_stage_history.sql b/schema/seeds/entity_deletion_stage_history.sql new file mode 100644 index 0000000000..014c480729 --- /dev/null +++ b/schema/seeds/entity_deletion_stage_history.sql @@ -0,0 +1,3 @@ +INSERT INTO _operator_global_tables (table_name, reason) +VALUES ('entity_deletion_stage_history', 'Operator-global deletion workflow history retained for recovery and audit') +ON CONFLICT (table_name) DO NOTHING; diff --git a/schema/tables/public/entity_deletion_approvals.sql b/schema/tables/public/entity_deletion_approvals.sql new file mode 100644 index 0000000000..eb71a2c889 --- /dev/null +++ b/schema/tables/public/entity_deletion_approvals.sql @@ -0,0 +1,9 @@ +-- Operator-global approval evidence; application transactions check the request. +CREATE TABLE entity_deletion_approvals ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + request_id UUID NOT NULL, + inventory_digest TEXT NOT NULL CHECK (inventory_digest ~ '^[0-9a-f]{64}$'), + approved_by TEXT NOT NULL CHECK (approved_by = btrim(approved_by) AND octet_length(approved_by) >= 1 AND octet_length(approved_by) <= 256), + approved_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (request_id, inventory_digest, approved_by) +); diff --git a/schema/tables/public/entity_deletion_requests.sql b/schema/tables/public/entity_deletion_requests.sql index a7788023c5..f949a10254 100644 --- a/schema/tables/public/entity_deletion_requests.sql +++ b/schema/tables/public/entity_deletion_requests.sql @@ -3,10 +3,20 @@ 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 = 'submitted'), + status TEXT NOT NULL DEFAULT 'submitted' CHECK (status IN ('submitted', 'inventoried', 'approved', 'running', 'retry_wait', 'completed')), requested_by TEXT NOT NULL CHECK (length(trim(requested_by)) BETWEEN 1 AND 256), reason TEXT CHECK (octet_length(reason) <= 4096), created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + inventory_manifest JSONB, + inventory_digest TEXT, + lease_owner TEXT, + lease_generation BIGINT NOT NULL DEFAULT 0, + lease_expires_at TIMESTAMPTZ, + heartbeat_at TIMESTAMPTZ, + retry_count INTEGER NOT NULL DEFAULT 0, + next_attempt_at TIMESTAMPTZ, + last_error TEXT, + completed_at TIMESTAMPTZ, UNIQUE (entity_type, target) ); 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..b277b4cee7 --- /dev/null +++ b/schema/tables/public/entity_deletion_stage_history.sql @@ -0,0 +1,20 @@ +-- Operator-global workflow history. Retained after community deletion as +-- orchestration evidence; target_community_id is provenance, not tenant ownership. +CREATE TABLE entity_deletion_stage_history ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + request_id UUID NOT NULL, + target_community_id UUID, + stage TEXT NOT NULL, + 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 '{}'::jsonb, + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE UNIQUE INDEX entity_deletion_one_current_stage + ON entity_deletion_stage_history (request_id) WHERE is_current = true; +CREATE INDEX entity_deletion_stage_history_request_idx + ON entity_deletion_stage_history (request_id);