From b20caa06804703adc5bbec0ad7d0de9df5444397 Mon Sep 17 00:00:00 2001 From: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta Date: Thu, 25 Jun 2026 15:07:56 -0400 Subject: [PATCH] Fix cross-pod membership notification fanout Co-authored-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta --- crates/buzz-relay/src/handlers/event.rs | 88 +++++++++++++++++-- .../buzz-relay/src/handlers/side_effects.rs | 23 ++++- 2 files changed, 100 insertions(+), 11 deletions(-) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index da664fdadb..b9338ab0f2 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -1021,7 +1021,7 @@ mod tests { use std::sync::Arc; use axum::extract::ws::Message; - use buzz_core::kind::KIND_PRESENCE_UPDATE; + use buzz_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_PRESENCE_UPDATE}; use buzz_pubsub::ChannelEvent; use nostr::{EventBuilder, Filter, Keys, Kind}; use tokio::sync::{mpsc, Mutex}; @@ -1035,9 +1035,11 @@ mod tests { super::fanout_access::test_state().await } - fn register_presence_sub( + fn register_global_sub( state: &AppState, sub_id: &str, + filter: Filter, + pubkey: Option>, ) -> (Uuid, mpsc::Receiver) { let conn_id = Uuid::new_v4(); let (tx, rx) = mpsc::channel(10); @@ -1049,21 +1051,58 @@ mod tests { Arc::new(Mutex::new(HashMap::new())), 3, ); - state.sub_registry.register( - conn_id, - sub_id.to_string(), - vec![Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16))], - None, - ); + if let Some(pubkey) = pubkey { + state.conn_manager.set_authenticated_pubkey(conn_id, pubkey); + } + state + .sub_registry + .register(conn_id, sub_id.to_string(), vec![filter], None); (conn_id, rx) } + fn register_presence_sub( + state: &AppState, + sub_id: &str, + ) -> (Uuid, mpsc::Receiver) { + register_global_sub( + state, + sub_id, + Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16)), + None, + ) + } + + fn register_membership_sub( + state: &AppState, + sub_id: &str, + target: &Keys, + ) -> (Uuid, mpsc::Receiver) { + register_global_sub( + state, + sub_id, + Filter::new() + .kind(Kind::Custom(KIND_MEMBER_ADDED_NOTIFICATION as u16)) + .pubkey(target.public_key()), + Some(target.public_key().to_bytes().to_vec()), + ) + } + fn presence_event(status: &str) -> nostr::Event { EventBuilder::new(Kind::Custom(KIND_PRESENCE_UPDATE as u16), status) .sign_with_keys(&Keys::generate()) .expect("sign presence") } + fn membership_event(target: &Keys, channel_id: Uuid) -> nostr::Event { + EventBuilder::new(Kind::Custom(KIND_MEMBER_ADDED_NOTIFICATION as u16), "{}") + .tags([ + nostr::Tag::parse(["p", &target.public_key().to_hex()]).expect("p tag"), + nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"), + ]) + .sign_with_keys(&Keys::generate()) + .expect("sign membership notification") + } + fn event_from_ws_message(msg: Message) -> nostr::Event { let Message::Text(text) = msg else { panic!("expected text ws message"); @@ -1116,6 +1155,39 @@ mod tests { ); } + #[tokio::test] + async fn global_membership_pubsub_event_fans_out_by_p_tag() { + let state = test_state().await; + let target = Keys::generate(); + let other = Keys::generate(); + let (_target_conn, mut target_rx) = + register_membership_sub(&state, "membership-target", &target); + let (_other_conn, mut other_rx) = + register_membership_sub(&state, "membership-other", &other); + let event = membership_event(&target, Uuid::new_v4()); + let event_id = event.id; + + fan_out_pubsub_event( + &state, + ChannelEvent { + channel_id: Uuid::nil(), + event, + }, + ) + .await; + + let delivered = event_from_ws_message( + target_rx + .try_recv() + .expect("target receives membership notification"), + ); + assert_eq!(delivered.id, event_id); + assert!( + other_rx.try_recv().is_err(), + "membership notification should only reach matching #p subscribers" + ); + } + async fn redis_url_if_available() -> Option { let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index f3f2810873..37aaf01d03 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -605,9 +605,26 @@ pub async fn emit_membership_notification( return Ok(()); } - // Fan-out only — skip search indexing and workflow evaluation. Routed - // through the guarded send path for uniformity; the access gate no-ops for - // these globally-scoped (channel_id = None) events. + // Fan-out only — skip search indexing and workflow evaluation. Publish through + // Redis before local fan-out so agents connected to other relay pods receive + // the global membership notification and can subscribe to the new channel. + // Use the nil UUID sentinel for globally-scoped events, matching + // `dispatch_persistent_event` and `fan_out_pubsub_event`. + state.mark_local_event(&stored.event.id); + if let Err(e) = state.pubsub.publish_event(Uuid::nil(), &stored.event).await { + state + .local_event_ids + .invalidate(&stored.event.id.to_bytes()); + warn!( + channel = %channel_id, + target = %target_hex, + kind = notification_kind, + "membership notification Redis publish failed: {e}" + ); + } + + // Routed through the guarded send path for uniformity; the access gate no-ops + // for these globally-scoped (channel_id = None) events. crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await; info!(