From 3255e910ef91c37dc274cf493f6d66d79a1aec54 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 21 Oct 2026 09:47:59 -0400 Subject: [PATCH] fix(relay): route live REQ events by each filter's own channel scope A live REQ took one request-wide routing scope from its filters' aggregate `#h`. If any filter lacked `#h`, the whole subscription became global, so: - `{kinds:[39000], "#d":[ch]}` got history and EOSE but never a later replacement, because relay-signed group-state events are stored under their channel and global subscriptions never see channel events; - in a mix of named and unnamed filters, the named filter lost live delivery too. Each filter now carries its own server-resolved live scope. A filter with `#h` routes under the channels it names and may read. A filter without `#h` keeps the global topic, and a group-state filter also routes under the readable channels its `#d` names. Fan-out delivers only through a filter that both matches the event and holds its channel, so a channel one filter names never reaches a `#P` or `#t` sibling (the confinement from #8469). The subscription retains each topic in its filters' union once; CLOSE, same-ID replacement and revocation release it once. Also runs the subscription registry unit tests in the unit lanes; no lane selected them before. Signed-off-by: Logan Johnson Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> --- Justfile | 2 +- NOSTR.md | 4 +- crates/buzz-relay/src/handlers/req.rs | 222 +++-- .../handlers/system_channel_postgres_tests.rs | 294 +++++++ .../src/handlers/system_channels.rs | 70 ++ crates/buzz-relay/src/subscription.rs | 773 ++++++++++++------ scripts/run-tests.sh | 3 + 7 files changed, 1026 insertions(+), 342 deletions(-) diff --git a/Justfile b/Justfile index 59e98d9c11..39fbfb0293 100644 --- a/Justfile +++ b/Justfile @@ -537,7 +537,7 @@ test-unit: # are infra-free; config parses env into a struct and never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^subscription::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/NOSTR.md b/NOSTR.md index 4212907295..aa3743a1e9 100644 --- a/NOSTR.md +++ b/NOSTR.md @@ -175,7 +175,9 @@ nak req -k 39002 --tag "d=" --auth --sec ws://localhost: > **Note:** Channel-scoped storage means live global subscriptions (`{kinds:[39000]}`) won't > receive these via fan-out. Clients discover groups via historical REQ queries. Live push for -> open-channel discovery is a future enhancement. +> open-channel discovery is a future enhancement. A group-state filter that names channels by +> `#d` (`{kinds:[39000], "#d":[]}`) does receive later replacements for the named +> channels it may read, after EOSE. ### Membership Notifications diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 646cfe375b..f5e5e88fd3 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -21,6 +21,7 @@ use buzz_auth::Scope; use crate::connection::{AuthState, ConnectionState}; use crate::protocol::RelayMessage; use crate::state::AppState; +use crate::subscription::SubscriptionScope; const MAX_SUBSCRIPTIONS: usize = 1024; @@ -384,15 +385,22 @@ pub async fn handle_req( } }; - let Some(owner) = claim_live_subscription( - &sub_id, - &filters, - authorized_requested_channels.as_deref(), - &conn, - &state, - ) - .await - else { + // Each filter routes live events under its own scope, so a channel one + // filter names never reaches a sibling filter (see `filter_live_scope`). + let live_filters = filters + .iter() + .map(|filter| { + ( + filter.clone(), + super::system_channels::filter_live_scope( + filter, + &accessible_channels, + &named_only, + ), + ) + }) + .collect(); + let Some(owner) = claim_live_subscription(&sub_id, live_filters, &conn, &state).await else { return; }; @@ -1509,12 +1517,13 @@ async fn handle_huddle_liveness_req( /// Claim `sub_id` for a live REQ under the lifecycle lock: record a fresh /// owner token, register for fan-out (replacing any same-ID entry), retain its -/// topics, and release the replaced scope's. Returns the owner token, or -/// `None` once the connection is closing — the caller then exits silently. +/// topics, and release the replaced scope's. Each filter carries its own +/// server-resolved live scope; the subscription retains each topic in their +/// union once. Returns the owner token, or `None` once the connection is +/// closing — the caller then exits silently. async fn claim_live_subscription( sub_id: &str, - filters: &[Filter], - channel_ids: Option<&[uuid::Uuid]>, + filters: Vec<(Filter, SubscriptionScope)>, conn: &ConnectionState, state: &AppState, ) -> Option { @@ -1524,38 +1533,24 @@ async fn claim_live_subscription( } let owner = super::close::next_owner(); subs.insert(sub_id.to_string(), owner); - let community = conn.tenant.community(); - let replaced = match channel_ids { - Some(ids) => state.sub_registry.register_channels_scoped( - community, - conn.conn_id, - sub_id.to_string(), - filters.to_vec(), - ids.to_vec(), - ), - None => state.sub_registry.register_scoped( - community, - conn.conn_id, - sub_id.to_string(), - filters.to_vec(), - None, - ), - }; - match channel_ids { - Some(ids) => { - for &channel_id in ids { - state - .pubsub - .retain_topic(&conn.tenant, EventTopic::Channel(channel_id)) - .await; - } - } - None => { - state - .pubsub - .retain_topic(&conn.tenant, EventTopic::Global) - .await; - } + let topics = SubscriptionScope::union(filters.iter().map(|(_, scope)| scope)); + let replaced = state.sub_registry.register_filters_scoped( + conn.tenant.community(), + conn.conn_id, + sub_id.to_string(), + filters, + ); + if topics.is_global() { + state + .pubsub + .retain_topic(&conn.tenant, EventTopic::Global) + .await; + } + for &channel_id in topics.channel_ids() { + state + .pubsub + .retain_topic(&conn.tenant, EventTopic::Channel(channel_id)) + .await; } if let Some(replaced) = replaced { super::close::release_scope_topics(state, &conn.tenant, &replaced.scope).await; @@ -1915,6 +1910,20 @@ mod tests { vec![Filter::new().kind(nostr::Kind::TextNote)] } + /// Give every filter the same live scope: `channels`, or global. + fn scoped( + filters: &[Filter], + channels: Option<&[uuid::Uuid]>, + ) -> Vec<(Filter, SubscriptionScope)> { + let scope = channels.map_or_else(SubscriptionScope::global, |channels| { + SubscriptionScope::channels(channels.iter().copied()) + }); + filters + .iter() + .map(|filter| (filter.clone(), scope.clone())) + .collect() + } + fn registered(state: &AppState, conn: &ConnectionState, sub_id: &str) -> bool { state .sub_registry @@ -1939,10 +1948,14 @@ mod tests { let (conn, mut send_rx) = lifecycle_conn(); let channel = uuid::Uuid::new_v4(); let topic = EventTopic::Channel(channel); - let owner = - claim_live_subscription("thread", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + let owner = claim_live_subscription( + "thread", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); assert_eq!(state.pubsub.topic_refcount(&conn.tenant, topic).await, 1); close_timed_out_subscription("thread", owner, &conn, &state).await; @@ -1974,14 +1987,22 @@ mod tests { EventTopic::Channel(old_channel), EventTopic::Channel(new_channel), ); - let old = - claim_live_subscription("x", &text_filters(), Some(&[old_channel]), &conn, &state) - .await - .expect("claim"); - let new = - claim_live_subscription("x", &text_filters(), Some(&[new_channel]), &conn, &state) - .await - .expect("claim"); + let old = claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[old_channel])), + &conn, + &state, + ) + .await + .expect("claim"); + let new = claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[new_channel])), + &conn, + &state, + ) + .await + .expect("claim"); assert_eq!( state.pubsub.topic_refcount(&conn.tenant, old_topic).await, 0 @@ -2013,9 +2034,14 @@ mod tests { let (conn, mut send_rx) = lifecycle_conn(); let channel = uuid::Uuid::new_v4(); let topic = EventTopic::Channel(channel); - claim_live_subscription("x", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); let search = claim_search_subscription("x", &conn, &state) .await @@ -2026,9 +2052,14 @@ mod tests { ); assert_eq!(state.pubsub.topic_refcount(&conn.tenant, topic).await, 0); - let live = claim_live_subscription("x", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + let live = claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); close_timed_out_subscription("x", search, &conn, &state).await; assert_eq!(conn.subscriptions.lock().await.get("x"), Some(&live)); @@ -2046,17 +2077,27 @@ mod tests { let conn = Arc::new(conn); let channel = uuid::Uuid::new_v4(); let topic = EventTopic::Channel(channel); - let stale = claim_live_subscription("x", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + let stale = claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); let tasks = (0..16).map(|i| { let (conn, state) = (Arc::clone(&conn), Arc::clone(&state)); tokio::spawn(async move { if i % 2 == 0 { - claim_live_subscription("x", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); } else { close_timed_out_subscription("x", stale, &conn, &state).await; } @@ -2107,7 +2148,7 @@ mod tests { let (conn, mut send_rx) = lifecycle_conn(); let conn = Arc::new(conn); let (a, b) = (uuid::Uuid::new_v4(), uuid::Uuid::new_v4()); - let old = claim_live_subscription("x", &text_filters(), Some(&[a]), &conn, &state) + let old = claim_live_subscription("x", scoped(&text_filters(), Some(&[a])), &conn, &state) .await .expect("claim"); @@ -2120,8 +2161,7 @@ mod tests { let (filters, b_scope) = (text_filters(), [b]); let mut replacement = Box::pin(claim_live_subscription( "x", - &filters, - Some(&b_scope), + scoped(&filters, Some(&b_scope)), &conn, &state, )); @@ -2171,7 +2211,7 @@ mod tests { crate::state::CommunityConnectionControl::new(conn.cancel.clone()), ); let (a, b) = (uuid::Uuid::new_v4(), uuid::Uuid::new_v4()); - claim_live_subscription("x", &text_filters(), Some(&[a]), &conn, &state) + claim_live_subscription("x", scoped(&text_filters(), Some(&[a])), &conn, &state) .await .expect("claim"); @@ -2188,8 +2228,7 @@ mod tests { let (filters, b_scope) = (text_filters(), [b]); let mut replacement = Box::pin(claim_live_subscription( "x", - &filters, - Some(&b_scope), + scoped(&filters, Some(&b_scope)), &conn, &state, )); @@ -2226,14 +2265,24 @@ mod tests { let (conn, mut send_rx) = lifecycle_conn(); let channel = uuid::Uuid::new_v4(); let topic = EventTopic::Channel(channel); - claim_live_subscription("live", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + claim_live_subscription( + "live", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); conn.cancel.cancel(); super::super::close::release_connection_subscriptions(&conn, &state).await; - let late_live = - claim_live_subscription("late", &text_filters(), Some(&[channel]), &conn, &state).await; + let late_live = claim_live_subscription( + "late", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await; let late_search = claim_search_subscription("search", &conn, &state).await; assert_eq!((late_live, late_search), (None, None)); @@ -2285,7 +2334,7 @@ mod tests { }); // Claim, then fill the send buffer so the next `send` returns false. - let owner = claim_live_subscription("x", &text_filters(), None, &conn, &state) + let owner = claim_live_subscription("x", scoped(&text_filters(), None), &conn, &state) .await .expect("claim"); // Occupy the one slot so try_send returns Full on the CLOSED. @@ -2360,9 +2409,14 @@ mod tests { ); let channel = uuid::Uuid::new_v4(); - claim_live_subscription("x", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + claim_live_subscription( + "x", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); // Fill the one send slot so the CLOSED restricted try_send returns Full. send_rx.try_recv().ok(); // drain any earlier messages 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 66662c9cfd..228a858e08 100644 --- a/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs @@ -1216,3 +1216,297 @@ async fn system_channels_are_found_by_type_or_creator() { "bridge /count outsider #t:system" ); } + +/// A live REQ through the production handler, on a connection that fan-out +/// can reach. `history` holds the event IDs sent before EOSE. +struct LiveReq { + conn: Arc, + rx: tokio::sync::mpsc::Receiver, + history: Vec, +} + +impl LiveReq { + async fn open(f: &Fixture, reader: &Keys, filters: Value) -> Self { + let (conn, rx) = test_conn(&f.tenant, reader); + f.state.conn_manager.register( + conn.conn_id, + conn.send_tx.clone(), + conn.ctrl_tx.clone(), + conn.terminal_ctrl_tx.clone(), + None, + conn.cancel.clone(), + f.tenant.community(), + conn.backpressure_count.clone(), + conn.subscriptions.clone(), + conn.grace_limit, + crate::state::CommunityConnectionControl::new(conn.cancel.clone()), + ); + f.state + .conn_manager + .set_authenticated_pubkey(conn.conn_id, reader.public_key().to_bytes().to_vec()); + let mut live = Self { + conn, + rx, + history: Vec::new(), + }; + live.req(f, filters).await; + live + } + + /// Send a REQ for sub `s` (replacing any earlier one) and drain to EOSE. + async fn req(&mut self, f: &Fixture, filters: Value) { + let filters: Vec = serde_json::from_value(filters).expect("filters"); + let before = vec![None; filters.len()]; + crate::handlers::req::handle_req( + "s".into(), + filters, + before, + self.conn.clone(), + f.state.clone(), + ) + .await; + self.history.clear(); + loop { + let (kind, frame) = self + .frame(std::time::Duration::from_secs(5)) + .await + .expect("EOSE"); + match kind.as_str() { + "EVENT" => self + .history + .push(frame[2]["id"].as_str().expect("id").to_owned()), + "EOSE" => return, + other => panic!("unexpected frame {other}: {frame}"), + } + } + } + + async fn frame(&mut self, wait: std::time::Duration) -> Option<(String, Value)> { + loop { + let frame = tokio::time::timeout(wait, self.rx.recv()).await.ok()??; + let axum::extract::ws::Message::Text(text) = frame else { + continue; + }; + let message: Value = serde_json::from_str(&text).expect("relay frame"); + return Some((message[0].as_str().expect("frame type").to_owned(), message)); + } + } + + /// The next live event, as `(kind, d or h tag, id)`. + async fn next_event(&mut self) -> (u16, String, String) { + let (kind, frame) = self + .frame(std::time::Duration::from_secs(5)) + .await + .expect("live event"); + assert_eq!(kind, "EVENT", "{frame}"); + let event: nostr::Event = serde_json::from_value(frame[2].clone()).expect("event"); + let scope = event + .tags + .iter() + .find(|tag| matches!(tag.as_slice().first().map(String::as_str), Some("d" | "h"))) + .and_then(|tag| tag.as_slice().get(1).cloned()) + .unwrap_or_default(); + (event.kind.as_u16(), scope, event.id.to_hex()) + } + + async fn assert_quiet(&mut self, who: &str) { + if let Some(frame) = self.frame(std::time::Duration::from_millis(200)).await { + panic!("{who}: unexpected frame {frame:?}"); + } + } + + async fn close(&self, f: &Fixture) { + crate::handlers::close::handle_close("s".into(), self.conn.clone(), f.state.clone()).await; + } +} + +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. +async fn fan_out_from_another_pod(f: &Fixture, channel: Uuid, name: &str) -> String { + let event = EventBuilder::new(Kind::Custom(39000), "") + .tags([ + Tag::parse(["d", &channel.to_string()]).unwrap(), + Tag::parse(["name", name]).unwrap(), + ]) + .sign_with_keys(&f.state.relay_keypair) + .expect("sign 39000"); + let id = event.id.to_hex(); + crate::handlers::event::fan_out_pubsub_event( + &f.state, + buzz_pubsub::ChannelEvent { + community_id: f.tenant.community(), + topic: buzz_pubsub::EventTopic::Channel(channel), + event, + }, + ) + .await; + id +} + +/// 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 +/// 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 +/// each retained topic once. +/// Mutation: give a filter without `#h` only the global scope in +/// `filter_live_scope` → RED (no live replacement after EOSE). +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn group_state_replacements_reach_live_d_subscribers() { + use buzz_pubsub::EventTopic; + let f = fixture().await; + let owner = Keys::generate(); + let outsider = Keys::generate(); + let private = create_channel(&f.state, &f.tenant, &owner, &[&["visibility", "private"]]).await; + let open = create_channel(&f.state, &f.tenant, &owner, &[&["visibility", "open"]]).await; + let by_d = |channel: Uuid| json!([{"kinds": [39000], "#d": [channel.to_string()]}]); + + let mut member = LiveReq::open(&f, &owner, by_d(private)).await; + let mut open_outsider = LiveReq::open(&f, &outsider, by_d(open)).await; + let mut private_outsider = LiveReq::open(&f, &outsider, by_d(private)).await; + assert_eq!(member.history, [group_state_id(&f, private, 39000).await]); + assert_eq!( + open_outsider.history, + [group_state_id(&f, open, 39000).await] + ); + assert!(private_outsider.history.is_empty()); + assert_eq!(refcount(&f, EventTopic::Channel(private)).await, 1); + assert_eq!(refcount(&f, EventTopic::Channel(open)).await, 1); + assert_eq!(refcount(&f, EventTopic::Global).await, 3); + + // This pod: the relay publishes a replacement after each edit. + for channel in [private, open] { + edit_channel( + &f.state, + &f.tenant, + &owner, + channel, + &[&["name", "renamed"]], + ) + .await + .expect("edit"); + } + assert_eq!( + member.next_event().await, + ( + 39000, + private.to_string(), + group_state_id(&f, private, 39000).await + ) + ); + assert_eq!( + open_outsider.next_event().await, + ( + 39000, + open.to_string(), + group_state_id(&f, open, 39000).await + ) + ); + + // Another pod, through Redis. + 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!( + member.next_event().await, + (39000, private.to_string(), private_id) + ); + assert_eq!( + open_outsider.next_event().await, + (39000, open.to_string(), open_id) + ); + private_outsider.assert_quiet("private outsider").await; + member.assert_quiet("member").await; + open_outsider.assert_quiet("open outsider").await; + + // Same-ID replacement moves the member's channel topic, once. + member.req(&f, by_d(open)).await; + assert_eq!(refcount(&f, EventTopic::Channel(private)).await, 0); + assert_eq!(refcount(&f, EventTopic::Channel(open)).await, 2); + assert_eq!(refcount(&f, EventTopic::Global).await, 3); + + for live in [&member, &open_outsider, &private_outsider] { + live.close(&f).await; + } + assert_eq!(refcount(&f, EventTopic::Channel(open)).await, 0); + assert_eq!(refcount(&f, EventTopic::Global).await, 0); +} + +/// In a REQ that mixes named and unnamed filters, the named filters keep +/// live delivery for their own channel and the unnamed filter keeps global +/// events. A `#P` or `#t:["system",