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 d8f3d7bc31..a80738052d 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/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/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/main.rs b/crates/buzz-relay/src/main.rs index dd0db58e51..ffda48bd59 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1287,37 +1287,12 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { &state, ))); - // Multi-node fan-out consumer: receive events from Redis pub/sub - // (published by other relay instances) and fan out to local WS subscribers. - { - let state_for_sub = Arc::clone(&state); - let mut rx = state_for_sub.pubsub.subscribe_local(); - tokio::spawn(async move { - loop { - match rx.recv().await { - Ok(channel_event) => { - buzz_relay::handlers::event::fan_out_pubsub_event( - &state_for_sub, - channel_event, - ) - .await; - } - Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { - metrics::counter!("buzz_multinode_fanout_lag_total").increment(n); - tracing::warn!("Multi-node fan-out lagged by {n} messages"); - // Skipped messages are not replayed (NIP-AR Sync rule 5). - state_for_sub - .artifact_syncs - .resync_all("resync: live delivery lagged"); - } - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - tracing::error!("Multi-node fan-out broadcast channel closed"); - break; - } - } - } - }); - } + // Multi-node fan-out: deliver events other relay instances published to + // Redis to local WS subscribers (see `handlers::multinode_fanout`). + buzz_relay::handlers::multinode_fanout::spawn( + Arc::clone(&state), + state.pubsub.subscribe_local(), + ); // Cross-pod cache-invalidation consumer: receive cache-key drops from Redis // pub/sub (published by other relay instances when membership/visibility diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index fa8309fe5e..d021774c8a 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -447,6 +447,10 @@ struct ConnEntry { grace_limit: u8, /// Lifecycle control used by `disconnect_nip_fi` for the denial transition. community_control: CommunityConnectionControl, + /// Set when this node lost a live event the connection should have + /// received (see [`ConnectionManager::fence_live_delivery`]). Written only + /// under the map's write guard, so it is a plain `bool`. + live_fenced: bool, } /// Community-scoped lifecycle registry shared by every long-lived socket type. @@ -874,6 +878,7 @@ impl ConnectionManager { admitted: AtomicBool::new(false), grace_limit, community_control, + live_fenced: false, }, ); // Insert-then-check pairs with drain_all's store-then-iterate: either @@ -1016,6 +1021,20 @@ impl ConnectionManager { .and_then(|entry| entry.authenticated_pubkey.read().ok()?.clone()) } + /// Stop live fan-out to a connection that lost a live event, until it + /// closes. A client resumes from the newest event it received; delivering + /// newer events after the loss would move that cursor past the lost one. + /// One-way: the connection must close for its client to recover. + /// + /// The fence takes the entry's write guard, and the live send path checks + /// it and queues its frame under one read guard. So a send already past + /// the check finishes before the fence lands, and no send starts after it. + pub(crate) fn fence_live_delivery(&self, conn_id: Uuid) { + if let Some(mut entry) = self.connections.get_mut(&conn_id) { + entry.live_fenced = true; + } + } + /// Cancel a single connection by ID. A no-op if the connection is not /// registered (already deregistered or never known). pub(crate) fn cancel_conn(&self, conn_id: Uuid) { @@ -1317,39 +1336,53 @@ impl ConnectionManager { /// Sends an already-serialized UTF-8 text payload to the given connection. /// /// The shared `Bytes` payload is cloned into the outbound WS message without - /// copying the frame body. Callers must only pass valid UTF-8 bytes. + /// copying the frame body. Callers must only pass valid UTF-8 bytes. This is + /// the live fan-out send path: a fenced connection (see + /// [`Self::fence_live_delivery`]) gets nothing. pub fn send_to_text_bytes(&self, conn_id: Uuid, msg: Arc) -> bool { + // Check the fence and queue the frame under one read guard, so a + // concurrent `fence_live_delivery` lands either before the check or + // after the frame is queued. + let Some(entry) = self.connections.get(&conn_id) else { + return false; + }; + // A fenced connection is closing so its client can replay from before + // a loss; the close accounts for this frame too. + if entry.live_fenced { + return true; + } let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref())) .expect("relay fan-out frames are serialized UTF-8 JSON"); - self.try_send_ws_message(conn_id, WsMessage::Text(text)) + Self::try_send_to_entry(conn_id, entry.value(), WsMessage::Text(text)) } fn try_send_ws_message(&self, conn_id: Uuid, msg: WsMessage) -> bool { - if let Some(entry) = self.connections.get(&conn_id) { - let conn = entry.value(); - match conn.tx.try_send(msg) { - Ok(_) => { - conn.backpressure_count.store(0, Ordering::Relaxed); - true - } - Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { - let count = conn.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1; - if count >= conn.grace_limit { - tracing::warn!(conn_id = %conn_id, count, "fan-out: sustained backpressure — cancelling slow client"); - metrics::counter!("buzz_ws_backpressure_disconnects_total").increment(1); - conn.cancel.cancel(); - } else { - tracing::warn!(conn_id = %conn_id, count, grace = conn.grace_limit, "fan-out: send buffer full — grace {count}/{}", conn.grace_limit); - } - false - } - Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { - tracing::debug!(conn_id = %conn_id, "fan-out: send channel closed"); - false - } + self.connections + .get(&conn_id) + .is_some_and(|entry| Self::try_send_to_entry(conn_id, entry.value(), msg)) + } + + fn try_send_to_entry(conn_id: Uuid, conn: &ConnEntry, msg: WsMessage) -> bool { + match conn.tx.try_send(msg) { + Ok(_) => { + conn.backpressure_count.store(0, Ordering::Relaxed); + true + } + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + let count = conn.backpressure_count.fetch_add(1, Ordering::Relaxed) + 1; + if count >= conn.grace_limit { + tracing::warn!(conn_id = %conn_id, count, "fan-out: sustained backpressure — cancelling slow client"); + metrics::counter!("buzz_ws_backpressure_disconnects_total").increment(1); + conn.cancel.cancel(); + } else { + tracing::warn!(conn_id = %conn_id, count, grace = conn.grace_limit, "fan-out: send buffer full — grace {count}/{}", conn.grace_limit); + } + false + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + tracing::debug!(conn_id = %conn_id, "fan-out: send channel closed"); + false } - } else { - false } } } diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 04ae0ce7fb..d3add0b736 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -533,6 +533,15 @@ impl SubscriptionRegistry { self.subs.len() } + /// Return every connection with at least one active subscription. + pub fn subscribed_connections(&self) -> Vec { + self.subs + .iter() + .filter(|entry| !entry.value().is_empty()) + .map(|entry| *entry.key()) + .collect() + } + /// Snapshot the number of active subscriptions per community. /// /// The usage poller sums it into `buzz_total_subscriptions`. diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index b37790861e..94a0d6c39c 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -263,6 +263,9 @@ run_unit_tests() { run_test_step "buzz-relay subscription registry tests" \ cargo test -p buzz-relay --lib subscription::tests:: -- --nocapture + run_test_step "buzz-relay multi-node fan-out tests" \ + cargo test -p buzz-relay --lib handlers::multinode_fanout::tests:: -- --nocapture + # telemetry and config tests mutate process env. libtest runs one process # with many threads (nextest, used by `just test-unit`, isolates each test), # so run them single-threaded here. `--skip` keeps the substring filter