jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 06:21:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(workflow): keep one state head across relay keys

Browse Source 
Signed-off-by: tornquist <tornquist@squareup.com>
This commit is contained in:
tornquist
2026-08-21 01:21:01 +00:00
committed by Luke Tornquist
parent 9d16ddb967
commit f3cfadc982
8 changed files with 154 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
crates/buzz-cli/src/commands/workflows.rs
View File

@@ -1,4 +1,4 @@
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use sha2::{Digest, Sha256};
@@ -21,16 +21,32 @@ fn json_tag(event: &serde_json::Value, name: &str) -> Option<String> {
}
fn current_workflow_events(events: &[serde_json::Value]) -> Vec<serde_json::Value> {
let projected: HashSet<String> = events
let mut state_heads: HashMap<String, &serde_json::Value> = HashMap::new();
for event in events
.iter()
.filter(|event| event.get("kind").and_then(|value| value.as_u64()) == Some(30623))
.filter_map(|event| json_tag(event, "d"))
.collect();
{
let Some(workflow_id) = json_tag(event, "d") else {
continue;
};
match state_heads.get(&workflow_id) {
Some(current) if !json_workflow_state_is_newer(event, current) => {}
_ => {
state_heads.insert(workflow_id, event);
}
}
}
let projected: HashSet<String> = state_heads.keys().cloned().collect();
events
.iter()
.filter(
|event| match event.get("kind").and_then(|value| value.as_u64()) {
Some(30623) => json_tag(event, "status").as_deref() != Some("deleted"),
Some(30623) => json_tag(event, "d").is_some_and(|workflow_id| {
state_heads
.get(&workflow_id)
.is_some_and(|head| head.get("id") == event.get("id"))
&& json_tag(event, "status").as_deref() != Some("deleted")
}),
Some(30620) => json_tag(event, "d").is_some_and(|id| !projected.contains(&id)),
_ => false,
},
@@ -39,6 +55,30 @@ fn current_workflow_events(events: &[serde_json::Value]) -> Vec<serde_json::Valu
.collect()
}
fn json_workflow_state_is_newer(
candidate: &serde_json::Value,
current: &serde_json::Value,
) -> bool {
let candidate_created = candidate
.get("created_at")
.and_then(serde_json::Value::as_u64)
.unwrap_or_default();
let current_created = current
.get("created_at")
.and_then(serde_json::Value::as_u64)
.unwrap_or_default();
let candidate_id = candidate
.get("id")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
let current_id = current
.get("id")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
candidate_created > current_created
|| (candidate_created == current_created && candidate_id < current_id)
}
fn workflow_owner(event: &serde_json::Value) -> Option<String> {
if event.get("kind").and_then(|value| value.as_u64()) == Some(30623) {
json_tag(event, "owner")
@@ -347,4 +387,22 @@ mod tests {
});
assert!(current_workflow_events(&[legacy, deleted]).is_empty());
}
#[test]
fn relay_state_fold_selects_one_head_after_key_rotation() {
let id = uuid::Uuid::new_v4().to_string();
let old = serde_json::json!({
"id": "bb", "kind": 30623, "pubkey": "old-relay", "created_at": 100, "content": "old",
"tags": [["d", id.clone()], ["owner", "11"], ["e", "aa"], ["status", "active"]]
});
let current = serde_json::json!({
"id": "dd", "kind": 30623, "pubkey": "new-relay", "created_at": 200, "content": "new",
"tags": [["d", id], ["owner", "11"], ["e", "cc"], ["status", "active"]]
});
assert_eq!(
current_workflow_events(&[old, current.clone()]),
vec![current]
);
}
}

2
crates/buzz-db/src/migration.rs
View File

@@ -1071,9 +1071,11 @@ 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"));
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"));
}
#[test]

3
crates/buzz-relay/src/handlers/workflow_state.rs
View File

@@ -93,12 +93,11 @@ pub(crate) async fn store_workflow_state(
sqlx::query(
"UPDATE events SET deleted_at = NOW() \
WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 \
WHERE community_id = $1 AND kind = $2 AND d_tag = $3 \
AND deleted_at IS NULL",
)
.bind(community_id.as_uuid())
.bind(KIND_WORKFLOW_STATE as i32)
.bind(pubkey.as_slice())
.bind(&workflow_id)
.execute(&mut **tx)
.await?;

33
desktop/src-tauri/src/commands/workflows.rs
View File

@@ -1,4 +1,4 @@
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use serde::Serialize;
use serde_json::Value;
@@ -460,18 +460,31 @@ fn webhook_secret_from_message(message: &str) -> Option<String> {
/// Fold relay-authored current state over legacy owner-authored definitions.
/// A deleted state is authoritative and suppresses the legacy fallback.
fn workflows_from_events(events: &[nostr::Event]) -> Vec<WorkflowWire> {
let projected_ids: HashSet<String> = events
.iter()
.filter(|event| event.kind.as_u16() == 30623)
.filter_map(|event| tag_value(event, "d"))
.collect();
let mut state_heads: HashMap<String, &nostr::Event> = HashMap::new();
for event in events.iter().filter(|event| event.kind.as_u16() == 30623) {
let Some(workflow_id) = tag_value(event, "d") else {
continue;
};
match state_heads.get(&workflow_id) {
Some(current) if !workflow_state_is_newer(event, current) => {}
_ => {
state_heads.insert(workflow_id, event);
}
}
}
let projected_ids: HashSet<String> = state_heads.keys().cloned().collect();
events
.iter()
.filter(|event| {
let is_state = event.kind.as_u16() == 30623;
if is_state {
return tag_value(event, "status").as_deref() != Some("deleted");
return tag_value(event, "d").is_some_and(|workflow_id| {
state_heads
.get(&workflow_id)
.is_some_and(|head| head.id == event.id)
&& tag_value(event, "status").as_deref() != Some("deleted")
});
}
event.kind.as_u16() == 30620
&& tag_value(event, "d").is_some_and(|id| !projected_ids.contains(&id))
@@ -480,6 +493,12 @@ fn workflows_from_events(events: &[nostr::Event]) -> Vec<WorkflowWire> {
.collect()
}
fn workflow_state_is_newer(candidate: &nostr::Event, current: &nostr::Event) -> bool {
candidate.created_at > current.created_at
|| (candidate.created_at == current.created_at
&& candidate.id.to_hex() < current.id.to_hex())
}
/// Convert a workflow definition or relay state event into a [`WorkflowWire`].
fn workflow_from_event(ev: &nostr::Event) -> WorkflowWire {
let id = tag_value(ev, "d").unwrap_or_default();

51
desktop/src-tauri/src/commands/workflows_tests.rs
View File

@@ -27,6 +27,19 @@ fn wf_state(
revision: &nostr::EventId,
status: &str,
yaml: &str,
) -> nostr::Event {
wf_state_at(d, h, owner, revision, status, yaml, 100)
}
#[allow(clippy::too_many_arguments)]
fn wf_state_at(
d: &str,
h: &str,
owner: &nostr::PublicKey,
revision: &nostr::EventId,
status: &str,
yaml: &str,
created_at: u64,
) -> nostr::Event {
let relay = Keys::generate();
let owner = owner.to_hex();
@@ -42,6 +55,7 @@ fn wf_state(
.map(|tag| Tag::parse(tag).expect("parse tag"))
.collect();
EventBuilder::new(Kind::Custom(30623), yaml)
.custom_created_at(nostr::Timestamp::from(created_at))
.tags(tags)
.sign_with_keys(&relay)
.expect("sign")
@@ -100,6 +114,43 @@ fn deleted_relay_state_suppresses_legacy_definition() {
assert!(workflows_from_events(&[legacy, deleted]).is_empty());
}
#[test]
fn relay_state_fold_selects_one_head_after_key_rotation() {
let owner = Keys::generate();
let first_request = EventBuilder::new(Kind::Custom(46021), "first")
.custom_created_at(nostr::Timestamp::from(100))
.tags([])
.sign_with_keys(&Keys::generate())
.expect("sign first request");
let second_request = EventBuilder::new(Kind::Custom(46021), "second")
.custom_created_at(nostr::Timestamp::from(200))
.tags([])
.sign_with_keys(&Keys::generate())
.expect("sign second request");
let first = wf_state_at(
WF,
CHAN,
&owner.public_key(),
&first_request.id,
"active",
"name: old\nsteps: []\n",
100,
);
let second = wf_state_at(
WF,
CHAN,
&owner.public_key(),
&second_request.id,
"active",
"name: new\nsteps: []\n",
200,
);
let workflows = workflows_from_events(&[first, second]);
assert_eq!(workflows.len(), 1);
assert_eq!(workflows[0].revision, second_request.id.to_hex());
}
#[test]
fn definition_is_parsed_into_object_with_nested_fields() {
let ev = wf_event(WF, CHAN, YAML);

2
docs/nips/NIP-WF.md
View File

@@ -55,7 +55,7 @@ After accepting a create, update, or deletion request, the relay publishes kind
}
```
Only the configured relay identity is authoritative for kind `30623`. Clients must not infer workflow ownership from the state event's relay pubkey; they use the `owner` tag.
Kind `30623` is a relay-owned projection. Relay signing-key rotation replaces the existing workflow coordinate instead of creating a second head; clients select one deterministic current head per `d` tag. Clients must not infer workflow ownership from the state event's relay pubkey; they use the `owner` tag.
A deleted state has empty content and `status=deleted`. Clients prefer valid relay-signed `30623` state over legacy `30620` definitions and hide deleted state.

7
migrations/0032_workflow_revision_event.sql
View File

@@ -26,3 +26,10 @@ 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;

3
schema/schema.sql
View File

@@ -270,6 +270,9 @@ 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