fix(relay): confirm every channel a REQ filter names

A REQ re-checked stale cache-negatives only for the all-`#h` channel set,
so a group-state `#d` filter, or an `#h` filter beside a channel-less
sibling, skipped the uncached membership check. On a pod whose access
cache lagged a join, such a filter dropped the channel from its history
and from its live scope, and a live scope is fixed at registration, so
updates stayed missing until the client resubscribed.

The REQ now confirms every distinct channel any filter names (`#h`, or
`#d` on group-state kinds), bounded by the explicit-channel limit before
any membership I/O. Token restrictions and per-filter scopes are kept.

Also narrows the test wording for the Redis receive handler: it covers
the handler for other pods' events, not publishing or transport.

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
This commit is contained in:
Logan Johnson
2026-10-21 10:48:35 -04:00
parent 3255e910ef
commit c25057e975
3 changed files with 102 additions and 5 deletions

View File

@@ -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=<just-added>` 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(),

View File

@@ -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);
}

View File

@@ -109,6 +109,18 @@ fn named_channels(filter: &Filter) -> impl Iterator<Item = Uuid> + '_ {
.filter_map(|value| value.parse::<Uuid>().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<Uuid> {
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.