jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-03 16:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(relay): defer post-commit event dispatch

Browse Source 
Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757
2026-07-01 22:11:01 -04:00
parent 88c089d3b6
commit 144f00bcdf
1 changed files with 53 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
crates/buzz-relay/src/handlers/event.rs
View File

@@ -241,13 +241,53 @@ pub async fn fan_out_pubsub_event(state: &Arc<AppState>, channel_event: buzz_pub
}
}
/// Publish a stored event to subscribers and kick off async side effects.
/// Schedule post-commit delivery/side effects for a stored event.
///
/// This intentionally returns after spawning the post-commit task: NIP-01 `OK`
/// means the event was durably accepted, not that Redis publish, local fan-out,
/// audit enqueue, or workflow triggering have completed. The spawned task still
/// runs the same guarded fan-out path, Redis publish, `mark_local_event` echo
/// dedupe, and delivery metrics as the former inline path.
pub(crate) async fn dispatch_persistent_event(
tenant: &TenantContext,
state: &Arc<AppState>,
stored_event: &StoredEvent,
kind_u32: u32,
actor_pubkey_hex: &str,
) -> usize {
let tenant = tenant.clone();
let state = Arc::clone(state);
let stored_event = stored_event.clone();
let actor_pubkey_hex = actor_pubkey_hex.to_owned();
let event_id_hex = stored_event.event.id.to_hex();
metrics::counter!("buzz_post_commit_dispatch_scheduled_total").increment(1);
tokio::spawn(async move {
let recipients = dispatch_persistent_event_inner(
&tenant,
&state,
&stored_event,
kind_u32,
&actor_pubkey_hex,
)
.await;
debug!(
event_id = %event_id_hex,
recipients,
"post-commit dispatch complete"
);
});
0
}
/// Run post-commit delivery/side effects for a stored event.
async fn dispatch_persistent_event_inner(
tenant: &TenantContext,
state: &Arc<AppState>,
stored_event: &StoredEvent,
kind_u32: u32,
actor_pubkey_hex: &str,
) -> usize {
// No `crate::conformance` emit here — the spec doesn't have a
// separate fan-out action. Acceptance was already recorded at the
@@ -285,8 +325,15 @@ pub(crate) async fn dispatch_persistent_event(
"Fan-out"
);
let event_json = serde_json::to_string(&stored_event.event)
.expect("nostr::Event serialization is infallible for well-formed events");
let event_json = match serde_json::to_string(&stored_event.event) {
Ok(json) => json,
Err(e) => {
error!(event_id = %event_id_hex, "Failed to serialize event for fan-out: {e}");
metrics::counter!("buzz_post_commit_dispatch_errors_total", "stage" => "serialize")
.increment(1);
return 0;
}
};
// For viewer-private snapshots (kind:30622), live fan-out must reach only the
// owner — a kindless `ids:[…]` subscription can otherwise match it. Pull paths
// (HTTP /query, WS historical) are gated separately by reader_authorized_for_event.
@@ -1459,7 +1506,7 @@ mod tests {
let event_id_hex = event.id.to_hex();
let stored = StoredEvent::new(event, None);
super::super::dispatch_persistent_event(
super::super::dispatch_persistent_event_inner(
&tenant,
&state,
&stored,
@@ -1559,7 +1606,7 @@ mod tests {
"test precondition: the two events must have distinct ids"
);
super::super::dispatch_persistent_event(
super::super::dispatch_persistent_event_inner(
&ta,
&state,
&a_stored,
@@ -1567,7 +1614,7 @@ mod tests {
&actor_hex,
)
.await;
super::super::dispatch_persistent_event(
super::super::dispatch_persistent_event_inner(
&tb,
&state,
&b_stored,