diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index bab06526ed..e0875618dd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -145,7 +145,7 @@ buzz-admin (operator CLI: relay membership + key generation) buzz-test-client (integration test harness + manual CLI) ``` -**Key architectural principle:** The relay is the single source of truth. `buzz-relay` orchestrates all subsystems by calling them directly — it imports `buzz-db`, `buzz-auth`, `buzz-pubsub`, `buzz-search`, `buzz-audit`, and `buzz-workflow`. However, those subsystems are isolated from each other: `buzz-workflow` never calls `buzz-pubsub`, `buzz-search` never calls `buzz-db`, etc. Cross-subsystem coordination happens only through the relay. In multi-community mode, the relay also owns propagation of `TenantContext`; service crates should receive community-scoped inputs rather than independently deriving tenancy from client-controlled event tags. +**Key architectural principle:** The relay is the single source of truth. `buzz-relay` orchestrates all subsystems by calling them directly — it imports `buzz-db`, `buzz-auth`, `buzz-pubsub`, `buzz-search`, `buzz-audit`, and `buzz-workflow`. However, those subsystems are isolated from each other: `buzz-workflow` never calls `buzz-pubsub`, `buzz-search` never calls `buzz-db`, etc. The one sanctioned exception is `buzz-audit`, which calls `buzz_db::deletion::guard_community_write` to admit each audit append into its community before taking the chain lock. Cross-subsystem coordination happens only through the relay. In multi-community mode, the relay also owns propagation of `TenantContext`; service crates should receive community-scoped inputs rather than independently deriving tenancy from client-controlled event tags. --- @@ -596,7 +596,7 @@ Tamper-evident append-only log with SHA-256 hash chaining. **Hash covers:** seq (big-endian bytes), timestamp (RFC3339), event_id, event_kind (big-endian), actor_pubkey, action string, channel_id (16 bytes or 16 zero bytes if None), canonical metadata JSON (BTreeMap for deterministic key ordering), prev_hash. -**Single-writer guarantee:** `pg_advisory_lock` before each transaction. Lock released in all branches including panic (`catch_unwind`). +**Single-writer guarantee:** each append runs in one transaction that first passes community write admission (the shared community deletion lock plus the active/undeleted check, via `buzz-db`'s `DeletionStore::guard_transaction`), then takes a per-community `pg_advisory_xact_lock`. Both locks release on commit or rollback, including a dropped future. **10 audit actions:** `EventCreated`, `EventDeleted`, `ChannelCreated`, `ChannelUpdated`, `ChannelDeleted`, `MemberAdded`, `MemberRemoved`, `AuthSuccess`, `AuthFailure`, `RateLimitExceeded`. diff --git a/Cargo.lock b/Cargo.lock index 0a924f4e36..27a35a4ae5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1058,8 +1058,8 @@ version = "0.1.0" dependencies = [ "buzz-core", "buzz-datastore-tracing", + "buzz-db", "chrono", - "futures-util", "hex", "metrics", "serde", diff --git a/crates/buzz-audit/Cargo.toml b/crates/buzz-audit/Cargo.toml index 766ade6505..a1af6aac0c 100644 --- a/crates/buzz-audit/Cargo.toml +++ b/crates/buzz-audit/Cargo.toml @@ -9,6 +9,7 @@ description = "Hash-chain audit log for Buzz" [dependencies] buzz-core = { workspace = true } +buzz-db = { workspace = true } buzz-datastore-tracing = { workspace = true } sqlx = { workspace = true } tokio = { workspace = true } @@ -21,4 +22,3 @@ metrics = { workspace = true } thiserror = { workspace = true } sha2 = { workspace = true } hex = { workspace = true } -futures-util = { workspace = true } diff --git a/crates/buzz-audit/src/error.rs b/crates/buzz-audit/src/error.rs index 6f28302b4b..9d63e15725 100644 --- a/crates/buzz-audit/src/error.rs +++ b/crates/buzz-audit/src/error.rs @@ -14,6 +14,10 @@ pub enum AuditError { #[error("database error: {0}")] Database(#[from] sqlx::Error), + /// Community write admission failed before the audit chain was locked. + #[error("audit write admission failed")] + Admission(#[from] buzz_db::DbError), + /// The `prev_hash` of an entry does not match the hash of the preceding /// entry in the same community's chain. #[error( @@ -77,6 +81,9 @@ mod tests { AuditError::HashMismatch { seq: 42 }, AuditError::UnknownAction, AuditError::UnsupportedHashVersion { version: 99 }, + AuditError::Admission(buzz_db::DbError::AccessDenied(format!( + "community {community} is write-fenced" + ))), ]; for err in &domain_errors { diff --git a/crates/buzz-audit/src/service.rs b/crates/buzz-audit/src/service.rs index 518ae33cec..3d2cb8d364 100644 --- a/crates/buzz-audit/src/service.rs +++ b/crates/buzz-audit/src/service.rs @@ -1,6 +1,5 @@ use chrono::{DateTime, Utc}; -use futures_util::FutureExt as _; -use sqlx::{Acquire, PgPool, Row}; +use sqlx::{PgPool, Row}; use tracing::{debug, warn}; use uuid::Uuid; @@ -17,16 +16,14 @@ use crate::{ /// The `created_at` stamped on a new entry. /// /// Reduced to the precision Postgres round-trips before it is hashed — see -/// [`to_storage_precision`]. Split out from [`AuditService::log_inner`] so the +/// [`to_storage_precision`]. Split out from [`AuditService::log`] so the /// invariant is testable without a database. fn log_timestamp() -> DateTime { to_storage_precision(Utc::now()) } -/// Per-community advisory lock key. Derived in Postgres from the community UUID -/// so two communities never serialize each other's audit writes (which would be -/// both a throughput bottleneck and a cross-tenant timing oracle). The lock is -/// taken with `pg_advisory_lock(hashtextextended(...))` — see [`AuditService::log`]. +/// Per-community audit-chain advisory lock key. Admission takes the community +/// deletion lock first; this key then serializes appends within that community. const AUDIT_LOCK_NAMESPACE: &str = "buzz_audit:"; /// Append-only, per-community hash-chain audit log backed by Postgres. @@ -47,50 +44,22 @@ impl AuditService { /// Append a new entry to the calling community's chain. /// - /// Serialized per-community via `pg_advisory_lock`. Postgres advisory locks - /// are session-scoped, so we acquire before the transaction and release - /// after commit (or on any error path). + /// Community admission precedes the transaction-scoped audit-chain lock. + /// Both locks are released on commit or rollback, including a dropped future. #[datastore_span( name = "audit_log", system = "postgresql", fields(action = %entry.action) )] pub async fn log(&self, entry: NewAuditEntry) -> Result { - let mut conn = self.pool.acquire().await?; - - // Per-community advisory lock: hash the namespaced community id to an - // i64 lock key inside Postgres. Communities lock independently. + let mut tx = self.pool.begin().await?; + buzz_db::deletion::guard_community_write(&mut tx, entry.community_id).await?; let lock_key = format!("{AUDIT_LOCK_NAMESPACE}{}", entry.community_id); - sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))") + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") .bind(&lock_key) - .execute(&mut *conn) + .execute(&mut *tx) .await?; - // Run the chain append and release the lock regardless of outcome. - // catch_unwind so a panic still releases the lock before the connection - // returns to the pool. - let result = std::panic::AssertUnwindSafe(self.log_inner(&mut conn, entry)) - .catch_unwind() - .await; - - let _ = sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))") - .bind(&lock_key) - .execute(&mut *conn) - .await; - - match result { - Ok(inner_result) => inner_result, - Err(panic_payload) => std::panic::resume_unwind(panic_payload), - } - } - - async fn log_inner( - &self, - conn: &mut sqlx::pool::PoolConnection, - entry: NewAuditEntry, - ) -> Result { - let mut tx = conn.begin().await?; - // The stored row keys on the raw UUID; the typed `CommunityId` on the // input is the provenance fence, dereferenced here at the DB boundary. let community_id = *entry.community_id.as_uuid(); @@ -321,6 +290,58 @@ mod postgres_tests { id } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn audit_append_requires_application_admission_with_triggers_bypassed() { + let _g = db_lock().lock().await; + let url = std::env::var("DATABASE_URL").expect("Postgres lane must provide DATABASE_URL"); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&url) + .await + .expect("connect test database"); + let active = make_community(&pool).await; + for state in ["quiescing", "fenced"] { + let id = make_community(&pool).await; + let mut tx = pool.begin().await.expect("begin lifecycle transition"); + sqlx::query( + "SELECT set_config('buzz.deletion_executor_community', $1, true), \ + set_config('buzz.deletion_fence_generation', \ + (SELECT deletion_fence_generation::text FROM communities WHERE id = $2), true)", + ) + .bind(id.to_string()) + .bind(id) + .execute(&mut *tx) + .await + .expect("authorize lifecycle transition"); + sqlx::query("UPDATE communities SET deletion_state = $2 WHERE id = $1") + .bind(id) + .bind(state) + .execute(&mut *tx) + .await + .expect("set lifecycle state"); + tx.commit().await.expect("commit lifecycle state"); + + sqlx::query("SET session_replication_role = replica") + .execute(&pool) + .await + .expect("disable ordinary triggers"); + let error = AuditService::new(pool.clone()) + .log(new_entry(id, AuditAction::EventCreated)) + .await + .expect_err("fenced audit append must reject before chain lock"); + assert!( + matches!(error, AuditError::Admission(buzz_db::DbError::AccessDenied(message)) if message.contains("write-fenced")), + "expected application admission failure" + ); + } + let entry = AuditService::new(pool.clone()) + .log(new_entry(active, AuditAction::EventCreated)) + .await + .expect("active audit append succeeds"); + assert_eq!(entry.seq, 1); + } + fn new_entry(community_id: Uuid, action: AuditAction) -> NewAuditEntry { NewAuditEntry { community_id: CommunityId::from_uuid(community_id), diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index f8adfefab7..2f306c35d0 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -60,8 +60,9 @@ pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionO pub const DB_CONNECTION_RAW_SERIES_PER_POD: usize = runtime::CONNECTION_RAW_SERIES_PER_POD; pub(crate) use runtime::{ begin_community_event_write_transaction, - begin_community_event_write_transaction_with_legacy_metrics, insert_mentions_in_transaction, - observability, route_proof, ReadSessionInner, RouteDecision, RoutePredicate, + begin_community_event_write_transaction_with_legacy_metrics, begin_community_write_transaction, + insert_mentions_in_transaction, observability, route_proof, ReadSessionInner, RouteDecision, + RoutePredicate, }; pub use store::{ admin_moderation, allowlist, api_token, archived_identities, artifact, channel, diff --git a/crates/buzz-db/src/runtime/admitted_tx.rs b/crates/buzz-db/src/runtime/admitted_tx.rs index fb0bb4f41f..eb4607be25 100644 --- a/crates/buzz-db/src/runtime/admitted_tx.rs +++ b/crates/buzz-db/src/runtime/admitted_tx.rs @@ -1,4 +1,4 @@ -//! Community-admitted event-write transactions. +//! Community-admitted write transactions. use std::collections::BTreeSet; @@ -9,12 +9,12 @@ use uuid::Uuid; use crate::deletion::{DeletionStore, ServingWriteLease}; use crate::Result; -/// An event-write transaction that has passed community admission. +/// A transaction that has passed community write admission. /// /// Its only constructors take the shared community admission lock (or /// validate a serving write lease) on the transaction they wrap, for the /// community they record, so construction and admission are one step. -/// Event-write helpers take `&mut AdmittedTx` and read the community from it, +/// Write helpers take `&mut AdmittedTx` and read the community from it, /// so the compiler rejects a raw [`sqlx::Transaction`] or a transaction /// admitted for a different community. /// @@ -154,7 +154,7 @@ impl AdmittedTx { } /// Commit the transaction. This is the only commit path for admitted - /// event writes. + /// writes. /// /// Refreshes the TTL of every channel that received an event first, as /// the last statement before COMMIT. diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 7be9402da4..266003f5a0 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -132,9 +132,9 @@ pub(crate) async fn insert_mentions_in_transaction( Ok(()) } -/// Start a tenant-local event-write transaction and take the shared community +/// Start a tenant-local write transaction and take the shared community /// deletion lock before any serving mutation. -async fn begin_community_event_write_transaction_with_metric_population( +async fn begin_community_write_transaction_with_metric_population( pool: &PgPool, community: CommunityId, operation: observability::WriterOperation, @@ -163,7 +163,7 @@ pub(crate) async fn begin_community_event_write_transaction( community: CommunityId, operation: observability::WriterOperation, ) -> Result { - begin_community_event_write_transaction_with_metric_population( + begin_community_write_transaction_with_metric_population( pool, community, operation, @@ -177,7 +177,7 @@ pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics( community: CommunityId, operation: observability::WriterOperation, ) -> Result { - begin_community_event_write_transaction_with_metric_population( + begin_community_write_transaction_with_metric_population( pool, community, operation, @@ -186,6 +186,22 @@ pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics( .await } +/// Begin a community write before any tenant mutation or row lock. The +/// admission lock is held by the returned transaction until commit or rollback. +pub(crate) async fn begin_community_write_transaction( + pool: &PgPool, + community: CommunityId, + operation: observability::WriterOperation, +) -> Result { + begin_community_write_transaction_with_metric_population( + pool, + community, + operation, + CommunityEventWriteMetricPopulation::TypedOnly, + ) + .await +} + /// Database handle. Clone is cheap (Arc-backed pool). #[derive(Clone, Debug)] pub struct Db { diff --git a/crates/buzz-db/src/store/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs new file mode 100644 index 0000000000..6f1d8cab68 --- /dev/null +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -0,0 +1,796 @@ +//! Live writer-admission regressions. A one-connection pool runs with ordinary +//! PostgreSQL triggers disabled, so a missing application check is observable. + +use crate::{Db, DbConfig, DbError}; +use buzz_core::CommunityId; +use uuid::Uuid; + +async fn fixture() -> (Db, [CommunityId; 3]) { + let config = DbConfig { + database_url: crate::test_support::database_url(), + max_connections: 1, + min_connections: 1, + ..DbConfig::default() + }; + let db = Db::new(&config).await.expect("connect test database"); + let pool = &db.pool; + let mut communities = [CommunityId::from_uuid(Uuid::nil()); 3]; + for community in &mut communities { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("admission-{id}.example")) + .execute(pool) + .await + .expect("create community"); + *community = CommunityId::from_uuid(id); + } + crate::test_support::set_deletion_state(pool, *communities[1].as_uuid(), "quiescing").await; + crate::test_support::set_deletion_state(pool, *communities[2].as_uuid(), "fenced").await; + + // One physical session serves every test API call. PostgreSQL's replica + // role skips the 32 community-write triggers as well as other ordinary + // triggers; the application precheck remains fully active. + sqlx::query("SET session_replication_role = replica") + .execute(pool) + .await + .expect("disable ordinary triggers on the test session"); + let role: String = sqlx::query_scalar("SHOW session_replication_role") + .fetch_one(pool) + .await + .expect("read replication role"); + assert_eq!(role, "replica"); + // Witness that the database backstop is truly bypassed: this raw fenced + // insert would be rejected with the community-write trigger enabled. + sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)") + .bind(communities[2].as_uuid()) + .bind(vec![0xfe_u8; 32]) + .execute(pool) + .await + .expect("raw fenced insert succeeds with triggers disabled"); + (db, communities) +} + +fn assert_fenced(error: DbError) { + assert!( + matches!(&error, DbError::AccessDenied(message) if message.contains("write-fenced")), + "expected application admission rejection, got {error:?}" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn api_token_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::api_token::create_api_token( + &db.pool, + *community.as_uuid(), + &[1; 32], + &[2; 32], + "test", + &[], + None, + None, + ) + .await + .expect_err("token create must reject"), + ); + assert_fenced( + super::api_token::revoke_all_tokens(&db.pool, *community.as_uuid(), &[2; 32], &[2; 32]) + .await + .expect_err("token revoke must reject"), + ); + } + super::api_token::create_api_token( + &db.pool, + *communities[0].as_uuid(), + &[3; 32], + &[2; 32], + "active", + &[], + None, + None, + ) + .await + .expect("active token create succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn archived_identity_writers_require_application_admission() { + let (db, communities) = fixture().await; + let pubkey = "a".repeat(64); + let actor = "b".repeat(64); + let event = "c".repeat(64); + for community in &communities[1..] { + assert_fenced( + super::archived_identities::archive( + &db.pool, *community, &pubkey, "self", &actor, None, None, &event, + ) + .await + .expect_err("archive must reject"), + ); + assert_fenced( + super::archived_identities::unarchive(&db.pool, *community, &pubkey) + .await + .expect_err("unarchive must reject"), + ); + } + assert!(super::archived_identities::archive( + &db.pool, + communities[0], + &pubkey, + "self", + &actor, + None, + None, + &event, + ) + .await + .expect("active archive succeeds")); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn allowlist_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + db.add_to_allowlist(*community, &[1; 32], &[2; 32], None) + .await + .expect_err("allowlist add must reject"), + ); + assert_fenced( + db.remove_from_allowlist(*community, &[1; 32]) + .await + .expect_err("allowlist remove must reject"), + ); + } + assert!(db + .add_to_allowlist(communities[0], &[1; 32], &[2; 32], None) + .await + .expect("active allowlist add succeeds")); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn git_name_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::git_repo::reserve_repo_name(&db.pool, *community, "repo", "owner") + .await + .expect_err("repo reserve must reject"), + ); + assert_fenced( + super::git_repo::release_repo_name(&db.pool, *community, "repo", "owner") + .await + .expect_err("repo release must reject"), + ); + } + assert_eq!( + super::git_repo::reserve_repo_name(&db.pool, communities[0], "repo", "owner") + .await + .expect("active repo reserve succeeds"), + super::git_repo::ReserveOutcome::Reserved + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn relay_member_writers_require_application_admission() { + let (db, communities) = fixture().await; + let pubkey = "d".repeat(64); + for community in &communities[1..] { + assert_fenced( + super::relay_members::add_relay_member(&db.pool, *community, &pubkey, "member", None) + .await + .expect_err("member add must reject"), + ); + assert_fenced( + super::relay_members::remove_relay_member(&db.pool, *community, &pubkey) + .await + .expect_err("member remove must reject"), + ); + } + assert!(super::relay_members::add_relay_member( + &db.pool, + communities[0], + &pubkey, + "member", + None, + ) + .await + .expect("active member add succeeds")); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn relay_invite_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::relay_invite::mint_relay_invite( + &db.pool, + *community, + &"e".repeat(64), + 3600, + Some(1), + ) + .await + .expect_err("invite mint must reject"), + ); + assert_fenced( + super::relay_invite::claim_relay_invite( + &db.pool, + *community, + &[3; 32], + &"f".repeat(64), + None, + ) + .await + .expect_err("invite claim must reject before row lock"), + ); + } + super::relay_invite::mint_relay_invite( + &db.pool, + communities[0], + &"e".repeat(64), + 3600, + Some(1), + ) + .await + .expect("active invite mint succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn user_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::user::ensure_user(&db.pool, *community, &[1; 32]) + .await + .expect_err("user ensure must reject"), + ); + assert_fenced( + super::user::update_user_profile( + &db.pool, + *community, + &[1; 32], + Some("name"), + None, + None, + None, + ) + .await + .expect_err("profile update must reject"), + ); + } + assert!(super::user::ensure_user(&db.pool, communities[0], &[1; 32]) + .await + .expect("active user ensure succeeds")); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn moderation_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::moderation::insert_report( + &db.pool, + *community, + super::moderation::NewReport { + report_event_id: &[3; 32], + reporter_pubkey: &[1; 32], + target: super::moderation::ReportTarget::Pubkey(vec![2; 32]), + channel_id: None, + report_type: "spam", + note: None, + }, + ) + .await + .expect_err("moderation report must reject"), + ); + assert_fenced( + super::moderation::insert_action( + &db.pool, + *community, + super::moderation::NewAction { + actor_pubkey: &[1; 32], + action: "ban", + target_pubkey: Some(&[2; 32]), + target_event_id: None, + channel_id: None, + reason_code: None, + public_reason: None, + private_reason: None, + matched_principal: None, + actor_authority: None, + }, + ) + .await + .expect_err("moderation action must reject"), + ); + assert_fenced( + super::moderation::ban_member(&db.pool, *community, &[1; 32], &[2; 32], None, None) + .await + .expect_err("ban must reject"), + ); + assert_fenced( + super::moderation::unban_member(&db.pool, *community, &[1; 32], &[2; 32]) + .await + .expect_err("unban must reject"), + ); + } + super::moderation::ban_member(&db.pool, communities[0], &[1; 32], &[2; 32], None, None) + .await + .expect("active ban succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn workflow_writers_require_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::workflow::create_workflow( + &db.pool, *community, None, &[1; 32], "workflow", "{}", &[2; 32], + ) + .await + .expect_err("workflow create must reject"), + ); + assert_fenced( + super::workflow::create_workflow_run(&db.pool, *community, Uuid::new_v4(), None, None) + .await + .expect_err("workflow run create must reject"), + ); + assert_fenced( + super::workflow::claim_scheduled_workflow_fire( + &db.pool, + *community, + Uuid::new_v4(), + chrono::Utc::now(), + ) + .await + .expect_err("schedule claim must reject"), + ); + assert_fenced( + super::workflow::attach_scheduled_workflow_run( + &db.pool, + *community, + Uuid::new_v4(), + chrono::Utc::now(), + Uuid::new_v4(), + ) + .await + .expect_err("schedule attach must reject"), + ); + assert_fenced( + super::workflow::create_approval( + &db.pool, + super::workflow::CreateApprovalParams { + community_id: *community, + token: "fenced-approval", + workflow_id: Uuid::new_v4(), + run_id: Uuid::new_v4(), + step_id: "step", + step_index: 0, + approver_spec: "owner", + expires_at: chrono::Utc::now() + chrono::Duration::hours(1), + }, + ) + .await + .expect_err("approval create must reject"), + ); + assert_fenced( + super::workflow::update_approval_by_stored_hash( + &db.pool, + *community, + &[4; 32], + super::workflow::ApprovalStatus::Granted, + Some(&[1; 32]), + None, + ) + .await + .expect_err("approval update must reject"), + ); + } + super::workflow::create_workflow( + &db.pool, + communities[0], + None, + &[1; 32], + "active", + "{}", + &[2; 32], + ) + .await + .expect("active workflow create succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn push_writers_require_application_admission() { + let (db, communities) = fixture().await; + let version = super::push::LeaseVersion { + source_event_id: &[3; 32], + source_created_at: chrono::Utc::now().timestamp(), + generation: 1, + expires_at: chrono::Utc::now().timestamp() + 3600, + }; + for community in &communities[1..] { + assert_fenced( + super::push::revoke_lease(&db.pool, *community, &[1; 32], "install", version) + .await + .expect_err("lease revoke must reject"), + ); + assert_fenced( + super::push::complete_match_batch(&db.pool, *community, Uuid::new_v4(), &[vec![2; 32]]) + .await + .expect_err("matcher complete must reject"), + ); + assert_fenced( + super::push::complete_wake(&db.pool, *community, Uuid::new_v4(), Uuid::new_v4()) + .await + .expect_err("wake complete must reject"), + ); + } + super::push::revoke_lease(&db.pool, communities[0], &[1; 32], "install", version) + .await + .expect("active lease revoke succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_and_membership_writers_require_application_admission() { + let (db, communities) = fixture().await; + let creator = [1_u8; 32]; + for community in &communities[1..] { + assert_fenced( + super::channel::create_channel( + &db.pool, + *community, + "fenced", + super::channel::ChannelType::Stream, + super::channel::ChannelVisibility::Open, + None, + &creator, + None, + ) + .await + .expect_err("channel create must reject"), + ); + assert_fenced( + super::channel::set_canvas(&db.pool, *community, Uuid::new_v4(), Some("text")) + .await + .expect_err("canvas update must reject"), + ); + assert_fenced( + super::channel_members::add_member( + &db.pool, + *community, + Uuid::new_v4(), + &[2; 32], + super::channel::MemberRole::Member, + None, + ) + .await + .expect_err("membership add must reject"), + ); + } + let channel = super::channel::create_channel( + &db.pool, + communities[0], + "active", + super::channel::ChannelType::Stream, + super::channel::ChannelVisibility::Open, + None, + &creator, + None, + ) + .await + .expect("active channel create succeeds"); + super::channel_members::add_member( + &db.pool, + communities[0], + channel.id, + &[2; 32], + super::channel::MemberRole::Member, + None, + ) + .await + .expect("active membership add succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dm_writers_require_application_admission() { + let (db, communities) = fixture().await; + let participants: [&[u8]; 2] = [&[1; 32], &[2; 32]]; + for community in &communities[1..] { + assert_fenced( + super::dm::create_dm(&db.pool, *community, &participants, participants[0]) + .await + .expect_err("DM create must reject"), + ); + assert_fenced( + super::dm::hide_dm(&db.pool, *community, Uuid::new_v4(), participants[0]) + .await + .expect_err("DM hide must reject"), + ); + } + super::dm::create_dm(&db.pool, communities[0], &participants, participants[0]) + .await + .expect("active DM create succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn thread_metadata_writer_requires_application_admission() { + let (db, communities) = fixture().await; + let timestamp = chrono::Utc::now(); + for community in &communities[1..] { + assert_fenced( + super::thread::insert_thread_metadata( + &db.pool, + *community, + &[1; 32], + timestamp, + Uuid::new_v4(), + None, + None, + None, + None, + 0, + false, + ) + .await + .expect_err("thread metadata must reject"), + ); + } + let channel = super::channel::create_channel( + &db.pool, + communities[0], + "thread", + super::channel::ChannelType::Stream, + super::channel::ChannelVisibility::Open, + None, + &[1; 32], + None, + ) + .await + .expect("active channel"); + super::thread::insert_thread_metadata( + &db.pool, + communities[0], + &[1; 32], + timestamp, + channel.id, + None, + None, + None, + None, + 0, + false, + ) + .await + .expect("active thread metadata insert succeeds"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn relay_admin_domain_writer_requires_application_admission() { + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_fenced( + super::relay_admin_actions::deploy_kick_member( + &db.pool, + *community, + Uuid::new_v4(), + &[1; 32], + &[2; 32], + ) + .await + .expect_err("deployment kick must reject"), + ); + } + let channel = super::channel::create_channel( + &db.pool, + communities[0], + "kick", + super::channel::ChannelType::Stream, + super::channel::ChannelVisibility::Open, + None, + &[1; 32], + None, + ) + .await + .expect("active channel"); + assert!(matches!( + super::relay_admin_actions::deploy_kick_member( + &db.pool, + communities[0], + channel.id, + &[1; 32], + &[2; 32], + ) + .await + .expect("active kick succeeds"), + super::relay_admin_actions::KickResult::Removed + )); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn personal_read_writer_requires_application_admission() { + use nostr::{EventBuilder, Keys, Kind}; + let (db, communities) = fixture().await; + let actor = Keys::generate(); + let target = super::personal_read::ReadTarget { + channel_id: Uuid::new_v4(), + root_id: None, + }; + let intent = super::personal_read::ReadIntent::MarkThrough { + target, + message_id: "a".repeat(64), + }; + for community in &communities[1..] { + assert_fenced( + db.apply_personal_read_intent(*community, &actor.public_key(), &intent) + .await + .expect_err("personal read must reject before account lock"), + ); + } + let channel = super::channel::create_channel( + &db.pool, + communities[0], + "reads", + super::channel::ChannelType::Stream, + super::channel::ChannelVisibility::Open, + None, + &actor.public_key().to_bytes(), + None, + ) + .await + .expect("active channel"); + let event = EventBuilder::new(Kind::Custom(9), "read") + .sign_with_keys(&Keys::generate()) + .expect("sign event"); + db.insert_event(communities[0], &event, Some(channel.id)) + .await + .expect("insert active event"); + let intent = super::personal_read::ReadIntent::MarkThrough { + target: super::personal_read::ReadTarget { + channel_id: channel.id, + root_id: None, + }, + message_id: event.id.to_hex(), + }; + assert_eq!( + db.apply_personal_read_intent(communities[0], &actor.public_key(), &intent) + .await + .expect("active personal read succeeds"), + super::personal_read::IntentOutcome::Applied + ); +} + +async fn count_relay_members(db: &Db, community: CommunityId) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM relay_members WHERE community_id = $1") + .bind(community.as_uuid()) + .fetch_one(&db.pool) + .await + .expect("count relay members") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn allowlist_backfill_requires_application_admission() { + let (db, communities) = fixture().await; + for community in &communities { + sqlx::query("INSERT INTO pubkey_allowlist (community_id, pubkey) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(vec![0x5a_u8; 32]) + .execute(&db.pool) + .await + .expect("seed legacy allowlist row"); + } + for community in &communities[1..] { + assert_fenced( + super::relay_members::backfill_from_allowlist(&db.pool, *community) + .await + .expect_err("backfill into a write-fenced community must reject"), + ); + assert_eq!(count_relay_members(&db, *community).await, 0); + } + assert_eq!( + super::relay_members::backfill_from_allowlist(&db.pool, communities[0]) + .await + .expect("active backfill succeeds"), + 1 + ); + assert_eq!(count_relay_members(&db, communities[0]).await, 1); + + // A community that already has members needs no write, so startup does + // not fail on admission even while the community is quiescing. + sqlx::query("INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')") + .bind(communities[1].as_uuid()) + .bind("e".repeat(64)) + .execute(&db.pool) + .await + .expect("seed existing member"); + assert_eq!( + super::relay_members::backfill_from_allowlist(&db.pool, communities[1]) + .await + .expect("no-op backfill does not need admission"), + 0 + ); + assert_eq!(count_relay_members(&db, communities[1]).await, 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn empty_allowlist_backfill_skips_admission() { + // No members and no legacy allowlist rows: there is nothing to write, so + // startup must not fail on admission while the community is fenced. + let (db, communities) = fixture().await; + for community in &communities[1..] { + assert_eq!( + super::relay_members::backfill_from_allowlist(&db.pool, *community) + .await + .expect("empty-allowlist backfill does not need admission"), + 0 + ); + assert_eq!(count_relay_members(&db, *community).await, 0); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn owner_mutations_report_pending_deletion_when_admission_refuses() { + let (db, communities) = fixture().await; + let quiescing = communities[1]; + sqlx::query( + "INSERT INTO community_deletion_requests (community_id, community_host, requested_by) \ + SELECT id, host, 'admission-test' FROM communities WHERE id = $1", + ) + .bind(quiescing.as_uuid()) + .execute(&db.pool) + .await + .expect("seed deletion request"); + let owner = "f".repeat(64); + let next_owner = "1".repeat(64); + + assert_eq!( + super::relay_members::transfer_ownership(&db.pool, quiescing, &next_owner, &owner) + .await + .expect("transfer returns a result"), + super::relay_members::TransferResult::DeletionPending + ); + assert_eq!( + db.provision_owner(quiescing, &owner) + .await + .expect("provision returns a result"), + super::relay_members::ProvisionOwnerResult::DeletionPending + ); + + // Without a deletion request the refusal stays a lifecycle conflict. + let fenced = communities[2]; + assert_eq!( + super::relay_members::transfer_ownership(&db.pool, fenced, &next_owner, &owner) + .await + .expect("transfer returns a result"), + super::relay_members::TransferResult::LifecycleConflict + ); + assert_eq!( + db.provision_owner(fenced, &owner) + .await + .expect("provision returns a result"), + super::relay_members::ProvisionOwnerResult::LifecycleConflict + ); + assert_eq!(count_relay_members(&db, quiescing).await, 0); + assert_eq!(count_relay_members(&db, fenced).await, 0); +} diff --git a/crates/buzz-db/src/store/allowlist.rs b/crates/buzz-db/src/store/allowlist.rs index 2b72780a62..c493cab22f 100644 --- a/crates/buzz-db/src/store/allowlist.rs +++ b/crates/buzz-db/src/store/allowlist.rs @@ -70,8 +70,9 @@ impl Db { added_by: &[u8], note: Option<&str>, ) -> Result { - let mut connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( &self.pool, + community, crate::observability::WriterOperation::Authorization, ) .await?; @@ -83,8 +84,9 @@ impl Db { .bind(pubkey) .bind(added_by) .bind(note) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -95,8 +97,9 @@ impl Db { community: CommunityId, pubkey: &[u8], ) -> Result { - let mut connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( &self.pool, + community, crate::observability::WriterOperation::Authorization, ) .await?; @@ -104,8 +107,9 @@ impl Db { sqlx::query("DELETE FROM pubkey_allowlist WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } diff --git a/crates/buzz-db/src/store/api_token.rs b/crates/buzz-db/src/store/api_token.rs index 41d4dcbad2..6c0eab0bf2 100644 --- a/crates/buzz-db/src/store/api_token.rs +++ b/crates/buzz-db/src/store/api_token.rs @@ -38,6 +38,12 @@ pub async fn create_api_token( }) .transpose()?; + let mut tx = crate::begin_community_write_transaction( + pool, + CommunityId::from_uuid(community_id), + crate::observability::WriterOperation::Authentication, + ) + .await?; sqlx::query( r#" INSERT INTO api_tokens @@ -53,9 +59,9 @@ pub async fn create_api_token( .bind(&scopes_json) .bind(&channel_ids_json) .bind(expires_at) - .execute(pool) + .execute(tx.conn()) .await?; - + tx.commit().await?; Ok(id) } @@ -94,6 +100,12 @@ pub async fn create_api_token_if_under_limit( // Conditional INSERT: only inserts if active (non-revoked, non-expired) token count < 10 // **for this (community, owner) pair**. The subquery and insert execute atomically -- // no separate count + insert race. + let mut tx = crate::begin_community_write_transaction( + pool, + CommunityId::from_uuid(community_id), + crate::observability::WriterOperation::Authentication, + ) + .await?; let result = sqlx::query( r#" INSERT INTO api_tokens @@ -118,8 +130,9 @@ pub async fn create_api_token_if_under_limit( .bind(&channel_ids_json) .bind(expires_at) .bind(owner_pubkey) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; if result.rows_affected() == 0 { // Limit exceeded -- the WHERE clause prevented the INSERT. @@ -279,6 +292,12 @@ pub async fn revoke_token( owner_pubkey: &[u8], revoked_by: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + CommunityId::from_uuid(community_id), + crate::observability::WriterOperation::Authentication, + ) + .await?; let result = sqlx::query( r#" UPDATE api_tokens @@ -293,9 +312,9 @@ pub async fn revoke_token( .bind(community_id) .bind(id) .bind(owner_pubkey) - .execute(pool) + .execute(tx.conn()) .await?; - + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -309,6 +328,12 @@ pub async fn revoke_all_tokens( owner_pubkey: &[u8], revoked_by: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + CommunityId::from_uuid(community_id), + crate::observability::WriterOperation::Authentication, + ) + .await?; let result = sqlx::query( r#" UPDATE api_tokens @@ -321,9 +346,9 @@ pub async fn revoke_all_tokens( .bind(revoked_by) .bind(community_id) .bind(owner_pubkey) - .execute(pool) + .execute(tx.conn()) .await?; - + tx.commit().await?; Ok(result.rows_affected()) } @@ -504,13 +529,20 @@ impl Db { /// Record a token usage (update `last_used_at`), scoped to community. #[datastore_span(name = "touch_api_token", system = "postgresql")] pub async fn touch_api_token(&self, community_id: CommunityId, hash: &[u8]) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + &self.pool, + community_id, + crate::observability::WriterOperation::Authentication, + ) + .await?; sqlx::query( "UPDATE api_tokens SET last_used_at = NOW() WHERE community_id = $1 AND token_hash = $2", ) .bind(community_id.as_uuid()) .bind(hash) - .execute(&self.pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } diff --git a/crates/buzz-db/src/store/archived_identities.rs b/crates/buzz-db/src/store/archived_identities.rs index b1636de31a..986985bc46 100644 --- a/crates/buzz-db/src/store/archived_identities.rs +++ b/crates/buzz-db/src/store/archived_identities.rs @@ -64,8 +64,9 @@ pub async fn archive( replaced_by: Option<&str>, request_event_id: &str, ) -> Result { - let mut connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community_id, crate::observability::WriterOperation::EventWrite, ) .await?; @@ -82,9 +83,9 @@ pub async fn archive( .bind(reason) .bind(replaced_by) .bind(request_event_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; - + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -93,8 +94,9 @@ pub async fn archive( /// Returns `true` if a row was deleted, `false` if the identity was not archived /// in that community. pub async fn unarchive(pool: &PgPool, community_id: CommunityId, pubkey: &str) -> Result { - let mut connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community_id, crate::observability::WriterOperation::EventWrite, ) .await?; @@ -102,9 +104,9 @@ pub async fn unarchive(pool: &PgPool, community_id: CommunityId, pubkey: &str) - sqlx::query("DELETE FROM archived_identities WHERE community_id = $1 AND pubkey = $2") .bind(community_id.as_uuid()) .bind(pubkey) - .execute(&mut *connection) + .execute(tx.conn()) .await?; - + tx.commit().await?; Ok(result.rows_affected() > 0) } diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 46375595cf..13a93263e9 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -29,27 +29,6 @@ pub use crate::channel_members::{ ChannelIdentity, LargeChannelRoster, LockedMemberSnapshot, MemberRecord, UserRecord, }; -async fn begin_event_write_transaction( - pool: &PgPool, -) -> Result> { - let connection = crate::observability::acquire_writer( - pool, - crate::observability::WriterOperation::EventWrite, - ) - .await?; - Ok(sqlx::Transaction::begin(connection, None).await?) -} - -async fn acquire_event_write_connection( - pool: &PgPool, -) -> Result> { - Ok(crate::observability::acquire_writer( - pool, - crate::observability::WriterOperation::EventWrite, - ) - .await?) -} - /// A channel row as returned from the database. #[derive(Debug, Clone)] pub struct ChannelRecord { @@ -135,7 +114,12 @@ pub async fn create_channel( let id = Uuid::new_v4(); - let mut tx = begin_event_write_transaction(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; sqlx::query( r#" @@ -152,7 +136,7 @@ pub async fn create_channel( .bind(description) .bind(created_by) .bind(ttl_seconds) - .execute(&mut *tx) + .execute(tx.conn()) .await?; sqlx::query( @@ -169,7 +153,7 @@ pub async fn create_channel( .bind(id) .bind(created_by) .bind(created_by) - .execute(&mut *tx) + .execute(tx.conn()) .await?; let row = sqlx::query( @@ -186,7 +170,7 @@ pub async fn create_channel( ) .bind(community_id.as_uuid()) .bind(id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let record = row_to_channel_record(row)?; @@ -228,7 +212,12 @@ pub async fn create_channel_with_id( return Err(DbError::InvalidData("channel name is required".into())); } - let mut tx = begin_event_write_transaction(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let rows_affected = sqlx::query( r#" @@ -246,7 +235,7 @@ pub async fn create_channel_with_id( .bind(description) .bind(created_by) .bind(ttl_seconds) - .execute(&mut *tx) + .execute(tx.conn()) .await? .rows_affected(); @@ -268,7 +257,7 @@ pub async fn create_channel_with_id( .bind(channel_id) .bind(created_by) .bind(created_by) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } @@ -286,7 +275,7 @@ pub async fn create_channel_with_id( ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let record = row_to_channel_record(row)?; @@ -361,17 +350,24 @@ pub async fn set_canvas( channel_id: Uuid, canvas: Option<&str>, ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let rows = sqlx::query( "UPDATE channels SET canvas = $1 WHERE community_id = $2 AND id = $3 AND deleted_at IS NULL", ) .bind(canvas) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(pool) + .execute(tx.conn()) .await?; if rows.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } + tx.commit().await?; Ok(()) } @@ -593,22 +589,33 @@ pub async fn update_channel( // this transition — whose own deadline reset is then the latest word. // Non-TTL updates don't touch the fast path and skip the lock. if updates.ttl_seconds.is_some() { - let mut tx = begin_event_write_transaction(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") .bind(channel_ttl_lock_key(community_id, channel_id)) - .execute(&mut *tx) + .execute(tx.conn()) .await?; - let result = q.execute(&mut *tx).await?; + let result = q.execute(tx.conn()).await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } tx.commit().await?; } else { - let mut connection = acquire_event_write_connection(pool).await?; - let result = q.execute(&mut *connection).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let result = q.execute(tx.conn()).await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } + tx.commit().await?; } get_channel_with_operation( @@ -628,7 +635,12 @@ pub async fn set_topic( topic: &str, set_by: &[u8], ) -> Result<()> { - let mut connection = acquire_event_write_connection(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let result = sqlx::query( "UPDATE channels SET topic = $1, topic_set_by = $2, topic_set_at = NOW() \ WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL", @@ -637,11 +649,12 @@ pub async fn set_topic( .bind(set_by) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } + tx.commit().await?; Ok(()) } @@ -653,7 +666,12 @@ pub async fn set_purpose( purpose: &str, set_by: &[u8], ) -> Result<()> { - let mut connection = acquire_event_write_connection(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let result = sqlx::query( "UPDATE channels SET purpose = $1, purpose_set_by = $2, purpose_set_at = NOW() \ WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL", @@ -662,11 +680,12 @@ pub async fn set_purpose( .bind(set_by) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } + tx.commit().await?; Ok(()) } @@ -679,14 +698,19 @@ pub async fn archive_channel( community_id: CommunityId, channel_id: Uuid, ) -> Result<()> { - let mut connection = acquire_event_write_connection(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; // First check: does the channel exist and what is its state? let row = sqlx::query( "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; match row { @@ -707,9 +731,10 @@ pub async fn archive_channel( ) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } @@ -722,14 +747,19 @@ pub async fn unarchive_channel( community_id: CommunityId, channel_id: Uuid, ) -> Result<()> { - let mut connection = acquire_event_write_connection(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; // First check: does the channel exist and what is its state? let row = sqlx::query( "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; match row { @@ -752,9 +782,10 @@ pub async fn unarchive_channel( ) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } @@ -767,15 +798,21 @@ pub async fn soft_delete_channel( community_id: CommunityId, channel_id: Uuid, ) -> Result { - let mut connection = acquire_event_write_connection(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let result = sqlx::query( "UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", ) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 7ff5146775..1994de14ed 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -4,7 +4,7 @@ //! roster snapshots hold that same lock through replacement publication. use chrono::{DateTime, Utc}; -use sqlx::{PgConnection, PgPool, Postgres, Row, Transaction}; +use sqlx::{PgConnection, PgPool, Row}; use uuid::Uuid; use crate::channel::{row_to_channel_record, ChannelRecord}; @@ -539,18 +539,18 @@ pub async fn add_member( ))); } - let connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community_id, crate::observability::WriterOperation::EventWrite, ) .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; // First statement: serialize the whole role-check / owner-count / upsert // sequence against concurrent membership writes on this channel. - acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?; + acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; - let channel = get_channel_tx(&mut tx, community_id, channel_id).await?; + let channel = get_channel_tx(tx.conn(), community_id, channel_id).await?; let effective_role = if channel.visibility == "private" { let inviter = invited_by.ok_or_else(|| { @@ -561,7 +561,7 @@ pub async fn add_member( let is_creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice(); if !is_creator_bootstrap { - let inviter_role_str = get_active_role_tx(&mut tx, community_id, channel_id, inviter) + let inviter_role_str = get_active_role_tx(tx.conn(), community_id, channel_id, inviter) .await? .ok_or_else(|| { DbError::AccessDenied("inviter is not an active member".to_string()) @@ -587,7 +587,7 @@ pub async fn add_member( // elevated roles. Self-join always gets Member. if role.is_elevated() { let granter_role = match invited_by { - Some(inv) => get_active_role_tx(&mut tx, community_id, channel_id, inv).await?, + Some(inv) => get_active_role_tx(tx.conn(), community_id, channel_id, inv).await?, None => None, }; match granter_role.as_deref() { @@ -619,10 +619,12 @@ pub async fn add_member( // current authority from a removed row would make soft-deleted ownership a // resurrection token: an owner removed by another owner could self-rejoin // via kind:9021 (`Member, None`) and silently regain ownership. - let current_role = get_active_role_tx(&mut tx, community_id, channel_id, pubkey).await?; + let current_role = get_active_role_tx(tx.conn(), community_id, channel_id, pubkey).await?; if let Some(current_role) = current_role.filter(|r| r != effective_role.as_str()) { let actor_role = match invited_by { - Some(inviter) => get_active_role_tx(&mut tx, community_id, channel_id, inviter).await?, + Some(inviter) => { + get_active_role_tx(tx.conn(), community_id, channel_id, inviter).await? + } None => None, }; let actor_role: Option = actor_role.and_then(|r| r.parse().ok()); @@ -642,7 +644,7 @@ pub async fn add_member( ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let owner_count: i64 = row.try_get("cnt")?; if owner_count <= 1 { @@ -668,7 +670,7 @@ pub async fn add_member( .bind(pubkey) .bind(effective_role.as_str()) .bind(invited_by) - .execute(&mut *tx) + .execute(tx.conn()) .await?; let row = sqlx::query( @@ -680,7 +682,7 @@ pub async fn add_member( .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let record = row_to_member_record(row)?; @@ -725,20 +727,20 @@ pub async fn remove_member( crate::user::is_agent_owner(pool, community_id, pubkey, actor_pubkey).await? }; - let connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community_id, crate::observability::WriterOperation::EventWrite, ) .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; // First statement: serialize the actor-role check, the last-owner count and // the UPDATE against concurrent membership writes on this channel (same key // as `add_member`). - acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?; + acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; if !is_self_remove { - let actor_role_str = get_active_role_tx(&mut tx, community_id, channel_id, actor_pubkey) + let actor_role_str = get_active_role_tx(tx.conn(), community_id, channel_id, actor_pubkey) .await? .ok_or_else(|| DbError::AccessDenied("actor is not an active member".to_string()))?; let actor_role: MemberRole = actor_role_str.parse().map_err(|_| { @@ -754,7 +756,7 @@ pub async fn remove_member( // Defense-in-depth: prevent removing the last owner regardless of caller. // Callers (REST handlers, NIP-29 handlers) also check this, but the DB // layer enforces it as the final safety net. - let target_role = get_active_role_tx(&mut tx, community_id, channel_id, pubkey).await?; + let target_role = get_active_role_tx(tx.conn(), community_id, channel_id, pubkey).await?; if target_role.as_deref() == Some("owner") { let row = sqlx::query( "SELECT COUNT(*) as cnt FROM channel_members \ @@ -762,7 +764,7 @@ pub async fn remove_member( ) .bind(community_id.as_uuid()) .bind(channel_id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let owner_count: i64 = row.try_get("cnt")?; if owner_count <= 1 { @@ -783,7 +785,7 @@ pub async fn remove_member( .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { @@ -900,7 +902,7 @@ pub struct MembershipRemovalFence { pub still_removed: bool, // Holds the advisory transaction lock. The caller commits via // `commit_disabling_workflows`; on drop without commit the tx rolls back. - tx: Transaction<'static, Postgres>, + tx: AdmittedTx, } impl MembershipRemovalFence { @@ -950,19 +952,27 @@ impl MembershipRemovalFence { /// lock (the transaction is implicitly rolled back if not committed). To /// durably write the workflow-disable and release the lock in one step, call /// [`Self::commit_disabling_workflows`]. +/// +/// Write admission is taken up front, before the membership lock, so the +/// lock order matches [`add_member`]. Taking it lazily in +/// [`Self::commit_disabling_workflows`] would invert that order. As a result, +/// a community that leaves `active` between the kick commit and this call +/// refuses the fence, and the caller skips its live side effects. That is +/// intended: the membership cache is already invalidated and fan-out +/// re-checks membership before delivery. pub async fn membership_removal_fence( pool: &PgPool, community_id: CommunityId, channel_id: Uuid, pubkey: &[u8], ) -> Result { - let connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community_id, crate::observability::WriterOperation::Authorization, ) .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; - acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?; + acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; let row = sqlx::query( "SELECT removed_at FROM channel_members \ @@ -971,7 +981,7 @@ pub async fn membership_removal_fence( .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; // Row absent → never joined or hard-deleted; treat as removed (safe to diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 2407535958..e084d8ed17 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -563,6 +563,12 @@ impl Db { return Ok(CreateCommunityWithOwnerResult::LimitReached); } + // The community row was just created in this transaction. Admit + // its first tenant row before inserting the initial owner. + self.deletion_store() + .guard_transaction(&mut tx, CommunityId::from_uuid(id)) + .await?; + sqlx::query( "INSERT INTO relay_members (community_id, pubkey, role, added_by) VALUES ($1, $2, 'owner', NULL)", ) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index eef0ac8b0b..d673c5d8d7 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -2955,27 +2955,14 @@ impl DeletionStore { } /// Take the shared community deletion lock inside an existing transaction. + /// + /// See [`guard_community_write`]. pub async fn guard_transaction( &self, tx: &mut Transaction<'_, Postgres>, community: CommunityId, ) -> Result<()> { - lock_community_deletion_shared(tx, community).await?; - let state: Option = sqlx::query_scalar( - "SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL", - ) - .bind(community.as_uuid()) - .fetch_optional(&mut **tx) - .await?; - match state.as_deref() { - Some("active") => Ok(()), - Some(other) => Err(DbError::AccessDenied(format!( - "community {community} is write-fenced ({other})" - ))), - None => Err(DbError::AccessDenied(format!( - "community {community} is missing or tombstoned" - ))), - } + guard_community_write(tx, community).await } /// Take the shared community deletion lock inside an existing transaction @@ -3342,6 +3329,36 @@ async fn lock_community_deletion( Ok(()) } +/// Admit a community write inside an existing transaction. +/// +/// Takes the shared community deletion lock, then requires the community to +/// be `active` and not tombstoned. Refusal is [`DbError::AccessDenied`]; +/// a lock timeout surfaces as the underlying `55P03` database error. The +/// lock is transaction-scoped, so commit, rollback, or a dropped transaction +/// releases it. Crates outside `buzz-db` that own their own transactions +/// (the audit chain) admit through this function. +pub async fn guard_community_write( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, +) -> Result<()> { + lock_community_deletion_shared(tx, community).await?; + let state: Option = sqlx::query_scalar( + "SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .fetch_optional(&mut **tx) + .await?; + match state.as_deref() { + Some("active") => Ok(()), + Some(other) => Err(DbError::AccessDenied(format!( + "community {community} is write-fenced ({other})" + ))), + None => Err(DbError::AccessDenied(format!( + "community {community} is missing or tombstoned" + ))), + } +} + pub(crate) async fn lock_community_deletion_shared( tx: &mut Transaction<'_, Postgres>, community: CommunityId, diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 89e4a0e522..8d51866e85 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -127,7 +127,12 @@ pub async fn create_dm( let hash = compute_participant_hash(participants); - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; // Idempotency check inside the transaction. let existing = sqlx::query( @@ -148,7 +153,7 @@ pub async fn create_dm( ) .bind(community_id.as_uuid()) .bind(hash.as_slice()) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; if let Some(row) = existing { @@ -177,7 +182,7 @@ pub async fn create_dm( .bind(&name) .bind(created_by) .bind(hash.as_slice()) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // Add all participants as members with role='member'. @@ -196,7 +201,7 @@ pub async fn create_dm( .bind(id) .bind(*pk) .bind(created_by) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } @@ -213,7 +218,7 @@ pub async fn create_dm( ) .bind(community_id.as_uuid()) .bind(id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let record = row_to_channel_record(row)?; @@ -402,6 +407,12 @@ pub async fn hide_dm( channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let result = sqlx::query( r#" UPDATE channel_members @@ -412,7 +423,7 @@ pub async fn hide_dm( .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) - .execute(pool) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { @@ -421,6 +432,7 @@ pub async fn hide_dm( ))); } + tx.commit().await?; Ok(()) } @@ -434,6 +446,12 @@ pub async fn unhide_dm( channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; sqlx::query( r#" UPDATE channel_members @@ -444,9 +462,10 @@ pub async fn unhide_dm( .bind(community_id.as_uuid()) .bind(channel_id) .bind(pubkey) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } diff --git a/crates/buzz-db/src/store/git_repo.rs b/crates/buzz-db/src/store/git_repo.rs index fac10c3f61..164b055a28 100644 --- a/crates/buzz-db/src/store/git_repo.rs +++ b/crates/buzz-db/src/store/git_repo.rs @@ -90,8 +90,9 @@ pub async fn reserve_repo_name( repo_id: &str, owner_pubkey: &str, ) -> Result { - let mut connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community, crate::observability::WriterOperation::Authorization, ) .await?; @@ -108,10 +109,11 @@ pub async fn reserve_repo_name( .bind(community.as_uuid()) .bind(repo_id) .bind(owner_pubkey) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; if inserted.is_some() { + tx.commit().await?; return Ok(ReserveOutcome::Reserved); } @@ -123,9 +125,9 @@ pub async fn reserve_repo_name( ) .bind(community.as_uuid()) .bind(repo_id) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; - + tx.commit().await?; match existing { Some(row) => { let holder: String = row @@ -183,8 +185,9 @@ pub async fn release_repo_name( repo_id: &str, owner_pubkey: &str, ) -> Result { - let mut connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community, crate::observability::WriterOperation::Authorization, ) .await?; @@ -195,8 +198,9 @@ pub async fn release_repo_name( .bind(community.as_uuid()) .bind(repo_id) .bind(owner_pubkey) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected()) } diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 91c4a6e282..efa46f1748 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -78,3 +78,6 @@ mod artifact_query; #[cfg(test)] mod artifact_postgres_tests; + +#[cfg(test)] +mod admission_postgres_tests; diff --git a/crates/buzz-db/src/store/moderation.rs b/crates/buzz-db/src/store/moderation.rs index cdcc339fe5..00bcb3af87 100644 --- a/crates/buzz-db/src/store/moderation.rs +++ b/crates/buzz-db/src/store/moderation.rs @@ -188,6 +188,12 @@ pub async fn insert_report( community: CommunityId, report: NewReport<'_>, ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let (target_kind, target_event_id, target_pubkey, target_blob_sha256) = match &report.target { ReportTarget::Event(id) => ("event", Some(id.as_slice()), None, None), ReportTarget::Pubkey(pubkey) => ("pubkey", None, Some(pubkey.as_slice()), None), @@ -223,8 +229,9 @@ pub async fn insert_report( .bind(report.report_type) .bind(report.note) .bind(initial_status) - .fetch_one(pool) + .fetch_one(tx.conn()) .await?; + tx.commit().await?; Ok(row.try_get("id")?) } @@ -313,6 +320,12 @@ pub async fn resolve_report( resolved_by: &[u8], action_id: Option, ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( r#" UPDATE moderation_reports @@ -325,8 +338,9 @@ pub async fn resolve_report( .bind(status) .bind(resolved_by) .bind(action_id) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -340,6 +354,12 @@ pub async fn ban_member( reason: Option<&str>, expires_at: Option>, ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; sqlx::query( r#" INSERT INTO community_bans ( @@ -358,8 +378,9 @@ pub async fn ban_member( .bind(expires_at) .bind(reason) .bind(actor) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } @@ -371,6 +392,12 @@ pub async fn unban_member( pubkey: &[u8], actor: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( r#" UPDATE community_bans @@ -383,8 +410,9 @@ pub async fn unban_member( .bind(community.as_uuid()) .bind(pubkey) .bind(actor) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -404,7 +432,12 @@ pub async fn unban_member_with_audit( actor: &[u8], actor_authority: &str, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( r#" @@ -418,7 +451,7 @@ pub async fn unban_member_with_audit( .bind(community.as_uuid()) .bind(pubkey) .bind(actor) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { @@ -438,7 +471,7 @@ pub async fn unban_member_with_audit( .bind(actor) .bind(pubkey) .bind(actor_authority) - .execute(&mut *tx) + .execute(tx.conn()) .await?; tx.commit().await?; @@ -456,7 +489,12 @@ pub async fn untimeout_member_with_audit( actor: &[u8], actor_authority: &str, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( r#" @@ -469,7 +507,7 @@ pub async fn untimeout_member_with_audit( .bind(community.as_uuid()) .bind(pubkey) .bind(actor) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { @@ -488,7 +526,7 @@ pub async fn untimeout_member_with_audit( .bind(actor) .bind(pubkey) .bind(actor_authority) - .execute(&mut *tx) + .execute(tx.conn()) .await?; tx.commit().await?; @@ -504,6 +542,12 @@ pub async fn timeout_member( muted_until: DateTime, reason: Option<&str>, ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; sqlx::query( r#" INSERT INTO community_bans ( @@ -521,8 +565,9 @@ pub async fn timeout_member( .bind(muted_until) .bind(reason) .bind(actor) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } @@ -534,6 +579,12 @@ pub async fn untimeout_member( pubkey: &[u8], actor: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( r#" UPDATE community_bans @@ -545,8 +596,9 @@ pub async fn untimeout_member( .bind(community.as_uuid()) .bind(pubkey) .bind(actor) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -704,6 +756,12 @@ pub async fn insert_action( community: CommunityId, action: NewAction<'_>, ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Authorization, + ) + .await?; let row = sqlx::query( r#" INSERT INTO moderation_actions ( @@ -725,8 +783,9 @@ pub async fn insert_action( .bind(action.private_reason) .bind(action.matched_principal) .bind(action.actor_authority.unwrap_or("community")) - .fetch_one(pool) + .fetch_one(tx.conn()) .await?; + tx.commit().await?; Ok(row.try_get("id")?) } diff --git a/crates/buzz-db/src/store/personal_read/postgres_tests.rs b/crates/buzz-db/src/store/personal_read/postgres_tests.rs index 136eb89148..cba0d992a4 100644 --- a/crates/buzz-db/src/store/personal_read/postgres_tests.rs +++ b/crates/buzz-db/src/store/personal_read/postgres_tests.rs @@ -258,7 +258,13 @@ async fn personal_read_intent_does_not_lock_shared_conversation_rows() { .execute(&pool) .await .unwrap(); - let mut held = pool.begin().await.unwrap(); + let mut held = crate::begin_community_write_transaction( + &pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .unwrap(); super::writes::lock_account(&mut held, community, &actor.public_key().to_bytes()) .await .unwrap(); diff --git a/crates/buzz-db/src/store/personal_read/writes.rs b/crates/buzz-db/src/store/personal_read/writes.rs index 9d4c4a6586..77df951f59 100644 --- a/crates/buzz-db/src/store/personal_read/writes.rs +++ b/crates/buzz-db/src/store/personal_read/writes.rs @@ -1,10 +1,10 @@ use super::model::*; use buzz_core::CommunityId; use chrono::{DateTime, Utc}; -use sqlx::{Acquire, PgConnection, Row}; +use sqlx::{PgConnection, Row}; use uuid::Uuid; -use crate::{observability, Db, Result}; +use crate::{observability, AdmittedTx, Db, Result}; pub(super) fn event_id(value: &str) -> Option> { if value.len() != 64 || value.bytes().any(|b| !b.is_ascii_hexdigit()) { @@ -26,7 +26,7 @@ pub(super) async fn deadlines(conn: &mut PgConnection) -> Result<()> { /// Start the account if this is the actor's first read intent, then serialize /// private frontier writes, never shared conversation rows. pub(super) async fn lock_account( - conn: &mut PgConnection, + tx: &mut AdmittedTx, community: CommunityId, actor: &[u8], ) -> Result<()> { @@ -37,14 +37,14 @@ pub(super) async fn lock_account( ) .bind(community.as_uuid()) .bind(actor) - .execute(&mut *conn) + .execute(tx.conn()) .await?; sqlx::query( "SELECT actor FROM personal_read_accounts WHERE community_id=$1 AND actor=$2 FOR UPDATE", ) .bind(community.as_uuid()) .bind(actor) - .fetch_one(conn) + .fetch_one(tx.conn()) .await?; Ok(()) } @@ -161,7 +161,7 @@ pub(super) async fn valid_target( /// Advance a monotone frontier. The anchor ID follows the greatest arrival; /// an equal arrival keeps the existing anchor. async fn frontier( - conn: &mut PgConnection, + tx: &mut AdmittedTx, community: CommunityId, actor: &[u8], target: &ReadTarget, @@ -179,12 +179,12 @@ async fn frontier( THEN EXCLUDED.through_message_id ELSE personal_read_frontiers.through_message_id END", ).bind(community.as_uuid()).bind(actor).bind(target.channel_id).bind(root) .bind(msg.received_at).bind(&msg.id) - .execute(&mut *conn).await?; + .execute(tx.conn()).await?; Ok(()) } pub(super) async fn apply( - conn: &mut PgConnection, + tx: &mut AdmittedTx, community: CommunityId, actor: &[u8], intent: &ReadIntent, @@ -194,11 +194,11 @@ pub(super) async fn apply( let Some(id) = event_id(message_id) else { return Ok(IntentOutcome::Invalid); }; - let Some(root) = valid_target(conn, community, actor, target).await? else { + let Some(root) = valid_target(tx.conn(), community, actor, target).await? else { return Ok(IntentOutcome::Blocked); }; let Some(msg) = message( - conn, + tx.conn(), community, target.channel_id, &id, @@ -214,7 +214,7 @@ pub(super) async fn apply( { return Ok(IntentOutcome::Blocked); } - frontier(conn, community, actor, target, &root, &msg).await?; + frontier(tx, community, actor, target, &root, &msg).await?; } } Ok(IntentOutcome::Applied) @@ -230,11 +230,13 @@ impl Db { actor: &nostr::PublicKey, intent: &ReadIntent, ) -> Result { - let mut conn = - observability::acquire_writer(&self.pool, observability::WriterOperation::EventWrite) - .await?; - let mut tx = conn.begin().await?; - deadlines(&mut tx).await?; + let mut tx = crate::begin_community_write_transaction( + &self.pool, + community, + observability::WriterOperation::EventWrite, + ) + .await?; + deadlines(tx.conn()).await?; let actor = actor.to_bytes(); lock_account(&mut tx, community, &actor).await?; let outcome = apply(&mut tx, community, &actor, intent).await?; diff --git a/crates/buzz-db/src/store/push.rs b/crates/buzz-db/src/store/push.rs index 1a95aecf70..4269c3f7e6 100644 --- a/crates/buzz-db/src/store/push.rs +++ b/crates/buzz-db/src/store/push.rs @@ -21,14 +21,6 @@ async fn acquire_operation_connection( Ok(crate::observability::acquire_writer(pool, operation).await?) } -async fn begin_operation_transaction( - pool: &PgPool, - operation: crate::observability::WriterOperation, -) -> Result> { - let connection = acquire_operation_connection(pool, operation).await?; - Ok(sqlx::Transaction::begin(connection, None).await?) -} - /// Namespace for the per-community push-gate advisory lock. The post-commit /// enqueue worker takes it SHARED, while lease transitions take it EXCLUSIVE. /// The worker checks receipt-time eligibility after obtaining the lock. Legacy @@ -485,11 +477,14 @@ async fn replace_lease( // lease" to "eligible"; serialize it against the event producer's shared // gate lock (gate → lease row, matching accept_lease_event's global order). // Revocations (is_active = false) never make eligibility true and skip it. - let mut tx = - begin_operation_transaction(pool, crate::observability::WriterOperation::EventWrite) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await?; if is_active { - acquire_push_gate_lock(&mut tx, community).await?; + acquire_push_gate_lock(tx.conn(), community).await?; } // The conflict predicate is the acceptance state machine. Keeping both @@ -537,7 +532,7 @@ async fn replace_lease( .bind(max_class) .bind(subscriptions) .bind(version.expires_at) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; if accepted.is_some() { @@ -553,7 +548,7 @@ async fn replace_lease( .bind(community.as_uuid()) .bind(author) .bind(installation_id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; tx.commit().await?; let current_created_at: i64 = current.try_get("source_created_at")?; @@ -635,9 +630,12 @@ pub async fn enqueue_wakes( if requests.is_empty() { return Ok(Vec::new()); } - let mut tx = - begin_operation_transaction(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; // 1. Lock and read the current lease row for every distinct requested // (author, installation), in deterministic order. @@ -666,7 +664,7 @@ pub async fn enqueue_wakes( .bind(community.as_uuid()) .bind(&lock_authors) .bind(&lock_installs) - .fetch_all(&mut *tx) + .fetch_all(tx.conn()) .await?; let mut leases: std::collections::HashMap<(Vec, String), (i64, Vec)> = std::collections::HashMap::with_capacity(lease_rows.len()); @@ -739,7 +737,7 @@ pub async fn enqueue_wakes( .bind(&ins_events) .bind(&ins_classes) .bind(&ins_expires) - .fetch_all(&mut *tx) + .fetch_all(tx.conn()) .await?; for row in inserted { let key: (Vec, Vec) = (row.try_get("endpoint_hash")?, row.try_get("event_id")?); @@ -762,7 +760,7 @@ pub async fn enqueue_wakes( .bind(community.as_uuid()) .bind(&ins_endpoints) .bind(&ins_events) - .fetch_all(&mut *tx) + .fetch_all(tx.conn()) .await?; for row in dup_rows { let key: (Vec, Vec) = @@ -932,9 +930,12 @@ where // recoverable after their claim lease expires. let gone: Vec> = attempts.into_keys().collect(); if !gone.is_empty() { - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; sqlx::query( "DELETE FROM push_match_queue \ WHERE community_id=$1 AND claim_id=$2 AND state='matching' AND event_id = ANY($3)", @@ -942,8 +943,9 @@ where .bind(community.as_uuid()) .bind(claim_id) .bind(&gone) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; } if jobs.is_empty() { return Ok(None); @@ -1014,19 +1016,24 @@ pub async fn complete_match_batch( if event_ids.is_empty() { return Ok(0); } - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; - Ok(sqlx::query( + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; + let affected = sqlx::query( "DELETE FROM push_match_queue \ WHERE community_id=$1 AND claim_id=$2 AND state='matching' AND event_id = ANY($3)", ) .bind(community.as_uuid()) .bind(claim_id) .bind(event_ids) - .execute(&mut *connection) + .execute(tx.conn()) .await? - .rows_affected()) + .rows_affected(); + tx.commit().await?; + Ok(affected) } /// Release fenced matcher claims from one batch for retry at the supplied @@ -1041,10 +1048,13 @@ pub async fn retry_match_batch( if event_ids.is_empty() { return Ok(0); } - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; - Ok(sqlx::query( + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; + let affected = sqlx::query( "UPDATE push_match_queue \ SET state='pending', claim_id=NULL, lease_until=NULL, next_attempt_at=$4 \ WHERE community_id=$1 AND claim_id=$2 AND state='matching' AND event_id = ANY($3)", @@ -1053,9 +1063,11 @@ pub async fn retry_match_batch( .bind(claim_id) .bind(event_ids) .bind(next) - .execute(&mut *connection) + .execute(tx.conn()) .await? - .rows_affected()) + .rows_affected(); + tx.commit().await?; + Ok(affected) } /// Claim due jobs for one community, recovering expired worker leases. @@ -1069,9 +1081,12 @@ pub async fn claim_due_wakes( lease_until: DateTime, ) -> Result> { let claim_id = Uuid::new_v4(); - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; let rows = sqlx::query( r#" WITH candidates AS ( @@ -1119,8 +1134,9 @@ pub async fn claim_due_wakes( .bind(limit) .bind(claim_id) .bind(lease_until) - .fetch_all(&mut *connection) + .fetch_all(tx.conn()) .await?; + tx.commit().await?; rows.into_iter().map(row_to_claimed_wake).collect() } @@ -1187,9 +1203,12 @@ pub async fn complete_wake( id: Uuid, claim_id: Uuid, ) -> Result { - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; let result = sqlx::query( "UPDATE push_wake_outbox \ SET state = 'delivered', claim_id = NULL, lease_until = NULL \ @@ -1198,8 +1217,9 @@ pub async fn complete_wake( .bind(community.as_uuid()) .bind(id) .bind(claim_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() == 1) } @@ -1211,9 +1231,12 @@ pub async fn retry_wake( claim_id: Uuid, next_attempt_at: DateTime, ) -> Result { - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; let result = sqlx::query( "UPDATE push_wake_outbox \ SET state = 'pending', next_attempt_at = $4, claim_id = NULL, lease_until = NULL \ @@ -1223,8 +1246,9 @@ pub async fn retry_wake( .bind(id) .bind(claim_id) .bind(next_attempt_at) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() == 1) } @@ -1235,9 +1259,12 @@ pub async fn fail_wake( id: Uuid, claim_id: Uuid, ) -> Result { - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; let result = sqlx::query( "UPDATE push_wake_outbox \ SET state = 'failed', claim_id = NULL, lease_until = NULL \ @@ -1246,8 +1273,9 @@ pub async fn fail_wake( .bind(community.as_uuid()) .bind(id) .bind(claim_id) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() == 1) } @@ -1262,9 +1290,12 @@ pub async fn disable_endpoint_generation( installation_id: &str, generation: i64, ) -> Result { - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; let result = sqlx::query( "UPDATE push_leases SET endpoint_enabled = false, updated_at = now() \ WHERE community_id = $1 AND author = $2 AND installation_id = $3 \ @@ -1274,8 +1305,9 @@ pub async fn disable_endpoint_generation( .bind(author) .bind(installation_id) .bind(generation) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() == 1) } @@ -1289,9 +1321,12 @@ pub async fn prune_wake_outbox( community: CommunityId, before: DateTime, ) -> Result { - let mut connection = - acquire_operation_connection(pool, crate::observability::WriterOperation::Maintenance) - .await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + crate::observability::WriterOperation::Maintenance, + ) + .await?; let result = sqlx::query( "DELETE FROM push_wake_outbox o \ WHERE o.community_id = $1 AND o.created_at < $2 \ @@ -1304,8 +1339,9 @@ pub async fn prune_wake_outbox( ) .bind(community.as_uuid()) .bind(before) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected()) } diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 0a876beb2a..e1cf4ca4c5 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -143,7 +143,12 @@ pub async fn resolve_report_decision_atomic( channel_id: Option, reason: Option<&str>, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // CAS: open → terminal. The update count tells us whether the report was open. let updated = sqlx::query( @@ -157,7 +162,7 @@ pub async fn resolve_report_decision_atomic( .bind(report_id) .bind(terminal_status) .bind(actor_pubkey) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if updated.rows_affected() == 0 { @@ -182,7 +187,7 @@ pub async fn resolve_report_decision_atomic( .bind(channel_id) .bind(reason) .bind(actor_authority) - .execute(&mut *tx) + .execute(tx.conn()) .await?; tx.commit().await?; @@ -221,7 +226,12 @@ pub async fn claim_report( target_event_id: Option<&[u8]>, channel_id: Option, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // Lock the report row to serialize concurrent claims on the same report. let report_row = sqlx::query( @@ -234,7 +244,7 @@ pub async fn claim_report( ) .bind(community_id.as_uuid()) .bind(report_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; let Some(report_row) = report_row else { @@ -258,7 +268,7 @@ pub async fn claim_report( .bind(community_id.as_uuid()) .bind(report_id) .bind(request_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; if let Some(row) = existing { @@ -296,7 +306,7 @@ pub async fn claim_report( .bind(timeout_until) .bind(target_pubkey) .bind(channel_id) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; let action_id: Uuid = action_row.try_get("id")?; @@ -318,7 +328,7 @@ pub async fn claim_report( .bind(channel_id) .bind(reason) .bind(actor_authority) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // CAS: set report to processing with active_action_id. @@ -332,7 +342,7 @@ pub async fn claim_report( .bind(community_id.as_uuid()) .bind(report_id) .bind(action_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if updated.rows_affected() == 0 { @@ -457,7 +467,12 @@ pub async fn execute_ban_with_marker( actor_pubkey: &[u8], reason: Option<&str>, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // Verify lease ownership first — abort without touching domain rows if the // lease is already gone. This prevents the commit entirely on a stale worker. @@ -474,7 +489,7 @@ pub async fn execute_ban_with_marker( ) .bind(action_id) .bind(lease_token) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; if !owned { @@ -488,7 +503,7 @@ pub async fn execute_ban_with_marker( // under the lock, not before it. sqlx::query("SELECT id FROM relay_admin_actions WHERE id = $1 FOR UPDATE") .bind(action_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; sqlx::query( @@ -505,7 +520,7 @@ pub async fn execute_ban_with_marker( .bind(target_pubkey) .bind(actor_pubkey) .bind(reason) - .execute(&mut *tx) + .execute(tx.conn()) .await?; let marker = sqlx::query( @@ -521,7 +536,7 @@ pub async fn execute_ban_with_marker( ) .bind(action_id) .bind(lease_token) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if marker.rows_affected() == 0 { @@ -548,7 +563,12 @@ pub async fn execute_timeout_with_marker( until: DateTime, reason: Option<&str>, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; let owned: bool = sqlx::query_scalar( r#" @@ -563,7 +583,7 @@ pub async fn execute_timeout_with_marker( ) .bind(action_id) .bind(lease_token) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; if !owned { @@ -574,7 +594,7 @@ pub async fn execute_timeout_with_marker( // Acquire the action-row lock before evaluating the wall-clock expiry check. sqlx::query("SELECT id FROM relay_admin_actions WHERE id = $1 FOR UPDATE") .bind(action_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; sqlx::query( @@ -593,7 +613,7 @@ pub async fn execute_timeout_with_marker( .bind(until) .bind(actor_pubkey) .bind(reason) - .execute(&mut *tx) + .execute(tx.conn()) .await?; let marker = sqlx::query( @@ -609,7 +629,7 @@ pub async fn execute_timeout_with_marker( ) .bind(action_id) .bind(lease_token) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if marker.rows_affected() == 0 { @@ -648,7 +668,12 @@ pub async fn execute_kick_with_marker( target_pubkey: &[u8], actor_pubkey: &[u8], ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; let owned: bool = sqlx::query_scalar( r#" @@ -663,7 +688,7 @@ pub async fn execute_kick_with_marker( ) .bind(action_id) .bind(lease_token) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; if !owned { @@ -676,7 +701,7 @@ pub async fn execute_kick_with_marker( // Acquire the action-row lock before evaluating the wall-clock expiry check. sqlx::query("SELECT id FROM relay_admin_actions WHERE id = $1 FOR UPDATE") .bind(action_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; let kick = sqlx::query( @@ -690,7 +715,7 @@ pub async fn execute_kick_with_marker( .bind(community_id.as_uuid()) .bind(channel_id) .bind(target_pubkey) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if kick.rows_affected() == 0 { @@ -711,7 +736,7 @@ pub async fn execute_kick_with_marker( ) .bind(action_id) .bind(lease_token) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if marker.rows_affected() == 0 { @@ -855,7 +880,12 @@ pub async fn finalize_success( reason: Option<&str>, timeout_until: Option>, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // Require step_marker = 'mutation_committed' to prevent premature finalization. let updated_action = sqlx::query( @@ -866,7 +896,7 @@ pub async fn finalize_success( "#, ) .bind(action_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if updated_action.rows_affected() == 0 { @@ -893,7 +923,7 @@ pub async fn finalize_success( .bind(terminal_status) .bind(actor_pubkey) .bind(action_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if updated_report.rows_affected() == 0 { @@ -930,7 +960,7 @@ pub async fn finalize_success( .bind(action_id) .bind(payload) .bind(format!("tombstone:{action_str}")) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } } @@ -953,7 +983,7 @@ pub async fn finalize_success( .bind(action_id) .bind(payload) .bind(format!("system_message:{action_str}")) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } } @@ -975,7 +1005,7 @@ pub async fn finalize_success( .bind(action_id) .bind(notice_payload) .bind(format!("reporter_notice:{action_str}")) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } @@ -1024,7 +1054,7 @@ pub async fn finalize_success( .bind(action_id) .bind(affected_payload) .bind(format!("affected_user_notice:{action_str}")) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } @@ -1124,7 +1154,12 @@ pub async fn cancel_action( report_id: Uuid, cancelled_by: &[u8], ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // Cancel only a pre-mutation `failed` action that BELONGS to the path // report and community. Fencing on report_id + report_community_id is what @@ -1146,7 +1181,7 @@ pub async fn cancel_action( .bind(report_id) .bind(community_id.as_uuid()) .bind(cancelled_by) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if updated.rows_affected() != 1 { @@ -1171,7 +1206,7 @@ pub async fn cancel_action( .bind(community_id.as_uuid()) .bind(report_id) .bind(action_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; if reopened.rows_affected() != 1 { @@ -1219,7 +1254,12 @@ pub async fn reopen_report( actor_role: &str, reason: Option<&str>, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // Lock the report row to serialize concurrent reopen/resolve on it. let report_row = sqlx::query( @@ -1232,7 +1272,7 @@ pub async fn reopen_report( ) .bind(community_id.as_uuid()) .bind(report_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; let Some(report_row) = report_row else { @@ -1252,7 +1292,7 @@ pub async fn reopen_report( .bind(community_id.as_uuid()) .bind(report_id) .bind(request_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; if existing.is_some() { @@ -1280,7 +1320,7 @@ pub async fn reopen_report( ) .bind(community_id.as_uuid()) .bind(report_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // Durable audit row. Inserted as 'succeeded' so the recovery worker never @@ -1299,7 +1339,7 @@ pub async fn reopen_report( .bind(actor_pubkey) .bind(actor_role) .bind(reason) - .execute(&mut *tx) + .execute(tx.conn()) .await?; tx.commit().await?; @@ -1643,6 +1683,12 @@ pub async fn deploy_kick_member( target_pubkey: &[u8], actor_pubkey: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; // Use a direct UPDATE to avoid the tenant ownership check in channel::remove_member. // This is the deployment-authority primitive: no actor role check. let result = sqlx::query( @@ -1656,8 +1702,9 @@ pub async fn deploy_kick_member( .bind(community_id.as_uuid()) .bind(channel_id) .bind(target_pubkey) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; if result.rows_affected() > 0 { Ok(KickResult::Removed) @@ -1765,7 +1812,12 @@ pub async fn claim_direct_action( pool: &PgPool, input: &DirectActionInput<'_>, ) -> Result { - let mut tx = pool.begin().await?; + let mut tx = crate::begin_community_write_transaction( + pool, + input.community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; let inserted = sqlx::query( r#" INSERT INTO relay_admin_actions ( @@ -1788,7 +1840,7 @@ pub async fn claim_direct_action( .bind(input.target_pubkey) .bind(input.target_event_id) .bind(input.channel_id) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; let Some(row) = inserted else { @@ -1818,7 +1870,7 @@ pub async fn claim_direct_action( .bind(input.channel_id) .bind(input.reason) .bind(input.actor_authority) - .execute(&mut *tx) + .execute(tx.conn()) .await?; let rec = row_to_action(row)?; tx.commit().await?; diff --git a/crates/buzz-db/src/store/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index 4bb48e121c..6487696151 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -116,15 +116,12 @@ pub async fn mint_relay_invite( // community-scoped database write. The trigger remains the final backstop, // but this typed guard keeps a quiescing community from surfacing as an // opaque SQLSTATE/HTTP 500 at the API boundary. - let connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community, crate::observability::WriterOperation::Authorization, ) .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; - crate::deletion::DeletionStore::new(pool.clone()) - .guard_transaction(&mut tx, community) - .await?; let row = sqlx::query( "INSERT INTO relay_invites (community_id, token_hash, max_uses, expires_at, created_by) \ VALUES ($1, $2, $3, $4, $5) \ @@ -135,7 +132,7 @@ pub async fn mint_relay_invite( .bind(max_uses) .bind(expires_at) .bind(created_by) - .fetch_one(&mut *tx) + .fetch_one(tx.conn()) .await?; tx.commit().await?; @@ -226,12 +223,12 @@ pub async fn claim_relay_invite( claimer_pubkey: &str, policy_version: Option<&str>, ) -> Result { - let connection = crate::observability::acquire_writer( + let mut tx = crate::begin_community_write_transaction( pool, + community, crate::observability::WriterOperation::Authorization, ) .await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; // 2. SELECT FOR UPDATE — lock the invite row for the duration of this txn. let row = sqlx::query( @@ -242,7 +239,7 @@ pub async fn claim_relay_invite( ) .bind(community.as_uuid()) .bind(token_hash) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; // 3. No matching invite. @@ -279,7 +276,7 @@ pub async fn claim_relay_invite( sqlx::query("SELECT 1 FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(claimer_pubkey) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; if existing.is_some() { @@ -292,7 +289,7 @@ pub async fn claim_relay_invite( .bind(community.as_uuid()) .bind(claimer_pubkey) .bind(version) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } tx.commit().await?; @@ -334,7 +331,7 @@ pub async fn claim_relay_invite( ) .bind(community.as_uuid()) .bind(claimer_pubkey) - .execute(&mut *tx) + .execute(tx.conn()) .await? .rows_affected() > 0; @@ -349,7 +346,7 @@ pub async fn claim_relay_invite( .bind(community.as_uuid()) .bind(claimer_pubkey) .bind(version) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } @@ -374,7 +371,7 @@ pub async fn claim_relay_invite( .bind(new_use_count) .bind(community.as_uuid()) .bind(invite_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // 11. Commit. diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 742e9173f2..6b21c7ce01 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -150,8 +150,12 @@ pub async fn add_relay_member( role: &str, added_by: Option<&str>, ) -> Result { - let mut connection = - observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ VALUES ($1, $2, $3, $4) ON CONFLICT (community_id, pubkey) DO NOTHING", @@ -160,8 +164,9 @@ pub async fn add_relay_member( .bind(pubkey) .bind(role) .bind(added_by) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -177,9 +182,12 @@ pub async fn claim_relay_membership( role: &str, policy_version: Option<&str>, ) -> Result { - let connection = - observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + observability::WriterOperation::Authorization, + ) + .await?; let inserted = sqlx::query( "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ VALUES ($1, $2, $3, 'invite') \ @@ -188,7 +196,7 @@ pub async fn claim_relay_membership( .bind(community.as_uuid()) .bind(pubkey) .bind(role) - .execute(&mut *tx) + .execute(tx.conn()) .await? .rows_affected() > 0; @@ -201,7 +209,7 @@ pub async fn claim_relay_membership( .bind(community.as_uuid()) .bind(pubkey) .bind(version) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } @@ -243,20 +251,21 @@ pub enum RemoveResult { RoleMismatch, } -/// Deletes a membership and its join-policy acceptances in one transaction. +/// Deletes a membership and its join-policy acceptances inside the caller's +/// admitted transaction. /// /// With `expected_role`, only a member holding exactly that role is removed; /// without it, any non-owner is. The member row is locked under that predicate /// first, so a concurrent acceptance insert (which must lock the member row for /// its foreign key) either commits before the child delete sees it or fails. -/// Children go before the parent so no foreign-key action is needed. +/// Children go before the parent so no foreign-key action is needed. The +/// caller commits. async fn delete_member_with_acceptances( - connection: &mut sqlx::PgConnection, - community: CommunityId, + tx: &mut crate::AdmittedTx, pubkey: &str, expected_role: Option<&str>, ) -> Result { - let mut tx = sqlx::Connection::begin(connection).await?; + let community = tx.community(); let locked = sqlx::query( "SELECT 1 FROM relay_members WHERE community_id = $1 AND pubkey = $2 \ AND CASE WHEN $3::text IS NULL THEN role <> 'owner' ELSE role = $3 END \ @@ -265,7 +274,7 @@ async fn delete_member_with_acceptances( .bind(community.as_uuid()) .bind(pubkey) .bind(expected_role) - .fetch_optional(&mut *tx) + .fetch_optional(tx.conn()) .await?; if locked.is_none() { return Ok(false); @@ -273,14 +282,13 @@ async fn delete_member_with_acceptances( sqlx::query("DELETE FROM join_policy_acceptances WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .execute(&mut *tx) + .execute(tx.conn()) .await?; sqlx::query("DELETE FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .execute(&mut *tx) + .execute(tx.conn()) .await?; - tx.commit().await?; Ok(true) } @@ -293,9 +301,14 @@ pub async fn remove_relay_member( community: CommunityId, pubkey: &str, ) -> Result { - let mut connection = - observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; - if delete_member_with_acceptances(&mut connection, community, pubkey, None).await? { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + observability::WriterOperation::Authorization, + ) + .await?; + if delete_member_with_acceptances(&mut tx, pubkey, None).await? { + tx.commit().await?; return Ok(RemoveResult::Removed); } @@ -304,9 +317,10 @@ pub async fn remove_relay_member( let exists = sqlx::query("SELECT 1 FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; + tx.commit().await?; if exists.is_some() { Ok(RemoveResult::IsOwner) } else { @@ -332,11 +346,14 @@ pub async fn remove_relay_member_if_role( pubkey: &str, expected_role: &str, ) -> Result { - let mut connection = - observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; - if delete_member_with_acceptances(&mut connection, community, pubkey, Some(expected_role)) - .await? - { + let mut tx = crate::begin_community_write_transaction( + pool, + community, + observability::WriterOperation::Authorization, + ) + .await?; + if delete_member_with_acceptances(&mut tx, pubkey, Some(expected_role)).await? { + tx.commit().await?; return Ok(RemoveResult::Removed); } @@ -345,9 +362,10 @@ pub async fn remove_relay_member_if_role( let row = sqlx::query("SELECT role FROM relay_members WHERE community_id = $1 AND pubkey = $2") .bind(community.as_uuid()) .bind(pubkey) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; + tx.commit().await?; match row { None => Ok(RemoveResult::NotFound), Some(r) => { @@ -372,8 +390,12 @@ pub async fn update_relay_member_role( pubkey: &str, new_role: &str, ) -> Result { - let mut connection = - observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community, + observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( "UPDATE relay_members SET role = $1, updated_at = now() \ WHERE community_id = $2 AND pubkey = $3 AND role <> 'owner'", @@ -381,8 +403,9 @@ pub async fn update_relay_member_role( .bind(new_role) .bind(community.as_uuid()) .bind(pubkey) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() > 0) } @@ -476,14 +499,7 @@ async fn lock_owner_mutation_admission( return Ok(OwnerMutationAdmission::Allowed(existing_owners)); } - let deletion_pending: bool = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM community_deletion_requests \ - WHERE community_id = $1 AND stage <> 'aborted')", - ) - .bind(community.as_uuid()) - .fetch_one(&mut **tx) - .await?; - if deletion_pending { + if deletion_pending(tx, community).await? { return Ok(OwnerMutationAdmission::DeletionPending); } @@ -497,6 +513,21 @@ async fn lock_owner_mutation_admission( Ok(OwnerMutationAdmission::Allowed(existing_owners)) } +/// Whether `community` has a non-aborted deletion request. +/// +/// A quiescing or fenced community always has one, so owner mutations that +/// admission refuses report `DeletionPending` rather than a generic lifecycle +/// conflict. +async fn deletion_pending(conn: &mut sqlx::PgConnection, community: CommunityId) -> Result { + Ok(sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM community_deletion_requests \ + WHERE community_id = $1 AND stage <> 'aborted')", + ) + .bind(community.as_uuid()) + .fetch_one(conn) + .await?) +} + async fn bootstrap_owner_with_operation( pool: &PgPool, community: CommunityId, @@ -507,6 +538,16 @@ async fn bootstrap_owner_with_operation( let connection = observability::acquire_writer(pool, operation).await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; + if let Err(error) = crate::deletion::guard_community_write(&mut tx, community).await { + return match error { + DbError::AccessDenied(_) if deletion_pending(&mut tx, community).await? => { + Ok(ProvisionOwnerResult::DeletionPending) + } + DbError::AccessDenied(_) => Ok(ProvisionOwnerResult::LifecycleConflict), + other => Err(other), + }; + } + match lock_owner_mutation_admission(&mut tx, community, &pubkey, OwnerMutationMode::Converge) .await? { @@ -730,6 +771,27 @@ pub async fn transfer_ownership( observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; + if let Err(error) = crate::deletion::guard_community_write(&mut tx, community).await { + return match error { + DbError::AccessDenied(_) if deletion_pending(&mut tx, community).await? => { + Ok(TransferResult::DeletionPending) + } + DbError::AccessDenied(_) => { + let exists: bool = + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1)") + .bind(community.as_uuid()) + .fetch_one(&mut *tx) + .await?; + if exists { + Ok(TransferResult::LifecycleConflict) + } else { + Ok(TransferResult::NoOwner) + } + } + other => Err(other), + }; + } + // 1. Serialize on the transferee so concurrent transfers to the same // recipient cannot both pass the ownership count check. crate::observability::observe_advisory_lock( @@ -825,40 +887,52 @@ pub async fn transfer_ownership( /// /// Converts BYTEA pubkeys to lowercase hex text and inserts them as members of /// `community`. Returns the number of rows inserted, or 0 if: -/// - the `pubkey_allowlist` table doesn't exist, or +/// - the `pubkey_allowlist` table doesn't exist, +/// - `pubkey_allowlist` has no rows for this community, or /// - `relay_members` already has rows for this community (migration ran in a /// prior startup). /// /// The empty-table guard prevents re-adding members that were intentionally /// removed by an admin after the initial backfill. pub async fn backfill_from_allowlist(pool: &PgPool, community: CommunityId) -> Result { + // The common startup case writes nothing: the legacy table is gone, it + // has no rows for this community, or the community already has members. + // Answer those with plain reads so a startup with nothing to backfill + // never needs write admission, and a quiescing deployment community keeps + // starting as it did before. let mut connection = observability::acquire_writer(pool, observability::WriterOperation::Bootstrap).await?; - // Check if pubkey_allowlist table exists. let exists: bool = sqlx::query_scalar( "SELECT EXISTS (SELECT 1 FROM information_schema.tables \ WHERE table_schema = 'public' AND table_name = 'pubkey_allowlist')", ) .fetch_one(&mut *connection) .await?; - - if !exists { + if !exists || community_has_members(&mut connection, community).await? { return Ok(0); } - - // Only backfill if this community's relay_members is empty — once it has - // rows (from a previous backfill or manual admin commands), we must not - // re-add members that were intentionally removed. - let has_members: bool = - sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM relay_members WHERE community_id = $1)") - .bind(community.as_uuid()) - .fetch_one(&mut *connection) - .await?; - - if has_members { + let has_allowlist_rows: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pubkey_allowlist WHERE community_id = $1)", + ) + .bind(community.as_uuid()) + .fetch_one(&mut *connection) + .await?; + if !has_allowlist_rows { return Ok(0); } + drop(connection); + let mut tx = crate::begin_community_write_transaction( + pool, + community, + observability::WriterOperation::Bootstrap, + ) + .await?; + // Re-check under admission: another pod may have backfilled meanwhile. + if community_has_members(tx.conn(), community).await? { + tx.commit().await?; + return Ok(0); + } let result = sqlx::query( "INSERT INTO relay_members (community_id, pubkey, role, added_by, created_at) \ SELECT $1, encode(pubkey, 'hex'), 'member', NULL, added_at \ @@ -867,12 +941,27 @@ pub async fn backfill_from_allowlist(pool: &PgPool, community: CommunityId) -> R ON CONFLICT (community_id, pubkey) DO NOTHING", ) .bind(community.as_uuid()) - .execute(&mut *connection) + .execute(tx.conn()) .await?; - + tx.commit().await?; Ok(result.rows_affected()) } +/// Whether `community` has any `relay_members` row. Once it does (from a +/// previous backfill or admin commands), the backfill must not re-add members +/// that were intentionally removed. +async fn community_has_members( + conn: &mut sqlx::PgConnection, + community: CommunityId, +) -> Result { + Ok( + sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM relay_members WHERE community_id = $1)") + .bind(community.as_uuid()) + .fetch_one(conn) + .await?, + ) +} + impl Db { /// Returns `true` if `pubkey` (64-char hex) is a member of `community`. /// diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 5b30f5df83..719861241c 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -11,23 +11,6 @@ use uuid::Uuid; use buzz_datastore_tracing::datastore_span; -async fn acquire_event_write_connection( - pool: &PgPool, -) -> Result> { - Ok(crate::observability::acquire_writer( - pool, - crate::observability::WriterOperation::EventWrite, - ) - .await?) -} - -async fn begin_event_write_transaction( - pool: &PgPool, -) -> Result> { - let connection = acquire_event_write_connection(pool).await?; - Ok(sqlx::Transaction::begin(connection, None).await?) -} - use buzz_core::CommunityId; use crate::{ @@ -148,7 +131,12 @@ pub async fn insert_thread_metadata( depth: i32, broadcast: bool, ) -> Result<()> { - let mut tx = begin_event_write_transaction(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let result = sqlx::query( r#" @@ -171,7 +159,7 @@ pub async fn insert_thread_metadata( .bind(root_event_created_at) .bind(depth) .bind(broadcast) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // Only bump reply counts if the row was actually inserted (not a duplicate). @@ -197,7 +185,7 @@ pub async fn insert_thread_metadata( .bind(parent_ts) .bind(pid) .bind(channel_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // Ensure the root also has a row (may differ from parent for nested replies). @@ -219,7 +207,7 @@ pub async fn insert_thread_metadata( .bind(root_ts) .bind(root_id) .bind(channel_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } } @@ -235,7 +223,7 @@ pub async fn insert_thread_metadata( ) .bind(community_id.as_uuid()) .bind(pid) - .execute(&mut *tx) + .execute(tx.conn()) .await?; // Increment root's total descendant count. @@ -249,7 +237,7 @@ pub async fn insert_thread_metadata( ) .bind(community_id.as_uuid()) .bind(root_id) - .execute(&mut *tx) + .execute(tx.conn()) .await?; } } diff --git a/crates/buzz-db/src/store/user.rs b/crates/buzz-db/src/store/user.rs index ec4b897a61..d767b56c75 100644 --- a/crates/buzz-db/src/store/user.rs +++ b/crates/buzz-db/src/store/user.rs @@ -57,7 +57,23 @@ async fn ensure_user_with_operation( pubkey: &[u8], operation: crate::observability::WriterOperation, ) -> Result { + // Almost every call finds the user already registered. Answer that with + // one read, so the hot path skips the admission round trips; no write + // happens, so the fence contract is unaffected. let mut connection = crate::observability::acquire_writer(pool, operation).await?; + let exists: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM users WHERE community_id = $1 AND pubkey = $2)", + ) + .bind(community_id.as_uuid()) + .bind(pubkey) + .fetch_one(&mut *connection) + .await?; + if exists { + return Ok(false); + } + drop(connection); + + let mut tx = crate::begin_community_write_transaction(pool, community_id, operation).await?; let result = sqlx::query( r#" INSERT INTO users (community_id, pubkey) @@ -67,8 +83,9 @@ async fn ensure_user_with_operation( ) .bind(community_id.as_uuid()) .bind(pubkey) - .execute(&mut *connection) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() == 1) } @@ -178,7 +195,14 @@ pub async fn update_user_profile( } query = query.bind(community_id.as_uuid()); query = query.bind(pubkey); - query.execute(pool).await?; + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + query.execute(tx.conn()).await?; + tx.commit().await?; Ok(()) } @@ -329,7 +353,7 @@ async fn set_agent_owner_with_operation( owner_pubkey: &[u8], operation: crate::observability::WriterOperation, ) -> Result { - let mut connection = crate::observability::acquire_writer(pool, operation).await?; + let mut tx = crate::begin_community_write_transaction(pool, community_id, operation).await?; // Conditional UPDATE: only set owner if currently NULL. This makes // "first mint wins" atomic — no TOCTOU race between concurrent mints. let result = sqlx::query( @@ -338,7 +362,7 @@ async fn set_agent_owner_with_operation( .bind(owner_pubkey) .bind(community_id.as_uuid()) .bind(agent_pubkey) - .execute(&mut *connection) + .execute(tx.conn()) .await?; if result.rows_affected() == 0 { @@ -347,7 +371,7 @@ async fn set_agent_owner_with_operation( let exists = sqlx::query(r#"SELECT 1 FROM users WHERE community_id = $1 AND pubkey = $2"#) .bind(community_id.as_uuid()) .bind(agent_pubkey) - .fetch_optional(&mut *connection) + .fetch_optional(tx.conn()) .await?; if exists.is_none() { return Err(crate::error::DbError::NotFound( @@ -355,8 +379,10 @@ async fn set_agent_owner_with_operation( )); } // Row exists but owner already set — return false (not an error). + tx.commit().await?; return Ok(false); } + tx.commit().await?; Ok(true) } @@ -461,14 +487,21 @@ pub async fn set_channel_add_policy( "invalid channel_add_policy: {policy}" ))); } + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::Authorization, + ) + .await?; let result = sqlx::query( r#"UPDATE users SET channel_add_policy = $1::channel_add_policy WHERE community_id = $2 AND pubkey = $3"#, ) .bind(policy) .bind(community_id.as_uuid()) .bind(pubkey) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; if result.rows_affected() == 0 { return Err(crate::error::DbError::NotFound( "pubkey not found in users table".into(), diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index c89bf0cbf0..8f63d85555 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -15,7 +15,7 @@ use std::str::FromStr; use chrono::{DateTime, Utc}; use sha2::{Digest, Sha256}; -use sqlx::{PgConnection, PgPool, Row}; +use sqlx::{PgPool, Row}; use uuid::Uuid; use buzz_core::CommunityId; @@ -290,6 +290,12 @@ pub async fn create_workflow( definition_json: &str, definition_hash: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let id = Uuid::new_v4(); sqlx::query( @@ -306,8 +312,9 @@ pub async fn create_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(id) } @@ -328,6 +335,12 @@ pub async fn upsert_workflow( definition_json: &str, definition_hash: &[u8], ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let row = sqlx::query( r#" INSERT INTO workflows @@ -350,8 +363,9 @@ pub async fn upsert_workflow( .bind(channel_id) .bind(definition_json) .bind(definition_hash) - .fetch_optional(pool) + .fetch_optional(tx.conn()) .await?; + tx.commit().await?; if row.is_none() { return Err(DbError::AccessDenied(format!( @@ -507,6 +521,12 @@ pub async fn claim_scheduled_workflow_fire( workflow_id: Uuid, scheduled_for: DateTime, ) -> Result> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let row = sqlx::query( r#" INSERT INTO scheduled_workflow_fires (community_id, workflow_id, scheduled_for) @@ -520,8 +540,9 @@ pub async fn claim_scheduled_workflow_fire( .bind(community_id.as_uuid()) .bind(workflow_id) .bind(scheduled_for) - .fetch_optional(pool) + .fetch_optional(tx.conn()) .await?; + tx.commit().await?; row.map(|row| { let community_id: Uuid = row.try_get("community_id")?; @@ -575,6 +596,12 @@ pub async fn attach_scheduled_workflow_run( scheduled_for: DateTime, workflow_run_id: Uuid, ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let result = sqlx::query( r#" UPDATE scheduled_workflow_fires @@ -589,8 +616,9 @@ pub async fn attach_scheduled_workflow_run( .bind(workflow_id) .bind(scheduled_for) .bind(workflow_run_id) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(result.rows_affected() == 1) } @@ -634,6 +662,12 @@ pub async fn update_workflow( definition_json: &str, definition_hash: &[u8], ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let affected = sqlx::query( r#" UPDATE workflows @@ -646,9 +680,10 @@ pub async fn update_workflow( .bind(definition_hash) .bind(community_id.as_uuid()) .bind(id) - .execute(pool) + .execute(tx.conn()) .await? .rows_affected(); + tx.commit().await?; if affected == 0 { return Err(DbError::NotFound(format!("workflow {id}"))); @@ -666,6 +701,12 @@ pub async fn update_workflow_status( id: Uuid, status: WorkflowStatus, ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let affected = sqlx::query( r#" UPDATE workflows @@ -676,9 +717,10 @@ pub async fn update_workflow_status( .bind(status.to_string()) .bind(community_id.as_uuid()) .bind(id) - .execute(pool) + .execute(tx.conn()) .await? .rows_affected(); + tx.commit().await?; if affected == 0 { return Err(DbError::NotFound(format!("workflow {id}"))); @@ -696,6 +738,12 @@ pub async fn set_workflow_enabled( id: Uuid, enabled: bool, ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let affected = sqlx::query( r#" UPDATE workflows @@ -706,9 +754,10 @@ pub async fn set_workflow_enabled( .bind(enabled) .bind(community_id.as_uuid()) .bind(id) - .execute(pool) + .execute(tx.conn()) .await? .rows_affected(); + tx.commit().await?; if affected == 0 { return Err(DbError::NotFound(format!("workflow {id}"))); @@ -729,6 +778,12 @@ pub async fn disable_workflows_for_owner_in_channel( channel_id: Uuid, owner_pubkey: &[u8], ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let affected = sqlx::query( r#" UPDATE workflows @@ -739,9 +794,10 @@ pub async fn disable_workflows_for_owner_in_channel( .bind(community_id.as_uuid()) .bind(channel_id) .bind(owner_pubkey) - .execute(pool) + .execute(tx.conn()) .await? .rows_affected(); + tx.commit().await?; Ok(affected) } @@ -753,11 +809,16 @@ pub async fn disable_workflows_for_owner_in_channel( /// so the disable executes on the fence's own connection, avoiding a second /// pool acquisition while the fence holds one connection. pub async fn disable_workflows_for_owner_in_channel_on_conn( - conn: &mut PgConnection, + tx: &mut crate::AdmittedTx, community_id: CommunityId, channel_id: Uuid, owner_pubkey: &[u8], ) -> Result { + if tx.community() != community_id { + return Err(DbError::AccessDenied( + "workflow write community mismatch".into(), + )); + } let affected = sqlx::query( r#" UPDATE workflows @@ -768,7 +829,7 @@ pub async fn disable_workflows_for_owner_in_channel_on_conn( .bind(community_id.as_uuid()) .bind(channel_id) .bind(owner_pubkey) - .execute(&mut *conn) + .execute(tx.conn()) .await? .rows_affected(); @@ -789,6 +850,12 @@ pub async fn create_workflow_run( trigger_event_id: Option<&[u8]>, trigger_context: Option<&serde_json::Value>, ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let id = Uuid::new_v4(); sqlx::query( @@ -803,8 +870,9 @@ pub async fn create_workflow_run( .bind(workflow_id) .bind(trigger_event_id) .bind(trigger_context) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(id) } @@ -907,6 +975,12 @@ pub async fn update_workflow_run( trace: &serde_json::Value, failure: Option>, ) -> Result<()> { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let status_str = status.to_string(); let (error_code, error) = failure .map(|failure| (Some(failure.code), Some(failure.message))) @@ -935,9 +1009,10 @@ pub async fn update_workflow_run( .bind(&status_str) // for completed_at CASE .bind(community_id.as_uuid()) .bind(id) - .execute(pool) + .execute(tx.conn()) .await? .rows_affected(); + tx.commit().await?; if affected == 0 { return Err(DbError::NotFound(format!("workflow_run {id}"))); @@ -983,6 +1058,12 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> expires_at, } = params; let token_hash = hash_approval_token(token); + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; sqlx::query( r#" @@ -999,8 +1080,9 @@ pub async fn create_approval(pool: &PgPool, params: CreateApprovalParams<'_>) -> .bind(step_index) .bind(approver_spec) .bind(expires_at) - .execute(pool) + .execute(tx.conn()) .await?; + tx.commit().await?; Ok(()) } @@ -1120,6 +1202,12 @@ pub async fn update_approval_by_stored_hash( approver_pubkey: Option<&[u8]>, note: Option<&str>, ) -> Result { + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; let status_str = status.to_string(); let affected = sqlx::query( r#" @@ -1139,9 +1227,10 @@ pub async fn update_approval_by_stored_hash( .bind(&status_str) // for denied_at CASE .bind(community_id.as_uuid()) .bind(token_hash) - .execute(pool) + .execute(tx.conn()) .await? .rows_affected(); + tx.commit().await?; Ok(affected > 0) } diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs index 1608779952..4d4edcf8e9 100644 --- a/crates/buzz-db/tests/observability_source.rs +++ b/crates/buzz-db/tests/observability_source.rs @@ -207,48 +207,47 @@ fn p0_pool_acquisitions_use_typed_operation_pairs_without_other() { assert!(!thread_metadata.contains("fetch_optional(pool)")); let channel = include_str!("../src/store/channel.rs"); - assert!(channel.contains("async fn begin_event_write_transaction(")); - assert!(channel.contains("async fn acquire_event_write_connection(")); + assert!(channel.contains("begin_community_write_transaction(")); for (start, end, expected) in [ ( "pub async fn create_channel(\n", "/// Creates a channel with a client-supplied UUID", - "begin_event_write_transaction(pool)", + "begin_community_write_transaction(", ), ( "pub async fn create_channel_with_id(\n", "/// Fetches a channel record by `(community_id, id)`", - "begin_event_write_transaction(pool)", + "begin_community_write_transaction(", ), ( "pub async fn update_channel(\n", "/// Sets the topic for a channel", - "begin_event_write_transaction(pool)", + "begin_community_write_transaction(", ), ( "pub async fn set_topic(\n", "/// Sets the purpose for a channel", - "acquire_event_write_connection(pool)", + "begin_community_write_transaction(", ), ( "pub async fn set_purpose(\n", "/// Archives a channel", - "acquire_event_write_connection(pool)", + "begin_community_write_transaction(", ), ( "pub async fn archive_channel(\n", "/// Unarchives a channel", - "acquire_event_write_connection(pool)", + "begin_community_write_transaction(", ), ( "pub async fn unarchive_channel(\n", "/// Soft-delete a channel", - "acquire_event_write_connection(pool)", + "begin_community_write_transaction(", ), ( "pub async fn soft_delete_channel(\n", "/// Archive ephemeral channels", - "acquire_event_write_connection(pool)", + "begin_community_write_transaction(", ), ] { let function = channel @@ -903,11 +902,13 @@ const GUARDED_TABLE_WRITE_MARKERS: [&str; 9] = [ ]; const COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_event_write_transaction("; +const GENERAL_COMMUNITY_CHOKEPOINT_MARKER: &str = "begin_community_write_transaction("; const COMMUNITY_CHOKEPOINT_LEGACY_MARKER: &str = "begin_community_event_write_transaction_with_legacy_metrics("; fn has_any_tenant_local_chokepoint(source: &str) -> bool { source.contains(COMMUNITY_CHOKEPOINT_MARKER) + || source.contains(GENERAL_COMMUNITY_CHOKEPOINT_MARKER) || source.contains(COMMUNITY_CHOKEPOINT_LEGACY_MARKER) } diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index b762af391d..be6743599b 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -8396,7 +8396,9 @@ pub(crate) mod postgres_tests { .await .expect("local Postgres and Redis"); // A schema whose `users` table can be taken away after the link, - // so only the owner-to-agent lookup fails at revoke time. + // so only the owner-to-agent lookup fails at revoke time. It + // carries `communities` and the deletion lock key so writes pass + // community write admission. let db_url = crate::test_support::database_url(); let schema = format!("late_owner_{}", uuid::Uuid::new_v4().simple()); sqlx::raw_sql(sqlx::AssertSqlSafe(format!( @@ -8405,7 +8407,10 @@ pub(crate) mod postgres_tests { CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL); \ CREATE TABLE {schema}.relay_members (LIKE public.relay_members INCLUDING ALL); \ CREATE TABLE {schema}.join_policy_acceptances \ - (LIKE public.join_policy_acceptances INCLUDING ALL);" + (LIKE public.join_policy_acceptances INCLUDING ALL); \ + CREATE TABLE {schema}.communities (LIKE public.communities INCLUDING ALL); \ + CREATE FUNCTION {schema}.community_deletion_lock_key(target UUID) RETURNS BIGINT \ + LANGUAGE SQL IMMUTABLE STRICT AS 'SELECT public.community_deletion_lock_key(target)';" ))) .execute(state.db.pool()) .await @@ -8423,6 +8428,17 @@ pub(crate) mod postgres_tests { let tenant = fresh_tenant("late-owner.test"); let other = fresh_tenant("late-owner-other.test"); + // Both tenants pass community write admission in the schema. + for community in [&tenant, &other] { + sqlx::query(sqlx::AssertSqlSafe(format!( + "INSERT INTO {schema}.communities (id, host) VALUES ($1, $2)" + ))) + .bind(community.community().as_uuid()) + .bind(community.host()) + .execute(&admin) + .await + .expect("seed admitted community"); + } let (owner, agent) = (Keys::generate(), Keys::generate()); let agent_bytes = agent.public_key().to_bytes(); diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 5aad9689aa..0324d6f2cc 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -7150,7 +7150,8 @@ mod postgres_tests { use tokio_util::sync::CancellationToken; // A schema with bans but no `users` table: the ban commits, and only - // the owned-agent lookup fails. + // the owned-agent lookup fails. It carries `communities` and the + // deletion lock key so the ban passes community write admission. let db_url = crate::test_support::database_url(); let admin = sqlx::PgPool::connect(&db_url) .await @@ -7158,7 +7159,10 @@ mod postgres_tests { let schema = format!("revoke_owner_{}", Uuid::new_v4().simple()); sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "CREATE SCHEMA {schema}; \ - CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL);" + CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL); \ + CREATE TABLE {schema}.communities (LIKE public.communities INCLUDING ALL); \ + CREATE FUNCTION {schema}.community_deletion_lock_key(target UUID) RETURNS BIGINT \ + LANGUAGE SQL IMMUTABLE STRICT AS 'SELECT public.community_deletion_lock_key(target)';" ))) .execute(&admin) .await @@ -7173,6 +7177,13 @@ mod postgres_tests { let state = build_canvas_ingest_state(&db_url, &pool).await; let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()); let tenant = TenantContext::resolved(community, "revoke-owner.test".to_string()); + sqlx::query(sqlx::AssertSqlSafe(format!( + "INSERT INTO {schema}.communities (id, host) VALUES ($1, 'revoke-owner.test')" + ))) + .bind(community.as_uuid()) + .execute(&admin) + .await + .expect("seed admitted community"); let (owner, agent, bystander) = ( nostr::Keys::generate(), nostr::Keys::generate(), diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 44183a8f1c..42930f6c1f 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -9,7 +9,7 @@ use buzz_core::filter::{filters_match, reader_authorized_for_event}; use chrono::{TimeDelta, Utc}; use nostr::Filter; use serde::{Deserialize, Serialize}; -use tracing::{error, warn}; +use tracing::{debug, error, warn}; use crate::{handlers::push_lease::Subscription, nip98::nip98_header, state::AppState}; @@ -398,6 +398,11 @@ async fn delivery_worker_loop(state: &AppState) { deliver_one(state, &http, wake).await; } } + // The community left `active` after the host scan; the + // next scan skips it, so this is not worth a warning. + Err(e @ buzz_db::DbError::AccessDenied(_)) => { + debug!(%community, "push wake claim refused: {e}") + } Err(e) => warn!(%community, "push wake claim failed: {e}"), } } diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 7e015a5dc8..1e88eb8652 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -2401,9 +2401,7 @@ async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::Ne metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64()); return; } - Err(buzz_audit::AuditError::Database(sqlx::Error::Database(database_error))) - if database_error.code().as_deref() == Some("55P03") => - { + Err(error) if is_lock_timeout(&error) => { retries += 1; metrics::counter!("buzz_audit_log_lock_retries_total").increment(1); tracing::warn!( @@ -2423,6 +2421,20 @@ async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::Ne } } +/// Whether an audit append failed on `lock_timeout` (`55P03`), either on the +/// audit-chain lock or on the community admission lock taken before it. Both +/// are transient; the entry is kept and retried. Admission refusal +/// (`AccessDenied`) is terminal: a quiescing or fenced community refuses for +/// good. +fn is_lock_timeout(error: &buzz_audit::AuditError) -> bool { + let database_error = match error { + buzz_audit::AuditError::Database(sqlx::Error::Database(e)) + | buzz_audit::AuditError::Admission(buzz_db::DbError::Sqlx(sqlx::Error::Database(e))) => e, + _ => return false, + }; + database_error.code().as_deref() == Some("55P03") +} + impl std::fmt::Debug for AppState { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("AppState") @@ -2929,7 +2941,25 @@ pub(crate) mod tests { Arc::new(state) } + /// The lock an audit-retry test holds while the worker times out on it. + #[derive(Clone, Copy)] + enum HeldAuditLock { + /// The per-community audit-chain lock (`buzz_audit:`). + AuditChain, + /// The community deletion lock, held EXCLUSIVE as `begin_quiescing`, + /// `fence` and `abort` do; audit admission takes it SHARED first. + CommunityDeletion, + } + async fn audit_worker_retries_lock_timeout_until_original_entry_is_appended_once() { + assert_audit_worker_retries_lock_timeout(HeldAuditLock::AuditChain).await; + } + + async fn audit_worker_retries_admission_lock_timeout_until_original_entry_is_appended_once() { + assert_audit_worker_retries_lock_timeout(HeldAuditLock::CommunityDeletion).await; + } + + async fn assert_audit_worker_retries_lock_timeout(held: HeldAuditLock) { let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL"); let observer = sqlx::PgPool::connect(&database_url) .await @@ -2974,11 +3004,21 @@ pub(crate) mod tests { // Mirrors buzz_audit::service::AUDIT_LOCK_NAMESPACE. let lock_key = format!("buzz_audit:{community_id}"); let mut holder = observer.acquire().await.expect("acquire lock holder"); - sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))") - .bind(&lock_key) - .execute(&mut *holder) - .await - .expect("hold community audit lock"); + match held { + HeldAuditLock::AuditChain => { + sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))") + .bind(&lock_key) + .execute(&mut *holder) + .await + } + HeldAuditLock::CommunityDeletion => { + sqlx::query("SELECT pg_advisory_lock(community_deletion_lock_key($1))") + .bind(community_id) + .execute(&mut *holder) + .await + } + } + .expect("hold lock"); let audit = Arc::new(AuditService::new(audit_pool)); let worker = tokio::spawn({ @@ -2997,7 +3037,7 @@ pub(crate) mod tests { "SELECT EXISTS (\ SELECT 1 FROM pg_stat_activity \ WHERE application_name = $1 \ - AND query LIKE 'SELECT pg_advisory_lock%' \ + AND query LIKE 'SELECT pg_advisory_xact_lock%' \ AND wait_event = 'advisory'\ )", ) @@ -3019,11 +3059,21 @@ pub(crate) mod tests { .await .expect("audit worker never retried after lock_timeout"); - sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))") - .bind(&lock_key) - .execute(&mut *holder) - .await - .expect("release community audit lock"); + match held { + HeldAuditLock::AuditChain => { + sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))") + .bind(&lock_key) + .execute(&mut *holder) + .await + } + HeldAuditLock::CommunityDeletion => { + sqlx::query("SELECT pg_advisory_unlock(community_deletion_lock_key($1))") + .bind(community_id) + .execute(&mut *holder) + .await + } + } + .expect("release lock"); tokio::time::timeout(std::time::Duration::from_secs(3), worker) .await .expect("audit worker did not finish after lock release") @@ -3058,6 +3108,14 @@ pub(crate) mod tests { async fn audit_worker_retries_lock_timeout_until_original_entry_is_appended_once() { super::audit_worker_retries_lock_timeout_until_original_entry_is_appended_once().await; } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn audit_worker_retries_admission_lock_timeout_until_original_entry_is_appended_once() + { + super::audit_worker_retries_admission_lock_timeout_until_original_entry_is_appended_once() + .await; + } } #[test]