diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 2196bb2a83..7993534160 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -7586,6 +7586,7 @@ mod postgres_tests { #[ignore = "requires Postgres"] async fn manifest_key_chunks_bind_freeze_execution_and_cleanup() { let (db, store) = store().await; + crate::test_support::retire_foreign_key_delete_actions(&db.pool).await; let (request, inventory) = inventoried_request(&db, &store).await; store .approve(request.id, "approver", None) @@ -7739,6 +7740,167 @@ mod postgres_tests { ); } + /// The whole-community purge is the only supported hard delete of channels, + /// personal-read accounts, and users. A child committed just before the + /// quiesce boundary must be included even with every FK action retired. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn purge_removes_parent_children_after_concurrent_insert_without_fk_actions() { + let (db, store) = store().await; + crate::test_support::retire_foreign_key_delete_actions(&db.pool).await; + let (request, inventory) = inventoried_request(&db, &store).await; + let community = request.community_id; + let owner = vec![7_u8; 32]; + let agent = vec![8_u8; 32]; + let channel = Uuid::new_v4(); + let feedback_id = Uuid::new_v4(); + sqlx::query("INSERT INTO users (community_id, pubkey) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(&owner) + .execute(&db.pool) + .await + .expect("seed owner"); + sqlx::query( + "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) VALUES ($1, $2, $3)", + ) + .bind(community.as_uuid()) + .bind(&agent) + .bind(&owner) + .execute(&db.pool) + .await + .expect("seed agent owned by user"); + sqlx::query( + "INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, 'purge', $3)", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(&owner) + .execute(&db.pool) + .await + .expect("seed channel"); + sqlx::query("INSERT INTO personal_read_accounts (community_id, actor) VALUES ($1, $2)") + .bind(community.as_uuid()) + .bind(&owner) + .execute(&db.pool) + .await + .expect("seed read account"); + sqlx::query( + "INSERT INTO product_feedback \ + (id, community_id, event_id, submitter_pubkey, body, event_created_at) \ + VALUES ($1, $2, $3, $4, 'feedback survives', now())", + ) + .bind(feedback_id) + .bind(community.as_uuid()) + .bind(vec![9_u8; 32]) + .bind(&owner) + .execute(&db.pool) + .await + .expect("seed operator feedback"); + sqlx::query( + "INSERT INTO community_deletion_executor_heartbeats \ + (executor_id, mode, request_id) VALUES ($1, 'worker', $2)", + ) + .bind(format!("purge-test-{}", Uuid::new_v4())) + .bind(request.id) + .execute(&db.pool) + .await + .expect("seed request heartbeat"); + + store + .approve(request.id, "approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + let mut writer = db + .begin_event_write_transaction(community) + .await + .expect("admit child inserter"); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey) VALUES ($1, $2, $3)", + ) + .bind(community.as_uuid()) + .bind(channel) + .bind(&owner) + .execute(writer.conn()) + .await + .expect("insert member before quiescing"); + sqlx::query( + "INSERT INTO personal_read_frontiers \ + (community_id, actor, channel_id, through_timestamp) VALUES ($1, $2, $3, now())", + ) + .bind(community.as_uuid()) + .bind(&owner) + .bind(channel) + .execute(writer.conn()) + .await + .expect("insert read frontier before quiescing"); + + let name = format!("purge-child-{}", Uuid::new_v4().simple()); + let quiesce_store = DeletionStore::new(crate::test_support::named_pool(&name).await); + let lease = claim.lease.clone(); + let quiescing = tokio::spawn(async move { quiesce_store.begin_quiescing(&lease).await }); + crate::test_support::wait_for_lock_wait(&db.pool, &name).await; + writer + .commit() + .await + .expect("commit child insert before quiesce"); + quiescing + .await + .expect("join quiesce") + .expect("quiesce after child commit"); + + 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 removed"); + let deleted = store + .purge_postgres(&token) + .await + .expect("purge without FK actions"); + for table in [ + "channel_members", + "personal_read_frontiers", + "personal_read_accounts", + "channels", + ] { + assert_eq!(deleted.get(table), Some(&1), "{table} explicitly removed"); + } + assert_eq!(deleted.get("users"), Some(&2), "owner and agent removed"); + let provenance: Option = + sqlx::query_scalar("SELECT community_id FROM product_feedback WHERE id = $1") + .bind(feedback_id) + .fetch_one(&db.pool) + .await + .expect("feedback remains"); + assert_eq!( + provenance, None, + "operator feedback loses tenant provenance" + ); + let heartbeat_request: Option = sqlx::query_scalar( + "SELECT request_id FROM community_deletion_executor_heartbeats \ + WHERE request_id = $1 LIMIT 1", + ) + .bind(request.id) + .fetch_one(&db.pool) + .await + .expect("heartbeat survives with retained request"); + assert_eq!(heartbeat_request, Some(request.id)); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn destructive_stages_serialize_with_migrations_and_fail_closed_on_new_scoped_tables() { diff --git a/crates/buzz-db/src/store/workflow.rs b/crates/buzz-db/src/store/workflow.rs index 8f63d85555..a800a9eab2 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -610,6 +610,10 @@ pub async fn attach_scheduled_workflow_run( AND workflow_id = $2 AND scheduled_for = $3 AND workflow_run_id IS NULL + AND EXISTS ( + SELECT 1 FROM workflow_runs + WHERE community_id = $1 AND workflow_id = $2 AND id = $4 + ) "#, ) .bind(community_id.as_uuid()) @@ -2475,6 +2479,43 @@ mod postgres_tests { ); } + /// A claim for workflow B cannot point at A's run. Such a link has a valid + /// run FK but would block A's explicit run delete with no FK action. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn attach_rejects_run_from_another_workflow() { + let pool = setup_pool().await; + let community = make_community(&pool).await; + let (workflow_a, _) = make_workflow_in(&pool, community).await; + let (workflow_b, _) = make_workflow_in(&pool, community).await; + let scheduled_for = Utc.with_ymd_and_hms(2026, 6, 27, 0, 3, 0).unwrap(); + claim_scheduled_workflow_fire(&pool, community, workflow_b, scheduled_for) + .await + .expect("claim B") + .expect("claim wins"); + let run_a = create_workflow_run(&pool, community, workflow_a, None, None) + .await + .expect("create A run"); + + assert!( + !attach_scheduled_workflow_run(&pool, community, workflow_b, scheduled_for, run_a) + .await + .expect("reject cross-workflow link"), + "B's fire must not reference A's run" + ); + let linked: Option = sqlx::query_scalar( + "SELECT workflow_run_id FROM scheduled_workflow_fires \ + WHERE community_id = $1 AND workflow_id = $2 AND scheduled_for = $3", + ) + .bind(community.as_uuid()) + .bind(workflow_b) + .bind(scheduled_for) + .fetch_one(&pool) + .await + .expect("read B fire"); + assert_eq!(linked, None); + } + /// Documents the retention-vs-interval coupling Sami flagged for ยง5c: /// pruning every claim below the workflow's interval makes /// `latest_scheduled_workflow_fire` return `None`, which re-introduces the