From 37e9b68c96f4f1f1de7f1a2456ecd6515bcba2c2 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 29 Sep 2026 17:16:50 +0000 Subject: [PATCH] fix(db): restore owner deletion approval authority fence Signed-off-by: Codex --- crates/buzz-db/src/store/deletion.rs | 195 +++++++++++++++++++++++++++ docs/operator-community-deletion.md | 13 +- 2 files changed, 202 insertions(+), 6 deletions(-) diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 31af26c257..74d463e297 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -1419,6 +1419,28 @@ impl DeletionStore { let storage = serde_json::to_value(&inventory.storage)?; let frozen = serde_json::to_value(inventory)?; let mut tx = self.pool.begin().await?; + + lock_community_deletion_shared(&mut tx, token.community_id).await?; + let community = sqlx::query( + "SELECT archived_at, deletion_state, deleted_at FROM communities \ + WHERE id = $1 FOR UPDATE", + ) + .bind(token.community_id.as_uuid()) + .fetch_optional(&mut *tx) + .await? + .ok_or_else(|| { + DbError::DeletionSafety(format!( + "owner deletion {} community is missing before automatic approval", + token.request_id + )) + })?; + let current_owners: Vec = sqlx::query_scalar( + "SELECT pubkey FROM relay_members \ + WHERE community_id = $1 AND role = 'owner' ORDER BY pubkey FOR UPDATE", + ) + .bind(token.community_id.as_uuid()) + .fetch_all(&mut *tx) + .await?; let request_row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE") .bind(token.request_id) @@ -1446,6 +1468,22 @@ impl DeletionStore { if !lease_matches { return Err(stale_lease_error(token)); } + let archived_at: Option> = community.try_get("archived_at")?; + let deletion_state: String = community.try_get("deletion_state")?; + let deleted_at: Option> = community.try_get("deleted_at")?; + let owner_authority_matches = request.owner_pubkey.as_ref().is_some_and(|owner| { + current_owners.len() == 1 && current_owners.first() == Some(owner) + }); + if archived_at.is_none() + || deletion_state != "active" + || deleted_at.is_some() + || !owner_authority_matches + { + return Err(DbError::DeletionSafety(format!( + "owner deletion {} community is no longer archived under the admitted owner", + token.request_id + ))); + } if request.stage == DeletionStage::Approved { let approval: Option<(Vec, String, String)> = sqlx::query_as( "SELECT inventory_digest, approved_by, approval_origin \ @@ -5128,6 +5166,77 @@ mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn privileged_abort_fences_a_live_owner_preparation_lease() { + let (db, store) = store().await; + let (host, owner, community) = archived_owned_community(&db).await; + 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 claim = store + .claim_specific_owner_submission(request_id, "preparer", DEFAULT_LEASE_DURATION) + .await + .expect("claim owner request") + .expect("owner request is preparable"); + let inventory = FrozenInventory { + schema: store + .inventory_schema(community) + .await + .expect("schema inventory"), + storage: empty_storage_manifest(community), + }; + let (lease_owner, lease_generation, lease_is_live): (Option, i64, bool) = + sqlx::query_as( + "SELECT lease_owner, lease_generation, lease_until >= now() \ + FROM community_deletion_requests WHERE id = $1", + ) + .bind(request_id) + .fetch_one(&db.pool) + .await + .expect("read live preparation lease"); + assert_eq!(lease_owner.as_deref(), Some("preparer")); + assert_eq!(lease_generation, claim.lease.generation); + assert!(lease_is_live, "preparation lease must be live before abort"); + + let aborted = store + .abort(request_id, "recovery-operator", "cancel live preparation") + .await + .expect("abort live preparation"); + assert_eq!(aborted.stage, DeletionStage::Aborted); + assert_eq!(aborted.lease_generation, claim.lease.generation + 1); + assert!(aborted.lease_owner.is_none()); + assert!(aborted.lease_until.is_none()); + + let heartbeat_error = store + .heartbeat_owner_submission(&claim.lease, "drain", DEFAULT_LEASE_DURATION, false) + .await + .expect_err("aborted preparation lease cannot heartbeat"); + assert!(is_stale_deletion_lease(&heartbeat_error)); + let completion_error = store + .complete_owner_preparation(&claim.lease, &inventory) + .await + .expect_err("aborted preparation lease cannot approve"); + assert!(is_stale_deletion_lease(&completion_error)); + + let request = store.get(request_id).await.expect("load aborted request"); + assert_eq!(request.stage, DeletionStage::Aborted); + assert!(request.inventory_digest.is_none()); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM community_deletion_approvals WHERE request_id = $1", + ) + .bind(request_id) + .fetch_one(&db.pool) + .await + .expect("count automatic approvals"), + 0 + ); + } + /// The reversible boundary extends through `fenced`. From `drained` /// onward, destruction may have begun, so abort must stay closed. #[tokio::test] @@ -5504,6 +5613,92 @@ mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn owner_preparation_rechecks_archived_current_owner_before_automatic_approval() { + enum AuthorityDrift { + Unarchived, + OwnerChanged, + } + + let mut failures = Vec::new(); + for drift in [AuthorityDrift::Unarchived, AuthorityDrift::OwnerChanged] { + let label = match drift { + AuthorityDrift::Unarchived => "unarchived", + AuthorityDrift::OwnerChanged => "owner-changed", + }; + let (db, store) = store().await; + let (host, owner, community) = archived_owned_community(&db).await; + 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 claim = store + .claim_specific_owner_submission(request_id, "preparer", DEFAULT_LEASE_DURATION) + .await + .expect("claim owner request") + .expect("owner request is preparable"); + let inventory = FrozenInventory { + schema: store + .inventory_schema(community) + .await + .expect("schema inventory"), + storage: empty_storage_manifest(community), + }; + + match drift { + AuthorityDrift::Unarchived => { + sqlx::query("UPDATE communities SET archived_at = NULL WHERE id = $1") + .bind(community.as_uuid()) + .execute(&db.pool) + .await + .expect("simulate stale archive authority"); + } + AuthorityDrift::OwnerChanged => { + sqlx::query( + "UPDATE relay_members SET role = 'member' \ + WHERE community_id = $1 AND pubkey = $2 AND role = 'owner'", + ) + .bind(community.as_uuid()) + .bind(&owner) + .execute(&db.pool) + .await + .expect("simulate stale owner authority"); + } + } + + let completion = store + .complete_owner_preparation(&claim.lease, &inventory) + .await; + let request = store.get(request_id).await.expect("load request"); + let approval_count = sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM community_deletion_approvals WHERE request_id = $1", + ) + .bind(request_id) + .fetch_one(&db.pool) + .await + .expect("count automatic approvals"); + if completion.is_ok() + || request.stage != DeletionStage::Submitted + || request.inventory_digest.is_some() + || approval_count != 0 + { + failures.push(format!( + "{label}: completion={completion:?}, stage={}, inventory_frozen={}, approvals={approval_count}", + request.stage, + request.inventory_digest.is_some(), + )); + } + } + + assert!( + failures.is_empty(), + "stale owner authority reached automatic approval: {failures:#?}" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn stale_owner_preparation_generation_cannot_freeze_or_approve() { diff --git a/docs/operator-community-deletion.md b/docs/operator-community-deletion.md index cb6b54f595..5a6dc4cde5 100644 --- a/docs/operator-community-deletion.md +++ b/docs/operator-community-deletion.md @@ -6,10 +6,11 @@ that command as a Kubernetes CronJob; it does not call relay HTTP and it does not add another queue or retry service. Postgres remains the handoff and source of truth. A run gives already-approved -work priority. When none is ready, it may claim an operator-attested owner-origin -request at `submitted`, build the existing bounded inventory, and atomically -freeze that inventory with a digest-bound `owner_automatic` approval. The same -lease then enters the unchanged executor and resumes from durable checkpoints. +work priority. When none is ready, it may claim an operator-attested +owner-origin request at `submitted`, build the existing bounded inventory, and +atomically freeze that inventory with a digest-bound `owner_automatic` +approval. The same lease then enters the unchanged executor and resumes from +durable checkpoints. Operator-origin requests never auto-progress. `concurrencyPolicy: Forbid` prevents scheduled pod overlap, `backoffLimit: 0` prevents Kubernetes Job retries, and the deletion store remains authoritative when a pod exits, reaches its deadline, or is @@ -144,8 +145,8 @@ handoff. Owner self-serve relay admission still records only a `submitted` row and does no inventory, approval, S3 work, or execution synchronously. A successful drain -has no human approval step or cooling-off period: operator-attested owner intent is -prepared automatically under privileged policy and becomes immediately +has no human approval step or cooling-off period: operator-attested owner intent +is prepared automatically under privileged policy and becomes immediately eligible for execution. Transient preparation failures use the existing retry schedule; permanent or exhausted failures block durably. Owner-facing admission has no cancellation endpoint.