mirror of
https://github.com/block/buzz.git
synced 2026-11-04 16:31:38 +00:00
buzz-db: pin explicit FK-child cleanup and same-workflow fire links
Audit every remaining ON DELETE CASCADE / SET NULL foreign key (BUZZ-320) against the supported parent-removal paths, with FK actions retired in the per-test schema: - attach_scheduled_workflow_run now links a scheduled fire only to a run of the same workflow, so a cross-workflow link can never block an explicit run delete once FK actions are dropped. - The community purge test seeds channel members, personal-read frontiers, owner/agent users, product feedback and an executor heartbeat, races a held child-insert transaction against quiescing, and requires explicit removal (or provenance clearing) of every child with no FK action. - The manifest-key lifecycle test runs on the retired-FK schema. FK constraints, actions, triggers and checks are unchanged; a later migration drops the actions. Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
This commit is contained in:
Elrond
@@ -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<Uuid> =
|
||||
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<Uuid> = 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() {
|
||||
|
||||
@@ -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<Uuid> = 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
|
||||
|
||||
Reference in New Issue
Block a user