diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 8d71a48daa0..ce823310fe6 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,12 +705,13 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 53); + assert_eq!(migrations.len(), 54); assert_eq!(migrations[48].version, 49); assert_eq!(migrations[49].version, 50); assert_eq!(migrations[50].version, 51); assert_eq!(migrations[51].version, 52); assert_eq!(migrations[52].version, 53); + assert_eq!(migrations[53].version, 54); assert!(migrations[48] .sql .as_str() @@ -724,6 +725,10 @@ mod postgres_tests { .sql .as_str() .contains("community_deletion_requests_owner_preparable")); + assert!(migrations[53] + .sql + .as_str() + .contains("community_deletion_requests_owner_quota_reservations")); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -1758,6 +1763,32 @@ mod postgres_tests { assert!(migration.contains("set local lock_timeout = '5s'")); } + #[test] + fn owner_deletion_quota_reservation_index_matches_desired_schema() { + let migration = MIGRATOR + .iter() + .find(|migration| migration.version == 54) + .expect("embedded migration 0054") + .sql + .as_ref() + .to_ascii_lowercase(); + let workspace_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(std::path::Path::parent) + .expect("workspace root"); + let schema = std::fs::read_to_string(workspace_root.join("schema/schema.sql")) + .expect("read schema/schema.sql") + .to_ascii_lowercase(); + + for sql in [&migration, &schema] { + assert!(sql.contains("community_deletion_requests_owner_quota_reservations")); + assert!(sql.contains("request_origin = 'owner'")); + assert!(sql.contains("stage <> 'aborted'")); + assert!(sql.contains("include (community_id, completed_at)")); + } + assert!(migration.contains("set local lock_timeout = '5s'")); + } + /// Structural parity between migration 0029's deletion surface and the /// desired-state bootstrap schema (`schema/schema.sql`). /// diff --git a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs index c6d38474e1d..e757b65d450 100644 --- a/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs @@ -311,7 +311,7 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer() production_result.is_ok(), "production migrator must preserve ingestion progress: {production_result:?}" ); - assert_eq!(version, 53); + assert_eq!(version, 54); assert_eq!(final_oid, oid, "prebuild must not be replaced"); assert_eq!(count, 4, "all writer witnesses must persist"); } diff --git a/crates/buzz-db/src/store/community.rs b/crates/buzz-db/src/store/community.rs index 461348a5cd3..cf4597662fd 100644 --- a/crates/buzz-db/src/store/community.rs +++ b/crates/buzz-db/src/store/community.rs @@ -44,7 +44,7 @@ pub enum CreateCommunityWithOwnerResult { Created(CreatedCommunityRecord), /// The host already belongs to another owner. HostExists, - /// The intended owner already owns the maximum number of communities. + /// The intended owner has reached the active or lifetime community limit. LimitReached, } @@ -61,6 +61,19 @@ pub struct OwnedCommunityRecord { pub archived_at: Option>, } +/// Owner-list rows plus the authoritative quota projection from one snapshot. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct OwnedCommunitiesPage { + /// Visible live owner memberships, excluding communities already under deletion. + pub communities: Vec, + /// De-duplicated live memberships and incomplete owner deletion reservations. + pub quota_used: i64, + /// Configured active limit only; the lifetime cap is reflected in `can_create`. + pub quota_limit: i64, + /// Whether the snapshot leaves room under both the active and lifetime caps. + pub can_create: bool, +} + /// Community row returned by an owner-authorized archive operation. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ArchivedCommunityRecord { @@ -206,13 +219,17 @@ impl Db { pub async fn list_communities_owned_by( &self, owner_pubkey: &str, - ) -> Result> { + ) -> Result { let owner_pubkey = owner_pubkey.to_ascii_lowercase(); - let mut connection = crate::observability::acquire_writer( + let connection = crate::observability::acquire_writer( &self.pool, crate::observability::WriterOperation::Authorization, ) .await?; + let mut tx = sqlx::Transaction::begin(connection, None).await?; + sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY") + .execute(&mut *tx) + .await?; let rows = sqlx::query( r#" SELECT c.id, c.host, c.created_at, c.archived_at @@ -227,11 +244,11 @@ impl Db { ORDER BY c.created_at ASC, c.host ASC "#, ) - .bind(owner_pubkey) - .fetch_all(&mut *connection) + .bind(&owner_pubkey) + .fetch_all(&mut *tx) .await?; - - rows.into_iter() + let communities = rows + .into_iter() .map(|row| { let id: Uuid = row.try_get("id")?; let host: String = row.try_get("host")?; @@ -244,7 +261,15 @@ impl Db { archived_at, }) }) - .collect() + .collect::>>()?; + let quota = relay_members::owner_quota_in_transaction(&mut tx, &owner_pubkey).await?; + tx.commit().await?; + Ok(OwnedCommunitiesPage { + communities, + quota_used: quota.active, + quota_limit: relay_members::max_communities_per_owner(), + can_create: quota.admits(), + }) } /// Returns the normalized host mapped to a community id, if the community @@ -450,14 +475,10 @@ impl Db { let host: String = row.try_get("host")?; // Enforce the limit before inserting the new owner row. - let owned_count: i64 = sqlx::query_scalar( - "SELECT count(*) FROM relay_members WHERE pubkey = $1 AND role = 'owner'", - ) - .bind(&owner_pubkey) - .fetch_one(&mut *tx) - .await?; - - if owned_count >= relay_members::max_communities_per_owner() { + if !relay_members::owner_quota_in_transaction(&mut tx, &owner_pubkey) + .await? + .admits() + { tx.rollback().await?; return Ok(CreateCommunityWithOwnerResult::LimitReached); } @@ -793,6 +814,7 @@ mod postgres_tests { "EnsuredCommunityRecord", "CreatedCommunityRecord", "OwnedCommunityRecord", + "OwnedCommunitiesPage", "ArchivedCommunityRecord", "UnarchivedCommunityRecord", ]; @@ -1093,8 +1115,168 @@ mod postgres_tests { .await .expect("list owned communities"); - assert_eq!(owned.len(), 1); - assert_eq!(owned[0].id, community_a); + assert_eq!(owned.communities.len(), 1); + assert_eq!(owned.communities[0].id, community_a); + assert_eq!(owned.quota_used, 1); + assert_eq!( + owned.quota_limit, + crate::relay_members::max_communities_per_owner() + ); + assert!(owned.can_create); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn owner_quota_serializes_concurrent_create_and_transfer_with_reservation() { + let db = setup_db().await; + let recipient = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + let source_owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple()); + + for index in 0..3 { + let host = format!("quota-live-{index}-{}.example", Uuid::new_v4().simple()); + assert!(matches!( + db.create_community_with_owner(&host, &recipient) + .await + .expect("create live quota fixture"), + CreateCommunityWithOwnerResult::Created(_) + )); + } + let reserved_host = format!("quota-held-{}.example", Uuid::new_v4().simple()); + let reserved = db + .create_community_with_owner(&reserved_host, &recipient) + .await + .expect("create reservation fixture"); + let CreateCommunityWithOwnerResult::Created(reserved) = reserved else { + panic!("expected reservation community") + }; + sqlx::query( + "INSERT INTO community_deletion_requests \ + (id, community_id, community_host, requested_by, request_origin, owner_pubkey, \ + mediating_operator_pubkey, acknowledgement_version) \ + VALUES ($1, $2, $3, $4, 'owner', $4, $5, 1)", + ) + .bind(Uuid::new_v4()) + .bind(reserved.id.as_uuid()) + .bind(&reserved_host) + .bind(&recipient) + .bind("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + .execute(&db.pool) + .await + .expect("insert reservation"); + sqlx::query("DELETE FROM relay_members WHERE community_id = $1") + .bind(reserved.id.as_uuid()) + .execute(&db.pool) + .await + .expect("simulate purged membership"); + + let transfer_host = format!("quota-transfer-{}.example", Uuid::new_v4().simple()); + let transfer_target = db + .create_community_with_owner(&transfer_host, &source_owner) + .await + .expect("create transfer target"); + let CreateCommunityWithOwnerResult::Created(transfer_target) = transfer_target else { + panic!("expected transfer target") + }; + let create_host = format!("quota-race-{}.example", Uuid::new_v4().simple()); + + let (create, transfer) = tokio::join!( + db.create_community_with_owner(&create_host, &recipient), + db.transfer_ownership(transfer_target.id, &recipient, &source_owner), + ); + let create = create.expect("concurrent create result"); + let transfer = transfer.expect("concurrent transfer result"); + let create_won = matches!(create, CreateCommunityWithOwnerResult::Created(_)); + let transfer_won = matches!( + transfer, + crate::relay_members::TransferResult::Transferred { .. } + ); + assert_ne!(create_won, transfer_won, "exactly one owner grant may win"); + assert!( + matches!(create, CreateCommunityWithOwnerResult::LimitReached) || create_won, + "create loser must observe the quota" + ); + assert!( + matches!(transfer, crate::relay_members::TransferResult::LimitReached) || transfer_won, + "transfer loser must observe the quota" + ); + assert_eq!( + db.list_communities_owned_by(&recipient) + .await + .expect("post-race quota") + .quota_used, + crate::relay_members::max_communities_per_owner() + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn completed_owner_deletions_count_toward_lifetime_cap() { + let db = setup_db().await; + let owner = format!("{:064x}", Uuid::new_v4().as_u128()); + let other_owner = format!("{:064x}", Uuid::new_v4().as_u128()); + let operator = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + + // Create and completely delete up to the lifetime cap: each tombstone + // frees its active slot but keeps its host and its lifetime count. + for i in 0..crate::relay_members::MAX_LIFETIME_COMMUNITIES_PER_OWNER { + let host = format!("lifetime-{i}-{}.example", Uuid::new_v4().simple()); + let CreateCommunityWithOwnerResult::Created(record) = db + .create_community_with_owner(&host, &owner) + .await + .expect("create under lifetime cap") + else { + panic!("create {i} must succeed below the lifetime cap") + }; + sqlx::query( + "INSERT INTO community_deletion_requests \ + (id, community_id, community_host, requested_by, request_origin, owner_pubkey, \ + mediating_operator_pubkey, acknowledgement_version, stage, completed_at) \ + VALUES ($1, $2, $3, $4, 'owner', $4, $5, 1, 'retention_pending', now())", + ) + .bind(Uuid::new_v4()) + .bind(record.id.as_uuid()) + .bind(&host) + .bind(&owner) + .bind(operator) + .execute(&db.pool) + .await + .expect("insert completed owner deletion"); + sqlx::query("DELETE FROM relay_members WHERE community_id = $1") + .bind(record.id.as_uuid()) + .execute(&db.pool) + .await + .expect("simulate purged membership"); + } + + let page = db + .list_communities_owned_by(&owner) + .await + .expect("owner list at lifetime cap"); + assert_eq!(page.quota_used, 0, "completed deletions free active slots"); + assert!(!page.can_create, "the lifetime cap still blocks creation"); + + let host = format!("lifetime-overflow-{}.example", Uuid::new_v4().simple()); + assert_eq!( + db.create_community_with_owner(&host, &owner) + .await + .expect("create past lifetime cap"), + CreateCommunityWithOwnerResult::LimitReached + ); + + let transfer_host = format!("lifetime-transfer-{}.example", Uuid::new_v4().simple()); + let CreateCommunityWithOwnerResult::Created(target) = db + .create_community_with_owner(&transfer_host, &other_owner) + .await + .expect("create transfer target") + else { + panic!("expected transfer target") + }; + assert_eq!( + db.transfer_ownership(target.id, &owner, &other_owner) + .await + .expect("transfer past lifetime cap"), + crate::relay_members::TransferResult::LimitReached + ); } #[tokio::test] diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 3cbbf05068c..96629dcdeb4 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -5104,6 +5104,7 @@ mod postgres_tests { db.list_communities_owned_by(&owner) .await .expect("owner list") + .communities .iter() .all(|row| row.id != community), "accepted deletion requests must not remain actionable archived rows" @@ -5160,6 +5161,14 @@ mod postgres_tests { .expect("privileged abort at the reversible submitted boundary"); assert_eq!(aborted.stage, DeletionStage::Aborted); assert_eq!(aborted.aborted_by.as_deref(), Some("recovery-operator")); + assert_eq!( + db.list_communities_owned_by(&owner) + .await + .expect("quota after abort") + .quota_used, + 1, + "abort releases the request reservation but preserved membership still counts" + ); // Abort reverses deletion intent, not the owner's archive decision. let (deletion_state, archived_at): (String, Option>) = @@ -5178,6 +5187,7 @@ mod postgres_tests { db.list_communities_owned_by(&owner) .await .expect("owner list after abort") + .communities .iter() .any(|row| row.id == community), "aborting the request must restore the owner's actionable archived row" @@ -5492,12 +5502,28 @@ mod postgres_tests { async fn owner_preparation_atomically_approves_exact_inventory_and_converges() { let (db, store) = store().await; let (host, owner, community) = archived_owned_community(&db).await; + assert_eq!( + db.list_communities_owned_by(&owner) + .await + .expect("initial owner quota") + .quota_used, + 1 + ); let operator = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; let request_id = Uuid::new_v4(); store .admit_owner_request(&host, &owner, operator, 1, request_id) .await .expect("admit owner request"); + let pending_quota = db + .list_communities_owned_by(&owner) + .await + .expect("pending owner quota"); + assert!(pending_quota.communities.is_empty()); + assert_eq!( + pending_quota.quota_used, 1, + "membership and incomplete request must deduplicate" + ); let claim = store .claim_specific_owner_submission(request_id, "preparer", DEFAULT_LEASE_DURATION) .await @@ -5562,6 +5588,73 @@ mod postgres_tests { .complete_owner_preparation(&claim.lease, &changed) .await .is_err()); + + store.begin_quiescing(&claim.lease).await.expect("quiesce"); + let generation = store.fence(&claim.lease).await.expect("fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + store + .freeze_destructive_storage_manifest(&token, &inventory.storage) + .await + .expect("freeze destructive storage"); + store.mark_drained(&token).await.expect("drain"); + store + .mark_bindings_removed(&token, serde_json::json!({"keys": 0})) + .await + .expect("bindings"); + store.purge_postgres(&token).await.expect("purge postgres"); + assert_eq!( + db.list_communities_owned_by(&owner) + .await + .expect("quota after membership purge") + .quota_used, + 1, + "the incomplete owner request must reserve the slot after membership purge" + ); + store + .mark_cache_purged(&token, serde_json::json!({"keys": 0})) + .await + .expect("cache"); + store + .verify_postgres_logically_deleted(&token) + .await + .expect("logical postgres verify"); + store + .mark_logically_verified(&token, serde_json::json!({"all": true})) + .await + .expect("mark verified"); + assert_eq!( + db.list_communities_owned_by(&owner) + .await + .expect("quota before logical completion") + .quota_used, + 1 + ); + store + .mark_retention_pending(&token, serde_json::json!({"shared_cas": "retained"})) + .await + .expect("production terminal transition"); + assert_eq!( + db.list_communities_owned_by(&owner) + .await + .expect("quota after logical completion") + .quota_used, + 0, + "the production logical-completion transition releases the slot" + ); + assert!(db + .lookup_community_by_host_for_management(&host) + .await + .expect("permanent tombstone lookup") + .is_some()); + assert_eq!( + db.create_community_with_owner(&host, &owner) + .await + .expect("recreate tombstoned host"), + CreateCommunityWithOwnerResult::HostExists + ); } #[tokio::test] diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 20309022cbb..f3ce399dd7e 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -544,7 +544,7 @@ pub enum TransferResult { LifecycleConflict, /// Durable deletion intent exists and wins over ownership mutation. DeletionPending, - /// The transferee already owns the maximum number of communities. + /// The transferee has reached the active or lifetime community limit. /// Enforced atomically inside the transfer transaction so concurrent /// transfers to the same recipient cannot both pass the limit. LimitReached, @@ -604,6 +604,72 @@ pub fn owner_count_advisory_lock_key(pubkey_hex: &str) -> i64 { h as i64 } +/// Lifetime cap on communities a pubkey may own, counting owner-deleted +/// communities whose tombstones permanently retain their hosts. Bounds +/// create-then-delete host squatting. Absolute: it does not scale with +/// `BUZZ_MAX_COMMUNITIES_PER_OWNER`. +pub const MAX_LIFETIME_COMMUNITIES_PER_OWNER: i64 = 20; + +/// One owner's quota usage, read inside the admitting transaction. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct OwnerQuota { + /// Live ownership plus incomplete owner-deletion reservations. + pub active: i64, + /// Live ownership plus every non-aborted owner deletion, completed or not. + pub lifetime: i64, +} + +impl OwnerQuota { + /// Whether this owner may gain one more community. + pub fn admits(self) -> bool { + self.active < max_communities_per_owner() + && self.lifetime < MAX_LIFETIME_COMMUNITIES_PER_OWNER + } +} + +/// Read an owner's active and lifetime community counts. +/// +/// `UNION` deliberately de-duplicates the live membership and deletion row +/// before PostgreSQL purges membership. An active reservation remains until +/// the logical-completion transition records `completed_at`; the lifetime +/// count keeps it forever. Aborted requests restore the community, which is +/// then counted through its live membership. +pub(crate) async fn owner_quota_in_transaction( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + owner_pubkey: &str, +) -> Result { + let (active, lifetime): (i64, i64) = sqlx::query_as( + r#" + WITH owned AS ( + SELECT community_id + FROM relay_members + WHERE pubkey = $1 AND role = 'owner' + ), deleted AS ( + SELECT community_id, completed_at + FROM community_deletion_requests + WHERE request_origin = 'owner' + AND owner_pubkey = $1 + AND stage <> 'aborted' + ) + SELECT + (SELECT count(*) FROM ( + SELECT community_id FROM owned + UNION + SELECT community_id FROM deleted WHERE completed_at IS NULL + ) active)::BIGINT, + (SELECT count(*) FROM ( + SELECT community_id FROM owned + UNION + SELECT community_id FROM deleted + ) lifetime)::BIGINT + "#, + ) + .bind(owner_pubkey) + .fetch_one(&mut **tx) + .await?; + Ok(OwnerQuota { active, lifetime }) +} + /// Atomically transfers ownership of `community` to `new_owner_pubkey`. /// /// Runs in a single transaction: @@ -618,8 +684,8 @@ pub fn owner_count_advisory_lock_key(pubkey_hex: &str) -> i64 { /// 3. Locks the current owner row `FOR UPDATE` and verifies /// `expected_owner_pubkey` matches. This prevents a stale-owner race where /// a delayed/retried request overwrites a completed transfer. -/// 4. Enforces the [`MAX_COMMUNITIES_PER_OWNER`] limit on the transferee by -/// counting owned communities inside the same transaction. +/// 4. Enforces the transferee's active and lifetime limits +/// ([`OwnerQuota::admits`]) inside the same transaction. /// 5. Upserts `new_owner_pubkey` as `owner` (insert or promote). /// 6. Demotes every other owner in this community to `member` — **not** /// `admin`, per product decision: the former owner retains no management @@ -698,14 +764,7 @@ pub async fn transfer_ownership( // 4. Enforce the transferee's community ownership limit inside the same // transaction that holds the advisory lock. This is the authoritative // check — kgoose's preflight count is advisory only. - let owned_count: i64 = sqlx::query_scalar( - "SELECT count(*) FROM relay_members WHERE pubkey = $1 AND role = 'owner'", - ) - .bind(&pubkey) - .fetch_one(&mut *tx) - .await?; - - if owned_count >= max_communities_per_owner() { + if !owner_quota_in_transaction(&mut tx, &pubkey).await?.admits() { tx.rollback().await?; return Ok(TransferResult::LimitReached); } @@ -1245,6 +1304,20 @@ mod postgres_tests { ); } + #[test] + fn owner_quota_admits_only_under_active_and_lifetime_caps() { + let quota = |active, lifetime| super::OwnerQuota { active, lifetime }; + let limit = super::max_communities_per_owner(); + let lifetime = super::MAX_LIFETIME_COMMUNITIES_PER_OWNER; + assert!(quota(0, 0).admits()); + assert!(quota(limit - 1, lifetime - 1).admits()); + assert!(!quota(limit, limit).admits(), "active cap"); + assert!( + !quota(0, lifetime).admits(), + "tombstones count toward lifetime" + ); + } + #[test] fn owner_limit_honors_positive_override() { assert_eq!(super::effective_owner_limit(Some("100")), 100); diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index c0dab3b09de..062a08cd43c 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -24,6 +24,17 @@ use crate::state::AppState; use super::{api_error, bridge, internal_error}; +fn deletion_api_error( + status: StatusCode, + code: &'static str, + message: &str, +) -> (StatusCode, Json) { + ( + status, + Json(serde_json::json!({ "error": message, "code": code })), + ) +} + /// Query parameters for `GET /operator/communities`. #[derive(Debug, Deserialize)] pub struct ListCommunitiesQuery { @@ -419,8 +430,9 @@ pub async fn unarchive_community( let record = match result { buzz_db::UnarchiveCommunityResult::Unarchived(record) => record, buzz_db::UnarchiveCommunityResult::DeletionPending => { - return Err(api_error( + return Err(deletion_api_error( StatusCode::CONFLICT, + "deletion_lifecycle_conflict", "community deletion is pending", )); } @@ -455,6 +467,12 @@ pub async fn unarchive_community( /// PostgreSQL-only transaction and returns `202`; inventory, approval, /// quiescing, object-store access, and executor work remain asynchronous. /// +/// Resubmitting the same UUID with the same host, owner, and acknowledgement +/// version returns `202` with that request's current `status`, at any stage and +/// even after membership purge, and never admits new work. Callers recover an +/// ambiguous submission by resending it; a different tuple under a known UUID +/// is `409 deletion_request_conflict`. +/// /// Owner consent is asserted by the operator, not proven to the relay. The /// operator authenticates the owner and collects the acknowledgement upstream; /// this request carries only the operator's NIP-98 signature. Authorization is @@ -470,23 +488,33 @@ pub async fn delete_community( let operator = authorize_operator_request(&state, &headers, "POST", PATH, None, Some(&body)).await?; let request: DeleteCommunityRequest = serde_json::from_slice(&body).map_err(|e| { - api_error( + deletion_api_error( StatusCode::BAD_REQUEST, + "invalid_request", &format!("invalid delete-community JSON: {e}"), ) })?; let normalized_host = normalize_candidate_host(&request.host) - .map_err(|msg| api_error(StatusCode::BAD_REQUEST, &msg))?; + .map_err(|msg| deletion_api_error(StatusCode::BAD_REQUEST, "invalid_request", &msg))?; + if normalized_host != request.host { + return Err(deletion_api_error( + StatusCode::BAD_REQUEST, + "invalid_request", + "host must use its exact canonical authority spelling", + )); + } let deployment_host = buzz_core::tenant::relay_url_authority(&state.config.relay_url); if normalized_host == deployment_host { - return Err(api_error( + return Err(deletion_api_error( StatusCode::CONFLICT, + "protected_community", "the deployment community cannot be deleted", )); } let owner = validate_pubkey_hex(&request.owner_pubkey).ok_or_else(|| { - api_error( + deletion_api_error( StatusCode::BAD_REQUEST, + "invalid_request", "invalid owner_pubkey: expected 64-char hex pubkey", ) })?; @@ -506,29 +534,37 @@ pub async fn delete_community( let accepted = match admission { buzz_db::deletion::OwnerDeletionAdmission::Accepted(request) => request, buzz_db::deletion::OwnerDeletionAdmission::NotFoundOrNotOwner => { - return Err(api_error(StatusCode::NOT_FOUND, "community not found")); + return Err(deletion_api_error( + StatusCode::NOT_FOUND, + "community_not_found", + "community not found", + )); } buzz_db::deletion::OwnerDeletionAdmission::NotArchived => { - return Err(api_error( + return Err(deletion_api_error( StatusCode::CONFLICT, + "community_not_archived", "community must be archived before deletion", )); } buzz_db::deletion::OwnerDeletionAdmission::LifecycleConflict => { - return Err(api_error( + return Err(deletion_api_error( StatusCode::CONFLICT, + "deletion_lifecycle_conflict", "community deletion lifecycle is already active", )); } buzz_db::deletion::OwnerDeletionAdmission::RequestConflict => { - return Err(api_error( + return Err(deletion_api_error( StatusCode::CONFLICT, + "deletion_request_conflict", "deletion request conflicts with existing intent", )); } buzz_db::deletion::OwnerDeletionAdmission::UnsupportedAcknowledgementVersion => { - return Err(api_error( + return Err(deletion_api_error( StatusCode::BAD_REQUEST, + "unsupported_acknowledgement_version", "unsupported acknowledgement_version", )); } @@ -539,6 +575,7 @@ pub async fn delete_community( "request_id": accepted.id, "community_id": accepted.community_id.to_string(), "host": accepted.community_host, + "acknowledgement_version": accepted.acknowledgement_version, "status": accepted.stage.to_string(), })), )) @@ -568,7 +605,7 @@ pub async fn list_owned_communities( ) })?; - let rows = state + let page = state .db .list_communities_owned_by(&owner_pubkey) .await @@ -576,12 +613,15 @@ pub async fn list_owned_communities( Ok(Json(serde_json::json!({ "owner_pubkey": owner_pubkey, - "communities": rows.into_iter().map(|row| serde_json::json!({ + "communities": page.communities.into_iter().map(|row| serde_json::json!({ "community_id": row.id.to_string(), "host": row.host, "created_at": row.created_at, "archived_at": row.archived_at, })).collect::>(), + "quota_used": page.quota_used, + "quota_limit": page.quota_limit, + "can_create": page.can_create, }))) } @@ -673,15 +713,16 @@ pub async fn transfer_community( )); } buzz_db::relay_members::TransferResult::DeletionPending => { - return Err(api_error( + return Err(deletion_api_error( StatusCode::CONFLICT, + "deletion_lifecycle_conflict", "community deletion is pending", )); } buzz_db::relay_members::TransferResult::LimitReached => { return Err(api_error( StatusCode::CONFLICT, - "limit_reached: transferee already owns the maximum number of communities", + "limit_reached: transferee has reached the community limit", )); } }; @@ -1111,6 +1152,7 @@ mod postgres_tests { assert_eq!(response.status(), StatusCode::ACCEPTED); let json = read_json(response).await; assert_eq!(json["request_id"], request_id.to_string()); + assert_eq!(json["acknowledgement_version"], 1); assert_eq!(json["status"], "submitted"); let request = state .db @@ -1127,6 +1169,221 @@ mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn owner_delete_admission_requires_exact_canonical_host_without_mutation() { + let operator = Keys::generate(); + let owner = Keys::generate(); + let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else { + return; + }; + let host = format!("community-{}.example", Uuid::new_v4().simple()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &owner) + .await + .status(), + StatusCode::OK + ); + archive_for_owner_deletion(&state, &host, &owner).await; + + for (label, repaired) in [ + ("whitespace", format!(" {host}")), + ("case", host.to_uppercase()), + ("url", format!("https://{host}")), + ] { + let request_id = Uuid::new_v4(); + let response = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(owner_delete_body(&repaired, &owner, request_id)), + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{label}"); + assert_eq!(read_json(response).await["code"], "invalid_request"); + assert_no_persisted_request(&state, request_id, label).await; + } + + let request_id = Uuid::new_v4(); + let accepted = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(owner_delete_body(&host, &owner, request_id)), + ) + .await; + assert_eq!(accepted.status(), StatusCode::ACCEPTED); + assert_eq!(read_json(accepted).await["host"], host); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn owner_delete_resubmission_reports_current_status_without_new_intent() { + let operator = Keys::generate(); + let outsider = Keys::generate(); + let owner = Keys::generate(); + let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else { + return; + }; + let host = format!("community-{}.example", Uuid::new_v4().simple()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &owner) + .await + .status(), + StatusCode::OK + ); + archive_for_owner_deletion(&state, &host, &owner).await; + let request_id = Uuid::new_v4(); + let body = owner_delete_body(&host, &owner, request_id); + assert_eq!( + signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(body.clone()), + ) + .await + .status(), + StatusCode::ACCEPTED + ); + let before = state + .db + .deletion_store() + .list(1_000) + .await + .expect("list requests before replays") + .into_iter() + .filter(|request| request.id == request_id) + .count(); + + let replay = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(body.clone()), + ) + .await; + assert_eq!(replay.status(), StatusCode::ACCEPTED); + let replay = read_json(replay).await; + assert_eq!(replay["request_id"], request_id.to_string()); + assert_eq!(replay["host"], host); + assert_eq!(replay["acknowledgement_version"], 1); + assert_eq!(replay["status"], "submitted"); + + // Recovery must survive membership purge: the replay converges on the + // stored tuple before any owner or archive check runs. + sqlx::query("DELETE FROM relay_members WHERE pubkey = $1 AND role = 'owner'") + .bind(owner.public_key().to_hex()) + .execute(state.db.pool()) + .await + .expect("simulate membership purge"); + let purged_replay = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(body.clone()), + ) + .await; + assert_eq!(purged_replay.status(), StatusCode::ACCEPTED); + assert_eq!(read_json(purged_replay).await["status"], "submitted"); + + for mismatch in [ + serde_json::json!({ + "host": format!("wrong-{host}"), + "owner_pubkey": owner.public_key().to_hex(), + "request_id": request_id, + "acknowledgement_version": 1, + }), + serde_json::json!({ + "host": host, + "owner_pubkey": Keys::generate().public_key().to_hex(), + "request_id": request_id, + "acknowledgement_version": 1, + }), + ] { + let response = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(mismatch.to_string()), + ) + .await; + assert_eq!(response.status(), StatusCode::CONFLICT); + assert_eq!( + read_json(response).await["code"], + "deletion_request_conflict" + ); + } + + // Version validation precedes the UUID lookup, so a changed version is + // rejected as unsupported rather than as a request conflict. + let changed_version = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some( + serde_json::json!({ + "host": host, + "owner_pubkey": owner.public_key().to_hex(), + "request_id": request_id, + "acknowledgement_version": 2, + }) + .to_string(), + ), + ) + .await; + assert_eq!(changed_version.status(), StatusCode::BAD_REQUEST); + assert_eq!( + read_json(changed_version).await["code"], + "unsupported_acknowledgement_version" + ); + + let outsider_response = signed_operator_request( + Arc::clone(&state), + &outsider, + "POST", + "/operator/communities/delete", + Some(body.clone()), + ) + .await; + assert_eq!(outsider_response.status(), StatusCode::FORBIDDEN); + + state + .db + .deletion_store() + .abort(request_id, &operator.public_key().to_hex(), "test abort") + .await + .expect("abort request"); + let aborted = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(body), + ) + .await; + assert_eq!(aborted.status(), StatusCode::ACCEPTED); + assert_eq!(read_json(aborted).await["status"], "aborted"); + + let after = state + .db + .deletion_store() + .list(1_000) + .await + .expect("list requests after replays") + .into_iter() + .filter(|request| request.id == request_id) + .count(); + assert_eq!(before, after, "replays must not add request rows"); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn owner_delete_endpoint_rejects_protected_host_and_malformed_or_mismatched_target() { @@ -1145,6 +1402,7 @@ mod postgres_tests { ) .await; assert_eq!(protected.status(), StatusCode::CONFLICT); + assert_eq!(read_json(protected).await["code"], "protected_community"); // Each malformed case keeps every unrelated field valid so it reaches // the guard under test instead of tripping an earlier one, and none of @@ -1170,6 +1428,7 @@ mod postgres_tests { ) .await; assert_eq!(bad_host.status(), StatusCode::BAD_REQUEST); + assert_eq!(read_json(bad_host).await["code"], "invalid_request"); assert_no_persisted_request(&state, bad_host_id, "unnormalizable host").await; let bad_pubkey_id = Uuid::new_v4(); @@ -1219,6 +1478,10 @@ mod postgres_tests { ) .await; assert_eq!(bad_version.status(), StatusCode::BAD_REQUEST); + assert_eq!( + read_json(bad_version).await["code"], + "unsupported_acknowledgement_version" + ); assert_no_persisted_request(&state, bad_version_id, "unsupported acknowledgement").await; let unknown_host_id = Uuid::new_v4(); @@ -1239,6 +1502,7 @@ mod postgres_tests { ) .await; assert_eq!(unknown_host.status(), StatusCode::NOT_FOUND); + assert_eq!(read_json(unknown_host).await["code"], "community_not_found"); assert_no_persisted_request(&state, unknown_host_id, "unprovisioned host").await; let bad_uuid = signed_operator_request( @@ -1924,6 +2188,128 @@ mod postgres_tests { assert_eq!(owner_member.role, "owner"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn unarchive_pending_deletion_returns_stable_conflict_without_mutation() { + let operator = Keys::generate(); + let owner = Keys::generate(); + let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else { + return; + }; + let host = format!("community-{}.example", Uuid::new_v4().simple()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &owner) + .await + .status(), + StatusCode::OK + ); + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("lookup community") + .expect("community exists"); + archive_for_owner_deletion(&state, &host, &owner).await; + let request_id = Uuid::new_v4(); + assert_eq!( + signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(owner_delete_body(&host, &owner, request_id)), + ) + .await + .status(), + StatusCode::ACCEPTED + ); + + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect lifecycle assertion pool"); + let before_lifecycle: ( + Option>, + String, + Option>, + ) = sqlx::query_as( + "SELECT archived_at, deletion_state, deleted_at FROM communities WHERE id = $1", + ) + .bind(community.id.as_uuid()) + .fetch_one(&pool) + .await + .expect("read lifecycle before unarchive conflict"); + let before_request = state + .db + .deletion_store() + .get(request_id) + .await + .expect("read deletion request before unarchive conflict"); + let owner_hex = owner.public_key().to_hex(); + let before_owner_role = state + .db + .get_relay_member(community.id, &owner_hex) + .await + .expect("read owner before unarchive conflict") + .expect("owner exists before unarchive conflict") + .role; + + let body = serde_json::json!({ + "host": host, + "owner_pubkey": owner_hex, + }) + .to_string(); + let response = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/unarchive", + Some(body), + ) + .await; + assert_eq!(response.status(), StatusCode::CONFLICT); + let json = read_json(response).await; + assert_eq!(json["code"], "deletion_lifecycle_conflict"); + assert_eq!(json["error"], "community deletion is pending"); + + let after_lifecycle: ( + Option>, + String, + Option>, + ) = sqlx::query_as( + "SELECT archived_at, deletion_state, deleted_at FROM communities WHERE id = $1", + ) + .bind(community.id.as_uuid()) + .fetch_one(&pool) + .await + .expect("read lifecycle after unarchive conflict"); + assert_eq!(after_lifecycle, before_lifecycle); + assert_eq!( + state + .db + .deletion_store() + .get(request_id) + .await + .expect("read deletion request after unarchive conflict"), + before_request + ); + assert_eq!( + state + .db + .get_relay_member(community.id, &owner.public_key().to_hex()) + .await + .expect("read owner after unarchive conflict") + .expect("owner exists after unarchive conflict") + .role, + before_owner_role + ); + assert!(state + .db + .lookup_community_by_host(&host) + .await + .expect("active lookup after unarchive conflict") + .is_none()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn archive_publish_failure_is_retryable_and_preserves_timestamp() { @@ -2027,6 +2413,7 @@ mod postgres_tests { .await .expect("owned communities"); let row = owned + .communities .iter() .find(|row| row.host == host) .expect("archived row"); @@ -2257,6 +2644,137 @@ mod postgres_tests { .await; } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn transfer_pending_deletion_returns_stable_conflict_without_mutation() { + let operator = Keys::generate(); + let initial_owner = Keys::generate(); + let new_owner = Keys::generate(); + let Some(state) = operator_test_state(std::slice::from_ref(&operator)).await else { + return; + }; + let host = format!("community-{}.example", Uuid::new_v4().simple()); + assert_eq!( + provision_community(Arc::clone(&state), &operator, &host, &initial_owner) + .await + .status(), + StatusCode::OK + ); + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("lookup community") + .expect("community exists"); + archive_for_owner_deletion(&state, &host, &initial_owner).await; + let request_id = Uuid::new_v4(); + assert_eq!( + signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/delete", + Some(owner_delete_body(&host, &initial_owner, request_id)), + ) + .await + .status(), + StatusCode::ACCEPTED + ); + + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect lifecycle assertion pool"); + let before_lifecycle: ( + Option>, + String, + Option>, + ) = sqlx::query_as( + "SELECT archived_at, deletion_state, deleted_at FROM communities WHERE id = $1", + ) + .bind(community.id.as_uuid()) + .fetch_one(&pool) + .await + .expect("read lifecycle before transfer conflict"); + let before_request = state + .db + .deletion_store() + .get(request_id) + .await + .expect("read deletion request before transfer conflict"); + let initial_owner_hex = initial_owner.public_key().to_hex(); + let new_owner_hex = new_owner.public_key().to_hex(); + let before_owner_role = state + .db + .get_relay_member(community.id, &initial_owner_hex) + .await + .expect("read owner before transfer conflict") + .expect("owner exists before transfer conflict") + .role; + assert!(state + .db + .get_relay_member(community.id, &new_owner_hex) + .await + .expect("read transferee before transfer conflict") + .is_none()); + + let body = serde_json::json!({ + "community_id": community.id.to_string(), + "new_owner_pubkey": new_owner_hex, + "expected_owner_pubkey": initial_owner_hex, + }) + .to_string(); + let response = signed_operator_request( + Arc::clone(&state), + &operator, + "POST", + "/operator/communities/transfer", + Some(body), + ) + .await; + assert_eq!(response.status(), StatusCode::CONFLICT); + let json = read_json(response).await; + assert_eq!(json["code"], "deletion_lifecycle_conflict"); + assert_eq!(json["error"], "community deletion is pending"); + + let after_lifecycle: ( + Option>, + String, + Option>, + ) = sqlx::query_as( + "SELECT archived_at, deletion_state, deleted_at FROM communities WHERE id = $1", + ) + .bind(community.id.as_uuid()) + .fetch_one(&pool) + .await + .expect("read lifecycle after transfer conflict"); + assert_eq!(after_lifecycle, before_lifecycle); + assert_eq!( + state + .db + .deletion_store() + .get(request_id) + .await + .expect("read deletion request after transfer conflict"), + before_request + ); + assert_eq!( + state + .db + .get_relay_member(community.id, &initial_owner.public_key().to_hex()) + .await + .expect("read owner after transfer conflict") + .expect("owner exists after transfer conflict") + .role, + before_owner_role + ); + assert!(state + .db + .get_relay_member(community.id, &new_owner.public_key().to_hex()) + .await + .expect("read transferee after transfer conflict") + .is_none()); + } + /// Transfer with an invalid community_id returns 400. #[tokio::test] #[ignore = "requires Postgres"] diff --git a/crates/buzz-relay/src/handlers/community_provisioning.rs b/crates/buzz-relay/src/handlers/community_provisioning.rs index 1b13e6d921a..e8a5ee90d10 100644 --- a/crates/buzz-relay/src/handlers/community_provisioning.rs +++ b/crates/buzz-relay/src/handlers/community_provisioning.rs @@ -287,10 +287,7 @@ pub async fn provision_community( return Err("community already exists".to_string()); } buzz_db::CreateCommunityWithOwnerResult::LimitReached => { - return Err( - "limit_reached: owner already owns the maximum number of communities" - .to_string(), - ); + return Err("limit_reached: owner has reached the community limit".to_string()); } }; diff --git a/docs/operator-community-deletion.md b/docs/operator-community-deletion.md index 0a3b1de5148..d4689006d58 100644 --- a/docs/operator-community-deletion.md +++ b/docs/operator-community-deletion.md @@ -151,6 +151,45 @@ eligible for execution. Transient preparation failures use the existing retry schedule; permanent or exhausted failures block durably. Owner-facing admission has no cancellation endpoint. +Admission is idempotent on the request UUID. Resending the same UUID with the +same host, owner, and acknowledgement version returns `202` with that request's +current `status` at any stage, including after membership purge, and admits no +new work. Clients recover an ambiguous submission by resending it. The same +UUID with a different tuple returns `409 deletion_request_conflict`. + +The acknowledgement version is a compile-time constant +(`OWNER_DELETION_ACKNOWLEDGEMENT_VERSION`), not operator configuration. +Admission validates it before the UUID lookup, and the executor claims and +leases only requests carrying the current version. Raising it therefore makes +resends of pending requests made under the old version fail with +`400 unsupported_acknowledgement_version`, and leaves already-admitted +old-version requests unclaimed in `submitted`. A version bump must ship code +that keeps admitting replays of, and executing, requests at the prior version +until none remain in a non-terminal stage. + +## Owner quota + +The relay enforces two per-owner caps on create and on transfer-in, both as +`limit_reached`: + +- **Active:** live ownership plus incomplete owner deletions + (`BUZZ_MAX_COMMUNITIES_PER_OWNER`, default 5). A deletion keeps its slot + until logical completion records `completed_at`. +- **Lifetime:** live ownership plus every non-aborted owner deletion, including + completed ones, capped at an absolute 20 regardless of the active limit. + Deleted communities keep their hosts as permanent tombstones, so this bounds + create-then-delete host squatting. Aborted deletions restore the community + and count only through its live membership. + +Owner-list responses carry `quota_used` (active), `quota_limit` (active), and +`can_create` (both caps). `can_create: false` is the only signal a client needs: +show a generic community-limit message and do not derive a reason from the +counts, because an owner at the lifetime cap can have `quota_used` below +`quota_limit`. The projection is advisory: clients may use it for +UX, but the relay's `limit_reached` is authoritative, and quota changes have no +deployment order. Keep owner deletion off until this relay and the drain +executor are live; on rollback, turn deletion off before rolling back the relay. + The chart has no existing PrometheusRule or provider-neutral CronJob alert integration. Operators must alert on failed/missed Jobs and long-running active Jobs in their deployment platform. Adding a chart-native alert abstraction is diff --git a/migrations/0054_owner_deletion_quota_reservation.sql b/migrations/0054_owner_deletion_quota_reservation.sql new file mode 100644 index 00000000000..c1487983eb2 --- /dev/null +++ b/migrations/0054_owner_deletion_quota_reservation.sql @@ -0,0 +1,10 @@ +-- Keep owner quota lookups bounded: incomplete owner deletions reserve an +-- active slot, and every non-aborted owner deletion counts toward the +-- lifetime cap after relay membership is purged. +SET LOCAL lock_timeout = '5s'; + +CREATE INDEX community_deletion_requests_owner_quota_reservations + ON community_deletion_requests (owner_pubkey) + INCLUDE (community_id, completed_at) + WHERE request_origin = 'owner' + AND stage <> 'aborted'; diff --git a/schema/schema.sql b/schema/schema.sql index 317044b0b56..3bf94907a89 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1309,6 +1309,11 @@ CREATE INDEX community_deletion_requests_owner_preparable WHERE request_origin = 'owner' AND stage = 'submitted' AND blocked_at IS NULL; +CREATE INDEX community_deletion_requests_owner_quota_reservations + ON community_deletion_requests (owner_pubkey) + INCLUDE (community_id, completed_at) + WHERE request_origin = 'owner' + AND stage <> 'aborted'; CREATE TABLE community_deletion_approvals ( request_id UUID PRIMARY KEY,