From 27ab6a085a6689da5c0c81ca74ee5ccc90ed2003 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Fri, 6 Nov 2026 14:33:16 -0500 Subject: [PATCH] fix(relay): deliver multi-node fan-out concurrently and never drop silently (#9195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem Each relay node runs one consumer for events that other nodes publish to Redis. It delivered them one at a time and awaited the access gate (visibility and membership lookups) for each event. Under load it fell more than 4,096 events behind the pub/sub broadcast. Each `Lagged(n)` then skipped `n` events for good: nothing replayed them, and plain subscribers were never told. In a 15-node load test, about 50% of cross-node events were skipped at 10x load and about 75% at 13x. Throughput at 13x was also *lower* than at 10x. Loadgen didn't show this, because it only times events that arrive. ## Change - **Concurrent delivery.** A dispatcher reads the broadcast and only routes. It drops local echoes and queues each event on one of 16 workers by `(community, topic)`. Each worker delivers its queue in order, so a channel's events keep their order while different channels deliver concurrently. `handlers/multinode_fanout.rs` - **Known loss closes the affected connections.** An event is lost when its worker's bounded queue (1,024 events) is full, or when it isn't delivered within 2 s of this node reading it from Redis, whether it waited in the broadcast, behind a backlog, or in a stalled delivery. `ChannelEvent` now carries `received_at`, stamped by the pub/sub subscriber. For a lost event, the relay sends `CLOSED "resync: live delivery lagged"` to every sync the event would have reached (NIP-AR Sync rule 5). It also closes every connection the event would have reached, so those clients reconnect and catch up from their cursor. Matching runs before 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.** Presence, typing and observer frames are never stored, so no reconnect can replay them. Losing one fences and closes nothing; it is still never delivered past its deadline. Without this, one stale global presence event would close nearly every mobile connection in the community, since every mobile client subscribes to presence. - **Unknown loss closes every subscribed connection.** If the dispatcher itself lags the broadcast, it can't know which events it skipped. It resyncs every sync (unchanged) and closes every connection that has a subscription. - **Immediate fence.** Each affected connection is fenced as soon as the loss is detected. The live fan-out send path (`send_to_text_bytes`, local and cross-node) delivers it nothing from then on. The fence is written under the connection map's write guard, and the send path checks it and queues its frame under one read guard, so no frame slips in between the check and the send. - **Why 2 s.** The mobile client resumes from the newest `created_at` it received, minus 5 s of slack. The desktop app (block/buzz-app) resumes 960 s before its connection last heard the relay, which already covers the full 900 s `created_at` drift. While an event waits, local fan-out and other workers keep moving that cursor. The fence stops the cursor, and the deadline means no withheld event is more than 2 s older than the fence, so a reconnect replays it. Without the deadline, an event stuck behind a long backlog could be withheld past the slack and never replayed. - **Paced closes.** Only the close is delayed. One closer task closes at most 50 connections per 10 ms (5,000/s per node). A drop in a hot channel therefore can't become a reconnect storm (Review-Proven Rule 9). - `fan_out_pubsub_event` is now split into `take_local_echo` and `deliver_pubsub_event`, so the dispatcher can check echoes before queueing. Its behavior is unchanged. - **Limits (mobile):** replay works while the 2 s deadline, plus the writer's clock skew and its publish delay, stays under the client's 5 s slack. The relay accepts `created_at` up to 15 min off, so a writer clock more than about 3 s slow can still put an event behind the cursor. An ordinary reconnect has the same limit today; closing it needs a client rewind instruction, which is out of scope here. The desktop app's 960 s resume window has no such gap. A dispatcher that falls behind without overflowing the broadcast still never delivers a late event, but it detects the loss, and fences, late. Events skipped by dispatcher lag are never seen. Neither case is bounded by the deadline, which is a cutoff for attempting delivery, not a guaranteed fence time. - **What a load-test rerun should watch:** under sustained overload the deadline turns backlog into steady closes (up to 5,000/s per node), and reconnecting clients replay from the same database pool. Watch `buzz_multinode_fanout_dropped_total{reason="stale"}`, `buzz_multinode_fanout_lag_closed_total` and the reconnect REQ rate next to pool wait. On dispatcher lag, the dispatcher scans the subscription registry and fences each subscribed connection synchronously, so also watch how long it stays behind after a `Lagged`. Metrics: - `buzz_multinode_fanout_lag_total` (dispatcher lag) and `buzz_multinode_fanout_total` (delivered) keep their meaning. - `buzz_multinode_fanout_dropped_total{reason}` counts lost events: `queue_full`, or `stale` for one not delivered by its deadline. A lost ephemeral event counts as `ephemeral` instead and closes nothing. - `buzz_multinode_fanout_lag_closed_total` counts close attempts for a loss, one batch at a time. ## Rollout - Staging first. The load-test rerun at 1x, 10x and 13x happens after merge, with the staging load tests planned once the current relay changes land. At 1x, expect zero `dropped_total` for stored kinds and zero `lag_closed_total`. Report delivered %, `dropped_total{reason}`, `lag_closed_total`, reconnect REQ rate and pool wait. - Then one production cluster before the rest. - **Rollback trigger:** `lag_closed_total` climbing steadily, or reconnect REQs pushing pool wait up, at normal load. Rollback is a plain revert and redeploy: there are no schema, config or wire changes. ## Ordering Order is kept within a `(community, topic)`. Events on different topics, for example a channel event and a global event, can now reach one subscriber in either order. Before this change they arrived in Redis arrival order. Nothing in the relay depends on order across topics: the access gate re-reads membership and visibility per event. Clients order by `created_at`. ## What got simpler / known debt - The fan-out loop in `main.rs` shrinks to one `spawn` call, and the lag handling lives next to the delivery code. - No new config knobs. The worker count and queue size are constants until a load test shows they need tuning. - **Known debt, not in this PR:** a Redis pub/sub reconnect can still lose events silently for plain subscribers. Syncs are already resynced in that case (`artifact_sync::watch_link`), but ordinary subscriptions are not. ## Tests `handlers::multinode_fanout::tests`, each checked with a mutant on Blox: - events reach their subscriber through the workers, in publish order - a full worker queue closes exactly the dropped event's recipients - a lost ephemeral event (full queue or stale) is counted as `ephemeral` and neither fences nor closes, while a stored event lost on the same queue still closes - a local echo on a full queue closes nothing - dispatcher lag closes every subscribed connection - after a known loss or dispatcher lag, a recipient gets no newer live event while it waits to be closed - the fence also withholds relay-local fan-out - an event that waited out its deadline before dispatch is lost, never delivered - an event behind a backlog is delivered inside its deadline, and past it is lost (its recipient fenced and queued for close), never delivered late (paused clock) - a delivery that stalls on a database that never answers is cut off at the deadline and handled as lost - the closer cancels in paced batches (paused clock), and counts each batch as it closes 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- ARCHITECTURE.md | 2 +- Justfile | 6 +- crates/buzz-pubsub/src/lib.rs | 3 + crates/buzz-pubsub/src/subscriber.rs | 1 + crates/buzz-relay/src/handlers/event.rs | 70 +- crates/buzz-relay/src/handlers/mod.rs | 2 + .../src/handlers/multinode_fanout.rs | 717 ++++++++++++++++++ crates/buzz-relay/src/main.rs | 37 +- crates/buzz-relay/src/state.rs | 85 ++- crates/buzz-relay/src/subscription.rs | 9 + scripts/run-tests.sh | 3 + 11 files changed, 855 insertions(+), 80 deletions(-) create mode 100644 crates/buzz-relay/src/handlers/multinode_fanout.rs 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