merge relay deletion review corrections

* private/community-delete-relay-followup:
  fix deletion admission and quota lifecycle proof

Signed-off-by: Codex <noreply@openai.com>

crates/buzz-db/src/store/community.rs

@@ -1127,122 +1127,6 @@ mod postgres_tests {
         assert!(owned.can_create);
     }
 
-    #[tokio::test]
-    #[ignore = "requires Postgres"]
-    async fn owner_quota_reservation_survives_purge_stages_and_releases_terminally() {
-        let db = setup_db().await;
-        let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
-        let host = format!("quota-reservation-{}.example", Uuid::new_v4().simple());
-        let created = db
-            .create_community_with_owner(&host, &owner)
-            .await
-            .expect("create owned community");
-        let CreateCommunityWithOwnerResult::Created(created) = created else {
-            panic!("expected created community")
-        };
-        let mut request_id = Uuid::new_v4();
-        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(request_id)
-        .bind(created.id.as_uuid())
-        .bind(&host)
-        .bind(&owner)
-        .bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
-        .execute(&db.pool)
-        .await
-        .expect("insert owner deletion request");
-
-        let pre_purge = db
-            .list_communities_owned_by(&owner)
-            .await
-            .expect("pre-purge quota");
-        assert!(
-            pre_purge.communities.is_empty(),
-            "pending row is suppressed"
-        );
-        assert_eq!(
-            pre_purge.quota_used, 1,
-            "membership and request deduplicate"
-        );
-
-        db.deletion_store()
-            .abort(
-                request_id,
-                "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
-                "quota reservation test abort",
-            )
-            .await
-            .expect("abort quota fixture");
-        let aborted = db
-            .list_communities_owned_by(&owner)
-            .await
-            .expect("aborted quota");
-        assert_eq!(aborted.communities.len(), 1, "abort restores the live row");
-        assert_eq!(
-            aborted.quota_used, 1,
-            "abort must fall back to the preserved membership"
-        );
-        request_id = Uuid::new_v4();
-        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(request_id)
-        .bind(created.id.as_uuid())
-        .bind(&host)
-        .bind(&owner)
-        .bind("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
-        .execute(&db.pool)
-        .await
-        .expect("insert replacement pending quota fixture");
-
-        sqlx::query("DELETE FROM relay_members WHERE community_id = $1")
-            .bind(created.id.as_uuid())
-            .execute(&db.pool)
-            .await
-            .expect("simulate membership purge");
-        for stage in ["postgres_purged", "cache_purged", "logically_verified"] {
-            sqlx::query(
-                "UPDATE community_deletion_requests SET stage = $2, completed_at = NULL WHERE id = $1",
-            )
-            .bind(request_id)
-            .bind(stage)
-            .execute(&db.pool)
-            .await
-            .expect("advance quota fixture");
-            assert_eq!(
-                db.list_communities_owned_by(&owner)
-                    .await
-                    .expect("reserved quota")
-                    .quota_used,
-                1,
-                "{stage} must retain the slot"
-            );
-        }
-
-        sqlx::query(
-            "UPDATE community_deletion_requests \
-             SET stage = 'retention_pending', completed_at = now() WHERE id = $1",
-        )
-        .bind(request_id)
-        .execute(&db.pool)
-        .await
-        .expect("complete quota fixture");
-        assert_eq!(
-            db.list_communities_owned_by(&owner)
-                .await
-                .expect("released quota")
-                .quota_used,
-            0
-        );
-    }
-
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn owner_quota_serializes_concurrent_create_and_transfer_with_reservation() {

crates/buzz-db/src/store/deletion.rs

@@ -5349,12 +5349,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
@@ -5419,6 +5435,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]

crates/buzz-relay/src/api/operator.rs

@@ -492,6 +492,13 @@ pub async fn delete_community(
     })?;
     let normalized_host = normalize_candidate_host(&request.host)
         .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(deletion_api_error(
@@ -1219,6 +1226,69 @@ 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 body = owner_delete_body(&host, &owner, request_id);
+        let accepted = signed_operator_request(
+            Arc::clone(&state),
+            &operator,
+            "POST",
+            "/operator/communities/delete",
+            Some(body.clone()),
+        )
+        .await;
+        assert_eq!(accepted.status(), StatusCode::ACCEPTED);
+        assert_eq!(read_json(accepted).await["host"], host);
+
+        let receipt = signed_operator_request(
+            Arc::clone(&state),
+            &operator,
+            "POST",
+            "/operator/communities/delete/receipt",
+            Some(body),
+        )
+        .await;
+        assert_eq!(receipt.status(), StatusCode::OK);
+        let receipt = read_json(receipt).await;
+        assert_eq!(receipt["host"], host);
+        assert_eq!(receipt["request_id"], request_id.to_string());
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn owner_delete_receipt_is_bound_read_only_and_reports_aborted() {

docs/operator-community-deletion.md

@@ -150,6 +150,23 @@ eligible for execution. Transient preparation failures use the existing retry
 schedule; permanent or exhausted failures block durably. Owner-facing
 admission has no cancellation endpoint.
 
+## Client compatibility and rollout order
+
+Owner-list responses now carry the authoritative quota trio: `quota_used`,
+`quota_limit`, and `can_create`. Quota-dependent clients fail closed when any
+member of that trio is absent: they disable new Create controls even when the
+separate community-deletion capability is absent or false. They do not derive
+capacity from the number of visible rows, because an in-progress deletion can
+hide a row while still reserving its owner's slot.
+
+Roll this contract out in dependency order: relay first, then KGoose, then
+Desktop and any other quota-dependent clients. This ordering is a compatibility
+requirement, not authorization to enable owner deletion or its drain job.
+During that rollout, keep the already-published migration 0052 channel-artifact
+surface and migration 0053 owner auto-approval surface byte-for-byte intact;
+do not rewrite their migration ledger entries or recreate a persistent
+database to introduce the quota projection.
+
 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