diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 2196bb2a833..51d5a68bc0d 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,157 @@ mod postgres_tests { ); } + /// The whole-community purge is the only supported hard delete of channels, + /// personal-read accounts, and users. With every FK action retired, each + /// child must still be deleted explicitly (the quiesce barrier itself is + /// pinned by `begin_quiescing_waits_for_open_admitted_writer_note_update`). + #[tokio::test] + #[ignore = "requires Postgres"] + async fn purge_removes_fk_children_explicitly_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"); + let executor_id = format!("purge-test-{}", Uuid::new_v4()); + sqlx::query( + "INSERT INTO community_deletion_executor_heartbeats \ + (executor_id, mode, request_id) VALUES ($1, 'worker', $2)", + ) + .bind(&executor_id) + .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 seed"); + 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("seed channel member"); + 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("seed read frontier"); + writer.commit().await.expect("commit seeded children"); + 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 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 executor_id = $1", + ) + .bind(&executor_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 8f63d85555e..486d378644c 100644 --- a/crates/buzz-db/src/store/workflow.rs +++ b/crates/buzz-db/src/store/workflow.rs @@ -589,6 +589,11 @@ pub async fn latest_scheduled_workflow_fire( /// boundary. If run creation succeeds, callers should attach the run id before /// spawning execution. If run creation fails, leaving `workflow_run_id` NULL is /// intentional: the schedule instant was claimed and must not duplicate later. +/// +/// Returns `false` when no unlinked claim matches, or the run is missing or +/// belongs to another workflow. Workflow delete removes only its own fires +/// before its runs, and the fire→run FK is NO ACTION, so a cross-workflow link +/// would block that delete. pub async fn attach_scheduled_workflow_run( pool: &PgPool, community_id: CommunityId, @@ -610,6 +615,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 +2484,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 diff --git a/schema/tables/public/scheduled_workflow_fires.sql b/schema/tables/public/scheduled_workflow_fires.sql index 507bde2d215..73be34a6993 100644 --- a/schema/tables/public/scheduled_workflow_fires.sql +++ b/schema/tables/public/scheduled_workflow_fires.sql @@ -10,7 +10,9 @@ -- The FK to workflow_runs uses NO ACTION (not SET NULL): community_id is shared -- with the claim PK and is NOT NULL, so SET NULL is unimplementable here; a future -- delete of a still-linked run is blocked rather than orphaning the at-most-once --- claim row. workflow_runs are not pruned today, so this is a guardrail, not a path. +-- claim row. Workflow delete removes the workflow's own fires before its runs, so +-- attach_scheduled_workflow_run only links a run of the same workflow; a +-- cross-workflow link would otherwise block that delete. CREATE TABLE scheduled_workflow_fires ( community_id UUID NOT NULL REFERENCES communities(id),