fix(workflow): serialize relay state projection

Signed-off-by: tornquist <tornquist@squareup.com>

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]

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();

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;

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