From eae24fa1fab912730f53eaf91af9a2d07c661124 Mon Sep 17 00:00:00 2001 From: tornquist Date: Fri, 21 Aug 2026 13:51:55 +0000 Subject: [PATCH] fix(workflow): serialize relay state projection Signed-off-by: tornquist --- crates/buzz-db/src/migration.rs | 7 +++- .../buzz-relay/src/handlers/workflow_state.rs | 39 +++++++++++++++++++ migrations/0032_workflow_revision_event.sql | 7 ---- schema/schema.sql | 3 -- 4 files changed, 44 insertions(+), 12 deletions(-) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 515929c672..26c8966614 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -1071,11 +1071,14 @@ mod tests { assert!(sql.contains("WHERE kind = 30620")); assert!(sql.contains("definition.id")); assert!(sql.contains("octet_length(revision_event_id) = 32")); - assert!(sql.contains("events_live_workflow_state_coordinate_idx")); + // `events` is partitioned by created_at, so PostgreSQL cannot enforce a + // unique workflow coordinate without weakening it with that partition + // key. The relay serializes projection writers with an advisory lock. + assert!(!sql.contains("events_live_workflow_state_coordinate_idx")); let desired_schema = include_str!("../../../schema/schema.sql"); assert!(desired_schema.contains("revision_event_id BYTEA CHECK")); assert!(desired_schema.contains("octet_length(revision_event_id) = 32")); - assert!(desired_schema.contains("events_live_workflow_state_coordinate_idx")); + assert!(!desired_schema.contains("events_live_workflow_state_coordinate_idx")); } #[test] diff --git a/crates/buzz-relay/src/handlers/workflow_state.rs b/crates/buzz-relay/src/handlers/workflow_state.rs index 63df7ef0d3..73bcbdbb39 100644 --- a/crates/buzz-relay/src/handlers/workflow_state.rs +++ b/crates/buzz-relay/src/handlers/workflow_state.rs @@ -91,6 +91,16 @@ pub(crate) async fn store_workflow_state( let tags = serde_json::to_value(&event.tags)?; let received_at = chrono::Utc::now(); + // NIP-33 normally keys replacement by author, but this relay-owned + // projection must keep one coordinate across relay signing-key rotation. + // Serialize every writer for (community, kind, d) before retiring the old + // head and inserting the new one. A hash collision only adds contention; + // it cannot weaken correctness. + sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(workflow_state_lock_key(community_id, &workflow_id)) + .execute(&mut **tx) + .await?; + sqlx::query( "UPDATE events SET deleted_at = NOW() \ WHERE community_id = $1 AND kind = $2 AND d_tag = $3 \ @@ -129,6 +139,21 @@ pub(crate) async fn store_workflow_state( )) } +fn workflow_state_lock_key(community_id: CommunityId, workflow_id: &str) -> i64 { + let mut hash: u64 = 0xcbf29ce484222325; + for byte in community_id + .as_uuid() + .as_bytes() + .iter() + .chain((KIND_WORKFLOW_STATE as i32).to_le_bytes().iter()) + .chain(workflow_id.as_bytes()) + { + hash ^= *byte as u64; + hash = hash.wrapping_mul(0x100000001b3); + } + hash as i64 +} + #[cfg(test)] mod tests { use super::*; @@ -148,6 +173,20 @@ mod tests { .collect() } + #[test] + fn state_lock_coordinate_does_not_depend_on_relay_key() { + let community = CommunityId::from_uuid(Uuid::new_v4()); + let workflow = Uuid::new_v4().to_string(); + assert_eq!( + workflow_state_lock_key(community, &workflow), + workflow_state_lock_key(community, &workflow) + ); + assert_ne!( + workflow_state_lock_key(community, &workflow), + workflow_state_lock_key(community, &Uuid::new_v4().to_string()) + ); + } + #[test] fn active_state_is_relay_authored_and_preserves_owner_and_actor() { let relay = Keys::generate(); diff --git a/migrations/0032_workflow_revision_event.sql b/migrations/0032_workflow_revision_event.sql index 72da2e6751..077708235f 100644 --- a/migrations/0032_workflow_revision_event.sql +++ b/migrations/0032_workflow_revision_event.sql @@ -26,10 +26,3 @@ WHERE definition.community_id = workflow.community_id ALTER TABLE workflows ADD CONSTRAINT workflows_revision_event_id_length CHECK (revision_event_id IS NULL OR octet_length(revision_event_id) = 32); - --- Workflow state is a relay-owned projection whose coordinate survives relay --- signing-key rotation. There may be only one live head per workflow, --- regardless of which relay key authored it. -CREATE UNIQUE INDEX events_live_workflow_state_coordinate_idx - ON events (community_id, kind, d_tag) - WHERE kind = 30623 AND d_tag IS NOT NULL AND deleted_at IS NULL; diff --git a/schema/schema.sql b/schema/schema.sql index 70c4dd6a2e..ab866d9863 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -270,9 +270,6 @@ CREATE INDEX idx_events_addressable CREATE INDEX idx_events_parameterized ON events (community_id, kind, pubkey, d_tag, created_at DESC, id) WHERE d_tag IS NOT NULL AND deleted_at IS NULL; -CREATE UNIQUE INDEX events_live_workflow_state_coordinate_idx - ON events (community_id, kind, d_tag) - WHERE kind = 30623 AND d_tag IS NOT NULL AND deleted_at IS NULL; CREATE INDEX idx_events_not_before ON events (community_id, not_before) WHERE not_before IS NOT NULL AND deleted_at IS NULL AND delivered_at IS NULL; -- Full-text search. Minimal GIN over the generated tsvector; community scoping