diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 5b43d443642..81e1d5ca242 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -37,6 +37,14 @@ pub enum DbError { #[error("access denied: {0}")] AccessDenied(String), + /// A permanent user-deletion restriction refuses this serving write. + #[error("identity is restricted for user deletion")] + UserDeletionRestricted, + + /// Ownership changed while acquiring event-author locks; retry in a new transaction. + #[error("agent ownership changed; retry the write")] + EventAuthorOwnershipChanged, + /// Invitation admission or minting was denied by authoritative restrictions. #[error("invitation principal is restricted")] InviteRestricted, diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index 2e4ce60a758..45d65181e86 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -145,6 +145,15 @@ impl AdmittedTx { &mut self.tx } + pub(crate) async fn guard_event_author(&mut self, event: &nostr::Event) -> Result<()> { + crate::store::user_deletion_write_guard::guard_event_author( + &mut self.tx, + self.community, + event, + ) + .await + } + /// Record that an event committed with this transaction belongs to /// `channel_id`, so [`AdmittedTx::commit`] refreshes the channel's TTL. pub(crate) fn record_channel_event(&mut self, channel_id: Option, kind: i32) { diff --git a/crates/buzz-db/src/store/archived_identities.rs b/crates/buzz-db/src/store/archived_identities.rs index 986985bc464..c1c337913ab 100644 --- a/crates/buzz-db/src/store/archived_identities.rs +++ b/crates/buzz-db/src/store/archived_identities.rs @@ -5,13 +5,14 @@ //! relay access, or repository permissions. //! All pubkey and event ID values are lowercase hex strings. -use buzz_core::CommunityId; +use buzz_core::{CommunityId, StoredEvent}; use buzz_datastore_tracing::datastore_span; use chrono::{DateTime, Utc}; +use nostr::Event; use sqlx::{PgPool, Row as _}; use crate::error::Result; -use crate::Db; +use crate::{AdmittedTx, Db}; /// A single archived identity record. #[derive(Debug, Clone)] @@ -70,6 +71,31 @@ pub async fn archive( crate::observability::WriterOperation::EventWrite, ) .await?; + let changed = archive_on( + &mut tx, + pubkey, + consent_path, + actor, + reason, + replaced_by, + request_event_id, + ) + .await?; + tx.commit().await?; + Ok(changed) +} + +#[allow(clippy::too_many_arguments)] +async fn archive_on( + tx: &mut AdmittedTx, + pubkey: &str, + consent_path: &str, + actor: &str, + reason: Option<&str>, + replaced_by: Option<&str>, + request_event_id: &str, +) -> Result { + let community_id = tx.community(); let result = sqlx::query( "INSERT INTO archived_identities \ (community_id, pubkey, consent_path, actor, reason, replaced_by, request_event_id) \ @@ -85,7 +111,6 @@ pub async fn archive( .bind(request_event_id) .execute(tx.conn()) .await?; - tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -100,13 +125,19 @@ pub async fn unarchive(pool: &PgPool, community_id: CommunityId, pubkey: &str) - crate::observability::WriterOperation::EventWrite, ) .await?; + let changed = unarchive_on(&mut tx, pubkey).await?; + tx.commit().await?; + Ok(changed) +} + +async fn unarchive_on(tx: &mut AdmittedTx, pubkey: &str) -> Result { + let community_id = tx.community(); let result = sqlx::query("DELETE FROM archived_identities WHERE community_id = $1 AND pubkey = $2") .bind(community_id.as_uuid()) .bind(pubkey) .execute(tx.conn()) .await?; - tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -148,7 +179,68 @@ fn row_to_archived_identity( }) } +/// Validated archive mutation accompanying a signed identity request. +pub struct IdentityArchiveChange<'a> { + /// Identity whose archive state changes. + pub target: &'a str, + /// Whether to archive (`true`) or unarchive (`false`). + pub archive: bool, + /// Validated consent path: self, owner, or admin. + pub consent_path: &'a str, + /// Optional reason supplied by the actor. + pub reason: Option<&'a str>, + /// Optional replacement identity. + pub replaced_by: Option<&'a str>, +} + impl Db { + /// Atomically persist the signed request, archive mutation, and relay delta. + /// + /// The request passes the shared author fence before any archive mutation. + /// Duplicate requests never reapply a mutation; the persisted delta is + /// returned for post-commit delivery when newly inserted. Stored deltas + /// remain available for replay; duplicate requests emit no new delta. + pub async fn apply_identity_archive_event( + &self, + community: CommunityId, + request: &Event, + change: IdentityArchiveChange<'_>, + delta: &Event, + ) -> Result<(StoredEvent, bool, Option)> { + let mut tx = self.begin_event_write_transaction(community).await?; + let (stored, inserted) = + crate::event::insert_event_in_transaction(&mut tx, request, None).await?; + let notification = if inserted { + let changed = if change.archive { + archive_on( + &mut tx, + change.target, + change.consent_path, + &request.pubkey.to_hex(), + change.reason, + change.replaced_by, + &request.id.to_hex(), + ) + .await? + } else { + unarchive_on(&mut tx, change.target).await? + }; + if changed { + Some( + crate::event::insert_event_in_transaction(&mut tx, delta, None) + .await? + .0, + ) + } else { + None + } + } else { + None + }; + tx.commit().await?; + Ok((stored, inserted, notification)) + } + /// Returns `true` if `pubkey` (64-char hex) is archived in `community_id`. #[datastore_span(name = "is_archived", system = "postgresql")] pub async fn is_archived(&self, community_id: CommunityId, pubkey: &str) -> Result { diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index 2f8a2e5f7a7..1786c25913c 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -699,6 +699,186 @@ async fn head_outside_authorized_channel_matches_missing_head() { )); } +/// Exercise the production Move transaction, including its second, relay-signed +/// insert after domain row writes. Exclusive holders do not touch domain rows. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn artifact_move_and_identity_fences_finish_without_partial_writes_or_deadlocks() { + use crate::user_deletion_fences::{self as fences, FenceIdentity}; + use std::{sync::Arc, time::Duration}; + for block_relay in [false, true] { + let f = Arc::new(Fixture::new().await); + let d = Uuid::new_v4(); + let create = f.revision(d, "create", f.a, None, &f.owner, vec![]); + f.accept(&create, None).await; + let moved = f.revision(d, "move", f.b, Some(&create), &f.owner, vec![]); + let move_id = moved.id; + let store = f.db.entity_deletion_store(); + let request = + f.db.user_deletion_store() + .submit(&format!("demo:{}", Uuid::new_v4()), "operator", None) + .await + .unwrap(); + store + .approve( + request.id, + "reviewer", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = store + .claim(request.id, "worker", Duration::from_secs(300)) + .await + .unwrap() + .unwrap(); + let identity = FenceIdentity::user(f.owner.public_key()); + let mut gate = f.db.pool.begin().await.unwrap(); + if block_relay { + fences::lock_exclusive(&mut gate, &[FenceIdentity::user(f.relay.public_key())]) + .await + .unwrap(); + } else { + fences::insert(&mut gate, identity, &lease, None) + .await + .unwrap(); + } + let gate_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *gate) + .await + .unwrap(); + let task_f = f.clone(); + let task = tokio::spawn(async move { + let env = artifact::validate(&moved).unwrap(); + task_f + .db + .accept_artifact( + task_f.community, + &moved, + &env, + Some(task_f.a), + &task_f.relay, + ) + .await + }); + wait_for_artifact_blocker(&f.db.pool, gate_pid).await; + let writer_pid: i32 = sqlx::query_scalar("SELECT pid FROM pg_stat_activity WHERE datname=current_database() AND $1=ANY(pg_blocking_pids(pid)) LIMIT 1") + .bind(gate_pid).fetch_one(&f.db.pool).await.unwrap(); + let coordinate_held: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_locks WHERE pid=$1 AND locktype='advisory' AND granted AND mode='ExclusiveLock' + AND classid=((hashtextextended($2,0) >> 32) & 4294967295)::oid AND objid=(hashtextextended($2,0) & 4294967295)::oid)") + .bind(writer_pid).bind(format!("artifact:{}:{}", f.community.as_uuid(), d)).fetch_one(&f.db.pool).await.unwrap(); + assert!( + coordinate_held, + "actual artifact caller must hold its domain lock before waiting" + ); + let pending_user_fence = if block_relay { + let mut user_gate = f.db.pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *user_gate) + .await + .unwrap(); + let fence_task = tokio::spawn(async move { + fences::insert(&mut user_gate, identity, &lease, None) + .await + .unwrap(); + user_gate.commit().await.unwrap(); + }); + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let blockers: Vec = sqlx::query_scalar("SELECT pg_blocking_pids($1)") + .bind(pid) + .fetch_one(&f.db.pool) + .await + .unwrap(); + if blockers.contains(&writer_pid) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("fence must wait on Move's already acquired author lock"); + Some(fence_task) + } else { + None + }; + gate.commit().await.unwrap(); + let outcome = tokio::time::timeout(Duration::from_secs(8), task) + .await + .unwrap() + .unwrap(); + if let Some(task) = pending_user_fence { + tokio::time::timeout(Duration::from_secs(8), task) + .await + .unwrap() + .unwrap(); + } + if block_relay { + let ArtifactOutcome::Accepted(events) = outcome.unwrap() else { + panic!("Move should finish before queued author fence"); + }; + assert_eq!(events.len(), 2); + assert_eq!(events[1].event.kind.as_u16(), 45011); + } else { + assert!(matches!( + outcome, + Err(crate::DbError::UserDeletionRestricted) + )); + } + let head: Vec = sqlx::query_scalar( + "SELECT event_id FROM artifact_heads WHERE community_id=$1 AND artifact_id=$2", + ) + .bind(f.community.as_uuid()) + .bind(d) + .fetch_one(&f.db.pool) + .await + .unwrap(); + assert_eq!( + head, + if block_relay { + move_id.as_bytes() + } else { + create.id.as_bytes() + } + ); + let revision_exists: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM artifact_revisions WHERE community_id=$1 AND event_id=$2)", + ) + .bind(f.community.as_uuid()) + .bind(move_id.as_bytes().as_slice()) + .fetch_one(&f.db.pool) + .await + .unwrap(); + assert_eq!(revision_exists, block_relay); + let removals: i64 = + sqlx::query_scalar("SELECT count(*) FROM events WHERE community_id=$1 AND kind=45011") + .bind(f.community.as_uuid()) + .fetch_one(&f.db.pool) + .await + .unwrap(); + assert_eq!(removals, i64::from(block_relay)); + let deadlocks: i64 = sqlx::query_scalar( + "SELECT deadlocks FROM pg_stat_database WHERE datname=current_database()", + ) + .fetch_one(&f.db.pool) + .await + .unwrap(); + assert_eq!(deadlocks, 0); + } +} + +async fn wait_for_artifact_blocker(pool: &PgPool, pid: i32) { + tokio::time::timeout(std::time::Duration::from_secs(3), async { + loop { + let waiting: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND $1=ANY(pg_blocking_pids(pid)))") + .bind(pid).fetch_one(pool).await.unwrap(); + if waiting { return; } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }).await.expect("artifact writer must wait on identity lock"); +} + /// Sync's complete-set read: keyset pages by `d` share one snapshot, so a /// change committed after it opens is in no page. Mutation: drop the /// `REPEATABLE READ` statement → the late create appears in a later page → diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index c85fb6fd017..604f439c310 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -477,6 +477,7 @@ async fn insert_event_on( return Err(DbError::EphemeralEventRejected(kind_u16)); } + tx.guard_event_author(event).await?; let id_bytes = event.id.as_bytes(); let pubkey_bytes = event.pubkey.to_bytes(); let sig_bytes = event.sig.serialize(); @@ -1880,6 +1881,7 @@ pub(crate) async fn insert_event_with_thread_metadata_tx( return Err(DbError::EphemeralEventRejected(kind_u16)); } + tx.guard_event_author(event).await?; let id_bytes = event.id.as_bytes(); let pubkey_bytes = event.pubkey.to_bytes(); let sig_bytes = event.sig.serialize(); diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index d5bbde43472..570efad620f 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -30,6 +30,10 @@ pub mod nip_fi_binding; pub mod user_deletion; /// Operator-global user restrictions and identity-lock primitives. pub mod user_deletion_fences; +pub(crate) mod user_deletion_write_guard; + +#[cfg(test)] +mod user_deletion_write_guard_postgres_tests; #[cfg(test)] mod event_follow_up_postgres_tests; @@ -98,3 +102,8 @@ mod unlisted_channels_postgres_tests; #[cfg(test)] mod admission_postgres_tests; + +#[cfg(test)] +mod user_deletion_lock_order_postgres_tests; +#[cfg(test)] +mod user_deletion_test_support; diff --git a/crates/buzz-db/src/store/user_deletion_fences.rs b/crates/buzz-db/src/store/user_deletion_fences.rs index 2d437fff094..04521be01d6 100644 --- a/crates/buzz-db/src/store/user_deletion_fences.rs +++ b/crates/buzz-db/src/store/user_deletion_fences.rs @@ -1,13 +1,26 @@ -//! Operator-global restriction storage and identity locks; no serving or worker -//! path calls these helpers yet. Installing fences does not itself enforce them. +//! Operator-global restriction storage and identity locks. Shared event insertion +//! paths consult these restrictions; the real user-deletion worker is not integrated. //! -//! 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. +//! Callers use writer-pool READ COMMITTED transactions. Community admission +//! precedes identity locking. Each lock call sorts global keys before scoped +//! keys, but this is NOT a transaction-wide order for serving writes: artifact +//! and channel writers can already hold domain row/advisory locks, and a later +//! relay-signed event can acquire another global identity lock in the same tx. +//! +//! Exclusive fencing transactions must lock the request first, acquire their +//! complete global identity set before their complete scoped set, then persist +//! only restriction/request/checkpoint state. They must not lock or mutate +//! users, channels, artifacts, events, or other serving-domain rows while +//! holding exclusive identity locks. Perform cleanup in later transactions. +//! Otherwise a writer holding a domain lock and waiting for a shared identity +//! lock can cycle with the fencing transaction. Ownership discovery under the +//! fence uses ordinary reads, not FOR UPDATE or owner mutation. +//! +//! Never upgrade shared identity locks to exclusive. Before enabling a worker, +//! audit its complete exclusive sets against multi-author event transactions; +//! per-call sorting alone does not prove arbitrary multi-key batches deadlock +//! free. Keep locks through the protected write/checkpoint; commit or rollback +//! releases them. The real deletion worker remains disabled. use buzz_core::CommunityId; use chrono::{DateTime, Utc}; diff --git a/crates/buzz-db/src/store/user_deletion_lock_order_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion_lock_order_postgres_tests.rs new file mode 100644 index 00000000000..462bac5df6f --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion_lock_order_postgres_tests.rs @@ -0,0 +1,132 @@ +use super::user_deletion_fences::{self as fences, FenceIdentity}; +use super::user_deletion_test_support::{message, restricted, wait_for_waiter, Fixture}; +use crate::{event, Db, DbError}; +use nostr::Keys; +use std::time::Duration; +use uuid::Uuid; + +/// Exercise the actual TTL-before-author settings writer against a message +/// holding the author's shared lock and a queued exclusive fence. PostgreSQL +/// may reorder soft wait edges; neither an error nor a swallowed TTL deadlock +/// is acceptable evidence of safe completion. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn message_then_fence_then_ttl_settings_complete_without_deadlock() { + let f = Fixture::new().await; + let user = Keys::generate(); + let channel = Uuid::new_v4(); + f.db.create_channel_with_id( + f.community, + channel, + "lock-order", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Open, + None, + user.public_key().as_bytes(), + Some(60), + ) + .await + .unwrap(); + let mut message_tx = crate::begin_community_event_write_transaction( + &f.pool, + f.community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .unwrap(); + let first = message(&user, 9); + event::insert_event_in_transaction(&mut message_tx, &first, Some(channel)) + .await + .unwrap(); + + let mut gate = f.pool.begin().await.unwrap(); + let gate_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *gate) + .await + .unwrap(); + let identity = FenceIdentity::user(user.public_key()); + let lease = f.lease.clone(); + let fence_task = tokio::spawn(async move { + fences::insert(&mut gate, identity, &lease, None).await?; + gate.commit().await?; + crate::Result::Ok(()) + }); + wait_for_waiter(&f.pool, gate_pid).await; + + // A one-connection pool lets us observe the settings writer's actual PID. + let settings_pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let settings_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&settings_pool) + .await + .unwrap(); + let settings_db = Db::from_pool(settings_pool); + let settings_event = message(&user, 9002); + let settings_id = settings_event.id; + let community = f.community; + let settings_task = tokio::spawn(async move { + let mut edit = crate::channel::ChannelMetadataEdit::default(); + edit.columns.ttl_seconds = Some(Some(3600)); + settings_db + .apply_channel_metadata_event(community, channel, &settings_event, edit, &[]) + .await + }); + wait_for_waiter(&f.pool, settings_pid).await; + // The settings transaction really holds the exclusive TTL lock, rather + // than merely waiting on some earlier lock in its path. + let holds_ttl: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE pid=$1 AND locktype='advisory' \ + AND mode='ExclusiveLock' AND granted \ + AND classid=((hashtextextended($2,0) >> 32) & 4294967295)::oid \ + AND objid=(hashtextextended($2,0) & 4294967295)::oid)", + ) + .bind(settings_pid) + .bind(crate::channel::channel_ttl_lock_key(f.community, channel)) + .fetch_one(&f.pool) + .await + .unwrap(); + assert!(holds_ttl); + + tokio::time::timeout(Duration::from_secs(8), async { + message_tx.commit().await.unwrap(); + fence_task.await.unwrap().unwrap(); + let settings = settings_task.await.unwrap(); + let accepted = match settings { + Ok(write) => { + assert!(write.inserted); + true + } + Err(DbError::UserDeletionRestricted) => false, + Err(other) => panic!("settings failed for a non-fence reason: {other:?}"), + }; + assert_eq!( + f.db.get_event_by_id(f.community, settings_id.as_bytes()) + .await + .unwrap() + .is_some(), + accepted + ); + }) + .await + .expect("all three production transactions must finish"); + assert!(f + .db + .get_event_by_id(f.community, first.id.as_bytes()) + .await + .unwrap() + .is_some()); + restricted( + f.db.insert_event(f.community, &message(&user, 9), Some(channel)) + .await, + ); + let deadlocks: i64 = sqlx::query_scalar( + "SELECT deadlocks FROM pg_stat_database WHERE datname=current_database()", + ) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!(deadlocks, 0, "include deadlocks swallowed by TTL refresh"); +} diff --git a/crates/buzz-db/src/store/user_deletion_test_support.rs b/crates/buzz-db/src/store/user_deletion_test_support.rs new file mode 100644 index 00000000000..624081f2b68 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion_test_support.rs @@ -0,0 +1,119 @@ +use super::user_deletion_fences::{self as fences, FenceIdentity}; +use crate::{Db, DbError}; +use buzz_core::CommunityId; +use nostr::{Event, EventBuilder, Keys, Kind}; +use sqlx::{PgPool, Postgres, Transaction}; +use std::time::Duration; +use uuid::Uuid; + +pub(super) struct Fixture { + pub(super) pool: PgPool, + pub(super) db: Db, + pub(super) community: CommunityId, + pub(super) lease: crate::entity_deletion::EntityLease, +} + +impl Fixture { + pub(super) async fn new() -> Self { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool.clone()); + let community = db + .ensure_configured_community(&format!("message-fences-{}.test", Uuid::new_v4())) + .await + .unwrap() + .id; + let request = db + .user_deletion_store() + .submit(&format!("demo:{}", Uuid::new_v4()), "operator", None) + .await + .unwrap(); + let store = db.entity_deletion_store(); + store + .approve( + request.id, + "reviewer", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = store + .claim(request.id, "worker", Duration::from_secs(300)) + .await + .unwrap() + .unwrap(); + Self { + pool, + db, + community, + lease, + } + } + + pub(super) async fn fence(&self, keys: &Keys, owner: Option<&Keys>) { + let mut tx = self.pool.begin().await.unwrap(); + self.fence_in(&mut tx, keys, owner).await; + tx.commit().await.unwrap(); + } + + pub(super) async fn fence_in( + &self, + tx: &mut Transaction<'_, Postgres>, + keys: &Keys, + owner: Option<&Keys>, + ) { + let identity = if owner.is_some() { + FenceIdentity::agent(self.community, keys.public_key()) + } else { + FenceIdentity::user(keys.public_key()) + }; + fences::insert(tx, identity, &self.lease, owner.map(Keys::public_key)) + .await + .unwrap(); + } + + pub(super) async fn agent(&self, owner: &Keys) -> Keys { + self.db + .ensure_user(self.community, owner.public_key().as_bytes()) + .await + .unwrap(); + let agent = Keys::generate(); + self.db + .ensure_user(self.community, agent.public_key().as_bytes()) + .await + .unwrap(); + self.db + .set_agent_owner( + self.community, + agent.public_key().as_bytes(), + owner.public_key().as_bytes(), + ) + .await + .unwrap(); + agent + } +} + +pub(super) fn message(keys: &Keys, kind: u16) -> Event { + EventBuilder::new(Kind::Custom(kind), Uuid::new_v4().to_string()) + .sign_with_keys(keys) + .unwrap() +} + +pub(super) fn restricted(result: crate::Result<(buzz_core::StoredEvent, bool)>) { + assert!( + matches!(result, Err(DbError::UserDeletionRestricted)), + "{result:?}" + ); +} + +pub(super) async fn wait_for_waiter(pool: &PgPool, pid: i32) { + tokio::time::timeout(Duration::from_secs(4), 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::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("production writer must wait for the identity lock"); +} diff --git a/crates/buzz-db/src/store/user_deletion_write_guard.rs b/crates/buzz-db/src/store/user_deletion_write_guard.rs new file mode 100644 index 00000000000..af08284859d --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion_write_guard.rs @@ -0,0 +1,76 @@ +//! Guard shared persistent-event insertion paths. +//! Other writes are not fenced here; real deletion must remain disabled until +//! every serving path and external-work drain has been integrated. + +use buzz_core::CommunityId; +use nostr::{Event, PublicKey}; +use sqlx::{Postgres, Transaction}; + +use super::user_deletion_fences::{lock_shared, FenceIdentity}; +use crate::{DbError, Result}; + +pub(crate) async fn guard_event_author( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + event: &Event, +) -> Result<()> { + let signer = event.pubkey; + let owner = owner(tx, community, signer).await?; + let mut identities = vec![ + FenceIdentity::user(signer), + FenceIdentity::agent(community, signer), + ]; + if let Some(owner) = owner { + identities.push(FenceIdentity::user(owner)); + } + // Never derive the owner from a cached session or an event tag. + lock_shared(tx, &identities).await?; + if owner != self::owner(tx, community, signer).await? { + // Restart in a new transaction: adding another owner lock now could + // invert the global-before-scoped ordering. The caller must roll back. + return Err(DbError::EventAuthorOwnershipChanged); + } + check_signer(tx, community, signer).await +} + +async fn owner( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + signer: PublicKey, +) -> Result> { + let owner: Option>> = sqlx::query_scalar( + "SELECT agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2", + ) + .bind(community.as_uuid()) + .bind(signer.to_bytes().as_slice()) + .fetch_optional(&mut **tx) + .await?; + owner.flatten().map(|key| parse_key(&key)).transpose() +} + +fn parse_key(key: &[u8]) -> Result { + PublicKey::from_slice(key) + .map_err(|_| DbError::InvalidData("invalid user-deletion identity key".into())) +} + +async fn check_signer( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + signer: PublicKey, +) -> Result<()> { + // Read the signer's restrictions, not the owner's. Check both possible + // scopes even if the profile/ownership row is absent: those mutable rows + // must not determine whether a permanent restriction still applies. + let restricted: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM user_deletion_fences + WHERE pubkey = $1 AND (scope_community_id IS NULL OR scope_community_id = $2))", + ) + .bind(signer.to_hex()) + .bind(community.as_uuid()) + .fetch_one(&mut **tx) + .await?; + if restricted { + return Err(DbError::UserDeletionRestricted); + } + Ok(()) +} diff --git a/crates/buzz-db/src/store/user_deletion_write_guard_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion_write_guard_postgres_tests.rs new file mode 100644 index 00000000000..7597fe5da75 --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion_write_guard_postgres_tests.rs @@ -0,0 +1,264 @@ +use super::user_deletion_fences::{self as fences, FenceIdentity}; +use super::user_deletion_test_support::{message, restricted, wait_for_waiter, Fixture}; +use crate::{event, DbError}; +use nostr::{Event, Keys}; +use std::time::Duration; +use uuid::Uuid; +async fn writer( + f: &Fixture, + event: Event, +) -> ( + i32, + tokio::task::JoinHandle>, +) { + let mut tx = crate::begin_community_event_write_transaction( + &f.pool, + f.community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .unwrap(); + let pid = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(tx.conn()) + .await + .unwrap(); + let task = tokio::spawn(async move { + let result = event::insert_event_in_transaction(&mut tx, &event, None).await; + if result.is_ok() { + tx.commit().await.unwrap(); + } else { + tx.rollback().await.unwrap(); + } + result + }); + (pid, task) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn message_writers_enforce_global_and_scoped_fences_even_without_profiles() { + let f = Fixture::new().await; + let user = Keys::generate(); + let agent = f.agent(&user).await; + let other = Keys::generate(); + let other_community = + f.db.ensure_configured_community(&format!("unrestricted-{}.test", Uuid::new_v4())) + .await + .unwrap() + .id; + for keys in [&user, &agent, &other] { + f.db.insert_event_with_thread_metadata(f.community, &message(keys, 9), None, None) + .await + .unwrap(); + } + f.fence(&user, None).await; + // The stored owner's restriction is not the agent's restriction. + f.db.insert_event(f.community, &message(&agent, 9), None) + .await + .unwrap(); + f.fence(&agent, Some(&user)).await; + for kind in [ + 9, 40002, 45001, 45003, 7, 40003, 45002, 9007, 9000, 9002, 40100, 5, 45010, 1234, + ] { + for keys in [&user, &agent] { + restricted( + f.db.insert_event(f.community, &message(keys, kind), None) + .await, + ); + restricted( + f.db.insert_event_with_thread_metadata( + f.community, + &message(keys, kind), + None, + None, + ) + .await, + ); + } + f.db.insert_event(f.community, &message(&other, kind), None) + .await + .unwrap(); + f.db.insert_event(other_community, &message(&agent, kind), None) + .await + .unwrap(); + restricted( + f.db.insert_event(other_community, &message(&user, kind), None) + .await, + ); + } + sqlx::query("DELETE FROM users WHERE community_id = $1 AND pubkey = $2") + .bind(f.community.as_uuid()) + .bind(agent.public_key().to_bytes().as_slice()) + .execute(&f.pool) + .await + .unwrap(); + restricted( + f.db.insert_event(f.community, &message(&agent, 9), None) + .await, + ); + f.db.ensure_user(f.community, agent.public_key().as_bytes()) + .await + .unwrap(); + restricted( + f.db.insert_event_with_thread_metadata(f.community, &message(&agent, 9), None, None) + .await, + ); + // Both scopes may match the same signer. The existence lookup must still + // refuse, and the global restriction must also cover another community. + f.fence(&agent, None).await; + for community in [f.community, other_community] { + restricted( + f.db.insert_event(community, &message(&agent, 9), None) + .await, + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn already_started_message_commits_before_fence_then_new_message_is_rejected() { + let f = Fixture::new().await; + let user = Keys::generate(); + let event = message(&user, 9); + let mut tx = crate::begin_community_event_write_transaction( + &f.pool, + f.community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .unwrap(); + event::insert_event_in_transaction(&mut tx, &event, None) + .await + .unwrap(); + let mut gate = f.pool.begin().await.unwrap(); + let pid = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *gate) + .await + .unwrap(); + let lease = f.lease.clone(); + let identity = FenceIdentity::user(user.public_key()); + let gating = tokio::spawn(async move { + fences::insert(&mut gate, identity, &lease, None) + .await + .unwrap(); + gate.commit().await.unwrap(); + }); + wait_for_waiter(&f.pool, pid).await; + assert!(f + .db + .get_event_by_id(f.community, event.id.as_bytes()) + .await + .unwrap() + .is_none()); + tx.commit().await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), gating) + .await + .unwrap() + .unwrap(); + assert!(f + .db + .get_event_by_id(f.community, event.id.as_bytes()) + .await + .unwrap() + .is_some()); + restricted( + f.db.insert_event(f.community, &message(&user, 9), None) + .await, + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn message_waits_for_human_or_agent_owner_fence_and_reads_after_commit() { + let f = Fixture::new().await; + for is_agent in [false, true] { + let user = Keys::generate(); + let sender = if is_agent { + f.agent(&user).await + } else { + user.clone() + }; + let event = message(&sender, 9); + let mut gate = f.pool.begin().await.unwrap(); + fences::lock_request(&mut gate, &f.lease).await.unwrap(); + fences::lock_exclusive(&mut gate, &[FenceIdentity::user(user.public_key())]) + .await + .unwrap(); + let (pid, sending) = writer(&f, event.clone()).await; + wait_for_waiter(&f.pool, pid).await; + f.fence_in(&mut gate, &user, None).await; + if is_agent { + f.fence_in(&mut gate, &sender, Some(&user)).await; + } + gate.commit().await.unwrap(); + restricted( + tokio::time::timeout(Duration::from_secs(5), sending) + .await + .unwrap() + .unwrap(), + ); + assert!(f + .db + .get_event_by_id(f.community, event.id.as_bytes()) + .await + .unwrap() + .is_none()); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn ownership_change_while_waiting_rejects_and_retry_uses_current_owner() { + let f = Fixture::new().await; + let agent = Keys::generate(); + let owner = Keys::generate(); + f.db.ensure_user(f.community, owner.public_key().as_bytes()) + .await + .unwrap(); + f.db.ensure_user(f.community, agent.public_key().as_bytes()) + .await + .unwrap(); + let mut assignment = f.pool.begin().await.unwrap(); + fences::lock_shared( + &mut assignment, + &[ + FenceIdentity::user(agent.public_key()), + FenceIdentity::user(owner.public_key()), + ], + ) + .await + .unwrap(); + fences::lock_exclusive( + &mut assignment, + &[FenceIdentity::agent(f.community, agent.public_key())], + ) + .await + .unwrap(); + let event = message(&agent, 9); + let (pid, sending) = writer(&f, event.clone()).await; + wait_for_waiter(&f.pool, pid).await; + sqlx::query("UPDATE users SET agent_owner_pubkey = $1 WHERE community_id = $2 AND pubkey = $3") + .bind(owner.public_key().to_bytes().as_slice()) + .bind(f.community.as_uuid()) + .bind(agent.public_key().to_bytes().as_slice()) + .execute(&mut *assignment) + .await + .unwrap(); + assignment.commit().await.unwrap(); + assert!(matches!( + sending.await.unwrap(), + Err(DbError::EventAuthorOwnershipChanged) + )); + assert!(f + .db + .get_event_by_id(f.community, event.id.as_bytes()) + .await + .unwrap() + .is_none()); + f.db.insert_event(f.community, &event, None).await.unwrap(); + f.fence(&agent, Some(&owner)).await; + restricted( + f.db.insert_event(f.community, &message(&agent, 9), None) + .await, + ); +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 10c88e3697c..e292e6fafde 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -9636,6 +9636,8 @@ pub(crate) mod postgres_tests { sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "CREATE SCHEMA {schema}; \ CREATE TABLE {schema}.users (LIKE public.users INCLUDING ALL); \ + CREATE TABLE {schema}.user_deletion_fences \ + (LIKE public.user_deletion_fences INCLUDING ALL); \ CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL); \ CREATE TABLE {schema}.relay_invites (LIKE public.relay_invites INCLUDING ALL); \ CREATE TABLE {schema}.relay_members (LIKE public.relay_members INCLUDING ALL); \ @@ -9783,6 +9785,8 @@ pub(crate) mod postgres_tests { .expect("removal commits"); } } + // Break only the revocation lookup. The membership writes above + // now also read users while checking their event author's fence. sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "ALTER TABLE {schema}.users RENAME TO users_unavailable" ))) diff --git a/crates/buzz-relay/src/handlers/artifact.rs b/crates/buzz-relay/src/handlers/artifact.rs index c52e71d6d4d..38d9f0e48dc 100644 --- a/crates/buzz-relay/src/handlers/artifact.rs +++ b/crates/buzz-relay/src/handlers/artifact.rs @@ -46,7 +46,10 @@ pub(crate) async fn accept( &state.relay_keypair, ) .await - .map_err(|e| IngestError::Internal(format!("artifact storage: {e}")))?; + .map_err(|e| { + super::ingest::event_write_error(&e) + .unwrap_or_else(|| IngestError::Internal(format!("artifact storage: {e}"))) + })?; match outcome { ArtifactOutcome::Accepted(events) => { for stored in &events { diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2c5a8ff5b5c..19baf278bab 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -184,7 +184,10 @@ async fn persist_command_event( let (_, was_inserted) = buzz_db::event::insert_event_in_transaction(&mut tx, event, channel_id) .await - .map_err(|e| IngestError::Internal(format!("error: insert event: {e}")))?; + .map_err(|e| { + super::ingest::event_write_error(&e) + .unwrap_or_else(|| IngestError::Internal(format!("error: insert event: {e}"))) + })?; if was_inserted { Ok(PersistResult::Inserted(tx)) } else { diff --git a/crates/buzz-relay/src/handlers/identity_archive.rs b/crates/buzz-relay/src/handlers/identity_archive.rs index 919c1467b69..e87b4b77d1c 100644 --- a/crates/buzz-relay/src/handlers/identity_archive.rs +++ b/crates/buzz-relay/src/handlers/identity_archive.rs @@ -1,22 +1,21 @@ //! NIP-IA identity archive request handler (kinds 9035–9036). //! -//! These events are processed before storage: the request mutates the -//! `archived_identities` table and may emit relay-signed NIP-IA deltas and a -//! snapshot, then the ingest pipeline stores the request itself for audit. +//! Validate consent before storage. Persist the request, archive mutation and +//! relay delta atomically; publish only after commit. use std::sync::Arc; use nostr::{Event, PublicKey}; -use tracing::{info, warn}; -use buzz_core::kind::{KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVE_REQUEST, KIND_PROFILE}; +use buzz_core::kind::{ + KIND_IA_ARCHIVED, KIND_IA_ARCHIVE_REQUEST, KIND_IA_UNARCHIVED, KIND_IA_UNARCHIVE_REQUEST, + KIND_PROFILE, +}; use buzz_core::tenant::TenantContext; -use buzz_core::CommunityId; +use buzz_core::{CommunityId, StoredEvent}; use buzz_db::EventQuery; -use crate::handlers::side_effects::{ - publish_nipia_archival_list, publish_nipia_archived, publish_nipia_unarchived, -}; +use crate::handlers::side_effects::{build_nipia_delta, publish_nipia_archival_list}; use crate::state::AppState; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -36,12 +35,20 @@ impl ConsentPath { } } -/// Validate and execute a NIP-IA archive/unarchive request. -pub async fn handle_identity_archive_event( +/// Validated archive parameters; no persistence occurs during validation. +pub(super) struct IdentityArchivePlan { + target: String, + consent: ConsentPath, + reason: Option, + replaced_by: Option, +} + +/// Validate a NIP-IA archive/unarchive request without changing state. +pub(super) async fn validate_identity_archive_event( tenant: &TenantContext, state: &Arc, event: &Event, -) -> Result<(), String> { +) -> Result { let kind = event.kind.as_u16() as u32; let actor_hex = event.pubkey.to_hex(); @@ -64,77 +71,83 @@ pub async fn handle_identity_archive_event( let reason = extract_tag_value(event, "reason"); let consent_path = determine_consent_path(tenant.community(), state, event, &target_hex, &actor_hex).await?; - let request_event_id = event.id.to_hex(); + Ok(IdentityArchivePlan { + target: target_hex, + consent: consent_path, + reason, + replaced_by, + }) +} - let changed = if kind == KIND_IA_ARCHIVE_REQUEST { - state - .db - .archive( - tenant.community(), - &target_hex, - consent_path.as_str(), - &actor_hex, - reason.as_deref(), - replaced_by.as_deref(), - &request_event_id, - ) - .await - .map_err(|e| format!("database error: {e}"))? +/// Commit archive state and its audit events before publishing notifications. +pub(super) async fn persist_identity_archive_event( + tenant: &TenantContext, + state: &Arc, + event: &Event, + plan: IdentityArchivePlan, +) -> anyhow::Result<(StoredEvent, bool)> { + let archive = u32::from(event.kind.as_u16()) == KIND_IA_ARCHIVE_REQUEST; + let delta_kind = if archive { + KIND_IA_ARCHIVED } else { - state - .db - .unarchive(tenant.community(), &target_hex) - .await - .map_err(|e| format!("database error: {e}"))? + KIND_IA_UNARCHIVED }; - - info!( - actor = %actor_hex, - target = %target_hex, - consent = consent_path.as_str(), - changed, - kind, - "identity archive request processed" - ); - - if !changed { - return Ok(()); - } - - let publish_delta = if kind == KIND_IA_ARCHIVE_REQUEST { - publish_nipia_archived( - tenant, - state, - &target_hex, - consent_path.as_str(), - &actor_hex, - &request_event_id, - &event.content, - reason.as_deref(), - replaced_by.as_deref(), + // Use the request timestamp for its corresponding relay notification. + let delta = build_nipia_delta( + state, + delta_kind, + &plan.target, + plan.consent.as_str(), + &event.pubkey.to_hex(), + &event.id.to_hex(), + &event.content, + event.created_at, + plan.reason.as_deref(), + plan.replaced_by.as_deref(), + )?; + let (stored, inserted, notification) = state + .db + .apply_identity_archive_event( + tenant.community(), + event, + buzz_db::archived_identities::IdentityArchiveChange { + target: &plan.target, + archive, + consent_path: plan.consent.as_str(), + reason: plan.reason.as_deref(), + replaced_by: plan.replaced_by.as_deref(), + }, + &delta, ) - .await - } else { - publish_nipia_unarchived( + .await?; + if let Some(notification) = notification { + super::event::dispatch_persistent_event( tenant, state, - &target_hex, - consent_path.as_str(), - &actor_hex, - &request_event_id, - &event.content, - reason.as_deref(), + ¬ification, + delta_kind, + &state.relay_keypair.public_key().to_hex(), + None, ) - .await - }; - - if let Err(e) = publish_delta { - warn!(error = %e, "failed to publish NIP-IA delta"); - } - if let Err(e) = publish_nipia_archival_list(tenant, state).await { - warn!(error = %e, "failed to publish NIP-IA archival list"); + .await; } + // Also refresh on duplicate retries: a previous post-commit refresh may + // have failed. Propagate failures so callers can retry without reapplying + // the archive mutation. The stored delta remains available for replay. + publish_nipia_archival_list(tenant, state).await?; + Ok((stored, inserted)) +} +#[cfg(test)] +async fn handle_identity_archive_event( + tenant: &TenantContext, + state: &Arc, + event: &Event, +) -> Result<(), String> { + let plan = validate_identity_archive_event(tenant, state, event).await?; + persist_identity_archive_event(tenant, state, event, plan) + .await + .map_err(|e| e.to_string())?; Ok(()) } diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 0e44eb8c1e5..e53c302ca0d 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -472,6 +472,19 @@ pub enum BanRefusal { Unavailable, } +/// Classify identity-fencing failures without changing unrelated write errors. +pub(crate) fn event_write_error(error: &buzz_db::DbError) -> Option { + match error { + buzz_db::DbError::UserDeletionRestricted => Some(IngestError::Rejected( + "restricted: identity is restricted for user deletion".into(), + )), + buzz_db::DbError::EventAuthorOwnershipChanged => Some(IngestError::Internal( + "error: agent ownership changed; retry the write".into(), + )), + _ => None, + } +} + impl IngestError { /// A ban refusal answered with `legacy` outside NIP-FI agent HTTP. pub(crate) fn ban(refusal: BanRefusal, legacy: IngestError) -> Self { @@ -3069,15 +3082,19 @@ async fn ingest_event_inner( None }; - // Processed here (verify consent, mutate archived_identities, emit the - // relay-signed 8002/8003 delta + 13535 snapshot), then — unlike the - // NIP-43 admin commands above — the request itself falls through to normal - // storage so the delta's `["e", request_id]` audit reference resolves. - if is_identity_archive_request_kind(kind_u32) { - crate::handlers::identity_archive::handle_identity_archive_event(tenant, state, &event) + // Consent validation has no side effects. Persistence below commits the + // request, archive mutation and relay delta in one guarded transaction. + let identity_archive = if is_identity_archive_request_kind(kind_u32) { + Some( + crate::handlers::identity_archive::validate_identity_archive_event( + tenant, state, &event, + ) .await - .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; - } + .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?, + ) + } else { + None + }; if kind_u32 == KIND_DELETION { // The target's archived channel blocks the deletion, so it must @@ -3466,8 +3483,10 @@ async fn ingest_event_inner( emoji, ) .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? - { + .map_err(|e| { + event_write_error(&e) + .unwrap_or_else(|| IngestError::Internal(format!("error: {e}"))) + })? { buzz_db::ReactionEventInsertOutcome::TargetMissing => { return Err(IngestError::Rejected(format!( "invalid: {REACTION_TARGET_NOT_FOUND}" @@ -3558,8 +3577,10 @@ async fn ingest_event_inner( .db .create_channel_with_event(tenant.community(), create.channel_id, channel, &event) .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? - { + .map_err(|e| { + event_write_error(&e) + .unwrap_or_else(|| IngestError::Internal(format!("error: {e}"))) + })? { buzz_db::channel::ChannelCreate::Created(stored) => { metrics::counter!( "buzz_channels_created_total", @@ -3584,13 +3605,31 @@ async fn ingest_event_inner( }); } } + } else if let Some(plan) = identity_archive { + crate::handlers::identity_archive::persist_identity_archive_event( + tenant, state, &event, plan, + ) + .await + .map_err(|e| { + e.downcast_ref::() + .and_then(event_write_error) + .unwrap_or_else(|| { + IngestError::Internal(format!("error: archive write failed: {e}")) + }) + })? } else if workflow_deletion { // A single commit owns public acceptance, domain mutation, and dispatch. // Failure rolls everything back; identical concurrent requests cannot // divide insertion and repair ownership between two relay workers. crate::handlers::side_effects::persist_workflow_deletion(tenant, &event, state) .await - .map_err(|e| IngestError::Internal(format!("error: workflow deletion failed: {e}")))? + .map_err(|e| { + e.downcast_ref::() + .and_then(event_write_error) + .unwrap_or_else(|| { + IngestError::Internal(format!("error: workflow deletion failed: {e}")) + }) + })? } else if let Some(plan) = put_user { // One commit stores the event and applies its membership changes. A // failure stores nothing, so the client is told and can resend it. @@ -3605,11 +3644,15 @@ async fn ingest_event_inner( thread_params, ) .await - .map_err(|e| match e.downcast_ref::() { - Some(buzz_db::DbError::AccessDenied(reason)) => { - IngestError::Rejected(format!("invalid: {reason}")) - } - _ => IngestError::Internal(format!("error: membership write failed: {e}")), + .map_err(|e| { + e.downcast_ref::() + .and_then(event_write_error) + .unwrap_or_else(|| match e.downcast_ref::() { + Some(buzz_db::DbError::AccessDenied(reason)) => { + IngestError::Rejected(format!("invalid: {reason}")) + } + _ => IngestError::Internal(format!("error: membership write failed: {e}")), + }) })?; message = ok_message; (stored, inserted) @@ -3620,11 +3663,15 @@ async fn ingest_event_inner( // that is not archived) is the client's error; anything else is ours. crate::handlers::side_effects::persist_edit_metadata(tenant, &event, state) .await - .map_err(|e| match e.downcast_ref::() { - Some(buzz_db::DbError::AccessDenied(reason)) => { - IngestError::Rejected(format!("invalid: {reason}")) - } - _ => IngestError::Internal(format!("error: channel update failed: {e}")), + .map_err(|e| { + e.downcast_ref::() + .and_then(event_write_error) + .unwrap_or_else(|| match e.downcast_ref::() { + Some(buzz_db::DbError::AccessDenied(reason)) => { + IngestError::Rejected(format!("invalid: {reason}")) + } + _ => IngestError::Internal(format!("error: channel update failed: {e}")), + }) })? } else if buzz_core::kind::is_replaceable(kind_u32) { // NIP-16 replaceable event — atomic replace with stale-write protection. @@ -3665,7 +3712,10 @@ async fn ingest_event_inner( .db .insert_canvas_head_checked(tenant.community(), &event, channel, precondition) .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))?; + .map_err(|e| { + event_write_error(&e) + .unwrap_or_else(|| IngestError::Internal(format!("error: {e}"))) + })?; match status { buzz_db::ChannelHeadWriteStatus::RevisionMissing => { return Err(IngestError::CanvasConflict( @@ -3703,7 +3753,9 @@ async fn ingest_event_inner( buzz_db::DbError::AuthEventRejected => { IngestError::Rejected("invalid: AUTH events cannot be stored".into()) } - other => IngestError::Internal(format!("error: database error: {other}")), + other => event_write_error(&other).unwrap_or_else(|| { + IngestError::Internal(format!("error: database error: {other}")) + }), }); } } diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index cb18b8525c8..3f9367b70e5 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -58,6 +58,8 @@ pub mod side_effects; mod system_channel_postgres_tests; /// Per-filter query scope for system channels. pub(crate) mod system_channels; +#[cfg(test)] +mod user_deletion_message_postgres_tests; /// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag, /// applying the server-side override when configured. @@ -87,3 +89,14 @@ pub fn resolve_ttl(event: &nostr::Event, ephemeral_ttl_override: Option) -> /// NIP-AR artifact lifecycle. pub mod artifact; + +#[cfg(test)] +mod user_deletion_archive_postgres_tests; +#[cfg(test)] +mod user_deletion_error_postgres_tests; +#[cfg(test)] +mod user_deletion_expanded_postgres_tests; +#[cfg(test)] +mod user_deletion_lifecycle_postgres_tests; +#[cfg(test)] +mod user_deletion_test_support; diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index de4ccfc673f..55cacfd0674 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -8,10 +8,10 @@ use uuid::Uuid; use buzz_core::kind::{ event_kind_u32, is_parameterized_replaceable, AUTHOR_ONLY_KINDS, KIND_AGENT_PROFILE, - KIND_DM_VISIBILITY, KIND_GIT_REPO_ANNOUNCEMENT, KIND_IA_ARCHIVED, KIND_IA_ARCHIVED_LIST, - KIND_IA_UNARCHIVED, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, - KIND_NIP29_GROUP_ADMINS, KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, - KIND_NIP43_MEMBERSHIP_LIST, KIND_REACTION, KIND_THREAD_SUMMARY, + KIND_DM_VISIBILITY, KIND_GIT_REPO_ANNOUNCEMENT, KIND_IA_ARCHIVED_LIST, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_NIP29_GROUP_ADMINS, + KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, KIND_NIP43_MEMBERSHIP_LIST, KIND_REACTION, + KIND_THREAD_SUMMARY, }; use buzz_core::StoredEvent; use buzz_db::channel::{MemberRecord, MemberRole}; @@ -4010,6 +4010,33 @@ pub async fn publish_nipia_archival_list( ..buzz_db::event::EventQuery::for_community(tenant.community()) }) .await?; + // A retry may need to repair a failed publication, but a current + // snapshot must retain its id/timestamp and emit no new notification. + // Tag order is not meaningful for the protected set of archived keys. + if previous.first().is_some_and(|snapshot| { + snapshot.event.content.is_empty() + && snapshot + .event + .tags + .iter() + .map(Tag::as_slice) + .collect::>() + == tags.iter().map(Tag::as_slice).collect() + }) { + // The canonical read can become stale before we inspect the head. + // Keep the same drift protection as the publication path below. + let current_archived = state.db.list_archived(tenant.community()).await?; + if archived + .iter() + .map(|identity| identity.pubkey.as_str()) + .eq(current_archived + .iter() + .map(|identity| identity.pubkey.as_str())) + { + return Ok(()); + } + continue; + } let created_at = previous .first() .map(|event| (event.event.created_at.as_secs() + 1).max(now)) @@ -4151,8 +4178,7 @@ pub async fn publish_dm_visibility_snapshot( } #[allow(clippy::too_many_arguments)] -async fn publish_nipia_delta( - tenant: &TenantContext, +pub(super) fn build_nipia_delta( state: &Arc, kind: u32, target_pubkey_hex: &str, @@ -4160,11 +4186,10 @@ async fn publish_nipia_delta( actor_pubkey_hex: &str, request_event_id: &str, content: &str, + created_at: nostr::Timestamp, reason: Option<&str>, replaced_by: Option<&str>, -) -> anyhow::Result<()> { - let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); - +) -> anyhow::Result { let mut tags = vec![ Tag::parse(["-"]).map_err(|e| anyhow::anyhow!("failed to build '-' tag: {e}"))?, Tag::parse(["p", target_pubkey_hex]) @@ -4190,82 +4215,12 @@ async fn publish_nipia_delta( let event = EventBuilder::new(Kind::Custom(kind as u16), content) .tags(tags) + .allow_self_tagging() + .custom_created_at(created_at) .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign kind:{kind}: {e}"))?; - let (stored, was_inserted) = state - .db - .insert_event(tenant.community(), &event, None) - .await?; - if !was_inserted { - return Ok(()); - } - - dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex, None).await; - - info!( - target = %target_pubkey_hex, - relay = %relay_pubkey_hex, - kind, - consent = %consent_path, - "NIP-IA delta event published" - ); - Ok(()) -} - -/// Publish a kind:8002 archived-identity delta event (NIP-IA). -#[allow(clippy::too_many_arguments)] -pub async fn publish_nipia_archived( - tenant: &TenantContext, - state: &Arc, - target_pubkey_hex: &str, - consent_path: &str, - actor_pubkey_hex: &str, - request_event_id: &str, - content: &str, - reason: Option<&str>, - replaced_by: Option<&str>, -) -> anyhow::Result<()> { - publish_nipia_delta( - tenant, - state, - KIND_IA_ARCHIVED, - target_pubkey_hex, - consent_path, - actor_pubkey_hex, - request_event_id, - content, - reason, - replaced_by, - ) - .await -} - -/// Publish a kind:8003 unarchived-identity delta event (NIP-IA). -#[allow(clippy::too_many_arguments)] -pub async fn publish_nipia_unarchived( - tenant: &TenantContext, - state: &Arc, - target_pubkey_hex: &str, - consent_path: &str, - actor_pubkey_hex: &str, - request_event_id: &str, - content: &str, - reason: Option<&str>, -) -> anyhow::Result<()> { - publish_nipia_delta( - tenant, - state, - KIND_IA_UNARCHIVED, - target_pubkey_hex, - consent_path, - actor_pubkey_hex, - request_event_id, - content, - reason, - None, - ) - .await + Ok(event) } #[cfg(test)] diff --git a/crates/buzz-relay/src/handlers/user_deletion_archive_postgres_tests.rs b/crates/buzz-relay/src/handlers/user_deletion_archive_postgres_tests.rs new file mode 100644 index 00000000000..d3214563671 --- /dev/null +++ b/crates/buzz-relay/src/handlers/user_deletion_archive_postgres_tests.rs @@ -0,0 +1,407 @@ +use super::user_deletion_test_support::Fixture; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use serde_json::Value; +use std::{sync::Arc, time::Duration}; +use uuid::Uuid; + +fn archive_request(key: &Keys, kind: u16) -> Event { + EventBuilder::new(Kind::Custom(kind), Uuid::new_v4().to_string()) + .allow_self_tagging() + .tags([ + Tag::parse(["-"]).unwrap(), + Tag::parse(["p", &key.public_key().to_hex()]).unwrap(), + ]) + .sign_with_keys(key) + .unwrap() +} + +async fn archive_snapshot(f: &Fixture) -> Vec { + let mut snapshot = Vec::new(); + for table in ["archived_identities", "events"] { + let rows: Value = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "SELECT coalesce(jsonb_agg(to_jsonb(t) ORDER BY to_jsonb(t)::text), '[]'::jsonb) FROM {table} t WHERE community_id=$1"))) + .bind(f.tenant.community().as_uuid()).fetch_one(&f.pool).await.unwrap(); + snapshot.push(rows); + } + snapshot +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archive_and_unarchive_refusals_leave_no_state_or_notification_events() { + for websocket in [false, true] { + for kind in [9035, 9036] { + for actor in [0, 1, 2] { + let f = Fixture::new().await; + let key = match actor { + 0 => &f.user, + 1 => &f.agent, + _ => &f.peer, + }; + // Before fencing, both operations must work through the real path. + let archive = archive_request(key, 9035); + let unarchive = archive_request(key, 9036); + for event in [&archive, &unarchive] { + let (accepted, reason) = f.send(key, event, websocket).await; + assert!(accepted, "baseline: {reason}"); + } + if kind == 9036 { + assert!(f.send(key, &archive_request(key, 9035), websocket).await.0); + } + f.fence().await; + let before = archive_snapshot(&f).await; + let event = archive_request(key, kind); + let (accepted, reason) = f.send(key, &event, websocket).await; + if actor == 2 { + assert!(accepted, "unaffected user: {reason}"); + assert_eq!( + f.state + .db + .is_archived(f.tenant.community(), &key.public_key().to_hex()) + .await + .unwrap(), + kind == 9035 + ); + } else { + assert!(!accepted); + assert_eq!( + reason, + "restricted: identity is restricted for user deletion" + ); + assert_eq!( + archive_snapshot(&f).await, + before, + "refused kind {kind} changed archive state or persisted notifications" + ); + } + } + } + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archive_replays_preserve_snapshot_identity_and_timestamp() { + for websocket in [false, true] { + let f = Fixture::new().await; + let archive = archive_request(&f.user, 9035); + let unarchive = archive_request(&f.user, 9036); + for event in [&archive, &unarchive] { + assert!(f.send(&f.user, event, websocket).await.0); + let before = archive_snapshot(&f).await; + for _ in 0..2 { + assert!(f.send(&f.user, event, websocket).await.0); + assert_eq!( + archive_snapshot(&f).await, + before, + "duplicate must preserve every row, including snapshot id and created_at" + ); + } + } + let before = archive_snapshot(&f).await; + assert!(f.send(&f.user, &archive, websocket).await.0); + assert_eq!( + archive_snapshot(&f).await, + before, + "old archive replay must not replace the newer unarchive snapshot" + ); + assert!( + f.send(&f.user, &archive_request(&f.user, 9035), websocket) + .await + .0 + ); + let before = archive_snapshot(&f).await; + assert!(f.send(&f.user, &unarchive, websocket).await.0); + assert_eq!( + archive_snapshot(&f).await, + before, + "old unarchive replay must not replace the newer archive snapshot" + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archive_delta_failure_rolls_back_request_and_mutation() { + let f = Fixture::new().await; + sqlx::raw_sql( + "CREATE FUNCTION fail_archive_delta() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'injected archive delta failure'; END $$; + CREATE TRIGGER fail_archive_delta BEFORE INSERT ON events + FOR EACH ROW WHEN (NEW.kind IN (8002, 8003)) EXECUTE FUNCTION fail_archive_delta();", + ) + .execute(&f.pool) + .await + .unwrap(); + for websocket in [false, true] { + for kind in [9035, 9036] { + if kind == 9036 { + f.state + .db + .archive( + f.tenant.community(), + &f.user.public_key().to_hex(), + "self", + &f.user.public_key().to_hex(), + None, + None, + &"a".repeat(64), + ) + .await + .unwrap(); + } + let before = archive_snapshot(&f).await; + let request = archive_request(&f.user, kind); + let (accepted, _, status) = f.submit(&f.user, &request, websocket).await; + assert!(!accepted, "injected database failure must reach the caller"); + if let Some(status) = status { + assert_eq!(status, axum::http::StatusCode::INTERNAL_SERVER_ERROR); + } + assert_eq!( + archive_snapshot(&f).await, + before, + "delta failure must roll back the request and archive change" + ); + f.state + .db + .unarchive(f.tenant.community(), &f.user.public_key().to_hex()) + .await + .unwrap(); + } + } + sqlx::raw_sql("DROP TRIGGER fail_archive_delta ON events; DROP FUNCTION fail_archive_delta();") + .execute(&f.pool) + .await + .unwrap(); + assert!( + f.send(&f.user, &archive_request(&f.user, 9035), false) + .await + .0 + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archive_snapshot_failure_is_retryable_without_reapplying_mutation() { + for (websocket, seed_snapshot) in [(false, false), (false, true), (true, false), (true, true)] { + let f = Fixture::new().await; + if seed_snapshot { + assert!( + f.send(&f.peer, &archive_request(&f.peer, 9035), websocket) + .await + .0 + ); + } + let archive = archive_request(&f.user, 9035); + sqlx::raw_sql( + "CREATE FUNCTION fail_archive_snapshot() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'injected archive snapshot failure'; END $$; + CREATE TRIGGER fail_archive_snapshot BEFORE INSERT ON events + FOR EACH ROW WHEN (NEW.kind = 13535) EXECUTE FUNCTION fail_archive_snapshot();", + ) + .execute(&f.pool) + .await + .unwrap(); + let (accepted, _, status) = f.submit(&f.user, &archive, websocket).await; + assert!(!accepted); + if let Some(status) = status { + assert_eq!(status, axum::http::StatusCode::INTERNAL_SERVER_ERROR); + } + assert!(f + .state + .db + .is_archived(f.tenant.community(), &f.user.public_key().to_hex()) + .await + .unwrap()); + assert!(f + .state + .db + .get_event_by_id(f.tenant.community(), archive.id.as_bytes()) + .await + .unwrap() + .is_some()); + sqlx::raw_sql( + "DROP TRIGGER fail_archive_snapshot ON events; DROP FUNCTION fail_archive_snapshot();", + ) + .execute(&f.pool) + .await + .unwrap(); + assert!( + f.send(&f.user, &archive, websocket).await.0, + "duplicate retry refreshes snapshot" + ); + let tags: Value = sqlx::query_scalar( + "SELECT tags FROM events WHERE community_id=$1 AND kind=13535 AND deleted_at IS NULL", + ) + .bind(f.tenant.community().as_uuid()) + .fetch_one(&f.pool) + .await + .unwrap(); + assert!( + tags.as_array() + .unwrap() + .contains(&serde_json::json!(["p", f.user.public_key().to_hex()])), + "retry must repair a missing or stale snapshot" + ); + let repaired = archive_snapshot(&f).await; + assert!(f.send(&f.user, &archive, websocket).await.0); + assert_eq!( + archive_snapshot(&f).await, + repaired, + "retry after repair must leave the snapshot unchanged" + ); + let unarchive = archive_request(&f.user, 9036); + assert!(f.send(&f.user, &unarchive, websocket).await.0); + assert!(f.send(&f.user, &archive, websocket).await.0); + assert!( + !f.state + .db + .is_archived(f.tenant.community(), &f.user.public_key().to_hex()) + .await + .unwrap(), + "replaying old archive must not undo later unarchive" + ); + let count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM events WHERE community_id=$1 AND kind IN (8002,8003)", + ) + .bind(f.tenant.community().as_uuid()) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!( + count, + if seed_snapshot { 3 } else { 2 }, + "retries must not create new deltas" + ); + assert!( + f.send(&f.user, &archive_request(&f.user, 9035), websocket) + .await + .0 + ); + assert!(f.send(&f.user, &unarchive, websocket).await.0); + assert!( + f.state + .db + .is_archived(f.tenant.community(), &f.user.public_key().to_hex()) + .await + .unwrap(), + "replaying old unarchive must not undo later archive" + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archive_snapshot_noop_rechecks_concurrent_state_changes() { + let f = Fixture::new().await; + assert!( + f.send(&f.user, &archive_request(&f.user, 9035), false) + .await + .0 + ); + let (reached, release) = super::side_effects::publish_test_hooks::arm(f.tenant.community()); + let state = f.state.clone(); + let tenant = f.tenant.clone(); + let publisher = tokio::spawn(async move { + super::side_effects::publish_nipia_archival_list(&tenant, &state).await + }); + tokio::time::timeout(Duration::from_secs(10), reached) + .await + .unwrap() + .unwrap(); + // Keep the old matching snapshot while changing canonical state. A stale + // no-op decision must not leave that snapshot in place. + f.state + .db + .unarchive(f.tenant.community(), &f.user.public_key().to_hex()) + .await + .unwrap(); + release.notify_one(); + tokio::time::timeout(Duration::from_secs(10), publisher) + .await + .unwrap() + .unwrap() + .unwrap(); + let tags: Value = sqlx::query_scalar( + "SELECT tags FROM events WHERE community_id=$1 AND kind=13535 AND deleted_at IS NULL", + ) + .bind(f.tenant.community().as_uuid()) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!( + tags, + serde_json::json!([["-"]]), + "publisher must repair the snapshot after detecting canonical drift" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archive_requests_wait_for_fence_commit_or_rollback() { + for websocket in [false, true] { + for kind in [9035, 9036] { + for commit in [false, true] { + let f = Arc::new(Fixture::new().await); + if kind == 9036 { + assert!( + f.send(&f.user, &archive_request(&f.user, 9035), websocket) + .await + .0 + ); + } + let before = archive_snapshot(&f).await; + let mut gate = f.pending_fence().await; + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *gate) + .await + .unwrap(); + let task_f = f.clone(); + let task = tokio::spawn(async move { + task_f + .send( + &task_f.user, + &archive_request(&task_f.user, kind), + websocket, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(4), async { + loop { + let waiting: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND $1=ANY(pg_blocking_pids(pid)))") + .bind(pid).fetch_one(&f.pool).await.unwrap(); + if waiting { break; } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("archive writer must wait for the exclusive identity lock"); + assert_eq!( + archive_snapshot(&f).await, + before, + "no partial state before fencing decides" + ); + if commit { + gate.commit().await.unwrap(); + } else { + gate.rollback().await.unwrap(); + } + let (accepted, reason) = tokio::time::timeout(Duration::from_secs(8), task) + .await + .unwrap() + .unwrap(); + assert_eq!(accepted, !commit, "{reason}"); + if commit { + assert_eq!(archive_snapshot(&f).await, before); + } else { + assert_eq!( + f.state + .db + .is_archived(f.tenant.community(), &f.user.public_key().to_hex()) + .await + .unwrap(), + kind == 9035 + ); + } + } + } + } +} diff --git a/crates/buzz-relay/src/handlers/user_deletion_error_postgres_tests.rs b/crates/buzz-relay/src/handlers/user_deletion_error_postgres_tests.rs new file mode 100644 index 00000000000..d7210dbe555 --- /dev/null +++ b/crates/buzz-relay/src/handlers/user_deletion_error_postgres_tests.rs @@ -0,0 +1,110 @@ +use super::group_state_postgres_tests::create_channel; +use super::user_deletion_test_support::Fixture; +use buzz_db::user_deletion_fences::{self as fences, FenceIdentity}; +use nostr::{EventBuilder, Kind, Tag}; +use std::{sync::Arc, time::Duration}; + +#[test] +fn identity_error_mapping_preserves_unrelated_failures() { + use crate::handlers::ingest::{event_write_error, IngestError}; + assert!( + matches!(event_write_error(&buzz_db::DbError::UserDeletionRestricted), + Some(IngestError::Rejected(message)) if message.starts_with("restricted:")) + ); + assert!( + matches!(event_write_error(&buzz_db::DbError::EventAuthorOwnershipChanged), + Some(IngestError::Internal(message)) if message.starts_with("error:")) + ); + for error in [ + buzz_db::DbError::AccessDenied("community is write-fenced (deleting)".into()), + buzz_db::DbError::AccessDenied( + "deletion-lock admission requires read committed isolation".into(), + ), + buzz_db::DbError::Sqlx(sqlx::Error::PoolTimedOut), + ] { + assert!(event_write_error(&error).is_none()); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn ownership_races_are_retryable_over_http_and_websocket() { + for websocket in [false, true] { + for kind in [9, 9000, 9002] { + let f = Arc::new(Fixture::new().await); + f.state + .db + .ensure_user(f.tenant.community(), f.peer.public_key().as_bytes()) + .await + .unwrap(); + let channel = create_channel(&f.state, &f.tenant, &f.agent, &[]).await; + let mut tags = vec![Tag::parse(["h", &channel.to_string()]).unwrap()]; + if kind == 9000 { + tags.push(Tag::parse(["p", &f.peer.public_key().to_hex(), "member"]).unwrap()); + } else if kind == 9002 { + tags.push(Tag::parse(["name", "renamed after retry"]).unwrap()); + } + let event = EventBuilder::new(Kind::Custom(kind), "owner race") + .tags(tags) + .sign_with_keys(&f.agent) + .unwrap(); + let mut assignment = f.pool.begin().await.unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *assignment) + .await + .unwrap(); + fences::lock_exclusive( + &mut assignment, + &[FenceIdentity::agent( + f.tenant.community(), + f.agent.public_key(), + )], + ) + .await + .unwrap(); + let writer_fixture = f.clone(); + let pending = event.clone(); + let sending = tokio::spawn(async move { + writer_fixture + .submit(&writer_fixture.agent, &pending, websocket) + .await + }); + tokio::time::timeout(Duration::from_secs(4), async { + loop { + let blocked: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid)))") + .bind(pid).fetch_one(&f.pool).await.unwrap(); + if blocked { break; } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("writer must reach the production identity guard"); + sqlx::query( + "UPDATE users SET agent_owner_pubkey = $1 WHERE community_id = $2 AND pubkey = $3", + ) + .bind(f.peer.public_key().to_bytes().as_slice()) + .bind(f.tenant.community().as_uuid()) + .bind(f.agent.public_key().to_bytes().as_slice()) + .execute(&mut *assignment) + .await + .unwrap(); + assignment.commit().await.unwrap(); + let (accepted, message, status) = sending.await.unwrap(); + assert!(!accepted); + if websocket { + assert_eq!(message, "error: internal server error"); + } else { + assert_eq!(status, Some(axum::http::StatusCode::INTERNAL_SERVER_ERROR)); + assert_eq!(message, "internal server error"); + } + assert!(f + .state + .db + .get_event_by_id(f.tenant.community(), event.id.as_bytes()) + .await + .unwrap() + .is_none()); + let (accepted, message) = f.send(&f.agent, &event, websocket).await; + assert!(accepted, "kind {kind}: {message}"); + } + } +} diff --git a/crates/buzz-relay/src/handlers/user_deletion_expanded_postgres_tests.rs b/crates/buzz-relay/src/handlers/user_deletion_expanded_postgres_tests.rs new file mode 100644 index 00000000000..24d7ac8b2fd --- /dev/null +++ b/crates/buzz-relay/src/handlers/user_deletion_expanded_postgres_tests.rs @@ -0,0 +1,243 @@ +//! Persistent insertion coverage through HTTP and WebSocket event handling. +use super::group_state_postgres_tests::create_channel; +use super::user_deletion_test_support::{action, snapshot, Fixture}; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use std::{sync::Arc, time::Duration}; +use uuid::Uuid; + +/// Each case must succeed before fencing, then leave every relevant table +/// unchanged on rejection. Both real HTTP routing and the WS EVENT handler run. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn expanded_actions_reject_fenced_humans_and_agents_without_partial_writes() { + let f = Fixture::new().await; + let mut fixtures = Vec::new(); + for websocket in [false, true] { + for key in [&f.user, &f.agent, &f.peer] { + let channel = create_channel(&f.state, &f.tenant, key, &[]).await; + let root = action(key, channel, 9, "edit target", vec![]); + let post = action(key, channel, 45001, "vote target", vec![]); + assert!(f.send(key, &root, websocket).await.0); + assert!(f.send(key, &post, websocket).await.0); + fixtures.push((websocket, key, channel, root, post)); + } + } + for fenced in [false, true] { + if fenced { + f.fence().await; + } + for (websocket, key, channel, root, post) in &fixtures { + let text = Uuid::new_v4().to_string(); + let member = Keys::generate().public_key().to_hex(); + let new_channel = Uuid::new_v4(); + let dm = EventBuilder::new(Kind::Custom(41010), text.clone()) + .tags([Tag::parse(["p", &Keys::generate().public_key().to_hex()]).unwrap()]) + .sign_with_keys(key) + .unwrap(); + let cases = vec![ + ("open DM", dm), + ( + "reaction", + action( + key, + *channel, + 7, + if fenced { "🚀" } else { "👍" }, + vec![vec!["e".into(), root.id.to_hex()]], + ), + ), + ( + "edit", + action( + key, + *channel, + 40003, + &text, + vec![vec!["e".into(), root.id.to_hex()]], + ), + ), + ( + "vote", + action( + key, + *channel, + 45002, + "+", + vec![ + vec!["e".into(), post.id.to_hex()], + vec!["nonce".into(), text.clone()], + ], + ), + ), + ("canvas", action(key, *channel, 40100, &text, vec![])), + ( + "settings", + action( + key, + *channel, + 9002, + "", + vec![ + vec!["name".into(), text.clone()], + vec!["topic".into(), text.clone()], + ], + ), + ), + ( + "ttl settings", + action( + key, + *channel, + 9002, + "", + vec![vec![ + "ttl".into(), + if fenced { "7200" } else { "3600" }.into(), + ]], + ), + ), + ( + "add member", + action( + key, + *channel, + 9000, + "", + vec![vec!["p".into(), member, "member".into()]], + ), + ), + ( + "create channel", + action( + key, + new_channel, + 9007, + "", + vec![vec!["name".into(), text.clone()]], + ), + ), + ( + "artifact", + action( + key, + *channel, + 45010, + &serde_json::json!({"text": text}).to_string(), + vec![ + vec!["ar".into(), "1".into()], + vec!["d".into(), Uuid::new_v4().to_string()], + vec!["type".into(), "buzz.task".into()], + vec!["op".into(), "create".into()], + vec!["title".into(), "Task".into()], + ], + ), + ), + ]; + for (name, event) in cases { + let should_accept = !fenced || key.public_key() == f.peer.public_key(); + let before = snapshot(&f).await; + let (accepted, reason) = f.send(key, &event, *websocket).await; + assert_eq!( + accepted, should_accept, + "{name} fenced={fenced} ws={websocket}: {reason}" + ); + assert_eq!( + f.state + .db + .get_event_by_id(f.tenant.community(), event.id.as_bytes()) + .await + .unwrap() + .is_some(), + should_accept, + "{name}" + ); + if !should_accept { + assert_eq!( + reason, "restricted: identity is restricted for user deletion", + "{name}" + ); + assert_eq!(snapshot(&f).await, before, "{name} must roll back domain writes, counters, mentions, and notifications"); + } + } + } + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn concurrent_expanded_writes_wait_for_fencing_commit_or_rollback() { + for commit in [false, true] { + let f = Arc::new(Fixture::new().await); + let channel = create_channel(&f.state, &f.tenant, &f.user, &[]).await; + let gate = f.pending_fence().await; + let mut tasks = Vec::new(); + for (index, kind) in [9, 40100, 9002, 9000, 9007].into_iter().enumerate() { + let f = f.clone(); + let channel = if kind == 9007 { + Uuid::new_v4() + } else { + channel + }; + let tags = match kind { + 9002 => vec![vec!["ttl".into(), "3600".into()]], + 9000 => vec![vec![ + "p".into(), + Keys::generate().public_key().to_hex(), + "member".into(), + ]], + 9007 => vec![vec!["name".into(), Uuid::new_v4().to_string()]], + _ => vec![], + }; + let event = action(&f.user, channel, kind, "concurrent test", tags); + tasks.push(tokio::spawn(async move { + let result = f.send(&f.user, &event, index % 2 == 0).await; + (event, result) + })); + } + // Wait for an actual database lock waiter, not a guessed sleep. + tokio::time::timeout(Duration::from_secs(4), async { + loop { + let waiting: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM pg_locks l JOIN pg_stat_activity a ON a.pid=l.pid WHERE a.datname=current_database() AND l.locktype='advisory' AND NOT l.granted)") + .fetch_one(&f.pool).await.unwrap(); + if waiting { break; } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }).await.expect("a real writer must wait on the fence"); + assert!(tasks.iter().all(|task| !task.is_finished())); + assert!( + f.send(&f.peer, &f.message(&f.peer, 9, None), false).await.0, + "unrelated user must proceed" + ); + if commit { + gate.commit().await.unwrap(); + } else { + gate.rollback().await.unwrap(); + } + for task in tasks { + let (event, (accepted, reason)) = tokio::time::timeout(Duration::from_secs(8), task) + .await + .unwrap() + .unwrap(); + assert_eq!(accepted, !commit, "kind {}: {reason}", event.kind.as_u16()); + assert_eq!( + f.state + .db + .get_event_by_id(f.tenant.community(), event.id.as_bytes()) + .await + .unwrap() + .is_some(), + !commit + ); + } + let deadlocks: i64 = sqlx::query_scalar( + "SELECT deadlocks FROM pg_stat_database WHERE datname=current_database()", + ) + .fetch_one(&f.pool) + .await + .unwrap(); + assert_eq!( + deadlocks, 0, + "no deadlocks, including swallowed TTL-refresh failures" + ); + } +} diff --git a/crates/buzz-relay/src/handlers/user_deletion_lifecycle_postgres_tests.rs b/crates/buzz-relay/src/handlers/user_deletion_lifecycle_postgres_tests.rs new file mode 100644 index 00000000000..139f5d80899 --- /dev/null +++ b/crates/buzz-relay/src/handlers/user_deletion_lifecycle_postgres_tests.rs @@ -0,0 +1,118 @@ +use super::group_state_postgres_tests::create_channel; +use super::user_deletion_test_support::{action, snapshot, Fixture}; +use uuid::Uuid; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn conditional_canvas_and_deletion_actions_enforce_sender_fences() { + for websocket in [false, true] { + for actor in [0, 1, 2] { + let f = Fixture::new().await; + let key = match actor { + 0 => &f.user, + 1 => &f.agent, + _ => &f.peer, + }; + let mut phases = Vec::new(); + // Prepare both targets before fencing, so authorization/target checks + // cannot accidentally make an unexercised guard look correct. + for _ in 0..2 { + let channel = create_channel(&f.state, &f.tenant, key, &[]).await; + let root = action(key, channel, 9, "delete me", vec![]); + assert!(f.send(key, &root, websocket).await.0); + let workflow = Uuid::new_v4(); + f.state + .db + .ensure_user(f.tenant.community(), key.public_key().as_bytes()) + .await + .unwrap(); + sqlx::query("INSERT INTO workflows (community_id,id,name,owner_pubkey,channel_id,definition,definition_hash) VALUES ($1,$2,'fence-test',$3,$4,'{}','\\x00')") + .bind(f.tenant.community().as_uuid()).bind(workflow).bind(key.public_key().as_bytes()).bind(channel) + .execute(&f.pool).await.unwrap(); + let definition = action( + key, + channel, + 30620, + "{}", + vec![vec!["d".into(), workflow.to_string()]], + ); + f.state + .db + .replace_parameterized_event( + f.tenant.community(), + &definition, + &workflow.to_string(), + Some(channel), + ) + .await + .unwrap(); + phases.push((channel, root, workflow)); + } + for (phase, (channel, root, workflow)) in phases.into_iter().enumerate() { + if phase == 1 { + f.fence().await; + } + let accepted = phase == 0 || actor == 2; + let cases = [ + ( + "conditional canvas", + action( + key, + channel, + 40100, + "canvas", + vec![vec!["expected-revision".into(), "none".into()]], + ), + ), + ( + "message deletion", + action( + key, + channel, + 5, + "", + vec![vec!["e".into(), root.id.to_hex()]], + ), + ), + ( + "workflow deletion", + action( + key, + channel, + 5, + "", + vec![vec![ + "a".into(), + format!("30620:{}:{workflow}", key.public_key()), + ]], + ), + ), + ("channel deletion", action(key, channel, 9008, "", vec![])), + ]; + for (name, event) in cases { + let before = snapshot(&f).await; + let (actual, reason) = f.send(key, &event, websocket).await; + assert_eq!( + actual, accepted, + "{name} actor={actor} phase={phase} ws={websocket}: {reason}" + ); + if !accepted { + assert_eq!( + reason, "restricted: identity is restricted for user deletion", + "{name}" + ); + assert_eq!( + snapshot(&f).await, + before, + "{name} changed state despite refusal" + ); + } else if name == "workflow deletion" { + let exists: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM workflows WHERE community_id=$1 AND id=$2)") + .bind(f.tenant.community().as_uuid()).bind(workflow).fetch_one(&f.pool).await.unwrap(); + assert!(!exists, "positive control must really delete the workflow"); + } + } + } + } + } +} diff --git a/crates/buzz-relay/src/handlers/user_deletion_message_postgres_tests.rs b/crates/buzz-relay/src/handlers/user_deletion_message_postgres_tests.rs new file mode 100644 index 00000000000..da4530e82c8 --- /dev/null +++ b/crates/buzz-relay/src/handlers/user_deletion_message_postgres_tests.rs @@ -0,0 +1,50 @@ +use super::user_deletion_test_support::Fixture; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn http_and_websocket_messages_honor_user_and_agent_fences() { + let f = Fixture::new().await; + let root = f.message(&f.peer, 9, None); + assert!(f.send(&f.peer, &root, false).await.0); + for websocket in [false, true] { + for key in [&f.user, &f.agent, &f.peer] { + assert!( + f.send(key, &f.message(key, 9, Some(&root)), websocket) + .await + .0 + ); + } + } + let counters: (i64, i64) = sqlx::query_as("SELECT reply_count::bigint, descendant_count::bigint FROM thread_metadata WHERE community_id = $1 AND event_id = $2").bind(f.tenant.community().as_uuid()).bind(root.id.as_bytes().as_slice()).fetch_one(&f.pool).await.unwrap(); + assert_eq!(counters, (6, 6), "fixture must create real replies"); + f.fence().await; + for websocket in [false, true] { + for key in [&f.user, &f.agent] { + for kind in [9, 40002, 45001, 45003] { + let event = f.message(key, kind, Some(&root)); + let (accepted, reason) = f.send(key, &event, websocket).await; + assert!(!accepted, "fenced event accepted"); + assert_eq!( + reason, + "restricted: identity is restricted for user deletion" + ); + assert!(f + .state + .db + .get_event_by_id(f.tenant.community(), event.id.as_bytes()) + .await + .unwrap() + .is_none()); + } + } + } + let after: (i64, i64) = sqlx::query_as("SELECT reply_count::bigint, descendant_count::bigint FROM thread_metadata WHERE community_id = $1 AND event_id = $2").bind(f.tenant.community().as_uuid()).bind(root.id.as_bytes().as_slice()).fetch_one(&f.pool).await.unwrap(); + assert_eq!(after, counters, "rejected replies must not update counters"); + for websocket in [false, true] { + assert!( + f.send(&f.peer, &f.message(&f.peer, 9, Some(&root)), websocket) + .await + .0 + ); + } +} diff --git a/crates/buzz-relay/src/handlers/user_deletion_test_support.rs b/crates/buzz-relay/src/handlers/user_deletion_test_support.rs new file mode 100644 index 00000000000..170d29f8775 --- /dev/null +++ b/crates/buzz-relay/src/handlers/user_deletion_test_support.rs @@ -0,0 +1,249 @@ +//! Shared fixtures for user-deletion transport regressions. + +use std::{sync::Arc, time::Duration}; + +use buzz_core::TenantContext; +use buzz_db::user_deletion_fences::{self as fences, FenceIdentity}; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use serde_json::Value; +use uuid::Uuid; + +use super::group_state_postgres_tests::{community, create_channel, test_conn}; +use crate::state::AppState; + +pub(super) struct Fixture { + pub(super) state: Arc, + pub(super) pool: sqlx::PgPool, + pub(super) tenant: TenantContext, + pub(super) channel: Uuid, + pub(super) user: Keys, + pub(super) agent: Keys, + pub(super) peer: Keys, + url: String, + server: tokio::task::JoinHandle<()>, +} + +impl Fixture { + pub(super) async fn new() -> Self { + let state = crate::api::bridge::postgres_tests::bridge_handler_test_state() + .await + .expect("Postgres and Redis"); + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let tenant = community(&state, "message-deletion").await; + let (user, agent, peer) = (Keys::generate(), Keys::generate(), Keys::generate()); + state + .db + .ensure_user(tenant.community(), user.public_key().as_bytes()) + .await + .unwrap(); + state + .db + .ensure_user(tenant.community(), agent.public_key().as_bytes()) + .await + .unwrap(); + state + .db + .set_agent_owner( + tenant.community(), + agent.public_key().as_bytes(), + user.public_key().as_bytes(), + ) + .await + .unwrap(); + let channel = create_channel( + &state, + &tenant, + &peer, + &[ + &["p", &user.public_key().to_hex()], + &["p", &agent.public_key().to_hex()], + ], + ) + .await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let router = crate::router::build_router(state.clone()); + let server = tokio::spawn(async move { + axum::serve( + listener, + router.into_make_service_with_connect_info::(), + ) + .await + .unwrap(); + }); + Self { + url, + server, + state, + pool, + tenant, + channel, + user, + agent, + peer, + } + } + + pub(super) fn message(&self, key: &Keys, kind: u16, root: Option<&Event>) -> Event { + let mut tags = vec![Tag::parse(["h", &self.channel.to_string()]).unwrap()]; + if let Some(root) = root { + tags.push(Tag::parse(["e", &root.id.to_hex(), "", "root"]).unwrap()); + tags.push(Tag::parse(["e", &root.id.to_hex(), "", "reply"]).unwrap()); + } + EventBuilder::new(Kind::Custom(kind), Uuid::new_v4().to_string()) + .tags(tags) + .sign_with_keys(key) + .unwrap() + } + + pub(super) async fn fence(&self) { + self.pending_fence().await.commit().await.unwrap(); + } + + pub(super) async fn pending_fence(&self) -> sqlx::Transaction<'static, sqlx::Postgres> { + let request = self + .state + .db + .user_deletion_store() + .submit(&format!("demo:{}", Uuid::new_v4()), "operator", None) + .await + .unwrap(); + let store = self.state.db.entity_deletion_store(); + store + .approve( + request.id, + "reviewer", + request.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); + let lease = store + .claim(request.id, "worker", Duration::from_secs(300)) + .await + .unwrap() + .unwrap(); + let mut tx = self.pool.begin().await.unwrap(); + let user = FenceIdentity::user(self.user.public_key()); + let agent = FenceIdentity::agent(self.tenant.community(), self.agent.public_key()); + fences::lock_request(&mut tx, &lease).await.unwrap(); + fences::lock_exclusive(&mut tx, &[user, agent]) + .await + .unwrap(); + fences::insert(&mut tx, user, &lease, None).await.unwrap(); + fences::insert(&mut tx, agent, &lease, Some(self.user.public_key())) + .await + .unwrap(); + tx + } + + pub(super) async fn send(&self, key: &Keys, event: &Event, websocket: bool) -> (bool, String) { + let (accepted, message, status) = self.submit(key, event, websocket).await; + if let Some(status) = status.filter(|s| !s.is_success()) { + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{message}"); + } + (accepted, message) + } + + pub(super) async fn submit( + &self, + key: &Keys, + event: &Event, + websocket: bool, + ) -> (bool, String, Option) { + if websocket { + let (conn, mut rx) = test_conn(&self.tenant, key); + if let crate::connection::AuthState::Authenticated(ctx) = + &mut *conn.auth_state.lock().unwrap() + { + ctx.scopes = buzz_auth::Scope::all_known(); + // Intentionally no cached owner: the write must use PostgreSQL. + ctx.agent_owner_pubkey = None; + } + super::event::handle_event(event.clone(), conn, self.state.clone()).await; + let axum::extract::ws::Message::Text(text) = rx.try_recv().unwrap() else { + panic!("expected OK"); + }; + let frame: Value = serde_json::from_str(&text).unwrap(); + assert_eq!(frame[0], "OK", "{frame}"); + assert_eq!(frame[1], event.id.to_hex()); + ( + frame[2].as_bool().unwrap(), + frame[3].as_str().unwrap().to_string(), + None, + ) + } else { + let response = reqwest::Client::new() + .post(format!("{}/events", self.url)) + .header("host", self.tenant.host()) + .header("x-pubkey", key.public_key().to_hex()) + .json(event) + .send() + .await + .unwrap(); + let status = response.status(); + let body: Value = response.json().await.unwrap(); + if status.is_success() { + ( + body["accepted"].as_bool().unwrap(), + body["message"].as_str().unwrap_or("").to_owned(), + Some(status), + ) + } else { + ( + false, + body["error"].as_str().unwrap().to_owned(), + Some(status), + ) + } + } + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.server.abort(); + } +} + +pub(super) fn action( + key: &Keys, + channel: Uuid, + kind: u16, + content: &str, + extra: Vec>, +) -> Event { + let mut tags = vec![Tag::parse(["h", &channel.to_string()]).unwrap()]; + tags.extend(extra.into_iter().map(|tag| Tag::parse(tag).unwrap())); + EventBuilder::new(Kind::Custom(kind), content) + .tags(tags) + .sign_with_keys(key) + .unwrap() +} + +pub(super) async fn snapshot(f: &Fixture) -> Vec { + let mut rows = Vec::new(); + for table in [ + "events", + "channels", + "channel_members", + "reactions", + "thread_metadata", + "event_mentions", + "operator_listener_outbox", + "artifact_heads", + "artifact_revisions", + "workflows", + ] { + let query = format!("SELECT COALESCE(jsonb_agg(to_jsonb(t) ORDER BY to_jsonb(t)::text), '[]'::jsonb) FROM {table} t WHERE community_id=$1"); + rows.push( + sqlx::query_scalar(sqlx::AssertSqlSafe(query)) + .bind(f.tenant.community().as_uuid()) + .fetch_one(&f.pool) + .await + .unwrap(), + ); + } + rows +} diff --git a/docs/user-deletion-message-fencing.md b/docs/user-deletion-message-fencing.md new file mode 100644 index 00000000000..01f3d580781 --- /dev/null +++ b/docs/user-deletion-message-fencing.md @@ -0,0 +1,114 @@ +# User-deletion fencing: event database writes + +The two shared event insertion functions in `crates/buzz-db/src/store/event.rs`, +`insert_event_on` and `insert_event_with_thread_metadata_tx`, refuse events +whose signer has a user-deletion restriction. Every event kind reaching these +functions is checked, including relay-signed events. + +## Transaction flow + +1. Enter a community-admitted, writer-pool READ COMMITTED transaction. +2. Read the signer's stored owner in this community, if present. +3. Acquire shared global signer/owner locks and the scoped signer lock. Each + call sorts its global keys before its scoped keys. +4. Read the owner again. If it changed while acquiring locks, return + `EventAuthorOwnershipChanged`; the caller must roll back and retry in a new + transaction. +5. Query the signer's global and community-scoped restrictions together. If + either exists, return `UserDeletionRestricted` and roll back. +6. Persist the event and its transactional derived changes. Hold the locks + until commit or rollback. + +Human restrictions apply across communities by public key. Agent restrictions +apply in their specified community. The signer's own restriction decides +whether the event is rejected; the owner's restriction is not an additional +check for an independently admitted agent. Missing or recreated profile rows do +not bypass restrictions. + +If a writer holds the shared identity lock first, an exclusive fence waits for +its transaction to finish. If the fence commits first, the waiting writer sees +the restriction and rolls back. If the fence rolls back, the writer may proceed. +No user-specific lease is involved in this database-write protocol. + +## Errors and atomicity + +`event_write_error` maps `UserDeletionRestricted` to the client refusal +`restricted: identity is restricted for user deletion`. Ownership changes use +the retryable internal-error category: HTTP responds with 500; WebSocket +responds with `error: internal server error`. This classification permits a +retry; it does not implement client retries. Unrelated database errors retain +the caller's existing mapping rather than becoming restriction errors. + +Archive/unarchive requests (9035/9036) validate consent without changing state. +Their persistence transaction stores the guarded request, changes +`archived_identities`, and stores the relay delta (8002/8003). A failure rolls +back all three. Notifications are dispatched after commit. Duplicate requests +do not reapply the archive change or create another delta. + +The archival snapshot (13535) is refreshed after commit only when missing or +stale. An already-current snapshot retains its event ID and timestamp and is +not published again, including on duplicate requests. A refresh failure +propagates to the caller; retrying the request refreshes the snapshot without +repeating the mutation. The committed request and delta remain available for +query/replay. This is not a durable background retry queue: a process exit +between commit and snapshot refresh can leave the snapshot stale until a later +refresh. Existing freshness/consent validation still applies to retries. + +## Actual lock order and exclusive-worker requirements + +Shared identity locking occurs at event insertion, not necessarily at the +start of the transaction. Artifact and channel writers can already hold domain +row/advisory locks. An artifact Move inserts the author's revision and then a +relay-signed removal, acquiring another global identity lock after the first +signer's scoped lock. Per-call sorting is not a transaction-wide lock order. + +An exclusive fencing transaction must: + +- Acquire community admission, then the deletion request row, then its complete + global identity set, then its complete scoped identity set. +- Write only restriction/request/checkpoint state while holding those locks. + Do not lock or mutate users, channels, artifacts, events or other serving + domain rows. Read ownership without row locks and perform cleanup in later + transactions. +- Never upgrade a shared identity lock to exclusive. Audit complete exclusive + sets against transactions containing multiple signers before wiring a worker; + per-call sorting alone does not prove arbitrary multi-key batches safe. + +The domain-row restriction prevents a fence from waiting on a row held by a +writer that is itself waiting on the fence. PostgreSQL regressions exercise +channel-TTL and artifact Move interleavings and assert that no deadlock was +recorded. Those cases are evidence for the tested paths, not a proof about every +future worker or lock combination. + +## Coverage and remaining work + +Persistence through the two guarded functions includes messages, replies, +reactions, edits, votes, canvas, channel changes, deletion events, artifacts, +and identity archive/unarchive requests. A refusal rolls back domain changes +performed in the same transaction, including counters and notification events. + +Separate replaceable-event writers (profiles, emoji collections, bookmarks and +workflow definitions), personal read state, uploads, git, audio and ephemeral +activity are outside this guard. Authentication, agent creation/owner +assignment, checks on affected users other than the signer, and background work +need separate coordination. + +The real user-deletion worker remains disabled. Before enabling it, coordinate +all remaining serving paths and external work, and atomically fence the user +and the approved owned-agent set. The owner reread detects changes observed +while obtaining locks; it is not a complete ownership-assignment protocol. +Restriction insertion behavior and the schema are unchanged by event guards. + +## Regression tests + +The directly registered `user_deletion_*_postgres_tests` modules in `buzz-db` +and `buzz-relay` cover scopes, missing profiles, unaffected actors, commit and +rollback races, ownership changes, transport errors and transactional state. +The artifact fence race lives in `artifact_postgres_tests`. + +With isolated PostgreSQL and Redis configured: + +```sh +scripts/postgres-test-run.sh -p buzz-db -p buzz-relay --lib \ + -E 'test(/user_deletion/) or test(/artifact_move_and_identity/)' --test-threads 4 +```