diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index f5e5e88fd3..a5af101dc9 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -111,6 +111,17 @@ pub async fn handle_req( } }; + // Every distinct channel a filter names, including group-state `#d`, is + // confirmed below; bound that set before any membership I/O. + let named_channels = super::system_channels::channels_named_by_filters(&filters); + if named_channels.len() > MAX_EXPLICIT_CHANNEL_VALUES { + conn.send(RelayMessage::closed( + &sub_id, + "restricted: too many explicit channels", + )); + return; + } + let mut accessible_channels = if filters_are_nip43_membership_only(&filters) { metrics::counter!("buzz_req_global_access_resolution_skips_total", "kind" => "13534") .increment(1); @@ -170,13 +181,18 @@ pub async fn handle_req( // `confirm_requested_channel_access`). Running this ahead of the search branch // is what fixes the search false-miss: a `#h=` search would // otherwise be scoped against the stale vector and return empty. - if let Some(requested) = requested_channel_ids.as_ref() { + // + // Confirm every channel any filter names (`#h`, or `#d` on group-state + // kinds), not only the all-`#h` set: each filter's history and live + // scope come from its own names, and a live scope is fixed at + // registration, so a stale miss would last until the client resubscribes. + if !named_channels.is_empty() { let repaired_from = accessible_channels.len(); let confirmed = match confirm_requested_channel_access( &state, conn.tenant.community(), &pubkey_bytes, - requested, + &named_channels, token_channel_ids.as_deref(), &mut accessible_channels, named_only.ids(), diff --git a/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs b/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs index 228a858e08..75423ca472 100644 --- a/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs @@ -1324,7 +1324,9 @@ async fn refcount(f: &Fixture, topic: buzz_pubsub::EventTopic) -> usize { f.state.pubsub.topic_refcount(&f.tenant, topic).await } -/// A relay-signed kind:39000 for `channel`, as another pod would publish it. +/// Hand a relay-signed kind:39000 for `channel` to the handler that receives +/// another pod's events from Redis. This covers the receive side only, not +/// publishing or Redis transport. async fn fan_out_from_another_pod(f: &Fixture, channel: Uuid, name: &str) -> String { let event = EventBuilder::new(Kind::Custom(39000), "") .tags([ @@ -1347,7 +1349,8 @@ async fn fan_out_from_another_pod(f: &Fixture, channel: Uuid, name: &str) -> Str } /// A `#d` group-state REQ keeps receiving its channel's replacements after -/// EOSE, on this pod and from Redis, for every reader who may read the +/// EOSE, from this pod and through the handler for other pods' events, for +/// every reader who may read the /// channel: a private channel's member and an open channel's outsider. A /// private channel's outsider gets neither history nor live updates, and /// never retains the channel's topic. CLOSE and same-ID replacement release @@ -1407,7 +1410,7 @@ async fn group_state_replacements_reach_live_d_subscribers() { ) ); - // Another pod, through Redis. + // Another pod's event, at the Redis receive handler. let private_id = fan_out_from_another_pod(&f, private, "from-redis").await; let open_id = fan_out_from_another_pod(&f, open, "from-redis").await; assert_eq!( @@ -1510,3 +1513,69 @@ async fn named_filters_keep_live_delivery_beside_unnamed_siblings() { assert_eq!(refcount(f, EventTopic::Global).await, 0); later.drop_database().await; } + +/// A cache-negative on a non-writer pod can lag a join. A REQ confirms every +/// channel its filters name against the database, so a recently joined +/// channel still enters the history and live scope of the filter that names +/// it, in both shapes that name a channel without every filter using `#h`: +/// a group-state `#d` filter, and an `#h` filter beside a channel-less one. +/// Mutation: confirm only `extract_channel_ids_from_filters_limited`'s +/// all-`#h` set in `handle_req` → RED (no history, no channel topic, no live +/// event). +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn named_channels_missing_from_a_stale_cache_still_route_live() { + use buzz_pubsub::EventTopic; + let f = fixture().await; + let reader = Keys::generate(); + let channel = create_channel(&f.state, &f.tenant, &reader, &[&["visibility", "private"]]).await; + let h = channel.to_string(); + let reader_hex = reader.public_key().to_hex(); + // This pod cached the reader's access set before the join. + let stale = || { + f.state.accessible_channels_cache.insert( + ( + f.tenant.community(), + reader.public_key().to_bytes().to_vec(), + ), + Vec::new(), + ); + }; + + stale(); + let mut by_d = LiveReq::open(&f, &reader, json!([{"kinds": [39000], "#d": [h]}])).await; + assert_eq!(by_d.history, [group_state_id(&f, channel, 39000).await]); + stale(); + let mut mixed = LiveReq::open( + &f, + &reader, + json!([ + {"kinds": [9], "#h": [h]}, + {"kinds": [1], "authors": [reader_hex]}, + ]), + ) + .await; + assert_eq!(refcount(&f, EventTopic::Channel(channel)).await, 2); + + let message = publish(&f, &reader, 9, "hello", &[&["h", &h]]).await; + assert_eq!(mixed.next_event().await, (9, h.clone(), message)); + edit_channel( + &f.state, + &f.tenant, + &reader, + channel, + &[&["name", "renamed"]], + ) + .await + .expect("edit"); + assert_eq!( + by_d.next_event().await, + (39000, h.clone(), group_state_id(&f, channel, 39000).await) + ); + by_d.assert_quiet("by #d").await; + mixed.assert_quiet("mixed").await; + + by_d.close(&f).await; + mixed.close(&f).await; + assert_eq!(refcount(&f, EventTopic::Channel(channel)).await, 0); +} diff --git a/crates/buzz-relay/src/handlers/system_channels.rs b/crates/buzz-relay/src/handlers/system_channels.rs index d6d569574e..8a981ad40e 100644 --- a/crates/buzz-relay/src/handlers/system_channels.rs +++ b/crates/buzz-relay/src/handlers/system_channels.rs @@ -109,6 +109,18 @@ fn named_channels(filter: &Filter) -> impl Iterator + '_ { .filter_map(|value| value.parse::().ok()) } +/// Every channel ID that some filter names (see [`named_channels`]), once +/// each, in request order. A request confirms these against the database, +/// because each one can enter its filter's history and live scope. +pub(crate) fn channels_named_by_filters(filters: &[Filter]) -> Vec { + let mut seen = HashSet::new(); + filters + .iter() + .flat_map(named_channels) + .filter(|id| seen.insert(*id)) + .collect() +} + /// Whether `filter` opts in to every readable system channel: its kinds are /// all group-state kinds, and it has `#t:["system"]` or any `#P`. It opts in /// to system channels only, never to another named-only type.