diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index b8ab2b2c12..df140755dd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -550,7 +550,7 @@ Subscriber → dedicated PubSub → PSUBSCRIBE buzz:channel:* The subscriber uses a **dedicated** `redis::aio::PubSub` connection — not from the pool. This is intentional: pool connections cannot hold `PSUBSCRIBE` state. -**Current state:** The subscriber loop is spawned in `buzz-relay/src/main.rs` and populates the broadcast channel. A consumer task subscribes via `pubsub.subscribe_local()`, calls `sub_registry.fan_out()` on each received event, and delivers matches to local WebSocket connections via `conn_manager.send_to()`. Multi-node fan-out is now wired end-to-end. Local-echo deduplication is implemented via `AppState.local_event_ids` — events published by the local relay instance are tracked and skipped when received via the Redis round-trip. +**Current state:** The subscriber loop is spawned in `buzz-relay/src/main.rs` and populates the broadcast channel. A dispatcher task subscribes via `pubsub.subscribe_local()` and routes each event to one of 16 delivery workers by `(community, topic)`, so one channel's events keep their order while channels deliver concurrently (`handlers/multinode_fanout.rs`). Each worker calls `sub_registry.fan_out()`, applies the access gate, and delivers matches to local WebSocket connections via `conn_manager.send_to()`. Live delivery never drops an event silently: when a worker's bounded queue is full, or an event is not delivered within 2 s of this node reading it from Redis, that event's local recipients get `CLOSED "resync: ..."` for syncs, are fenced from every further live EVENT frame as soon as the loss is detected (so a client's resume cursor stops moving), and are closed so clients catch up from their cursor; when the dispatcher itself lags the broadcast, every subscribed connection is treated the same way. Only the fence is immediate: closes are paced (at most 5,000/s per node) so a loss cannot become a reconnect storm. Clients resume from the newest `created_at` they received minus 5 s, so a closed client replays a missed event only while the 2 s deadline plus the event's writer clock skew and publish delay stays under that 5 s; an ordinary reconnect has the same limit. A dispatcher that falls behind without overflowing the broadcast detects a loss late, and dispatcher lag loses events it never sees; neither is bounded by the deadline. Multi-node fan-out is now wired end-to-end. Local-echo deduplication is implemented via `AppState.local_event_ids` — events published by the local relay instance are tracked and skipped when received via the Redis round-trip. **Reconnection:** exponential backoff 1s → 30s (`backoff_secs * 2`). Backoff resets to 1s only after a clean stream end, not on each reconnect attempt. diff --git a/Justfile b/Justfile index 96be962855..22e9944c4b 100644 --- a/Justfile +++ b/Justfile @@ -556,9 +556,11 @@ test-unit: # never connects. # `^config::` is anchored so it does not also select nip_fi_config. # subscription::tests covers the in-memory registry and its fan-out - # indexes; it is infra-free. + # indexes; it is infra-free. handlers::multinode_fanout::tests covers + # cross-node fan-out loss handling; its stalled-delivery case fakes a + # database that never answers on loopback, and none use DATABASE_URL. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^subscription::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::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_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + 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(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + 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::failed_search_closes_without_eose_or_trying_later_filters) + 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::count_fallback_fetches_one_extra_candidate) + 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) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^subscription::tests::/) + test(/^handlers::multinode_fanout::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::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_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + 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(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + 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::failed_search_closes_without_eose_or_trying_later_filters) + 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::count_fallback_fetches_one_extra_candidate) + 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) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (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 b3d628786f..629c4f808a 100644 --- a/NOSTR.md +++ b/NOSTR.md @@ -329,7 +329,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-cli/src/client.rs b/crates/buzz-cli/src/client.rs index fcc102e4ce..bed59b5ae0 100644 --- a/crates/buzz-cli/src/client.rs +++ b/crates/buzz-cli/src/client.rs @@ -4220,6 +4220,51 @@ mod ws_deadline_tests { assert_eq!(accepted.load(Ordering::SeqCst), 1); } + /// The NIP-42 AUTH event names the configured community URL, path and + /// trailing slash kept, as the relay checks it. + #[tokio::test] + async fn the_auth_event_names_the_relay_with_its_trailing_slash() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let named = Arc::new(std::sync::Mutex::new(None::)); + let log = named.clone(); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + let _ = ws + .send(Message::Text(r#"["AUTH","challenge"]"#.into())) + .await; + while let Some(Ok(Message::Text(text))) = ws.next().await { + let frame: serde_json::Value = serde_json::from_str(&text).unwrap(); + if frame[0] == "AUTH" { + let relay = frame[1]["tags"] + .as_array() + .unwrap() + .iter() + .find(|t| t[0] == "relay") + .map(|t| t[1].as_str().unwrap().to_owned()); + *log.lock().unwrap() = relay; + } + let ok = serde_json::json!(["OK", frame[1]["id"], true, ""]); + let _ = ws.send(Message::Text(ok.to_string().into())).await; + } + }); + let keys = Keys::generate(); + let c = BuzzClient::new( + format!("http://{addr}/team/"), + Credentials::from_keys(keys.clone()), + ) + .unwrap(); + let event = EventBuilder::new(Kind::Custom(20001), "") + .sign_with_keys(&keys) + .unwrap(); + c.publish_ephemeral_event(event).await.unwrap(); + assert_eq!( + named.lock().unwrap().as_deref(), + Some(format!("ws://{addr}/team/").as_str()) + ); + } + #[test] fn auth_keeps_the_configured_relay_path() { let client = BuzzClient::new( diff --git a/crates/buzz-cli/src/credentials.rs b/crates/buzz-cli/src/credentials.rs index 033ba022f8..c042e3b487 100644 --- a/crates/buzz-cli/src/credentials.rs +++ b/crates/buzz-cli/src/credentials.rs @@ -81,11 +81,37 @@ pub async fn resolve(settings: Settings, relay_url: &str) -> Result CliError { + let fix = "set BUZZ_PRIVATE_KEY, pass --private-key, or run `buzz auth desktop pair` \ + with Buzz running"; + #[cfg(unix)] + let this_one = crate::desktop_provider::provider_relay_url(relay_url) + .is_ok_and(|relay| paired.contains(&relay)); + #[cfg(not(unix))] + let this_one = false; + CliError::Auth(if paired.is_empty() { + format!("no credentials for {relay_url}: {fix}") + } else if this_one { + format!( + "no credentials for {relay_url}: this CLI is paired for it, but no running Buzz \ + desktop accepts the pairing; open Buzz and retry, or check `buzz auth desktop list`" + ) + } else { + format!( + "no credentials for {relay_url}: this CLI is paired with Buzz only for {}; pass \ + --relay (or set BUZZ_RELAY_URL) to use a pairing, or {fix}", + paired.join(", ") + ) + }) } /// The explicit-key link, with its headless evidence source. Without an @@ -802,4 +828,33 @@ mod tests { ); assert_eq!(id.len(), 32); } + + #[test] + fn no_credentials_names_the_relay_and_the_paired_ones() { + let message = |relay: &str, paired: &[&str]| { + let paired: Vec = paired.iter().map(|r| (*r).to_owned()).collect(); + match no_credentials(relay, &paired) { + CliError::Auth(m) => m, + other => panic!("{other:?}"), + } + }; + let none = message("http://localhost:3000", &[]); + assert!(none.starts_with("no credentials for http://localhost:3000: set BUZZ_PRIVATE_KEY")); + + let elsewhere = message("http://localhost:3000", &["wss://team.example/"]); + assert!( + elsewhere.contains("paired with Buzz only for wss://team.example/"), + "{elsewhere}" + ); + assert!(elsewhere.contains("pass --relay"), "{elsewhere}"); + + #[cfg(unix)] + { + let here = message("https://team.example", &["wss://team.example/"]); + assert!( + here.contains("no running Buzz desktop accepts the pairing"), + "{here}" + ); + } + } } diff --git a/crates/buzz-cli/src/desktop_provider.rs b/crates/buzz-cli/src/desktop_provider.rs index 0f864e0e98..c688ec7186 100644 --- a/crates/buzz-cli/src/desktop_provider.rs +++ b/crates/buzz-cli/src/desktop_provider.rs @@ -70,6 +70,9 @@ pub enum ProviderError { Transport(String), /// The same-user check ran out of time before anything was sent. Unconfirmed, + /// The provider port is held by another user, or no owner was found: + /// nothing was sent. + OtherUser, } impl ProviderError { @@ -112,6 +115,10 @@ impl ProviderError { Self::Unconfirmed => CliError::Unavailable( "Buzz desktop: the same-user check timed out; nothing was sent".into(), ), + Self::OtherUser => CliError::Auth( + "Buzz desktop: could not confirm the provider runs as this user; nothing was sent" + .into(), + ), } } } @@ -293,6 +300,10 @@ async fn connect(port: u16, check: bool) -> Result { .local_addr() .map_err(|_| ProviderError::Transport("no local address".into()))? .port(); + #[cfg(test)] + if tests::SAME_USER_HANGS.try_with(|_| ()).is_ok() { + return Err(ProviderError::Unconfirmed); + } confirm_same_user(server_side_uid(port, local)).await?; } Ok(stream) @@ -305,9 +316,7 @@ async fn confirm_same_user( ) -> Result<(), ProviderError> { match tokio::time::timeout(SAME_USER_TIMEOUT, owner).await { Ok(Some(uid)) if uid == current_uid() => Ok(()), - Ok(_) => Err(ProviderError::Transport( - "could not confirm the provider runs as this user; nothing was sent".into(), - )), + Ok(_) => Err(ProviderError::OtherUser), Err(_) => Err(ProviderError::Unconfirmed), } } @@ -568,7 +577,9 @@ impl Provider { Ok(raw) => raw, Err(ProviderError::Revoked) => return Ok(Err(Skip::Revoked)), Err(ProviderError::NoCommonVersion) => return Ok(Err(Skip::NotApplicable)), - Err(ProviderError::Transport(_)) => return Ok(Err(Skip::Unreachable)), + Err(ProviderError::Transport(_) | ProviderError::OtherUser) => { + return Ok(Err(Skip::Unreachable)) + } Err(ProviderError::Unconfirmed) => return Ok(Err(Skip::Unconfirmed)), Err(e) => return Err(e.into_cli("describe")), }; @@ -1064,9 +1075,7 @@ pub async fn inherited_relay() -> Result, CliError> { .unwrap_or_default(); let mut distinct: Vec = Vec::new(); for relay in relays { - if !distinct.iter().any(|known| same_relay(known, &relay)) { - distinct.push(provider_relay_url(&relay).unwrap_or(relay)); - } + push_distinct(&mut distinct, relay); } match distinct.len() { 0 => Ok(None), @@ -1079,6 +1088,213 @@ pub async fn inherited_relay() -> Result, CliError> { } } +/// Add `relay` to `list` in bound form, unless it is already there. +fn push_distinct(list: &mut Vec, relay: String) { + if !list.iter().any(|known| same_relay(known, &relay)) { + list.push(provider_relay_url(&relay).unwrap_or(relay)); + } +} + +/// The relay a paired person's command given none acts on: the one relay a +/// running desktop still accepts this CLI's pairing for, by its `describe`. +/// A stored pairing alone does not count, since its desktop may have revoked +/// it or stopped. `None` for an agent (its inherited token decides), or when +/// no relay or several qualify. +pub async fn paired_relay() -> Result, CliError> { + if inherited()?.is_some() { + return Ok(None); + } + accepted_relay(&Keychain, advertisements()).await +} + +async fn accepted_relay( + store: &dyn PairingStore, + advertised: Vec<(String, Endpoint)>, +) -> Result, CliError> { + let Ok(Ok(pairings)) = store.read().map(parse_pairings) else { + return Ok(None); + }; + let mut accepted = Vec::new(); + let mut transient = None; + for (instance, endpoint) in advertised { + let Some(pairing) = pairings.0.get(&instance) else { + continue; + }; + match probe(&endpoint, pairing, RATE_LIMIT_RETRIES).await { + Probe::Accepted(relays) => { + for relay in relays { + push_distinct(&mut accepted, relay); + } + } + Probe::NotAccepted | Probe::NotRunning => {} + // A desktop that answered with a refusal has decided. + Probe::Refused(e) => return Err(e), + Probe::Unconfirmed(e) | Probe::Throttled(e) => keep_transient(&mut transient, e), + } + } + match (accepted.len(), transient) { + // A desktop whose check failed may accept a relay of its own: report + // it rather than pick, or fall back to, a relay it might contradict. + (0 | 1, Some(e)) => Err(e), + (1, None) => Ok(accepted.pop()), + _ => Ok(None), + } +} + +/// Keep the first transient failure, except that a retryable one replaces an +/// identity mismatch: the command may succeed once that desktop answers. +fn keep_transient(kept: &mut Option, e: CliError) { + let better = kept.as_ref().is_none_or(|k| { + !crate::error::is_retryable_error(k) && crate::error::is_retryable_error(&e) + }); + if better { + *kept = Some(e); + } +} + +/// What a running desktop says about a stored pairing now. +enum Probe { + /// The pairing's relays it still binds for the token. + Accepted(Vec), + /// It answered: the token is revoked, or binds none of the relays. + NotAccepted, + /// The advertised endpoint could not be reached or did not answer. + NotRunning, + /// The same-user check timed out, or the port's owner is another user. + Unconfirmed(CliError), + /// It is still rate-limiting after `retries` waits. + Throttled(CliError), + /// It refused `describe` (`denied`, `unavailable`, ...): final. + Refused(CliError), +} + +/// Ask the desktop at `endpoint`, on a same-user connection, which of +/// `pairing`'s relays it binds for its token. The one check behind both the +/// default relay and `auth desktop list`. +async fn probe(endpoint: &Endpoint, pairing: &Pairing, retries: u32) -> Probe { + let raw = match call_with( + endpoint, + Some(&pairing.token), + true, + "describe", + retries, + &Vec::new, + ) + .await + { + Ok(raw) => raw, + Err(ProviderError::Revoked | ProviderError::NoCommonVersion) => return Probe::NotAccepted, + Err(ProviderError::Transport(_)) => return Probe::NotRunning, + Err(e @ (ProviderError::Unconfirmed | ProviderError::OtherUser)) => { + return Probe::Unconfirmed(e.into_cli("describe")) + } + Err( + e @ ProviderError::Code { + code: "rate_limited", + .. + }, + ) => return Probe::Throttled(e.into_cli("describe")), + Err(e) => return Probe::Refused(e.into_cli("describe")), + }; + let Ok(describe) = serde_json::from_str::(&raw) else { + return Probe::Refused(CliError::Auth( + "Buzz desktop: unreadable `describe` answer".into(), + )); + }; + let mut accepted = Vec::new(); + for relay in describe.relays { + let bound = provider_relay_url(&relay).is_ok_and(|bound| { + pairing + .relays + .iter() + .any(|paired| same_relay(&bound, paired)) + }); + if bound { + push_distinct(&mut accepted, relay); + } + } + if accepted.is_empty() { + Probe::NotAccepted + } else { + Probe::Accepted(accepted) + } +} + +/// This CLI's pairings, without their tokens, each with what a running +/// desktop says about it now: `accepted`, `not accepted` (revoked or no +/// longer bound), `not running`, `unconfirmed` (same-user check failed), +/// `rate limited` or `refused`. Local and read-only. Desktops are asked at +/// once, each without waiting out throttling. +pub async fn list() -> Result, CliError> { + list_with(&Keychain, advertisements()).await +} + +async fn list_with( + store: &dyn PairingStore, + advertised: Vec<(String, Endpoint)>, +) -> Result, CliError> { + let pairings = parse_pairings(store.read()?)?; + let rows = pairings.0.into_iter().map(|(instance, pairing)| { + let running = advertised + .iter() + .find(|(i, _)| *i == instance) + .map(|(_, e)| e.clone()); + async move { + let (status, error) = match running { + None => ("not running", None), + Some(endpoint) => match probe(&endpoint, &pairing, 0).await { + Probe::Accepted(_) => ("accepted", None), + Probe::NotAccepted => ("not accepted", None), + Probe::NotRunning => ("not running", None), + Probe::Unconfirmed(e) => ("unconfirmed", Some(e)), + Probe::Throttled(e) => ("rate limited", Some(e)), + Probe::Refused(e) => ("refused", Some(e)), + }, + }; + let mut row = serde_json::json!({ + "instance": instance, + "relays": pairing.relays, + "pubkey": pairing.pubkey, + "status": status, + }); + if let Some(e) = error { + row["error"] = e.to_string().into(); + } + row + } + }); + let asked: Vec<_> = rows.map(tokio::spawn).collect(); + let mut listed = Vec::with_capacity(asked.len()); + for row in asked { + listed.push( + row.await + .map_err(|e| CliError::Other(format!("listing failed: {e}")))?, + ); + } + Ok(listed) +} + +/// The relays this CLI's stored pairings are for, in bound form, without +/// asking any desktop. For error hints only: a stored pairing may be stale. +pub fn paired_relays() -> Vec { + // Tests never read the real keychain. + #[cfg(test)] + return Vec::new(); + #[cfg(not(test))] + paired_relays_in(&Keychain) +} + +fn paired_relays_in(store: &dyn PairingStore) -> Vec { + let Ok(Ok(pairings)) = store.read().map(parse_pairings) else { + return Vec::new(); + }; + let mut relays = Vec::new(); + for relay in pairings.0.into_values().flat_map(|p| p.relays) { + push_distinct(&mut relays, relay); + } + relays +} + async fn discover_with( inherited: Option<(Endpoint, String)>, store: &dyn PairingStore, @@ -1122,6 +1338,7 @@ async fn paired_link( }; let relay = provider_relay_url(relay_url)?; let mut revoked = Vec::new(); + let mut unconfirmed = false; let mut found = Ok(None); for (instance, endpoint) in advertised { let Some(pairing) = pairings.0.get(&instance) else { @@ -1143,7 +1360,10 @@ async fn paired_link( ); } // A stale or unconfirmed advertisement decides nothing: skip it. - Ok(Err(Skip::NotApplicable | Skip::Unreachable | Skip::Unconfirmed)) => {} + Ok(Err(Skip::NotApplicable | Skip::Unreachable)) => {} + // A same-user check that ran out of time decided nothing either, + // but if nothing else serves, the command is worth retrying. + Ok(Err(Skip::Unconfirmed)) => unconfirmed = true, // The desktop answered (denied, unavailable, ...): that is final. Err(e) => { found = Err(e); @@ -1152,7 +1372,10 @@ async fn paired_link( } } forget_revoked(store, &revoked); - found + match found { + Ok(None) if unconfirmed => Err(ProviderError::Unconfirmed.into_cli("describe")), + found => found, + } } #[derive(serde::Deserialize)] @@ -1170,7 +1393,24 @@ pub async fn pair( client_name: &str, instance: Option<&str>, ) -> Result { - let mut found = advertisements(); + pair_among( + &Keychain, + advertisements(), + relay_url, + client_name, + instance, + ) + .await +} + +/// [`pair`] among the `found` desktops, storing in `store`. +async fn pair_among( + store: &dyn PairingStore, + mut found: Vec<(String, Endpoint)>, + relay_url: &str, + client_name: &str, + instance: Option<&str>, +) -> Result { if let Some(want) = instance { found.retain(|(i, _)| i == want); } @@ -1189,7 +1429,7 @@ pub async fn pair( ))); } }; - pair_with(&Keychain, instance, endpoint, relay_url, client_name).await + pair_with(store, instance, endpoint, relay_url, client_name).await } async fn pair_with( @@ -1282,7 +1522,10 @@ pub(crate) mod tests { tokio::task_local! { /// Pretend the caller runs as this user ID. pub(crate) static CALLER_UID: u32; + /// Make every same-user check run out of time. + pub(crate) static SAME_USER_HANGS: (); } + use std::sync::Mutex as StdMutex; #[test] @@ -2152,6 +2395,279 @@ pub(crate) mod tests { assert!(store.json.lock().unwrap().is_none()); } + /// A paired desktop whose user cannot be confirmed is skipped without a + /// byte sent, so the chain moves on; a check that only ran out of time + /// leaves the command retryable rather than without credentials. + #[tokio::test] + async fn an_unconfirmed_paired_desktop_is_skipped() { + let (endpoint, seen) = desktop(nostr::Keys::generate(), vec![RELAY.into()]).await; + let store = MemoryStore::with(&[("a", "ta", &[RELAY])]); + let ads = || vec![("a".to_owned(), endpoint.clone())]; + let found = CALLER_UID + .scope(u32::MAX - 1, discover_with(None, &store, ads, RELAY)) + .await + .unwrap(); + assert!(found.is_none()); + + let ads = || vec![("a".to_owned(), endpoint.clone())]; + let err = SAME_USER_HANGS + .scope((), discover_with(None, &store, ads, RELAY)) + .await + .unwrap_err(); + assert!(crate::error::is_retryable_error(&err), "{err:?}"); + assert_eq!(crate::error::exit_code(&err), 2); + assert!(seen.lock().unwrap().is_empty()); + assert_eq!(store.instances(), ["a"]); + } + + /// A command given no relay uses the one relay a running desktop still + /// accepts a pairing for, never a stored pairing alone. + #[tokio::test] + async fn the_default_relay_is_the_one_a_running_desktop_accepts() { + const OTHER: &str = "wss://other.example/"; + let (good, _) = desktop(nostr::Keys::generate(), vec![RELAY.into()]).await; + let (other, _) = desktop(nostr::Keys::generate(), vec![OTHER.into()]).await; + let (revoked, _) = + provider(|_, _| serde_json::json!({"result": "", "error": "revoked"})).await; + let gone = Endpoint::parse("http://127.0.0.1:1/rpc").unwrap(); + let store = MemoryStore::with(&[ + ("a", "ta", &[RELAY]), + ("b", "tb", &[OTHER]), + ("c", "tc", &[OTHER]), + ("d", "td", &[OTHER]), + ]); + let ad = |i: &str, e: &Endpoint| (i.to_owned(), e.clone()); + + let one = vec![ad("a", &good), ad("c", &revoked), ad("d", &gone)]; + assert_eq!( + accepted_relay(&store, one).await.unwrap().as_deref(), + Some("wss://community.example/") + ); + // Several accepted relays, or none, keep the built-in default. + let two = vec![ad("a", &good), ad("b", &other)]; + assert_eq!(accepted_relay(&store, two).await.unwrap(), None); + let none = vec![ad("c", &revoked), ad("d", &gone)]; + assert_eq!(accepted_relay(&store, none).await.unwrap(), None); + // A desktop that no longer binds the paired relay does not count. + let rebound = vec![ad("b", &good)]; + assert_eq!(accepted_relay(&store, rebound).await.unwrap(), None); + + assert_eq!( + paired_relays_in(&store), + ["wss://community.example/", OTHER] + ); + } + + /// With no relay given, a desktop that refuses is final, and a check that + /// may pass on retry stays retryable, never a fallback to localhost. + #[tokio::test] + async fn the_default_relay_keeps_desktop_failures() { + let store = MemoryStore::with(&[("a", "ta", &[RELAY])]); + let (denied, _) = provider( + |_, _| serde_json::json!({"result": "", "error": "denied", "message": "locked"}), + ) + .await; + let err = accepted_relay(&store, vec![("a".into(), denied)]) + .await + .unwrap_err(); + assert!(!crate::error::is_retryable_error(&err), "{err:?}"); + assert!(err.to_string().contains("locked"), "{err}"); + + let (throttled, _) = provider( + |_, _| serde_json::json!({"result": "", "error": "rate_limited", "retry_after": 1}), + ) + .await; + let err = accepted_relay(&store, vec![("a".into(), throttled)]) + .await + .unwrap_err(); + assert!(crate::error::is_retryable_error(&err), "{err:?}"); + + let (good, _) = desktop(nostr::Keys::generate(), vec![RELAY.into()]).await; + let err = SAME_USER_HANGS + .scope((), accepted_relay(&store, vec![("a".into(), good.clone())])) + .await + .unwrap_err(); + assert_eq!(crate::error::exit_code(&err), 2, "{err:?}"); + + // A still-throttled desktop might add a second relay, so one + // accepted relay elsewhere does not settle the choice. + let (throttled, _) = provider( + |_, _| serde_json::json!({"result": "", "error": "rate_limited", "retry_after": 1}), + ) + .await; + let store = MemoryStore::with(&[("a", "ta", &[RELAY]), ("b", "tb", &[RELAY])]); + let ads = vec![("a".into(), good), ("b".into(), throttled)]; + let err = accepted_relay(&store, ads).await.unwrap_err(); + assert_eq!(crate::error::exit_code(&err), 2, "{err:?}"); + } + + /// With no usable relay, a retryable failure is reported over an + /// identity mismatch, in either order. + #[test] + fn a_retryable_failure_outranks_an_identity_mismatch() { + let mismatch = || ProviderError::OtherUser.into_cli("describe"); + let timeout = || ProviderError::Unconfirmed.into_cli("describe"); + for order in [[mismatch(), timeout()], [timeout(), mismatch()]] { + let mut kept = None; + for e in order { + keep_transient(&mut kept, e); + } + assert_eq!(crate::error::exit_code(&kept.unwrap()), 2); + } + let mut kept = None; + keep_transient(&mut kept, mismatch()); + assert_eq!(crate::error::exit_code(&kept.unwrap()), 3); + } + + /// Through the command's relay selection with no `--relay`: a desktop's + /// refusal and a retryable failure both fail the command. + #[tokio::test] + async fn a_command_given_no_relay_reports_desktop_failures() { + use clap::Parser; + let cli = crate::Cli::try_parse_from(["buzz", "users", "get"]).unwrap(); + assert!(cli.relay.is_none()); + let store = MemoryStore::with(&[("a", "ta", &[RELAY])]); + let (denied, _) = provider( + |_, _| serde_json::json!({"result": "", "error": "denied", "message": "locked"}), + ) + .await; + let paired = accepted_relay(&store, vec![("a".into(), denied)]); + let err = crate::default_relay_with(&cli.command, false, async { Ok(None) }, paired) + .await + .unwrap_err(); + assert!(err.to_string().contains("locked"), "{err}"); + + let (good, _) = desktop(nostr::Keys::generate(), vec![RELAY.into()]).await; + let paired = accepted_relay(&store, vec![("a".into(), good.clone())]); + let err = SAME_USER_HANGS + .scope( + (), + crate::default_relay_with(&cli.command, false, async { Ok(None) }, paired), + ) + .await + .unwrap_err(); + assert_eq!(crate::error::exit_code(&err), 2, "{err:?}"); + + let paired = accepted_relay(&store, vec![("a".into(), good)]); + let relay = crate::default_relay_with(&cli.command, false, async { Ok(None) }, paired) + .await + .unwrap(); + assert_eq!(relay, "wss://community.example/"); + } + + /// `auth desktop pair --instance b` with no `--relay`, through relay + /// selection and command dispatch: desktop A, which would refuse, is + /// never asked, and B gets exactly one pair request. + #[tokio::test] + async fn pairing_a_chosen_desktop_ignores_another_desktops_refusal() { + use clap::Parser; + let cli = + crate::Cli::try_parse_from(["buzz", "auth", "desktop", "pair", "--instance", "b"]) + .unwrap(); + let store = MemoryStore::with(&[("a", "ta", &[RELAY])]); + let (refusing, refused) = provider( + |_, _| serde_json::json!({"result": "", "error": "denied", "message": "locked"}), + ) + .await; + let pubkey = nostr::Keys::generate().public_key().to_hex(); + let answer = serde_json::json!({ + "token": "tok", "instance": "b", "pubkey": pubkey, "relays": [RELAY], + }); + let reply = answer.to_string(); + let (b, seen) = provider(move |_, _| serde_json::json!({ "result": reply })).await; + let found = vec![("a".to_owned(), refusing.clone()), ("b".to_owned(), b)]; + + let paired = accepted_relay(&store, vec![("a".into(), refusing)]); + let relay = crate::default_relay_with(&cli.command, false, async { Ok(None) }, paired) + .await + .unwrap(); + let result = std::cell::RefCell::new(None); + let connected = crate::connect( + &relay, + cli.command, + async |sub, relay| { + let crate::AuthCmd::Desktop(crate::DesktopAuthCmd::Pair { name, instance }) = sub + else { + unreachable!() + }; + let out = pair_among(&store, found, relay, &name, instance.as_deref()).await?; + *result.borrow_mut() = Some(out); + Ok(()) + }, + async |_| unreachable!(), + ) + .await + .unwrap(); + assert!(connected.is_none()); + + assert!(refused.lock().unwrap().is_empty(), "A was asked"); + let asked = seen.lock().unwrap(); + assert_eq!(asked.len(), 1); + assert_eq!(asked[0].2["method"], "pair"); + assert_eq!( + asked[0].2["params"], + serde_json::json!(["buzz CLI", "ws://localhost:3000/"]) + ); + assert_eq!( + result.into_inner().unwrap(), + serde_json::json!({"instance": "b", "pubkey": pubkey, "relays": [RELAY]}) + ); + } + + /// A crashed desktop's leftover advertisement reads as not running, and + /// a throttled one is reported at once, not waited out. + #[tokio::test] + async fn list_tells_a_stale_advertisement_from_a_rejection() { + let closed = { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); + Endpoint::parse(&format!("http://127.0.0.1:{port}/rpc")).unwrap() + }; + let (throttled, seen) = provider( + |_, _| serde_json::json!({"result": "", "error": "rate_limited", "retry_after": 30}), + ) + .await; + let store = MemoryStore::with(&[("a", "ta", &[RELAY]), ("b", "tb", &[RELAY])]); + let ads = vec![("a".to_owned(), closed), ("b".to_owned(), throttled)]; + let started = std::time::Instant::now(); + let listed = list_with(&store, ads).await.unwrap(); + assert!(started.elapsed() < Duration::from_secs(5)); + assert_eq!(listed[0]["status"], "not running"); + assert_eq!(listed[1]["status"], "rate limited"); + assert_eq!(seen.lock().unwrap().len(), 1, "no retries"); + } + + /// The listing never shows a token and reports each pairing's state the + /// way the default relay judges it. + #[tokio::test] + async fn list_shows_each_pairing_without_its_token() { + assert!(list_with(&MemoryStore::default(), Vec::new()) + .await + .unwrap() + .is_empty()); + let (good, _) = desktop(nostr::Keys::generate(), vec![RELAY.into()]).await; + let (revoked, _) = + provider(|_, _| serde_json::json!({"result": "", "error": "revoked"})).await; + let store = MemoryStore::with(&[ + ("a", "secret-a", &[RELAY]), + ("b", "secret-b", &[RELAY]), + ("c", "secret-c", &[RELAY]), + ]); + let ads = vec![("a".to_owned(), good), ("b".to_owned(), revoked)]; + let listed = list_with(&store, ads).await.unwrap(); + let text = serde_json::to_string(&listed).unwrap(); + assert!(!text.contains("secret"), "{text}"); + let status: Vec<_> = listed.iter().map(|p| p["status"].clone()).collect(); + assert_eq!(status, ["accepted", "not accepted", "not running"]); + assert_eq!(listed[0]["relays"], serde_json::json!([RELAY])); + assert_eq!( + store.instances(), + ["a", "b", "c"], + "listing changes nothing" + ); + } + /// When this user cannot be confirmed on the server side, not one byte /// (so no token) reaches the endpoint. #[tokio::test] diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 546fb76ef2..973fd0d2a5 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -129,7 +129,10 @@ Buzz CLI — interact with a Buzz relay Configuration (flags override env vars): BUZZ_RELAY_URL Relay base URL [default: the relay an agent's inherited - Buzz desktop credential acts on, else http://localhost:3000] + Buzz desktop credential acts on; else, with no private key, + the one relay a running Buzz desktop accepts this CLI's + pairing for (see `buzz auth desktop list`); else + http://localhost:3000] BUZZ_PRIVATE_KEY Nostr private key (hex or nsec) [unless a Buzz desktop provides it] BUZZ_AUTH_TAG NIP-OA auth tag JSON [optional] BUZZ_FEDERATED_IDENTITY_TOKEN Your NIP-FA sign-in session token [optional] @@ -147,7 +150,8 @@ Errors are JSON on stderr: {\"error\": \"\", \"message\": \"\" struct Cli { /// Relay URL (http:// or https://). Overrides BUZZ_RELAY_URL env var. /// Without either, the relay an agent's inherited Buzz desktop credential - /// acts on; otherwise http://localhost:3000. + /// acts on; else, with no private key, the one relay a running Buzz + /// desktop accepts this CLI's pairing for; otherwise http://localhost:3000. #[arg(long, env = "BUZZ_RELAY_URL")] relay: Option, @@ -376,23 +380,58 @@ enum DesktopAuthCmd { #[arg(long)] instance: Option, }, + /// List this CLI's pairings: instance, relays, pubkey, and whether a + /// running Buzz desktop accepts each now. Never prints tokens; needs no + /// relay. + List, } /// The relay for a command given none: the one an agent's inherited Buzz -/// desktop credential acts on, else `http://localhost:3000`. Local-only -/// unpairing never asks the desktop. -async fn default_relay(command: &Cmd) -> Result { +/// desktop credential acts on; else, without an explicit key, the one relay +/// a running Buzz desktop still accepts this CLI's pairing for; else +/// `http://localhost:3000`. Local-only unpairing never asks the desktop, +/// and pairing never asks for other pairings. +async fn default_relay(command: &Cmd, explicit_key: bool) -> Result { + #[cfg(unix)] + let (inherited, paired) = ( + desktop_provider::inherited_relay(), + desktop_provider::paired_relay(), + ); + #[cfg(not(unix))] + let (inherited, paired) = (async { Ok(None) }, async { Ok(None) }); + default_relay_with(command, explicit_key, inherited, paired).await +} + +/// [`default_relay`], with the agent and paired-person selectors (each +/// awaited only when needed) parameters so a test can run it against a fake +/// desktop. +pub(crate) async fn default_relay_with( + command: &Cmd, + explicit_key: bool, + inherited: impl std::future::Future, CliError>>, + paired: impl std::future::Future, CliError>>, +) -> Result { #[cfg(unix)] if !matches!( command, Cmd::Auth(AuthCmd::Desktop(DesktopAuthCmd::Unpair { .. })) ) { - if let Some(relay) = desktop_provider::inherited_relay().await? { + if let Some(relay) = inherited.await? { return Ok(relay); } + // Pairing asks its own desktop; another desktop's answer is moot. + let pairing = matches!( + command, + Cmd::Auth(AuthCmd::Desktop(DesktopAuthCmd::Pair { .. })) + ); + if !explicit_key && !pairing { + if let Some(relay) = paired.await? { + return Ok(relay); + } + } } #[cfg(not(unix))] - let _ = command; + let _ = (command, explicit_key, inherited, paired); Ok("http://localhost:3000".to_owned()) } @@ -404,11 +443,34 @@ async fn run_auth(cmd: AuthCmd, relay_url: &str) -> Result<(), CliError> { desktop_provider::pair(relay_url, &name, instance.as_deref()).await? } DesktopAuthCmd::Unpair { instance } => desktop_provider::unpair(instance.as_deref())?, + DesktopAuthCmd::List => unreachable!("listing runs before any relay is chosen"), }; println!("{out}"); Ok(()) } +/// `auth desktop list`, in `format`: `compact` drops the pubkey. +#[cfg(unix)] +async fn list_pairings(format: &OutputFormat) -> Result<(), CliError> { + let mut listed = desktop_provider::list().await?; + if matches!(format, OutputFormat::Compact) { + for pairing in &mut listed { + if let Some(fields) = pairing.as_object_mut() { + fields.remove("pubkey"); + } + } + } + println!("{}", serde_json::Value::Array(listed)); + Ok(()) +} + +#[cfg(not(unix))] +async fn list_pairings(_: &OutputFormat) -> Result<(), CliError> { + Err(CliError::Usage( + "the Buzz desktop credential provider is not available on Windows yet".into(), + )) +} + #[cfg(not(unix))] async fn run_auth(_: AuthCmd, _: &str) -> Result<(), CliError> { Err(CliError::Usage( @@ -2566,9 +2628,14 @@ async fn run(cli: Cli) -> Result<(), CliError> { ..Default::default() } .with_env_evidence(); + // Listing pairings is local: it needs no relay and runs no discovery. + if let Cmd::Auth(AuthCmd::Desktop(DesktopAuthCmd::List)) = cli.command { + return list_pairings(&cli.format).await; + } + let explicit_key = settings.private_key.is_some(); let relay = match cli.relay { Some(relay) => relay, - None => default_relay(&cli.command).await?, + None => default_relay(&cli.command, explicit_key).await?, }; let Some((command, client)) = connect(&relay, cli.command, run_auth, async |relay| { credentials::resolve(settings, relay).await @@ -2681,6 +2748,38 @@ mod tests { Cli::command().debug_assert(); } + /// Relay selection asks only what a command needs: an explicit key never + /// asks for a pairing, unpairing asks nothing, pairing asks for no other + /// pairing, and an agent's inherited relay comes first. `None` for + /// `inherited` means it must not be asked; the paired lookup never is. + #[cfg(unix)] + #[tokio::test] + async fn relay_selection_asks_only_what_a_command_needs() { + use clap::Parser; + const INHERITED: &str = "wss://inherited.example/"; + const DEFAULT: &str = "http://localhost:3000"; + let cases = [ + ("users get", true, Some(None), DEFAULT), + ("auth desktop unpair", false, None, DEFAULT), + ("auth desktop pair", false, Some(None), DEFAULT), + ("users get", false, Some(Some(INHERITED)), INHERITED), + ]; + for (args, explicit_key, inherited, want) in cases { + let argv = std::iter::once("buzz").chain(args.split(' ')); + let cli = Cli::try_parse_from(argv).unwrap(); + let inherited = async move { + Ok(inherited + .unwrap_or_else(|| panic!("`{args}` asked for an inherited relay")) + .map(str::to_owned)) + }; + let paired = async { panic!("`{args}` asked for a paired relay") }; + let relay = default_relay_with(&cli.command, explicit_key, inherited, paired) + .await + .unwrap(); + assert_eq!(relay, want, "`{args}`"); + } + } + #[test] fn messages_send_from_thread_conflicts_with_reply_and_broadcast() { let base = [ diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index d3c09338b9..586a8514da 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -3050,6 +3050,7 @@ mod e_tag_filter_shape_tests { let split = crate::store::recent_pass::Split { floor: at(10), ceiling: Some(at(12)), + older: true, }; let floor = "'2026-10-01T00:00:00Z'::timestamptz"; let ceiling = "'2026-12-01T00:00:00Z'::timestamptz"; @@ -3648,6 +3649,39 @@ mod postgres_tests { assert_eq!(ids(page), expected[..7]); } + /// The `events` partitions one pass of `q` over `range` locks, run in a + /// transaction of its own on `conn`, after checking that it returns + /// `rows` rows. Run it under `plan_cache_mode = force_generic_plan` to see + /// what a cached generic plan locks. + async fn locked_partitions( + conn: &mut sqlx::PgConnection, + q: &EventQuery, + range: Range, + rows: usize, + ) -> std::collections::BTreeSet { + let mut tx = sqlx::Connection::begin(conn).await.expect("begin"); + let mut qb = build_query_events_sql(q, range, 10); + let read = qb + .build() + .persistent(true) + .fetch_all(&mut *tx) + .await + .expect("pass"); + assert_eq!(read.len(), rows, "the pass reads every row: {range:?}"); + let locked: Vec = sqlx::query_scalar( + "SELECT c.relname::text FROM pg_locks l \ + JOIN pg_class c ON c.oid = l.relation \ + JOIN pg_inherits i ON i.inhrelid = c.oid \ + WHERE l.pid = pg_backend_pid() \ + AND i.inhparent = 'events'::regclass", + ) + .fetch_all(&mut *tx) + .await + .expect("locked partitions"); + tx.rollback().await.expect("rollback"); + locked.into_iter().collect() + } + /// The recent pass of a regular-kind read locks only the partitions /// between its floor and ceiling, even as a cached generic plan. Without /// the ceiling (a read that may match latest-state kinds) it also locks @@ -3712,42 +3746,129 @@ mod postgres_tests { channel_id: Some(channel), ..EventQuery::for_community(community) }; - async fn locked( - conn: &mut sqlx::PgConnection, - q: &EventQuery, - split: crate::store::recent_pass::Split, - ) -> std::collections::BTreeSet { - let mut tx = sqlx::Connection::begin(conn).await.expect("begin"); - let mut qb = build_query_events_sql(q, Range::Recent(split), 10); - let rows = qb - .build() - .persistent(true) - .fetch_all(&mut *tx) - .await - .expect("recent pass"); - assert_eq!(rows.len(), 3, "the recent pass reads every row"); - let locked: Vec = sqlx::query_scalar( - "SELECT c.relname::text FROM pg_locks l \ - JOIN pg_class c ON c.oid = l.relation \ - JOIN pg_inherits i ON i.inhrelid = c.oid \ - WHERE l.pid = pg_backend_pid() \ - AND i.inhparent = 'events'::regclass", - ) - .fetch_all(&mut *tx) - .await - .expect("locked partitions"); - tx.rollback().await.expect("rollback"); - locked.into_iter().collect() - } - let ceiled_locks = locked(&mut conn, &q, ceiled).await; + let ceiled_locks = locked_partitions(&mut conn, &q, Range::Recent(ceiled), 3).await; assert_eq!(ceiled_locks, expected, "partitions below the ceiling"); - let open_locks = locked(&mut conn, &q, open).await; + let open_locks = locked_partitions(&mut conn, &q, Range::Recent(open), 3).await; assert!( open_locks.is_superset(&ceiled_locks) && open_locks.len() > ceiled_locks.len(), "without the ceiling the pass also locks the later partitions: {open_locks:?}" ); } + /// A read whose own `since` is at or above the floor, such as a `#p` + /// mention REQ from an agent reconnecting, is one recent pass: it returns + /// every matching row and runs no older pass. As a cached generic plan, + /// its literal floor and ceiling lock only the partitions between them, + /// where the `since` bind alone (the read whole) locks every partition. + /// Mutation: drop the `since` case from `split_for`, or run the older + /// pass after it → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn since_bounded_mention_read_locks_only_recent_partitions() { + use crate::store::recent_pass::{ceiling_at, floor_at, next}; + let pool = setup_pool().await; + let db = crate::Db::from_pool(pool.clone()); + db.maintain_partitions( + crate::partition::PARTITION_MANAGER_MONTHS_AHEAD, + crate::partition::PartitionMaintenancePolicy { + create_enabled: true, + advance_enabled: true, + }, + ) + .await + .expect("maintain partitions"); + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let channel = make_test_channel(&pool, community_uuid, None).await; + + // Mentions of the target at the floor, now and a second below the + // ceiling, newest first, and one the day before the floor. + let now = Utc::now(); + let floor = floor_at(now); + let author = Keys::generate(); + let target = "ab".repeat(32); + let mut recent = Vec::new(); + let mut expected = std::collections::BTreeSet::new(); + let last = ceiling_at(now) - chrono::Duration::seconds(1); + for at in [last, now, floor, floor - chrono::Duration::days(1)] { + let event = EventBuilder::new(Kind::Custom(9), format!("at {at}")) + .tags([Tag::parse(["p", target.as_str()]).expect("p tag")]) + .custom_created_at(nostr::Timestamp::from(at.timestamp() as u64)) + .sign_with_keys(&author) + .expect("sign"); + db.insert_event(community, &event, Some(channel)) + .await + .expect("insert"); + if at >= floor { + recent.push(event.id); + let partition: String = + sqlx::query_scalar("SELECT tableoid::regclass::text FROM events WHERE id = $1") + .bind(event.id.as_bytes().to_vec()) + .fetch_one(&pool) + .await + .expect("partition"); + expected.insert(partition); + } + } + + let q = EventQuery { + kinds: Some(vec![9]), + p_tag_hex: Some(target.clone()), + since: Some(floor), + limit: Some(10), + ..EventQuery::for_community(community) + }; + let split = recent_pass_split(&q, now).expect("a since-bounded read has a recent pass"); + assert!(!split.older, "rows below `since` cannot match"); + assert_eq!(next(Range::Recent(split), recent.len(), 10), None); + + // The production read: every matching row, from exactly the recent + // pass's statement, so the lock checks below bind what it prepared. + let mut conn = pool.acquire().await.expect("acquire"); + sqlx::Connection::clear_cached_statements(&mut *conn) + .await + .expect("clear prepared statements"); + let page = query_events_on(&mut conn, &q, ReadDeadline::PerRead) + .await + .expect("mention page"); + let ids: Vec<_> = page.into_iter().map(|r| r.event.id).collect(); + assert_eq!(ids, recent); + let prepared: Vec = sqlx::query_scalar( + "SELECT statement FROM pg_prepared_statements WHERE statement LIKE $1", + ) + .bind("%JOIN event_mentions%") + .fetch_all(&mut *conn) + .await + .expect("read prepared statements"); + let recent_pass = build_query_events_sql(&q, Range::Recent(split), 10) + .sql() + .as_str() + .to_owned(); + assert_eq!(prepared, [recent_pass], "one pass, bounded by literals"); + + sqlx::query("SET plan_cache_mode = force_generic_plan") + .execute(&mut *conn) + .await + .expect("force generic plans"); + let bounded = locked_partitions(&mut conn, &q, Range::Recent(split), recent.len()).await; + assert_eq!( + bounded, expected, + "partitions between the floor and ceiling" + ); + let bind_only = locked_partitions(&mut conn, &q, Range::All, recent.len()).await; + let every: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_inherits WHERE inhparent = 'events'::regclass", + ) + .fetch_one(&mut *conn) + .await + .expect("partitions"); + assert_eq!( + bind_only.len() as i64, + every, + "the `since` bind alone locks every partition: {bind_only:?}" + ); + } + /// A newest-first page read split at the floor returns the rows the /// unsplit read returns: the recent pass alone when it fills the page, /// otherwise followed by the older pass for just the missing rows. diff --git a/crates/buzz-db/src/store/recent_pass.rs b/crates/buzz-db/src/store/recent_pass.rs index 96d7c72035..a4dd5d2ef3 100644 --- a/crates/buzz-db/src/store/recent_pass.rs +++ b/crates/buzz-db/src/store/recent_pass.rs @@ -22,6 +22,13 @@ //! passes read one snapshot (the stores run them in one `REPEATABLE READ` //! transaction, `ReadTransaction`). //! +//! A read whose own `since` is at or above the floor matches no row below +//! it, so its recent pass is the whole read and no older pass follows. The +//! floor literal is still worth adding: `since` is a bind parameter, which a +//! cached generic plan prunes only at execution, after it has locked every +//! partition. Live REQs with a recent `since` (agents reconnecting, `#p` +//! mention reads) are that read. +//! //! The bounds are SQL literals, so Postgres prunes when it plans the //! statement, cached generic plans included. They change about once a month, //! so the prepared-statement cache keeps one statement per pass and shape. @@ -52,6 +59,10 @@ pub(crate) struct Split { /// The recent pass reads below it, when set. No row the read can match /// sits at or above it. pub(crate) ceiling: Option>, + /// Whether rows below the floor can match, so a short recent pass is + /// followed by the older pass. False when the read's own `since` is at + /// or above the floor: the recent pass is then the whole read. + pub(crate) older: bool, } /// The `created_at` range one pass of a page read covers. @@ -122,11 +133,13 @@ pub(crate) fn ceiling_at(now: DateTime) -> DateTime { month_start(now + MAX_FUTURE, 1) } -/// Where to split a page read, or `None` to read it whole: its own -/// bounds already exclude everything older than the floor or everything -/// newer, or its rows are latest-state records that rarely fill a page above -/// the floor, so a split would only add a second statement. The recent pass -/// has a ceiling only when every kind is regular (see [`MAX_FUTURE`]). +/// Where to split a page read, or `None` to read it whole: its own bounds +/// already exclude everything newer than the floor, or its rows are +/// latest-state records that rarely fill a page above the floor, so a split +/// would only add a second statement. A read whose `since` is at or above the +/// floor gets a recent pass with no older pass after it (see the module +/// docs), whatever its kinds. The recent pass has a ceiling only when every +/// kind is regular (see [`MAX_FUTURE`]). /// /// `kinds` are the read's kinds (`None`: any); `since`/`until` its bounds. pub(crate) fn split_for( @@ -146,10 +159,15 @@ pub(crate) fn split_for( let regular_only = kinds.is_some_and(|ks| !ks.iter().any(|&k| latest_state(k))); let since_prunes = since.is_some_and(|s| s >= floor); let until_before_floor = until.is_some_and(|u| u < floor); - (!latest_state_only && !since_prunes && !until_before_floor).then(|| Split { + let split = |older| Split { floor, ceiling: regular_only.then(|| ceiling_at(now)), - }) + older, + }; + if since_prunes { + return (!until_before_floor).then(|| split(false)); + } + (!latest_state_only && !until_before_floor).then(|| split(true)) } /// The pass to run after `range` once the read holds `found` of its `limit` @@ -160,7 +178,7 @@ pub(crate) fn next(range: Range, found: usize, limit: i64) -> Option<(Range, i64 return None; }; let found = i64::try_from(found).unwrap_or(i64::MAX); - (found < limit).then(|| (Range::Older(split.floor), limit - found)) + (split.older && found < limit).then(|| (Range::Older(split.floor), limit - found)) } #[cfg(test)] @@ -219,6 +237,7 @@ mod tests { let split = Split { floor, ceiling: Some(midnight(2026, 12, 1)), + older: true, }; assert_eq!(Range::All.predicate("e.created_at"), None); assert_eq!( @@ -242,13 +261,18 @@ mod tests { ); } - /// Which reads are split, and which recent passes stop at the ceiling: - /// (kinds, since, until) → `WHOLE` (not split), `CEILED` or `OPEN`. + /// Which reads are split, which recent passes stop at the ceiling, and + /// which are the whole read: (kinds, since, until) → `WHOLE` (not split), + /// `CEILED` or `OPEN` (recent pass, then the older pass if short), or + /// `CEILED_ONLY` or `OPEN_ONLY` (recent pass, no older pass). #[test] fn split_eligibility() { - const WHOLE: Option = None; - const CEILED: Option = Some(true); - const OPEN: Option = Some(false); + // (ceiled, older) + const WHOLE: Option<(bool, bool)> = None; + const CEILED: Option<(bool, bool)> = Some((true, true)); + const OPEN: Option<(bool, bool)> = Some((false, true)); + const CEILED_ONLY: Option<(bool, bool)> = Some((true, false)); + const OPEN_ONLY: Option<(bool, bool)> = Some((false, false)); let now = at(2026, 11, 4); let floor = floor_at(now); let day = Duration::days(1); @@ -258,7 +282,7 @@ mod tests { Option<&'a [i32]>, Option>, Option>, - Option, + Option<(bool, bool)>, ); let cases: &[Case] = &[ ("unbounded messages", Some(&[9, 40002]), None, None, CEILED), @@ -272,17 +296,38 @@ mod tests { CEILED, ), ( - "since at the floor already prunes", + "since at the floor", Some(&[9]), Some(floor), None, - WHOLE, + CEILED_ONLY, ), ( - "recent since already prunes", + "recent since", Some(&[9]), Some(now - day), None, + CEILED_ONLY, + ), + ( + "recent since, any kind", + None, + Some(now - day), + None, + OPEN_ONLY, + ), + ( + "recent since, addressable only", + Some(&[39002]), + Some(now - day), + None, + OPEN_ONLY, + ), + ( + "recent since, until before the floor", + Some(&[9]), + Some(now - day), + Some(floor - day), WHOLE, ), ("until at the floor", Some(&[9]), None, Some(floor), CEILED), @@ -318,9 +363,10 @@ mod tests { ), ]; for &(label, kinds, since, until, outcome) in cases { - let expected = outcome.map(|ceiled| Split { + let expected = outcome.map(|(ceiled, older)| Split { floor, ceiling: ceiled.then(|| ceiling_at(now)), + older, }); assert_eq!(split_for(now, kinds, since, until), expected, "{label}"); } @@ -350,12 +396,25 @@ mod tests { let ceiled = Split { floor, ceiling: Some(ceiling), + older: true, }; let open = Split { ceiling: None, ..ceiled }; - for split in [None, Some(ceiled), Some(open)] { + // A read whose `since` is at or above the floor: only the rows + // at or above it match. + let since_floor = Split { + older: false, + ..ceiled + }; + for split in [None, Some(ceiled), Some(open), Some(since_floor)] { + let expected: Vec<_> = match split { + Some(s) if !s.older => { + expected.iter().copied().filter(|&t| t >= floor).collect() + } + _ => expected.clone(), + }; let mut passes = Vec::new(); let mut rows = Vec::new(); let mut pass = Some((split.map_or(Range::All, Range::Recent), limit)); @@ -377,7 +436,7 @@ mod tests { assert_eq!(rows, expected, "split {split:?}, limit {limit}"); let expected_passes = match split { None => vec![(Range::All, limit)], - Some(s) if limit <= recent => vec![(Range::Recent(s), limit)], + Some(s) if limit <= recent || !s.older => vec![(Range::Recent(s), limit)], Some(s) => vec![ (Range::Recent(s), limit), (Range::Older(s.floor), limit - recent), diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 57c6b90560..5eb1897cea 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -75,6 +75,9 @@ pub struct ChannelEvent { pub topic: EventTopic, /// The Nostr event payload. pub event: nostr::Event, + /// When this node read the event from Redis. Delivery deadlines start + /// here, so time spent waiting in the local broadcast counts. + pub received_at: tokio::time::Instant, } /// Health of this node's Redis event subscription link. diff --git a/crates/buzz-pubsub/src/subscriber.rs b/crates/buzz-pubsub/src/subscriber.rs index 0c34dba8d1..b43ccea461 100644 --- a/crates/buzz-pubsub/src/subscriber.rs +++ b/crates/buzz-pubsub/src/subscriber.rs @@ -187,6 +187,7 @@ async fn connect_and_subscribe( community_id: topic_key.community_id, topic: topic_key.topic, event, + received_at: tokio::time::Instant::now(), }; if let Err(_e) = broadcast_tx.send(channel_event) { diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index a9067da122..8cf56875a2 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -2556,8 +2556,10 @@ pub(crate) mod tests { let topic = EventTopic::Channel(channel); handlers::req::claim_live_subscription( "s", - &[nostr::Filter::new().kind(Kind::TextNote)], - Some(&[channel]), + vec![( + nostr::Filter::new().kind(Kind::TextNote), + crate::subscription::SubscriptionScope::channels([channel]), + )], &conn, &state, ) @@ -2652,8 +2654,10 @@ pub(crate) mod tests { let topic = EventTopic::Channel(channel); handlers::req::claim_live_subscription( "s", - &[nostr::Filter::new().kind(Kind::TextNote)], - Some(&[channel]), + vec![( + nostr::Filter::new().kind(Kind::TextNote), + crate::subscription::SubscriptionScope::channels([channel]), + )], &conn, &state, ) @@ -2742,8 +2746,10 @@ pub(crate) mod tests { let (conn, _rx) = test_conn_with_auth(AuthState::Failed); handlers::req::claim_live_subscription( "s", - &[nostr::Filter::new().kind(Kind::TextNote)], - None, + vec![( + nostr::Filter::new().kind(Kind::TextNote), + crate::subscription::SubscriptionScope::global(), + )], &conn, &state, ) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 3393b01c66..7c93733262 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -367,31 +367,55 @@ pub(crate) async fn fan_out_event_to_local_subscribers( /// Fan out one event received from Redis pub/sub to this relay's local subscribers. #[tracing::instrument(skip_all)] pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pubsub::ChannelEvent) { - // The Redis topic carries the tenant-local routing scope explicitly: - // `Channel(id)` for a per-channel event, `Global` for a channel-less one. - // Convert back to the `Option` channel id `fan_out()` indexes on — - // `Global` selects the global subscriber index. + if take_local_echo(state, &channel_event) { + return; + } + let community_id = channel_event.community_id; + deliver_pubsub_event(state, community_id, &pubsub_stored_event(channel_event)).await; +} + +/// Report whether `channel_event` is the Redis echo of an event this relay +/// already fanned out in-process, consuming the echo record if so. +/// +/// The dedup key is `(community_id, event_id)` — a same-id event arriving for a +/// *different* community is a distinct delivery and must not be suppressed. +/// The cache has TTL-based eviction (60s) so entries are bounded regardless of +/// subscriber health. +pub(crate) fn take_local_echo(state: &AppState, channel_event: &buzz_pubsub::ChannelEvent) -> bool { + let echo_key = ( + channel_event.community_id, + channel_event.event.id.to_bytes(), + ); + if state.local_event_ids.get(&echo_key).is_some() { + state.local_event_ids.invalidate(&echo_key); + return true; + } + false +} + +/// The `StoredEvent` a Redis event fans out as. +/// +/// The Redis topic carries the tenant-local routing scope explicitly: +/// `Channel(id)` for a per-channel event, `Global` for a channel-less one. +/// Convert back to the `Option` channel id `fan_out()` indexes on — +/// `Global` selects the global subscriber index. +pub(crate) fn pubsub_stored_event(channel_event: buzz_pubsub::ChannelEvent) -> StoredEvent { let channel_id = match channel_event.topic { buzz_pubsub::EventTopic::Channel(id) => Some(id), buzz_pubsub::EventTopic::Global => None, }; - let community_id = channel_event.community_id; - let stored = StoredEvent::new(channel_event.event, channel_id); + StoredEvent::new(channel_event.event, channel_id) +} - // Skip events that were already fanned out in-process (local echo). The - // dedup key is `(community_id, event_id)` — a same-id event arriving for a - // *different* community is a distinct delivery and must not be suppressed. - // The cache has TTL-based eviction (60s) so entries are bounded regardless - // of subscriber health. - let event_id_bytes = stored.event.id.to_bytes(); - let echo_key = (community_id, event_id_bytes); - if state.local_event_ids.get(&echo_key).is_some() { - state.local_event_ids.invalidate(&echo_key); - return; - } - - let matches = state.sub_registry.fan_out_scoped(community_id, &stored); - let matches = route_fanout_matches(state, community_id, &stored, matches, None).await; +/// Deliver one Redis event that is not a local echo to this relay's local +/// subscribers, through the access gate. +pub(crate) async fn deliver_pubsub_event( + state: &Arc, + community_id: CommunityId, + stored: &StoredEvent, +) { + let matches = state.sub_registry.fan_out_scoped(community_id, stored); + let matches = route_fanout_matches(state, community_id, stored, matches, None).await; metrics::counter!("buzz_multinode_fanout_total").increment(1); if matches.is_empty() { return; @@ -2464,6 +2488,7 @@ mod tests { community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), topic: EventTopic::Global, event, + received_at: tokio::time::Instant::now(), }, ) .await; @@ -2559,6 +2584,7 @@ mod tests { community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), topic: EventTopic::Global, event, + received_at: tokio::time::Instant::now(), }, ) .await; @@ -2582,6 +2608,7 @@ mod tests { community_id: community, topic: EventTopic::Global, event, + received_at: tokio::time::Instant::now(), }, ) .await; @@ -2624,6 +2651,7 @@ mod tests { community_id: community_b, topic: EventTopic::Global, event, + received_at: tokio::time::Instant::now(), }, ) .await; @@ -2653,6 +2681,7 @@ mod tests { community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()), topic: EventTopic::Global, event, + received_at: tokio::time::Instant::now(), }, ) .await; @@ -3474,6 +3503,7 @@ mod tests { community_id, topic: buzz_pubsub::EventTopic::Channel(channel_id), event: artifact(channel_id).event, + received_at: tokio::time::Instant::now(), }, ) .await; diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index d15623f5bf..7f5d11082b 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -42,6 +42,8 @@ pub mod moderation_authz; pub mod moderation_commands; /// Relay-signed moderation notice DMs. pub mod moderation_notices; +/// Multi-node fan-out: Redis events to local subscribers, with lossless lag handling. +pub mod multinode_fanout; /// Product-feedback validation + deployment sidecar persistence. pub mod product_feedback; #[allow(dead_code, missing_docs)] diff --git a/crates/buzz-relay/src/handlers/multinode_fanout.rs b/crates/buzz-relay/src/handlers/multinode_fanout.rs new file mode 100644 index 0000000000..15eacdd9ee --- /dev/null +++ b/crates/buzz-relay/src/handlers/multinode_fanout.rs @@ -0,0 +1,717 @@ +//! Multi-node fan-out: deliver events other relay nodes published to Redis to +//! this node's local subscribers. +//! +//! One dispatcher reads the pub/sub broadcast and only routes. Delivery runs on +//! [`WORKERS`] workers, each with a bounded queue. Every event of one +//! `(community, topic)` goes to the same worker, so each channel's events keep +//! their order while different channels are delivered concurrently. +//! +//! Live delivery never loses an event silently. An event is lost when its +//! worker's queue is full, or when it is not delivered within +//! [`MAX_QUEUE_DELAY`] of this node reading it from Redis. Either way the syncs +//! it would have reached get `CLOSED "resync: ..."` (NIP-AR Sync rule 5), and +//! the connections it would have reached are fenced and closed so their +//! clients catch up from their cursor. When the dispatcher itself lags the +//! broadcast, it cannot tell which events it skipped, so it does the same for +//! every subscribed connection. +//! +//! An ephemeral event (presence, typing, observer frames) is never stored, so +//! no reconnect can replay it. Losing one is only counted: closing its +//! recipients would recover nothing, and a global presence event reaches most +//! of a community's connections. +//! +//! A client resumes from the newest `created_at` it received, minus a few +//! seconds of slack. Recovery depends on that cursor never getting further +//! ahead of a lost event than the slack: +//! - A fence is immediate and atomic with the live send path, so after it no +//! newer frame is queued to the connection. Frames queued before it were +//! produced no later than the fence. +//! - [`MAX_QUEUE_DELAY`] is the cutoff for attempting delivery: no event is +//! delivered, and so none moves a cursor, later than that after this node +//! read it from Redis. A worker that has not delivered an event by then +//! fences its recipients at once. +//! +//! So a missed event is replayed as long as `MAX_QUEUE_DELAY`, plus its +//! writer's clock skew and its publish-to-Redis delay, stays under the +//! client's slack. Three cases fall outside that bound: +//! - The relay accepts `created_at` far outside the slack, and an ordinary +//! reconnect has the same limit. Only a client-side rewind can recover an +//! event whose writer's clock is further behind. +//! - A dispatcher that falls behind without overflowing the broadcast detects +//! the loss late: the event is still never delivered, but its fence lands +//! only when the dispatcher reaches it. +//! - Dispatcher lag, where the skipped events are never seen. +//! +//! The socket close itself goes through one paced closer so a hot channel +//! cannot turn into a reconnect storm (Review-Proven Rule 9). +use std::collections::HashSet; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::sync::{Arc, Mutex, PoisonError}; +use std::time::Duration; + +use buzz_core::kind::{event_kind_u32, is_ephemeral}; +use buzz_core::CommunityId; +use buzz_pubsub::ChannelEvent; +use tokio::sync::mpsc::error::TrySendError; +use tokio::sync::{broadcast, mpsc, Notify}; +use tokio::time::Instant; +use tracing::warn; + +use super::event::{deliver_pubsub_event, pubsub_stored_event, take_local_echo}; +use crate::state::AppState; +use crate::subscription::ConnId; +use buzz_core::event::StoredEvent; + +/// Delivery workers per node. +pub const WORKERS: usize = 16; +/// Events each worker may hold before the dispatcher drops new ones. +pub const WORKER_QUEUE: usize = 1024; +/// The cutoff for delivering an event, from when this node read it from Redis. +/// An event not delivered by then counts as lost. +/// It must stay well under the clients' replay slack (5 s), so a client whose +/// cursor moved on through faster paths still replays what it missed. +pub const MAX_QUEUE_DELAY: Duration = Duration::from_secs(2); +/// Connections the closer closes per tick. +const CLOSE_BATCH: usize = 50; +/// The closer's tick: with [`CLOSE_BATCH`], at most 5,000 closes per second. +const CLOSE_TICK: Duration = Duration::from_millis(10); +/// The resync reason for a sync whose live entry was lost here. +const LAG_RESYNC: &str = "resync: live delivery lagged"; + +/// An event waiting for its worker, and when it counts as lost: its deadline +/// runs from when this node read it from Redis, not from dispatch. +struct Queued { + community_id: CommunityId, + stored: StoredEvent, + deadline: Instant, +} + +impl Queued { + fn new(channel_event: ChannelEvent) -> Self { + Self { + community_id: channel_event.community_id, + deadline: channel_event.received_at + MAX_QUEUE_DELAY, + stored: pubsub_stored_event(channel_event), + } + } +} + +/// Start multi-node fan-out for `rx`: the dispatcher, its workers, and the +/// closer. The returned handle ends when the broadcast closes. +pub fn spawn( + state: Arc, + rx: broadcast::Receiver, +) -> tokio::task::JoinHandle<()> { + let (dispatcher, queues) = Dispatcher::new(WORKERS, WORKER_QUEUE); + for queue in queues { + tokio::spawn(run_worker( + Arc::clone(&state), + Arc::clone(&dispatcher.closer), + queue, + )); + } + tokio::spawn(Arc::clone(&dispatcher.closer).run(Arc::clone(&state))); + tokio::spawn(dispatcher.run(state, rx)) +} + +/// Delivers one worker's queue in order. +async fn run_worker( + state: Arc, + closer: Arc, + mut queue: mpsc::Receiver, +) { + while let Some(queued) = queue.recv().await { + deliver_or_lose(&state, &closer, queued).await; + } +} + +/// Deliver a queued event by its deadline, or handle it as lost. A delivery +/// cut off at the deadline may have reached some recipients; they close too. +async fn deliver_or_lose(state: &Arc, closer: &LagCloser, queued: Queued) { + let Queued { + community_id, + stored, + deadline, + } = queued; + let delivered = Instant::now() < deadline + && tokio::time::timeout_at(deadline, deliver_pubsub_event(state, community_id, &stored)) + .await + .is_ok(); + if !delivered { + lose(state, closer, community_id, &stored, "stale"); + } +} + +/// One known event will not be delivered: resync the syncs and fence and +/// close the connections it would have reached. Matching skips the access +/// gate, so a recipient that would have been filtered out may be closed too; +/// that costs a reconnect, never a leak. A lost ephemeral event is only +/// counted, since no reconnect can replay it. +fn lose( + state: &AppState, + closer: &LagCloser, + community_id: CommunityId, + stored: &StoredEvent, + reason: &'static str, +) { + if is_ephemeral(event_kind_u32(&stored.event)) { + metrics::counter!("buzz_multinode_fanout_dropped_total", "reason" => "ephemeral") + .increment(1); + return; + } + metrics::counter!("buzz_multinode_fanout_dropped_total", "reason" => reason).increment(1); + let matches = state.sub_registry.fan_out_scoped(community_id, stored); + state.artifact_syncs.resync_matches(&matches, LAG_RESYNC); + closer.queue(state, matches.into_iter().map(|(conn_id, _)| conn_id)); +} + +/// Routes broadcast events to worker queues and handles every loss. +struct Dispatcher { + queues: Vec>, + closer: Arc, +} + +impl Dispatcher { + /// A dispatcher over `workers` queues of `capacity` events, and the + /// receiving ends for the workers. + fn new(workers: usize, capacity: usize) -> (Self, Vec>) { + let (queues, receivers) = (0..workers.max(1)) + .map(|_| mpsc::channel(capacity.max(1))) + .unzip(); + let dispatcher = Self { + queues, + closer: Arc::new(LagCloser::default()), + }; + (dispatcher, receivers) + } + + async fn run(self, state: Arc, mut rx: broadcast::Receiver) { + loop { + match rx.recv().await { + Ok(channel_event) => self.dispatch(&state, channel_event), + Err(broadcast::error::RecvError::Lagged(skipped)) => self.lagged(&state, skipped), + Err(broadcast::error::RecvError::Closed) => { + tracing::error!("Multi-node fan-out broadcast channel closed"); + return; + } + } + } + } + + /// Queue one event on its worker, or handle it as lost when that worker + /// is full. A local echo is no loss, so it is dropped before queueing. + fn dispatch(&self, state: &AppState, channel_event: ChannelEvent) { + if take_local_echo(state, &channel_event) { + return; + } + let queue = &self.queues[self.worker_for(&channel_event)]; + match queue.try_send(Queued::new(channel_event)) { + Ok(()) => {} + Err(TrySendError::Full(lost) | TrySendError::Closed(lost)) => { + lose( + state, + &self.closer, + lost.community_id, + &lost.stored, + "queue_full", + ); + } + } + } + + fn worker_for(&self, channel_event: &ChannelEvent) -> usize { + let mut hasher = DefaultHasher::new(); + (channel_event.community_id, channel_event.topic).hash(&mut hasher); + (hasher.finish() % self.queues.len() as u64) as usize + } + + /// The dispatcher skipped `skipped` events it never saw. Any subscription + /// may have missed one, so every sync resyncs and every subscribed + /// connection is fenced and closes. + fn lagged(&self, state: &AppState, skipped: u64) { + metrics::counter!("buzz_multinode_fanout_lag_total").increment(skipped); + warn!("Multi-node fan-out lagged by {skipped} messages"); + state.artifact_syncs.resync_all(LAG_RESYNC); + self.closer + .queue(state, state.sub_registry.subscribed_connections()); + } +} + +/// Fences connections that lost a live event at once, then closes them at most +/// [`CLOSE_BATCH`] per [`CLOSE_TICK`]. The pending set holds each connection +/// once and never more than this node's connections. +#[derive(Default)] +struct LagCloser { + pending: Mutex>, + wake: Notify, +} + +impl LagCloser { + /// Fence `conns` now and queue them for a paced close. Fencing happens + /// before this returns, so no event dispatched after the loss reaches them. + fn queue(&self, state: &AppState, conns: impl IntoIterator) { + let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner); + let before = pending.len(); + for conn_id in conns { + state.conn_manager.fence_live_delivery(conn_id); + pending.insert(conn_id); + } + if pending.len() > before { + self.wake.notify_one(); + } + } + + /// Close up to [`CLOSE_BATCH`] pending connections, and count them. + /// Returns how many it closed. + fn close_batch(&self, state: &AppState) -> usize { + let batch: Vec = { + let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner); + let batch: Vec = pending.iter().take(CLOSE_BATCH).copied().collect(); + for conn_id in &batch { + pending.remove(conn_id); + } + batch + }; + for conn_id in &batch { + state.conn_manager.cancel_conn(*conn_id); + } + metrics::counter!("buzz_multinode_fanout_lag_closed_total").increment(batch.len() as u64); + batch.len() + } + + async fn run(self: Arc, state: Arc) { + loop { + self.wake.notified().await; + let mut closed = 0usize; + loop { + let batch = self.close_batch(&state); + if batch == 0 { + break; + } + closed += batch; + tokio::time::sleep(CLOSE_TICK).await; + } + warn!( + closed, + "multi-node fan-out: closed connections that missed live events" + ); + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::atomic::AtomicU8; + + use axum::extract::ws::Message; + use buzz_core::kind::KIND_PRESENCE_UPDATE; + use buzz_core::tenant::CommunityId; + use buzz_pubsub::EventTopic; + use nostr::{EventBuilder, Filter, Keys, Kind}; + use tokio_util::sync::CancellationToken; + use uuid::Uuid; + + use super::*; + use crate::handlers::event::fan_out_event_to_local_subscribers; + + fn community() -> CommunityId { + CommunityId::from_uuid(Uuid::nil()) + } + + /// A stored kind: a lost event of it can be replayed, so its loss closes. + const STORED: u16 = 1; + /// An ephemeral kind: a lost event of it cannot be replayed. + const PRESENCE: u16 = KIND_PRESENCE_UPDATE as u16; + + /// A connection with one subscription to `kind`, its frame receiver, and + /// its cancellation token. + fn subscriber( + state: &AppState, + kind: u16, + ) -> (ConnId, mpsc::Receiver, CancellationToken) { + let conn_id = Uuid::new_v4(); + let (tx, rx) = mpsc::channel(64); + let cancel = CancellationToken::new(); + state.conn_manager.register( + conn_id, + tx, + mpsc::channel(10).0, + mpsc::channel(1).0, + None, + cancel.clone(), + community(), + Arc::new(AtomicU8::new(0)), + Arc::new(tokio::sync::Mutex::new(HashMap::new())), + 3, + crate::state::CommunityConnectionControl::new(CancellationToken::new()), + ); + state.sub_registry.register( + conn_id, + "sub".to_string(), + vec![Filter::new().kind(Kind::Custom(kind))], + None, + ); + (conn_id, rx, cancel) + } + + fn event(kind: u16, n: usize) -> ChannelEvent { + ChannelEvent { + community_id: community(), + topic: EventTopic::Global, + event: EventBuilder::new(Kind::Custom(kind), n.to_string()) + .sign_with_keys(&Keys::generate()) + .expect("sign event"), + received_at: Instant::now(), + } + } + + fn pending(dispatcher: &Dispatcher) -> HashSet { + dispatcher.closer.pending.lock().expect("pending").clone() + } + + /// Events reach their subscriber through the workers, in publish order. + #[tokio::test] + async fn delivers_one_topic_in_order_through_workers() { + let state = crate::state::tests::test_state().await; + let (_conn, mut frames, cancel) = subscriber(&state, STORED); + let (tx, rx) = broadcast::channel(64); + let handle = spawn(Arc::clone(&state), rx); + + let sent: Vec<_> = (0..20).map(|n| event(STORED, n)).collect(); + for event in &sent { + tx.send(event.clone()).expect("broadcast"); + } + for event in &sent { + let frame = tokio::time::timeout(Duration::from_secs(5), frames.recv()) + .await + .expect("delivered in time") + .expect("frame"); + let Message::Text(text) = frame else { + panic!("expected an EVENT frame"); + }; + let v: serde_json::Value = serde_json::from_str(&text).expect("frame JSON"); + assert_eq!(v[2]["id"], event.event.id.to_hex()); + } + assert!(!cancel.is_cancelled(), "no loss, no close"); + drop(tx); + handle + .await + .expect("dispatcher ends when the broadcast closes"); + } + + /// A full worker queue drops the event and closes exactly the connections + /// it would have reached. + #[tokio::test] + async fn full_worker_queue_closes_the_dropped_events_recipients() { + let state = crate::state::tests::test_state().await; + let (reached, _frames, _cancel) = subscriber(&state, STORED); + let other = Uuid::new_v4(); + state.sub_registry.register( + other, + "other-kind".to_string(), + vec![Filter::new().kind(Kind::Custom(7))], + None, + ); + // Hold the only queue's receiver without draining it. + let (dispatcher, _queues) = Dispatcher::new(1, 1); + + dispatcher.dispatch(&state, event(STORED, 0)); + assert!(pending(&dispatcher).is_empty(), "the first event fits"); + dispatcher.dispatch(&state, event(STORED, 1)); + assert_eq!(pending(&dispatcher), HashSet::from([reached])); + } + + /// A local echo arriving on a full queue was already delivered, so it is + /// not a loss. + #[tokio::test] + async fn local_echo_on_a_full_queue_closes_nothing() { + let state = crate::state::tests::test_state().await; + let _subscriber = subscriber(&state, STORED); + let (dispatcher, _queues) = Dispatcher::new(1, 1); + dispatcher.dispatch(&state, event(STORED, 0)); + + let echo = event(STORED, 1); + state.mark_local_event(echo.community_id, &echo.event.id); + dispatcher.dispatch(&state, echo); + assert!(pending(&dispatcher).is_empty()); + } + + /// An unknown loss closes every subscribed connection, and only those. + #[tokio::test] + async fn dispatcher_lag_closes_every_subscribed_connection() { + let state = crate::state::tests::test_state().await; + let (first, _f1, _c1) = subscriber(&state, STORED); + let (second, _f2, _c2) = subscriber(&state, STORED); + let (dispatcher, _queues) = Dispatcher::new(1, 1); + + dispatcher.lagged(&state, 7); + assert_eq!(pending(&dispatcher), HashSet::from([first, second])); + } + + /// The closer cancels queued connections, at most one batch per tick. + #[tokio::test(start_paused = true)] + async fn closer_cancels_queued_connections_in_paced_batches() { + let state = crate::state::tests::test_state().await; + let subscribers: Vec<_> = (0..CLOSE_BATCH + 1) + .map(|_| subscriber(&state, STORED)) + .collect(); + let closer = Arc::new(LagCloser::default()); + tokio::spawn(Arc::clone(&closer).run(Arc::clone(&state))); + + closer.queue(&state, subscribers.iter().map(|(conn_id, _, _)| *conn_id)); + let cancelled = || { + subscribers + .iter() + .filter(|(_, _, cancel)| cancel.is_cancelled()) + .count() + }; + tokio::time::sleep(CLOSE_TICK / 2).await; + assert_eq!(cancelled(), CLOSE_BATCH, "one batch before the first tick"); + tokio::time::sleep(CLOSE_TICK).await; + assert_eq!(cancelled(), CLOSE_BATCH + 1, "the rest after one tick"); + } + + /// Deliver `event` the way a worker does, and report whether the + /// subscriber got a frame. + async fn worker_delivers( + state: &Arc, + closer: &LagCloser, + frames: &mut mpsc::Receiver, + event: ChannelEvent, + ) -> bool { + deliver_or_lose(state, closer, Queued::new(event)).await; + frames.try_recv().is_ok() + } + + /// After a known loss, a recipient gets no newer live event while it + /// waits for its paced close, so its client cannot resume past the loss. + #[tokio::test] + async fn dropped_event_fences_its_recipients_before_their_close() { + let state = crate::state::tests::test_state().await; + let (_conn, mut frames, cancel) = subscriber(&state, STORED); + let (dispatcher, _queues) = Dispatcher::new(1, 1); + assert!( + worker_delivers(&state, &dispatcher.closer, &mut frames, event(STORED, 0)).await, + "delivers before any loss" + ); + dispatcher.dispatch(&state, event(STORED, 1)); + dispatcher.dispatch(&state, event(STORED, 2)); // dropped + + assert!( + !worker_delivers(&state, &dispatcher.closer, &mut frames, event(STORED, 3)).await, + "an event after the loss must not reach the fenced connection" + ); + assert!( + !cancel.is_cancelled(), + "the close itself waits for the closer" + ); + } + + /// The same fence holds after an unknown loss. + #[tokio::test] + async fn dispatcher_lag_fences_every_subscribed_connection() { + let state = crate::state::tests::test_state().await; + let (_conn, mut frames, cancel) = subscriber(&state, STORED); + let (dispatcher, _queues) = Dispatcher::new(1, 1); + dispatcher.lagged(&state, 3); + + assert!(!worker_delivers(&state, &dispatcher.closer, &mut frames, event(STORED, 0)).await); + assert!(!cancel.is_cancelled()); + } + + /// Gandalf's backlog case: R waits behind a backlog on its worker while + /// local fan-out and other workers keep moving its recipient's cursor. R + /// must reach the recipient by its deadline, or the recipient must be + /// fenced and closed so its client replays R. It is never delivered late. + #[tokio::test(start_paused = true)] + async fn backlogged_event_is_delivered_by_its_deadline_or_lost() { + let state = crate::state::tests::test_state().await; + let (conn, mut frames, cancel) = subscriber(&state, STORED); + let (dispatcher, mut queues) = Dispatcher::new(1, 4); + let queue = &mut queues[0]; + + dispatcher.dispatch(&state, event(STORED, 0)); + tokio::time::advance(MAX_QUEUE_DELAY - Duration::from_millis(1)).await; + let in_time = queue.try_recv().expect("queued"); + deliver_or_lose(&state, &dispatcher.closer, in_time).await; + assert!(frames.try_recv().is_ok(), "delivered inside its deadline"); + + dispatcher.dispatch(&state, event(STORED, 1)); // R + tokio::time::advance(MAX_QUEUE_DELAY).await; + let late = queue.try_recv().expect("queued"); + deliver_or_lose(&state, &dispatcher.closer, late).await; + assert!(frames.try_recv().is_err(), "R is not delivered late"); + assert_eq!(pending(&dispatcher), HashSet::from([conn])); + assert!( + !worker_delivers(&state, &dispatcher.closer, &mut frames, event(STORED, 2)).await, + "R's recipient is fenced until it closes" + ); + assert!(!cancel.is_cancelled(), "the close waits for the closer"); + } + + /// The deadline runs from Redis receipt: an event that already waited out + /// its deadline in the broadcast before dispatch is lost, not delivered. + #[tokio::test(start_paused = true)] + async fn event_that_waited_before_dispatch_is_lost_not_delivered() { + let state = crate::state::tests::test_state().await; + let (conn, mut frames, _cancel) = subscriber(&state, STORED); + let (dispatcher, mut queues) = Dispatcher::new(1, 4); + + let received = event(STORED, 0); + tokio::time::advance(MAX_QUEUE_DELAY).await; + dispatcher.dispatch(&state, received); + let queued = queues[0].try_recv().expect("queued"); + deliver_or_lose(&state, &dispatcher.closer, queued).await; + assert!(frames.try_recv().is_err(), "not delivered late"); + assert_eq!(pending(&dispatcher), HashSet::from([conn])); + } + + /// A delivery that stalls (here on a database that accepts connections + /// but never answers) is cut off at its deadline and handled as lost. + #[tokio::test(start_paused = true)] + async fn delivery_stalled_past_its_deadline_is_lost() { + let silent_db = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let url = format!( + "postgres://buzz@{}/buzz", + silent_db.local_addr().expect("addr") + ); + let state = crate::state::tests::test_state_with_database_url_and_pool_options( + &url, + sqlx::postgres::PgPoolOptions::new().acquire_timeout(Duration::from_secs(3600)), + ) + .await; + let channel_id = Uuid::new_v4(); + let (conn, mut frames, _cancel) = subscriber(&state, STORED); + state.sub_registry.register( + conn, + "channel".to_string(), + vec![Filter::new().kind(Kind::Custom(9))], + Some(channel_id), + ); + let event = ChannelEvent { + community_id: community(), + topic: EventTopic::Channel(channel_id), + event: EventBuilder::new(Kind::Custom(9), "stalls") + .sign_with_keys(&Keys::generate()) + .expect("sign"), + received_at: Instant::now(), + }; + let closer = LagCloser::default(); + + let started = Instant::now(); + deliver_or_lose(&state, &closer, Queued::new(event)).await; + assert_eq!( + started.elapsed(), + MAX_QUEUE_DELAY, + "cut off at the deadline" + ); + assert!(frames.try_recv().is_err()); + assert_eq!( + closer.pending.lock().expect("pending").clone(), + HashSet::from([conn]) + ); + } + + /// Each batch is counted when it closes, not when the backlog drains. + #[tokio::test] + async fn each_close_batch_is_counted_as_it_closes() { + let state = crate::state::tests::test_state().await; + let subscribers: Vec<_> = (0..CLOSE_BATCH + 1) + .map(|_| subscriber(&state, STORED)) + .collect(); + let closer = LagCloser::default(); + closer.queue(&state, subscribers.iter().map(|(conn_id, _, _)| *conn_id)); + + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let closed = metrics::with_local_recorder(&recorder, || closer.close_batch(&state)); + assert_eq!(closed, CLOSE_BATCH); + assert_eq!(closer.pending.lock().expect("pending").len(), 1); + let counted: u64 = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_multinode_fanout_lag_closed_total") + .map(|(.., value)| match value { + metrics_util::debugging::DebugValue::Counter(n) => n, + other => panic!("expected a counter, got {other:?}"), + }) + .sum(); + assert_eq!(counted, CLOSE_BATCH as u64); + } + + /// A lost ephemeral event cannot be replayed, so it is only counted: its + /// recipients are neither fenced nor closed. A stored event lost on the + /// same full queue still closes them. + #[tokio::test(start_paused = true)] + async fn lost_ephemeral_event_is_counted_and_closes_nothing() { + let state = crate::state::tests::test_state().await; + let (conn, mut frames, cancel) = subscriber(&state, PRESENCE); + state.sub_registry.register( + conn, + "stored".to_string(), + vec![Filter::new().kind(Kind::Custom(STORED))], + None, + ); + let (dispatcher, _queues) = Dispatcher::new(1, 1); + dispatcher.dispatch(&state, event(PRESENCE, 0)); + + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + dispatcher.dispatch(&state, event(PRESENCE, 1)); // queue full + }); + let reasons: Vec = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_multinode_fanout_dropped_total") + .flat_map(|(key, ..)| { + key.key() + .labels() + .map(|label| label.value().to_string()) + .collect::>() + }) + .collect(); + assert_eq!(reasons, ["ephemeral"], "counted as an ephemeral drop"); + assert!( + pending(&dispatcher).is_empty(), + "a full queue closes nothing" + ); + + let stale = event(PRESENCE, 2); + tokio::time::advance(MAX_QUEUE_DELAY).await; + deliver_or_lose(&state, &dispatcher.closer, Queued::new(stale)).await; + assert!(frames.try_recv().is_err(), "still never delivered late"); + assert!( + pending(&dispatcher).is_empty(), + "a stale one closes nothing" + ); + + assert!( + worker_delivers(&state, &dispatcher.closer, &mut frames, event(PRESENCE, 3)).await, + "nor fences the connection" + ); + dispatcher.dispatch(&state, event(STORED, 4)); // queue full + assert_eq!(pending(&dispatcher), HashSet::from([conn])); + assert!(!cancel.is_cancelled()); + } + + /// The fence holds on relay-local fan-out too, not only on the workers. + #[tokio::test] + async fn fence_withholds_local_fan_out() { + let state = crate::state::tests::test_state().await; + let (conn, mut frames, _cancel) = subscriber(&state, STORED); + let local = |n| pubsub_stored_event(event(STORED, n)); + + fan_out_event_to_local_subscribers(&state, community(), &local(0)).await; + assert!(frames.try_recv().is_ok(), "delivers before the fence"); + state.conn_manager.fence_live_delivery(conn); + fan_out_event_to_local_subscribers(&state, community(), &local(1)).await; + assert!( + frames.try_recv().is_err(), + "a fenced connection gets nothing" + ); + } +} diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 989c2957a2..b5e9347941 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; pub(crate) const MAX_SUBSCRIPTIONS: usize = 1024; @@ -118,6 +119,13 @@ pub async fn handle_req( } }; + // Every distinct channel a filter names, including group-state `#d`, is + // confirmed below. `#h` is bounded above; past the same limit, `#d` + // names are left out of confirmation and live routing rather than + // refused, as before `#d` routed live. + let (named_channels, live_names) = + super::system_channels::channels_named_by_filters(&filters, MAX_EXPLICIT_CHANNEL_VALUES); + let mut accessible_channels = if filters_are_nip43_membership_only(&filters) { metrics::counter!("buzz_req_global_access_resolution_skips_total", "kind" => "13534") .increment(1); @@ -179,13 +187,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(), @@ -418,15 +431,23 @@ 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, + live_names, + ), + ) + }) + .collect(); + let Some(owner) = claim_live_subscription(&sub_id, live_filters, &conn, &state).await else { return; }; @@ -1646,12 +1667,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. pub(crate) async fn claim_live_subscription( sub_id: &str, - filters: &[Filter], - channel_ids: Option<&[uuid::Uuid]>, + filters: Vec<(Filter, SubscriptionScope)>, conn: &ConnectionState, state: &AppState, ) -> Option { @@ -1664,38 +1686,24 @@ pub(crate) async fn claim_live_subscription( // A sync that held the ID stops; the registry entry and topics below are // replaced in place. state.artifact_syncs.remove(conn.conn_id, sub_id); - 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; @@ -2084,6 +2092,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 @@ -2113,8 +2135,7 @@ mod tests { state.artifact_syncs.insert_idle(conn.conn_id, "s"); claim_live_subscription( "s", - &text_filters(), - Some(&[uuid::Uuid::new_v4()]), + scoped(&text_filters(), Some(&[uuid::Uuid::new_v4()])), &conn, &state, ) @@ -2149,9 +2170,14 @@ mod tests { (Err("sync requires #h"), "invalid: sync requires #h"), (valid, "auth-required: not authenticated"), ] { - claim_live_subscription("s", &text_filters(), Some(&[channel]), &conn, &state) - .await - .expect("claim"); + claim_live_subscription( + "s", + scoped(&text_filters(), Some(&[channel])), + &conn, + &state, + ) + .await + .expect("claim"); let permit = Arc::new(tokio::sync::Semaphore::new(1)) .acquire_owned() .await @@ -2180,10 +2206,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; @@ -2215,14 +2245,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 @@ -2254,9 +2292,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 @@ -2267,9 +2310,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)); @@ -2420,17 +2468,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; } @@ -2481,7 +2539,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"); @@ -2494,8 +2552,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, )); @@ -2545,7 +2602,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"); @@ -2562,8 +2619,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, )); @@ -2600,14 +2656,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)); @@ -2660,7 +2726,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. @@ -2736,9 +2802,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 710fbdbe90..530f6db3bb 100644 --- a/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs @@ -1424,3 +1424,394 @@ async fn workflows_in_system_channels_follow_channel_access() { ); } } + +/// 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 +} + +/// 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([ + 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, 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 +/// 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'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!( + 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",