diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index c85b7dc6b9..9205bdc899 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -69,7 +69,8 @@ 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, user_deletion, workflow, + replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion, + user_deletion_fences, workflow, }; pub use allowlist::AllowlistEntry; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 3540a52b5d..38ed0014c4 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -549,6 +549,7 @@ mod postgres_tests { "entity_deletion_approvals", "entity_deletion_requests", "entity_deletion_stage_history", + "user_deletion_fences", "community_deletion_approvals", "community_deletion_checkpoints", "community_deletion_manifest_keys", @@ -775,7 +776,8 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 73); + assert_eq!(migrations.len(), 74); + assert_eq!(migrations[73].version, 74); assert_eq!(migrations[72].version, 73); assert_eq!(migrations[71].version, 72); assert_eq!(migrations[70].version, 71); @@ -2905,6 +2907,7 @@ mod postgres_tests { "entity_deletion_approvals", "entity_deletion_requests", "entity_deletion_stage_history", + "user_deletion_fences", ] { assert_eq!( columns(&desired, table).await, @@ -2941,6 +2944,7 @@ mod postgres_tests { for table in [ "entity_deletion_requests", "entity_deletion_stage_history", + "user_deletion_fences", "entity_deletion_approvals", ] { assert_eq!( diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 3ef4fdfb33..d1aaf28a1f 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -4400,6 +4400,116 @@ mod postgres_tests { use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; use std::str::FromStr; + #[tokio::test] + #[ignore = "requires Postgres"] + async fn user_fences_survive_community_and_profile_deletion() { + assert_user_fence_retention().await; + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn migration_schema_user_fences_survive_community_and_profile_deletion() { + assert_user_fence_retention().await; + } + + async fn assert_user_fence_retention() { + use crate::user_deletion_fences::{self as fences, FenceIdentity}; + let (db, store) = store().await; + let (request, inventory) = inventoried_request(&db, &store).await; + let entity = db + .user_deletion_store() + .submit("demo:fence-retention", "operator", None) + .await + .unwrap(); + let entities = db.entity_deletion_store(); + entities + .approve( + entity.id, + "approver", + entity.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let entity_lease = entities + .claim(entity.id, "fence-worker", Duration::from_secs(300)) + .await + .unwrap() + .unwrap(); + let human_key = nostr::Keys::generate().public_key(); + let agent_key = nostr::Keys::generate().public_key(); + let human = FenceIdentity::user(human_key); + let agent = FenceIdentity::agent(request.community_id, agent_key); + sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)") + .bind(request.community_id.as_uuid()) + .bind(human_key.to_bytes().to_vec()) + .execute(&db.pool) + .await + .unwrap(); + let mut tx = db.pool.begin().await.unwrap(); + fences::lock_request(&mut tx, &entity_lease).await.unwrap(); + fences::lock_exclusive(&mut tx, &[human, agent]) + .await + .unwrap(); + let global = fences::insert(&mut tx, human, &entity_lease, None) + .await + .unwrap(); + let scoped = fences::insert(&mut tx, agent, &entity_lease, Some(human_key)) + .await + .unwrap(); + tx.commit().await.unwrap(); + sqlx::query("DELETE FROM users WHERE community_id = $1 AND pubkey = $2") + .bind(request.community_id.as_uuid()) + .bind(human_key.to_bytes().to_vec()) + .execute(&db.pool) + .await + .unwrap(); + let mut tx = db.pool.begin().await.unwrap(); + assert_eq!( + fences::find(&mut tx, human).await.unwrap(), + Some(global.clone()) + ); + tx.commit().await.unwrap(); + + store.approve(request.id, "approver", None).await.unwrap(); + let claim = store + .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) + .await + .unwrap() + .unwrap(); + store.begin_quiescing(&claim.lease).await.unwrap(); + let generation = store.fence(&claim.lease).await.unwrap(); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + store + .freeze_destructive_storage_manifest(&token, &inventory.storage) + .await + .unwrap(); + store.mark_drained(&token).await.unwrap(); + store + .mark_bindings_removed(&token, serde_json::json!({"keys": 0})) + .await + .unwrap(); + store.purge_postgres(&token).await.unwrap(); + // This database-only fixture models the external cache adapter's + // completion before exercising PostgreSQL's logical-absence verifier. + store + .mark_cache_purged(&token, serde_json::json!({"test_cache": "empty"})) + .await + .unwrap(); + store + .verify_postgres_logically_deleted(&token) + .await + .unwrap(); + // Full row equality also proves that scoped restrictions were not + // widened to global ones by clearing scope_community_id. + let mut tx = db.pool.begin().await.unwrap(); + assert_eq!(fences::find(&mut tx, human).await.unwrap(), Some(global)); + assert_eq!(fences::find(&mut tx, agent).await.unwrap(), Some(scoped)); + tx.commit().await.unwrap(); + } + async fn store() -> (Db, DeletionStore) { let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) diff --git a/crates/buzz-db/src/store/entity_deletion/leases.rs b/crates/buzz-db/src/store/entity_deletion/leases.rs index 92fcc0b789..bffb6e4579 100644 --- a/crates/buzz-db/src/store/entity_deletion/leases.rs +++ b/crates/buzz-db/src/store/entity_deletion/leases.rs @@ -140,7 +140,10 @@ impl EntityDeletionStore { } } -pub(super) async fn lock_owned(tx: &mut Tx<'_>, lease: &EntityLease) -> Result { +pub(in crate::store) 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!( @@ -167,7 +170,7 @@ pub(super) async fn lock_owned(tx: &mut Tx<'_>, lease: &EntityLease) -> Result, request: &EntityRequest, ) -> Result<()> { diff --git a/crates/buzz-db/src/store/entity_deletion/mod.rs b/crates/buzz-db/src/store/entity_deletion/mod.rs index e78c5d719c..f9b840ce82 100644 --- a/crates/buzz-db/src/store/entity_deletion/mod.rs +++ b/crates/buzz-db/src/store/entity_deletion/mod.rs @@ -25,6 +25,7 @@ mod leases; pub(crate) mod requests; mod stages; pub use approvals::EntityApproval; +pub(super) use leases::{lock_owned, validate_approved_inventory}; pub use leases::{EntityExecution, EntityLease}; pub use requests::{inventory_digest, EntityInspection, EntityRequest}; use requests::{validate_operator, validated_inventory_digest}; diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index f3ea643ba4..d36fa4d089 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -28,6 +28,8 @@ pub(crate) mod event_follow_up; pub mod nip_fi_binding; /// User-specific deletion submission and inventory preparation. pub mod user_deletion; +/// Operator-global user restrictions and identity-lock primitives. +pub mod user_deletion_fences; #[cfg(test)] mod event_follow_up_postgres_tests; diff --git a/crates/buzz-db/src/store/user_deletion_fences.rs b/crates/buzz-db/src/store/user_deletion_fences.rs new file mode 100644 index 0000000000..2d437fff09 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion_fences.rs @@ -0,0 +1,260 @@ +//! Operator-global restriction storage and identity locks; no serving or worker +//! path calls these helpers yet. Installing fences does not itself enforce them. +//! +//! Callers use a writer-pool READ COMMITTED transaction. Acquire existing +//! community guards first, then `lock_request` for the worker, then all human +//! locks, then scoped agent locks. +//! Within each group acquire the whole set in one call (sorted here), never +//! upgrade shared locks to exclusive. Gate workers may discover agents after +//! locking humans, then lock the complete agent set. Keep the transaction open +//! through the protected write/checkpoint; commit or rollback releases the locks. + +use buzz_core::CommunityId; +use chrono::{DateTime, Utc}; +use nostr::PublicKey; +use sha2::{Digest, Sha256}; +use sqlx::{Postgres, Transaction}; +use uuid::Uuid; + +use super::entity_deletion::{lock_owned, validate_approved_inventory, EntityLease}; +use crate::{DbError, Result}; + +/// A stable lock/restriction identity, independent of mutable profile rows. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct FenceIdentity { + pubkey: PublicKey, + scope: Option, +} + +impl FenceIdentity { + /// A human key restricted across the operator's hosted communities. + pub fn user(pubkey: PublicKey) -> Self { + Self { + pubkey, + scope: None, + } + } + + /// An agent key restricted only within its ownership community. + pub fn agent(community: CommunityId, pubkey: PublicKey) -> Self { + Self { + pubkey, + scope: Some(community), + } + } + + fn scope_id(self) -> Option { + self.scope.map(|scope| *scope.as_uuid()) + } + + fn lock_key(self) -> i64 { + let mut hash = Sha256::new(); + hash.update(b"buzz:user-deletion-fence:v1\0"); + hash.update([u8::from(self.scope.is_some())]); + if let Some(scope) = self.scope { + hash.update(scope.as_uuid().as_bytes()); + } + hash.update(self.pubkey.to_bytes()); + let digest = hash.finalize(); + let mut bytes = [0; 8]; + bytes.copy_from_slice(&digest[..8]); + i64::from_be_bytes(bytes) + } +} + +/// Durable evidence of the first restriction on an identity and scope. +#[derive(Clone, Debug, PartialEq, Eq, sqlx::FromRow)] +pub struct UserDeletionFence { + /// Stable restriction identifier. + pub id: Uuid, + /// Canonical lowercase public key. + pub pubkey: String, + /// Agent scope, or NULL for a global human restriction. + pub scope_community_id: Option, + /// Request that first installed this restriction. + pub request_id: Uuid, + /// Verified owner at fencing time; absent for a human restriction. + pub source_owner_pubkey: Option, + /// Time of the first insertion; retries preserve it. + pub blocked_at: DateTime, +} + +/// Acquire shared identity locks for normal writes. Does not check restrictions. +pub async fn lock_shared( + tx: &mut Transaction<'_, Postgres>, + identities: &[FenceIdentity], +) -> Result<()> { + lock(tx, identities, false).await +} + +/// Acquire exclusive identity locks for fencing. Does not install restrictions. +pub async fn lock_exclusive( + tx: &mut Transaction<'_, Postgres>, + identities: &[FenceIdentity], +) -> Result<()> { + lock(tx, identities, true).await +} + +async fn lock( + tx: &mut Transaction<'_, Postgres>, + identities: &[FenceIdentity], + exclusive: bool, +) -> Result<()> { + if identities.len() > 4096 { + return Err(DbError::DeletionSafety( + "too many user-deletion identity locks".into(), + )); + } + let (isolation, timeout): (String, String) = sqlx::query_as( + "SELECT current_setting('transaction_isolation'), current_setting('lock_timeout')", + ) + .fetch_one(&mut **tx) + .await?; + if isolation != "read committed" { + return Err(DbError::DeletionSafety( + "user-deletion locks require READ COMMITTED".into(), + )); + } + // Bound each lock wait without widening a caller's shorter timeout. Restore + // the original setting on success; a SQL error requires caller rollback. + sqlx::query( + "SELECT set_config('lock_timeout', + (EXTRACT(EPOCH FROM LEAST(NULLIF(current_setting('lock_timeout')::interval, + interval '0'), interval '5 seconds')) * 1000)::bigint::text, true)", + ) + .execute(&mut **tx) + .await?; + let mut keys: Vec<_> = identities + .iter() + .map(|identity| (identity.scope.is_some(), identity.lock_key())) + .collect(); + keys.sort_unstable(); + keys.dedup(); + for (_, key) in keys { + let query = if exclusive { + "SELECT pg_advisory_xact_lock($1)" + } else { + "SELECT pg_advisory_xact_lock_shared($1)" + }; + sqlx::query(query).bind(key).execute(&mut **tx).await?; + } + sqlx::query("SELECT set_config('lock_timeout', $1, true)") + .bind(timeout) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Lock and read the exact identity restriction in the caller's transaction. +/// +/// For multiple identities, acquire the full shared set first. The caller must +/// refuse a restricted operation and retain this transaction through any write. +/// An agent lookup is scoped; it does not resolve or check the human owner. +/// The planned agent-write path locks the verified owner and agent together, +/// then reads the agent restriction only. This requires the gate to install the +/// human and all owned-agent restrictions atomically, and creation/assignment +/// paths to reject restricted owners. These callers are not wired in yet. +pub async fn find( + tx: &mut Transaction<'_, Postgres>, + identity: FenceIdentity, +) -> Result> { + lock_shared(tx, &[identity]).await?; + read(tx, identity).await +} + +async fn read( + tx: &mut Transaction<'_, Postgres>, + identity: FenceIdentity, +) -> Result> { + // Explicit predicates let PostgreSQL use the matching partial unique index. + let sql = if identity.scope.is_some() { + "SELECT id, pubkey, scope_community_id, request_id, source_owner_pubkey, blocked_at + FROM user_deletion_fences WHERE pubkey = $1 AND scope_community_id = $2" + } else { + "SELECT id, pubkey, scope_community_id, request_id, source_owner_pubkey, blocked_at + FROM user_deletion_fences WHERE pubkey = $1 AND scope_community_id IS NULL" + }; + let query = sqlx::query_as(sql).bind(identity.pubkey.to_hex()); + let query = if let Some(scope) = identity.scope_id() { + query.bind(scope) + } else { + query + }; + Ok(query.fetch_optional(&mut **tx).await?) +} + +/// Lock the parent request and validate the worker's lease and approved inventory. +/// +/// Call before acquiring any identity locks in a fencing transaction. Each +/// insertion rechecks this guard; holding the row lock prevents another worker +/// from claiming the request until this transaction ends. The caller must still +/// verify that the identities and agent ownership belong to this request. +pub async fn lock_request(tx: &mut Transaction<'_, Postgres>, lease: &EntityLease) -> Result<()> { + let request = lock_owned(tx, lease).await?; + if request.entity_type != "user" { + return Err(DbError::DeletionSafety( + "fences require a user deletion request".into(), + )); + } + validate_approved_inventory(tx, &request).await +} + +/// Insert an idempotent restriction under a valid worker lease and identity lock. +/// +/// For multiple identities, first call `lock_request`, then acquire the complete exclusive +/// identity set in order, verify agent ownership, and save its checkpoint in this +/// same transaction. A retry preserves the first row; conflicting provenance +/// errors. On any error roll back, including other fences/checkpoints. +pub async fn insert( + tx: &mut Transaction<'_, Postgres>, + identity: FenceIdentity, + lease: &EntityLease, + source_owner: Option, +) -> Result { + if identity.scope.is_some() != source_owner.is_some() { + return Err(DbError::DeletionSafety( + "agent fences require an owner; user fences do not".into(), + )); + } + lock_request(tx, lease).await?; + let request_id = lease.request_id; + lock_exclusive(tx, &[identity]).await?; + let owner = source_owner.map(|key| key.to_hex()); + sqlx::query( + "INSERT INTO user_deletion_fences (pubkey, scope_community_id, request_id, source_owner_pubkey) + VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + ) + .bind(identity.pubkey.to_hex()).bind(identity.scope_id()).bind(request_id).bind(&owner) + .execute(&mut **tx).await?; + let saved = read(tx, identity) + .await? + .ok_or_else(|| DbError::DeletionSafety("fence insertion returned no row".into()))?; + if saved.request_id != request_id || saved.source_owner_pubkey != owner { + return Err(DbError::DeletionSafety( + "identity is already fenced by different provenance".into(), + )); + } + Ok(saved) +} + +#[cfg(test)] +#[path = "user_deletion_fences_postgres_tests.rs"] +mod postgres_tests; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lock_mapping_is_stable_across_processes_and_releases() { + // A change to these values would split the lock domain during rollout. + let key = + PublicKey::from_hex("79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798") + .unwrap(); + assert_eq!(FenceIdentity::user(key).lock_key(), -4153709860967480559); + assert_eq!( + FenceIdentity::agent(CommunityId::from_uuid(Uuid::from_u128(1)), key).lock_key(), + -3253902268089525902, + ); + } +} diff --git a/crates/buzz-db/src/store/user_deletion_fences_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion_fences_postgres_tests.rs new file mode 100644 index 0000000000..27281150db --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion_fences_postgres_tests.rs @@ -0,0 +1,453 @@ +use super::*; +use sqlx::PgPool; +use std::time::Duration; + +async fn pool() -> PgPool { + PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap() +} + +async fn request(pool: &PgPool) -> EntityLease { + let db = crate::Db::from_pool(pool.clone()); + let request = db + .user_deletion_store() + .submit( + &format!("demo:{}", Uuid::new_v4().simple()), + "operator", + None, + ) + .await + .unwrap(); + let store = db.entity_deletion_store(); + store + .approve( + request.id, + "approver", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + store + .claim(request.id, "worker", Duration::from_secs(300)) + .await + .unwrap() + .unwrap() +} + +fn user() -> FenceIdentity { + FenceIdentity::user(nostr::Keys::generate().public_key()) +} + +async fn assert_plain_schema(pool: &PgPool) { + let guards: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_trigger WHERE tgrelid = 'user_deletion_fences'::regclass", + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(guards, 0); + let fks: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_constraint + WHERE conrelid = 'user_deletion_fences'::regclass AND contype = 'f'", + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(fks, 0); + crate::Db::from_pool(pool.clone()) + .validate_deletion_catalog() + .await + .unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn desired_schema_is_global_and_has_no_cascades_or_triggers() { + assert_plain_schema(&pool().await).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_is_global_and_has_no_cascades_or_triggers() { + let pool = pool().await; + crate::Db::from_pool(pool.clone()).migrate().await.unwrap(); + assert_plain_schema(&pool).await; + let registered: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM _operator_global_tables WHERE table_name = 'user_deletion_fences')", + ).fetch_one(&pool).await.unwrap(); + assert!(registered); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn fences_are_atomic_idempotent_scoped_and_preserve_provenance() { + let pool = pool().await; + let lease = request(&pool).await; + let human = user(); + let agent_key = nostr::Keys::generate().public_key(); + let a = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), agent_key); + let b = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), agent_key); + let mut tx = pool.begin().await.unwrap(); + lock_request(&mut tx, &lease).await.unwrap(); + lock_exclusive(&mut tx, &[a, human]).await.unwrap(); + insert(&mut tx, human, &lease, None).await.unwrap(); + insert(&mut tx, a, &lease, Some(human.pubkey)) + .await + .unwrap(); + tx.rollback().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(find(&mut tx, human).await.unwrap().is_none()); + assert!(find(&mut tx, a).await.unwrap().is_none()); + tx.commit().await.unwrap(); + + let mut tx = pool.begin().await.unwrap(); + lock_request(&mut tx, &lease).await.unwrap(); + lock_exclusive(&mut tx, &[a, human]).await.unwrap(); + let first = insert(&mut tx, human, &lease, None).await.unwrap(); + let scoped = insert(&mut tx, a, &lease, Some(human.pubkey)) + .await + .unwrap(); + tx.commit().await.unwrap(); + // A new connection/transaction reads durable state, not an in-memory cache. + let mut tx = pool.begin().await.unwrap(); + assert_eq!(insert(&mut tx, human, &lease, None).await.unwrap(), first); + assert_eq!( + insert(&mut tx, a, &lease, Some(human.pubkey)) + .await + .unwrap(), + scoped + ); + tx.commit().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM user_deletion_fences") + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!( + count, 2, + "retries must not create duplicate global or scoped rows" + ); + assert!(find(&mut tx, b).await.unwrap().is_none()); + assert!(find(&mut tx, FenceIdentity::user(agent_key)) + .await + .unwrap() + .is_none()); + tx.commit().await.unwrap(); + let other_lease = request(&pool).await; + let mut tx = pool.begin().await.unwrap(); + assert!(insert(&mut tx, human, &other_lease, None).await.is_err()); + tx.rollback().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(insert(&mut tx, a, &lease, Some(user().pubkey)) + .await + .is_err()); + tx.rollback().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert_eq!(find(&mut tx, human).await.unwrap(), Some(first)); + assert_eq!(find(&mut tx, a).await.unwrap(), Some(scoped)); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn rejects_missing_requests_and_invalid_owner_shape() { + let pool = pool().await; + let lease = request(&pool).await; + let human = user(); + let agent = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), user().pubkey); + let mut tx = pool.begin().await.unwrap(); + let missing = EntityLease { + request_id: Uuid::new_v4(), + ..lease.clone() + }; + assert!(insert(&mut tx, human, &missing, None).await.is_err()); + assert!(insert(&mut tx, human, &lease, Some(user().pubkey)) + .await + .is_err()); + assert!(insert(&mut tx, agent, &lease, None).await.is_err()); + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM user_deletion_fences") + .fetch_one(&mut *tx) + .await + .unwrap(); + assert_eq!(count, 0); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn identity_locks_share_conflict_and_release_at_transaction_end() { + let pool = pool().await; + let human = user(); + let mut first = pool.begin().await.unwrap(); + lock_shared(&mut first, &[human]).await.unwrap(); + let mut second = pool.begin().await.unwrap(); + lock_shared(&mut second, &[human]).await.unwrap(); + second.commit().await.unwrap(); + let mut second = pool.begin().await.unwrap(); + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(&mut *second) + .await + .unwrap(); + let error = lock_exclusive(&mut second, &[human]).await.unwrap_err(); + assert!(error.to_string().contains("lock timeout"), "{error}"); + second.rollback().await.unwrap(); + first.commit().await.unwrap(); + + let mut gate = pool.begin().await.unwrap(); + lock_exclusive(&mut gate, &[human]).await.unwrap(); + let mut normal = pool.begin().await.unwrap(); + // An unrelated identity is not blocked. + lock_shared(&mut normal, &[user()]).await.unwrap(); + sqlx::query("SET LOCAL lock_timeout = '100ms'") + .execute(&mut *normal) + .await + .unwrap(); + assert!(lock_shared(&mut normal, &[human]).await.is_err()); + normal.rollback().await.unwrap(); + gate.rollback().await.unwrap(); + let mut normal = pool.begin().await.unwrap(); + lock_shared(&mut normal, &[human]).await.unwrap(); + let timeout: String = sqlx::query_scalar("SHOW lock_timeout") + .fetch_one(&mut *normal) + .await + .unwrap(); + assert_eq!(timeout, "0"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn read_waits_for_gate_then_observes_committed_restriction() { + let pool = pool().await; + let lease = request(&pool).await; + let human = user(); + let mut gate = pool.begin().await.unwrap(); + let saved = insert(&mut gate, human, &lease, None).await.unwrap(); + let mut normal = pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *normal) + .await + .unwrap(); + let reader = tokio::spawn(async move { find(&mut normal, human).await.unwrap() }); + // Observe the actual lock wait before committing, rather than assuming a + // sleep was long enough for the competing transaction to reach the guard. + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE pid = $1 AND locktype = 'advisory' AND NOT granted)", + ).bind(pid).fetch_one(&pool).await.unwrap(); + if waiting { break; } + tokio::task::yield_now().await; + } + }).await.unwrap(); + gate.commit().await.unwrap(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), reader) + .await + .unwrap() + .unwrap(), + Some(saved) + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn batch_order_is_stable_and_agent_locks_are_community_scoped() { + let pool = pool().await; + let human = user(); + let a = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), user().pubkey); + let b = FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), a.pubkey); + let mut first = pool.begin().await.unwrap(); + lock_exclusive(&mut first, &[a]).await.unwrap(); + let mut second = pool.begin().await.unwrap(); + lock_exclusive(&mut second, &[b]).await.unwrap(); + second.commit().await.unwrap(); + first.commit().await.unwrap(); + let run = |identities: Vec| { + let pool = pool.clone(); + async move { + let mut tx = pool.begin().await.unwrap(); + lock_exclusive(&mut tx, &identities).await.unwrap(); + tx.commit().await.unwrap(); + } + }; + tokio::time::timeout(Duration::from_secs(10), async { + tokio::join!(run(vec![a, human, b, human]), run(vec![b, a, human])); + }) + .await + .unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn rejects_stale_snapshot_isolation_and_oversized_lock_sets() { + let pool = pool().await; + let mut tx = pool.begin().await.unwrap(); + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") + .execute(&mut *tx) + .await + .unwrap(); + assert!(find(&mut tx, user()).await.is_err()); + tx.rollback().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(lock_exclusive(&mut tx, &vec![user(); 4097]).await.is_err()); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn insert_requires_approved_inventory_and_current_unexpired_worker() { + let pool = pool().await; + let db = crate::Db::from_pool(pool.clone()); + let store = db.entity_deletion_store(); + let submitted = db + .user_deletion_store() + .submit("demo:unapproved", "operator", None) + .await + .unwrap(); + assert!(store + .claim(submitted.id, "worker", Duration::from_secs(300)) + .await + .unwrap() + .is_none()); + let unapproved = EntityLease { + request_id: submitted.id, + owner: "worker".into(), + generation: 0, + }; + let lease = request(&pool).await; + let wrong_worker = EntityLease { + owner: "other-worker".into(), + ..lease.clone() + }; + let wrong_generation = EntityLease { + generation: lease.generation + 1, + ..lease.clone() + }; + let identity = user(); + for rejected in [&unapproved, &wrong_worker, &wrong_generation] { + let mut tx = pool.begin().await.unwrap(); + assert!(matches!( + insert(&mut tx, identity, rejected, None).await, + Err(DbError::AccessDenied(_)) + )); + tx.rollback().await.unwrap(); + } + // Expire deterministically instead of racing a short wall-clock timeout. + sqlx::query("UPDATE entity_deletion_requests SET lease_expires_at = clock_timestamp() - interval '1 second' WHERE id = $1") + .bind(lease.request_id).execute(&pool).await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(matches!( + insert(&mut tx, identity, &lease, None).await, + Err(DbError::AccessDenied(_)) + )); + tx.rollback().await.unwrap(); + let successor = store + .claim(lease.request_id, "successor", Duration::from_secs(300)) + .await + .unwrap() + .unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(matches!( + insert(&mut tx, identity, &lease, None).await, + Err(DbError::AccessDenied(_)) + )); + tx.rollback().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(find(&mut tx, identity).await.unwrap().is_none()); + tx.commit().await.unwrap(); + + let mut tx = pool.begin().await.unwrap(); + let saved = insert(&mut tx, identity, &successor, None).await.unwrap(); + tx.commit().await.unwrap(); + // Exercise the insertion guard independently of claim-time validation. + sqlx::query("DELETE FROM entity_deletion_approvals WHERE request_id = $1") + .bind(successor.request_id) + .execute(&pool) + .await + .unwrap(); + let mut tx = pool.begin().await.unwrap(); + assert!(matches!( + insert(&mut tx, user(), &successor, None).await, + Err(DbError::DeletionSafety(_)) + )); + tx.rollback().await.unwrap(); + let rows: Vec = sqlx::query_as( + "SELECT id, pubkey, scope_community_id, request_id, source_owner_pubkey, blocked_at FROM user_deletion_fences" + ).fetch_all(&pool).await.unwrap(); + assert_eq!(rows, vec![saved]); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn competing_inserts_wait_then_preserve_or_reject_original_provenance() { + let pool = pool().await; + let lease = request(&pool).await; + let other_lease = request(&pool).await; + for (scoped, conflict) in [(false, false), (false, true), (true, false), (true, true)] { + let owner = user().pubkey; + let identity = if scoped { + FenceIdentity::agent(CommunityId::from_uuid(Uuid::new_v4()), user().pubkey) + } else { + user() + }; + let source_owner = scoped.then_some(owner); + let mut first = pool.begin().await.unwrap(); + let saved = insert(&mut first, identity, &lease, source_owner) + .await + .unwrap(); + let next_lease = if conflict && !scoped { + other_lease.clone() + } else { + lease.clone() + }; + let next_owner = if conflict && scoped { + Some(user().pubkey) + } else { + source_owner + }; + let mut second = pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *second) + .await + .unwrap(); + let writer = tokio::spawn(async move { + let result = insert(&mut second, identity, &next_lease, next_owner).await; + if result.is_ok() { + second.commit().await.unwrap(); + } else { + second.rollback().await.unwrap(); + } + result + }); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE pid = $1 AND NOT granted)", + ) + .bind(pid) + .fetch_one(&pool) + .await + .unwrap(); + if waiting { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + first.commit().await.unwrap(); + let result = tokio::time::timeout(Duration::from_secs(5), writer) + .await + .unwrap() + .unwrap(); + if conflict { + assert!( + matches!(result, Err(DbError::DeletionSafety(ref message)) if message.contains("different provenance")) + ); + } else { + assert_eq!(result.unwrap(), saved); + } + let mut tx = pool.begin().await.unwrap(); + assert_eq!(find(&mut tx, identity).await.unwrap(), Some(saved)); + tx.commit().await.unwrap(); + } +} diff --git a/migrations/0074_user_deletion_fences.sql b/migrations/0074_user_deletion_fences.sql new file mode 100644 index 0000000000..cf0a806ac8 --- /dev/null +++ b/migrations/0074_user_deletion_fences.sql @@ -0,0 +1,23 @@ +SET LOCAL lock_timeout = '5s'; + +-- Operator-global restrictions, deliberately retained after profile and community +-- deletion. scope_community_id is the scope of an agent restriction, not ownership +-- of this row. Never clear it during community deletion: NULL means a global user. +CREATE TABLE user_deletion_fences ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + pubkey TEXT NOT NULL CHECK (pubkey ~ '^[0-9a-f]{64}$'), + scope_community_id UUID, + request_id UUID NOT NULL, + source_owner_pubkey TEXT CHECK (source_owner_pubkey ~ '^[0-9a-f]{64}$'), + blocked_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ((scope_community_id IS NULL) = (source_owner_pubkey IS NULL)) +); + +CREATE UNIQUE INDEX user_deletion_fences_global_key + ON user_deletion_fences (pubkey) WHERE scope_community_id IS NULL; +CREATE UNIQUE INDEX user_deletion_fences_scoped_key + ON user_deletion_fences (pubkey, scope_community_id) WHERE scope_community_id IS NOT NULL; + +INSERT INTO _operator_global_tables (table_name, reason) +VALUES ('user_deletion_fences', 'Operator-global user and agent restrictions retained after profile and community deletion') +ON CONFLICT (table_name) DO NOTHING; diff --git a/schema/schema.sql b/schema/schema.sql index ce60e11a20..649e761aaf 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -125,6 +125,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \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/user_deletion_fences.sql \i tables/public/community_deletion_approvals.sql \i tables/public/community_deletion_checkpoints.sql \i tables/public/community_deletion_manifest_keys.sql @@ -200,6 +201,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i seeds/entity_deletion_approvals.sql \i seeds/entity_deletion_requests.sql \i seeds/entity_deletion_stage_history.sql +\i seeds/user_deletion_fences.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/user_deletion_fences.sql b/schema/seeds/user_deletion_fences.sql new file mode 100644 index 0000000000..b8f67069eb --- /dev/null +++ b/schema/seeds/user_deletion_fences.sql @@ -0,0 +1,3 @@ +INSERT INTO _operator_global_tables (table_name, reason) +VALUES ('user_deletion_fences', 'Operator-global user and agent restrictions retained after profile and community deletion') +ON CONFLICT (table_name) DO NOTHING; diff --git a/schema/tables/public/user_deletion_fences.sql b/schema/tables/public/user_deletion_fences.sql new file mode 100644 index 0000000000..6b3a448bb6 --- /dev/null +++ b/schema/tables/public/user_deletion_fences.sql @@ -0,0 +1,17 @@ +-- Operator-global restrictions, deliberately retained after profile and community +-- deletion. scope_community_id is the scope of an agent restriction, not ownership +-- of this row. Never clear it during community deletion: NULL means a global user. +CREATE TABLE user_deletion_fences ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + pubkey TEXT NOT NULL CHECK (pubkey ~ '^[0-9a-f]{64}$'), + scope_community_id UUID, + request_id UUID NOT NULL, + source_owner_pubkey TEXT CHECK (source_owner_pubkey ~ '^[0-9a-f]{64}$'), + blocked_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK ((scope_community_id IS NULL) = (source_owner_pubkey IS NULL)) +); + +CREATE UNIQUE INDEX user_deletion_fences_global_key + ON user_deletion_fences (pubkey) WHERE scope_community_id IS NULL; +CREATE UNIQUE INDEX user_deletion_fences_scoped_key + ON user_deletion_fences (pubkey, scope_community_id) WHERE scope_community_id IS NOT NULL;