mirror of
https://github.com/block/buzz.git
synced 2026-11-06 01:11:33 +00:00
Add user deletion fence storage and locks
Signed-off-by: Ravneet Arora <rarora@squareup.com>
This commit is contained in:
@@ -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;
|
||||
|
||||
@@ -546,6 +546,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",
|
||||
@@ -772,7 +773,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);
|
||||
@@ -2902,6 +2904,7 @@ mod postgres_tests {
|
||||
"entity_deletion_approvals",
|
||||
"entity_deletion_requests",
|
||||
"entity_deletion_stage_history",
|
||||
"user_deletion_fences",
|
||||
] {
|
||||
assert_eq!(
|
||||
columns(&desired, table).await,
|
||||
@@ -2938,6 +2941,7 @@ mod postgres_tests {
|
||||
for table in [
|
||||
"entity_deletion_requests",
|
||||
"entity_deletion_stage_history",
|
||||
"user_deletion_fences",
|
||||
"entity_deletion_approvals",
|
||||
] {
|
||||
assert_eq!(
|
||||
|
||||
@@ -4400,6 +4400,101 @@ 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 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_exclusive(&mut tx, &[human, agent])
|
||||
.await
|
||||
.unwrap();
|
||||
let global = fences::insert(&mut tx, human, entity.id, None)
|
||||
.await
|
||||
.unwrap();
|
||||
let scoped = fences::insert(&mut tx, agent, entity.id, 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"))
|
||||
|
||||
@@ -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;
|
||||
|
||||
243
crates/buzz-db/src/store/user_deletion_fences.rs
Normal file
243
crates/buzz-db/src/store/user_deletion_fences.rs
Normal file
@@ -0,0 +1,243 @@
|
||||
//! 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/worker guards first, 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 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<CommunityId>,
|
||||
}
|
||||
|
||||
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<Uuid> {
|
||||
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<Uuid>,
|
||||
/// 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<String>,
|
||||
/// Time of the first insertion; retries preserve it.
|
||||
pub blocked_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// 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.
|
||||
pub async fn find(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
identity: FenceIdentity,
|
||||
) -> Result<Option<UserDeletionFence>> {
|
||||
lock_shared(tx, &[identity]).await?;
|
||||
read(tx, identity).await
|
||||
}
|
||||
|
||||
async fn read(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
identity: FenceIdentity,
|
||||
) -> Result<Option<UserDeletionFence>> {
|
||||
// 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?)
|
||||
}
|
||||
|
||||
/// Insert an idempotent restriction under its exclusive identity lock.
|
||||
///
|
||||
/// The caller must validate approval/worker ownership and verified agent
|
||||
/// ownership, acquire the complete exclusive identity set in order, and save
|
||||
/// its checkpoint in this same transaction. This helper does not authorize a
|
||||
/// deletion. A retry preserves the first row; conflicting provenance errors.
|
||||
/// On any error roll back the transaction, including other fences/checkpoints.
|
||||
pub async fn insert(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
identity: FenceIdentity,
|
||||
request_id: Uuid,
|
||||
source_owner: Option<PublicKey>,
|
||||
) -> Result<UserDeletionFence> {
|
||||
if identity.scope.is_some() != source_owner.is_some() {
|
||||
return Err(DbError::DeletionSafety(
|
||||
"agent fences require an owner; user fences do not".into(),
|
||||
));
|
||||
}
|
||||
let exists: bool = sqlx::query_scalar(
|
||||
"SELECT EXISTS (SELECT 1 FROM entity_deletion_requests WHERE id = $1 AND entity_type = 'user')",
|
||||
).bind(request_id).fetch_one(&mut **tx).await?;
|
||||
if !exists {
|
||||
return Err(DbError::NotFound("user deletion request".into()));
|
||||
}
|
||||
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,
|
||||
);
|
||||
}
|
||||
}
|
||||
275
crates/buzz-db/src/store/user_deletion_fences_postgres_tests.rs
Normal file
275
crates/buzz-db/src/store/user_deletion_fences_postgres_tests.rs
Normal file
@@ -0,0 +1,275 @@
|
||||
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) -> Uuid {
|
||||
crate::Db::from_pool(pool.clone())
|
||||
.user_deletion_store()
|
||||
.submit(
|
||||
&format!("demo:{}", Uuid::new_v4().simple()),
|
||||
"operator",
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.id
|
||||
}
|
||||
|
||||
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 request_id = 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_exclusive(&mut tx, &[a, human]).await.unwrap();
|
||||
insert(&mut tx, human, request_id, None).await.unwrap();
|
||||
insert(&mut tx, a, request_id, 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_exclusive(&mut tx, &[a, human]).await.unwrap();
|
||||
let first = insert(&mut tx, human, request_id, None).await.unwrap();
|
||||
let scoped = insert(&mut tx, a, request_id, 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, request_id, None).await.unwrap(),
|
||||
first
|
||||
);
|
||||
assert_eq!(
|
||||
insert(&mut tx, a, request_id, 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_request = request(&pool).await;
|
||||
let mut tx = pool.begin().await.unwrap();
|
||||
assert!(insert(&mut tx, human, other_request, None).await.is_err());
|
||||
tx.rollback().await.unwrap();
|
||||
let mut tx = pool.begin().await.unwrap();
|
||||
assert!(insert(&mut tx, a, request_id, 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 request_id = 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();
|
||||
assert!(insert(&mut tx, human, Uuid::new_v4(), None).await.is_err());
|
||||
assert!(insert(&mut tx, human, request_id, Some(user().pubkey))
|
||||
.await
|
||||
.is_err());
|
||||
assert!(insert(&mut tx, agent, request_id, 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 request_id = request(&pool).await;
|
||||
let human = user();
|
||||
let mut gate = pool.begin().await.unwrap();
|
||||
let saved = insert(&mut gate, human, request_id, 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<FenceIdentity>| {
|
||||
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());
|
||||
}
|
||||
25
migrations/0074_user_deletion_fences.sql
Normal file
25
migrations/0074_user_deletion_fences.sql
Normal file
@@ -0,0 +1,25 @@
|
||||
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;
|
||||
CREATE INDEX user_deletion_fences_request
|
||||
ON user_deletion_fences (request_id);
|
||||
|
||||
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;
|
||||
@@ -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
|
||||
|
||||
3
schema/seeds/user_deletion_fences.sql
Normal file
3
schema/seeds/user_deletion_fences.sql
Normal file
@@ -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;
|
||||
19
schema/tables/public/user_deletion_fences.sql
Normal file
19
schema/tables/public/user_deletion_fences.sql
Normal file
@@ -0,0 +1,19 @@
|
||||
-- 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;
|
||||
CREATE INDEX user_deletion_fences_request
|
||||
ON user_deletion_fences (request_id);
|
||||
Reference in New Issue
Block a user