jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 12:40:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Scope relay runtime state by community (#1658)

Browse Source 
Co-authored-by: jm <jm@block.xyz>
This commit is contained in:
Jordan Mecom
2026-07-16 11:07:08 -07:00
committed by GitHub
parent 3dd236eb6f
commit d52dedb06f
7 changed files with 241 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
crates/buzz-relay/src/audio/handler.rs
View File

@@ -341,7 +341,9 @@ async fn handle_active_audio_connection(
}
}
let room = state.audio_rooms.get_or_create(channel_id);
let room = state
.audio_rooms
.get_or_create(tenant.community(), channel_id);
// Re-check archived status after obtaining the room. This closes the
// cross-boundary race: a joiner that passed ensure_membership before
@@ -359,12 +361,16 @@ async fn handle_active_audio_connection(
.into(),
))
.await;
state.audio_rooms.cleanup_if_empty(channel_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
Err(e) => {
warn!(channel_id = %channel_id, "pre-join channel check failed (fail-closed): {e}");
state.audio_rooms.cleanup_if_empty(channel_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
Ok(_) => {} // Channel exists and is not archived — proceed.
@@ -438,7 +444,9 @@ async fn handle_active_audio_connection(
remote_rejection_ws_error(&reason).to_string().into(),
))
.await;
state.audio_rooms.cleanup_if_empty(channel_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
Err(crate::audio::join::DialError::Mesh(e)) => {
@@ -453,7 +461,9 @@ async fn handle_active_audio_connection(
.into(),
))
.await;
state.audio_rooms.cleanup_if_empty(channel_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
}
@@ -587,7 +597,9 @@ async fn handle_active_audio_connection(
.is_err()
{
room.remove_peer(peer_id);
state.audio_rooms.cleanup_if_empty(channel_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
} else {
@@ -797,7 +809,9 @@ async fn handle_active_audio_connection(
room_emptied = false;
}
Ok(()) => {
room_emptied = state.audio_rooms.cleanup_if_empty(channel_id);
room_emptied = state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
emit_participant_event(
&state,
@@ -811,7 +825,9 @@ async fn handle_active_audio_connection(
}
}
} else {
room_emptied = state.audio_rooms.cleanup_if_empty(channel_id);
room_emptied = state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
}
// Owner path: release this room's lease when the room empties, so a new

27
crates/buzz-relay/src/audio/join.rs
View File

@@ -1171,7 +1171,13 @@ impl<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
let msg = match event {
Ok(delta) => roster_delta_msg(delta),
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
let Some(room) = self.rooms.get(session_id) else {
let Some(community_id) = stream_community else {
break Ok(());
};
let Some(room) = self.rooms.get(
CommunityId::from_uuid(community_id),
session_id,
) else {
break Ok(());
};
roster_snapshot_msg(&room)
@@ -1245,7 +1251,7 @@ impl<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
teardown_reason = Some(GoodbyeReason::Draining);
break Ok(());
}
let room = self.rooms.get_or_create(session_id);
let room = self.rooms.get_or_create(community, session_id);
// Subscribe before admission; PeerRegistered carries a
// snapshot after admission, and queued deltas at or below
// that revision are ignored by the receiver.
@@ -1282,7 +1288,10 @@ impl<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
}
HuddleControlMsg::UnregisterPeer { pubkey } => {
if let Some(peer_id) = registered.remove(&pubkey) {
if let Some(room) = self.rooms.get(session_id) {
if let Some(room) = stream_community.and_then(|community_id| {
self.rooms
.get(CommunityId::from_uuid(community_id), session_id)
}) {
let peer_index = room.peers.get(&peer_id).map(|peer| peer.peer_index);
room.remove_peer(peer_id);
if let Some(peer_index) = peer_index {
@@ -1299,7 +1308,10 @@ impl<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
}
}
HuddleControlMsg::RosterResync => {
let Some(room) = self.rooms.get(session_id) else {
let Some(room) = stream_community.and_then(|community_id| {
self.rooms
.get(CommunityId::from_uuid(community_id), session_id)
}) else {
break Ok(());
};
stream
@@ -1335,7 +1347,10 @@ impl<D: HuddleDirectory + ?Sized> HuddleControlAcceptor<D> {
// Teardown: drop every peer this stream registered, regardless of how
// the loop ended. Dropping the peer drops its `audio_tx`, which ends the
// matching `spawn_remote_peer_sink` task.
if let Some(room) = self.rooms.get(session_id) {
if let Some(room) = stream_community.and_then(|community_id| {
self.rooms
.get(CommunityId::from_uuid(community_id), session_id)
}) {
for (pubkey, peer_id) in registered {
let peer_index = room.peers.get(&peer_id).map(|peer| peer.peer_index);
room.remove_peer(peer_id);
@@ -2261,7 +2276,7 @@ mod tests {
let session_id = Uuid::new_v4();
let fenced = fenced_owned_by(owner_rt, session_id);
let rooms = Arc::new(AudioRoomManager::new());
let room = rooms.get_or_create(session_id);
let room = rooms.get_or_create(community(), session_id);
let (_local_id, _local_index, _audio_rx, mut local_ctrl_rx) =
room.add_peer("owner-local".into(), 2).unwrap();
// Discard the local peer's own roster delta; this assertion targets the

6
crates/buzz-relay/src/audio/mesh.rs
View File

@@ -223,9 +223,11 @@ impl MeshAudioRouter {
return verdict;
}
let Some(room) = self.rooms.get(session_id) else {
let Some(room) = self.rooms.get_unambiguous_by_channel(session_id) else {
// No local room for this session: nothing to deliver to. Not an
// error — membership can race a datagram in flight.
// error — membership can race a datagram in flight. An ambiguous
// same-UUID room collision is also dropped because the current
// media envelope has no community label.
return verdict;
};

104
crates/buzz-relay/src/audio/room.rs
View File

@@ -8,6 +8,7 @@
//! Frames are opaque Opus bytes — the relay never decodes audio.
//! `try_send` is used throughout: real-time audio tolerates drops, never queues.
use buzz_core::CommunityId;
use bytes::Bytes;
use dashmap::DashMap;
use std::sync::Arc;
@@ -117,8 +118,8 @@ struct AdmissionGuard {
///
/// Pin is per-`Room`-instance and clears when the manager evicts the
/// Room via [`AudioRoomManager::cleanup_if_empty`] — the next
/// `get_or_create` for the same channel id then constructs a fresh
/// `Room` with a fresh `AdmissionGuard` (and therefore `None` pin),
/// `get_or_create` for the same community-local channel then constructs a
/// fresh `Room` with a fresh `AdmissionGuard` (and therefore `None` pin),
/// so a new generation of joiners can negotiate a new version.
/// A momentarily-empty-but-not-yet-cleaned-up Room keeps its pin so
/// reconnecting peers don't accidentally renegotiate mid-call. See
@@ -158,6 +159,8 @@ impl AdmissionGuard {
/// A single audio room for one channel.
pub struct Room {
/// Community this room belongs to.
pub community_id: CommunityId,
/// Channel UUID this room belongs to.
pub channel_id: Uuid,
/// Connected peers keyed by peer UUID.
@@ -170,10 +173,11 @@ pub struct Room {
}
impl Room {
/// Create an empty room for the given channel.
pub fn new(channel_id: Uuid) -> Self {
/// Create an empty room for the given community-local channel.
pub fn new(community_id: CommunityId, channel_id: Uuid) -> Self {
let (roster_tx, _) = broadcast::channel(64);
Self {
community_id,
channel_id,
peers: DashMap::new(),
guard: std::sync::Mutex::new(AdmissionGuard::new()),
@@ -487,7 +491,7 @@ impl Room {
/// Global registry of active audio rooms.
pub struct AudioRoomManager {
rooms: DashMap<Uuid, Arc<Room>>,
rooms: DashMap<(CommunityId, Uuid), Arc<Room>>,
}
impl AudioRoomManager {
@@ -499,25 +503,47 @@ impl AudioRoomManager {
}
/// Get an existing room or create a new one.
pub fn get_or_create(&self, channel_id: Uuid) -> Arc<Room> {
///
/// Channel UUIDs are only unique inside a community. The room key must
/// carry both labels so two tenants that legitimately reuse the same UUID
/// never share peer lists, protocol pins, or audio frames.
pub fn get_or_create(&self, community_id: CommunityId, channel_id: Uuid) -> Arc<Room> {
self.rooms
.entry(channel_id)
.or_insert_with(|| Arc::new(Room::new(channel_id)))
.entry((community_id, channel_id))
.or_insert_with(|| Arc::new(Room::new(community_id, channel_id)))
.clone()
}
/// Look up an existing room without creating one. Returns `None` when this
/// pod hosts no room for the channel — the cross-pod media path uses this
/// so an inbound datagram for a session we don't participate in is dropped,
/// never materializing an empty room.
pub fn get(&self, channel_id: Uuid) -> Option<Arc<Room>> {
self.rooms.get(&channel_id).map(|r| r.clone())
/// Look up an existing community-local room without creating one.
pub fn get(&self, community_id: CommunityId, channel_id: Uuid) -> Option<Arc<Room>> {
self.rooms
.get(&(community_id, channel_id))
.map(|room| room.clone())
}
/// Look up a room for a mesh datagram that carries only a channel UUID.
///
/// The current mesh media envelope does not carry a community identifier.
/// If two active communities use the same channel UUID, routing would be
/// ambiguous, so fail closed instead of delivering one community's audio
/// to the other. Control-path lookups always use [`Self::get`].
pub fn get_unambiguous_by_channel(&self, channel_id: Uuid) -> Option<Arc<Room>> {
let mut matches = self
.rooms
.iter()
.filter(|entry| entry.key().1 == channel_id)
.map(|entry| Arc::clone(entry.value()));
let room = matches.next()?;
if matches.next().is_some() {
return None;
}
Some(room)
}
/// Remove the room if it has no peers. Returns `true` if the room was removed.
pub fn cleanup_if_empty(&self, channel_id: Uuid) -> bool {
pub fn cleanup_if_empty(&self, community_id: CommunityId, channel_id: Uuid) -> bool {
self.rooms
.remove_if(&channel_id, |_, room| room.is_empty())
.remove_if(&(community_id, channel_id), |_, room| room.is_empty())
.is_some()
}
}
@@ -533,7 +559,7 @@ mod tests {
use super::*;
fn fresh_room() -> Room {
Room::new(Uuid::new_v4())
Room::new(CommunityId::from_uuid(Uuid::new_v4()), Uuid::new_v4())
}
#[test]
@@ -644,9 +670,10 @@ mod tests {
#[test]
fn manager_cleanup_resets_version_pin() {
let manager = AudioRoomManager::new();
let community_id = CommunityId::from_uuid(Uuid::new_v4());
let channel_id = Uuid::new_v4();
let room1 = manager.get_or_create(channel_id);
let room1 = manager.get_or_create(community_id, channel_id);
let (peer_id, _, _, _) = room1
.add_peer("alice".to_string(), 2)
.expect("first peer admits");
@@ -655,16 +682,53 @@ mod tests {
.remove_peer_and_check_ended(peer_id)
.expect("peer existed");
assert!(ended, "single-peer room should end on its last departure");
assert!(manager.cleanup_if_empty(channel_id));
assert!(manager.cleanup_if_empty(community_id, channel_id));
// Next joiner with a different version on the same channel id gets a
// brand-new room (no v=2 pin carried over from the prior generation).
let room2 = manager.get_or_create(channel_id);
let room2 = manager.get_or_create(community_id, channel_id);
let _ = room2
.add_peer("bob".to_string(), 1)
.expect("fresh room must accept any version");
}
#[test]
fn manager_isolates_same_channel_uuid_across_communities() {
let manager = AudioRoomManager::new();
let channel_id = Uuid::new_v4();
let community_a = CommunityId::from_uuid(Uuid::new_v4());
let community_b = CommunityId::from_uuid(Uuid::new_v4());
let room_a = manager.get_or_create(community_a, channel_id);
assert!(Arc::ptr_eq(
&manager
.get_unambiguous_by_channel(channel_id)
.expect("one matching room is unambiguous"),
&room_a
));
let room_b = manager.get_or_create(community_b, channel_id);
assert!(
!Arc::ptr_eq(&room_a, &room_b),
"same channel UUID in two communities must create distinct rooms"
);
assert_eq!(room_a.community_id, community_a);
assert_eq!(room_b.community_id, community_b);
assert!(
manager.get_unambiguous_by_channel(channel_id).is_none(),
"community-free mesh lookup must fail closed on a UUID collision"
);
room_a
.add_peer("alice".to_string(), 1)
.expect("A peer admits");
assert_eq!(room_a.peer_pubkeys(), vec![("alice".to_string(), 0)]);
assert!(
room_b.peer_pubkeys().is_empty(),
"A room peers must not appear in B's same-UUID room"
);
}
/// Peer-index reuse: after a peer leaves, their index is released; a new
/// peer joining the same (still-pinned) room reuses the freed index.
/// Version pin must persist across this reuse — the room generation

6
crates/buzz-relay/src/handlers/auth.rs
View File

@@ -304,7 +304,11 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
// Successfully materialized — this owner is authoritative.
auth_ctx.agent_owner_pubkey = Some(owner);
// Pre-warm the observer cache to avoid stale negatives.
let cache_key = (pubkey.to_bytes().to_vec(), owner.to_bytes().to_vec());
let cache_key = (
conn.tenant.community(),
pubkey.to_bytes().to_vec(),
owner.to_bytes().to_vec(),
);
state.observer_owner_cache.insert(cache_key, true);
}
Ok(false) => {

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

@@ -983,7 +983,11 @@ async fn handle_agent_observer_event(
let agent_bytes = route.agent.to_bytes().to_vec();
let owner_bytes = route.owner.to_bytes().to_vec();
let cache_key = (agent_bytes.clone(), owner_bytes.clone());
let cache_key = (
conn.tenant.community(),
agent_bytes.clone(),
owner_bytes.clone(),
);
let is_owner = if session_owner_match {
true
} else {
@@ -1127,6 +1131,8 @@ fn single_tag_content<'a>(event: &'a Event, tag_name: &str) -> Result<&'a str, S
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::sync::atomic::AtomicU8;
use std::sync::Arc;
use buzz_core::kind::{
@@ -1138,6 +1144,9 @@ mod tests {
OBSERVER_FRAME_TELEMETRY,
};
use nostr::{EventBuilder, Keys, Kind, Tag};
use tokio::sync::{mpsc, Mutex, RwLock};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
#[test]
fn fanout_event_frame_matches_legacy_format_byte_for_byte() {
@@ -1303,6 +1312,95 @@ mod tests {
);
}
#[tokio::test]
async fn observer_owner_cache_is_scoped_to_community() {
let state = fanout_access::test_state().await;
let agent = Keys::generate();
let owner = Keys::generate();
let agent_bytes = agent.public_key().to_bytes().to_vec();
let owner_bytes = owner.public_key().to_bytes().to_vec();
let community_a = buzz_core::CommunityId::from_uuid(Uuid::new_v4());
let community_b = buzz_core::CommunityId::from_uuid(Uuid::new_v4());
state.observer_owner_cache.insert(
(community_a, agent_bytes.clone(), owner_bytes.clone()),
true,
);
assert_eq!(
state.observer_owner_cache.get(&(
community_b,
agent_bytes.clone(),
owner_bytes.clone()
)),
None,
"A cached allow must not populate B's observer authorization key"
);
state.observer_owner_cache.insert(
(community_b, agent_bytes.clone(), owner_bytes.clone()),
false,
);
let encrypted = encrypt_observer_payload(
&agent,
&owner.public_key(),
&serde_json::json!({"type": "acp_read"}),
)
.expect("encrypt observer payload");
let event = EventBuilder::new(Kind::Custom(KIND_AGENT_OBSERVER_FRAME as u16), encrypted)
.tags([
Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"),
Tag::parse([OBSERVER_AGENT_TAG, &agent.public_key().to_hex()]).expect("agent tag"),
Tag::parse([OBSERVER_FRAME_TAG, OBSERVER_FRAME_TELEMETRY]).expect("frame tag"),
])
.sign_with_keys(&agent)
.expect("sign event");
let (send_tx, mut send_rx) = mpsc::channel(1);
let (ctrl_tx, _ctrl_rx) = mpsc::channel(1);
let conn = Arc::new(crate::connection::ConnectionState {
conn_id: Uuid::new_v4(),
tenant: buzz_core::TenantContext::resolved(community_b, "b.example"),
remote_addr: "127.0.0.1:1234".parse().expect("socket addr"),
auth_state: RwLock::new(crate::connection::AuthState::Authenticated(
buzz_auth::AuthContext {
pubkey: agent.public_key(),
scopes: vec![],
channel_ids: None,
auth_method: buzz_auth::AuthMethod::Nip42,
agent_owner_pubkey: None,
},
)),
subscriptions: Arc::new(Mutex::new(HashMap::new())),
send_tx,
ctrl_tx,
cancel: CancellationToken::new(),
backpressure_count: Arc::new(AtomicU8::new(0)),
grace_limit: 3,
});
super::handle_agent_observer_event(
event.clone(),
conn.conn_id,
&event.id.to_hex(),
conn,
state,
)
.await;
let axum::extract::ws::Message::Text(text) =
send_rx.try_recv().expect("observer rejection sent")
else {
panic!("expected text relay message");
};
let frame: serde_json::Value = serde_json::from_str(&text).expect("relay frame JSON");
assert_eq!(frame[0], "OK");
assert_eq!(frame[2], false);
assert_eq!(
frame[3],
"restricted: observer frame is not authorized for this agent owner"
);
}
mod pubsub_fanout {
use std::collections::HashMap;
use std::sync::atomic::AtomicU8;

7
crates/buzz-relay/src/state.rs
View File

@@ -535,11 +535,12 @@ pub struct AppState {
/// Current in-flight media uploads per (community, uploader pubkey).
pub media_uploads_in_flight: Arc<DashMap<ScopedPubkeyKey, u32>>,
/// Cache for observer agent-owner authorization (kind 24200).
/// Key: (agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner.
/// agent_owner_pubkey is immutable so a long TTL (5 min) is safe.
/// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner.
/// `agent_owner_pubkey` is immutable inside one community, so a long TTL
/// (5 min) is safe once the community label is part of the key.
/// Prevents repeated DB lookups from bursty observer traffic.
#[allow(clippy::type_complexity)]
pub observer_owner_cache: Arc<moka::sync::Cache<(Vec<u8>, Vec<u8>), bool>>,
pub observer_owner_cache: Arc<moka::sync::Cache<(CommunityId, Vec<u8>, Vec<u8>), bool>>,
/// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`]
/// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to