From efcdf00101f56e11998e072a62502f1bcadc313c Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 29 Oct 2026 18:48:55 -0400 Subject: [PATCH] relay: open the search pool lazily so a dead reader can't block boot (#8662) --- Justfile | 2 +- crates/buzz-relay/src/handlers/req.rs | 74 ++++++++++++++++- crates/buzz-relay/src/main.rs | 52 ++++++++---- crates/buzz-relay/src/startup_steps.rs | 4 - crates/buzz-relay/tests/boot_lifecycle.rs | 98 +++++++++++++++++------ scripts/run-tests.sh | 1 + 6 files changed, 187 insertions(+), 44 deletions(-) diff --git a/Justfile b/Justfile index e171a015c6..719b19dff7 100644 --- a/Justfile +++ b/Justfile @@ -546,7 +546,7 @@ test-unit: # never connects. # `^config::` is anchored so it does not also select nip_fi_config. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::admin::/) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::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_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^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::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + 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(/^api::admin::/) + 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_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^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::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-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index fc7ebf6a80..6676dd19ad 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -876,8 +876,19 @@ async fn handle_search_req( let search_result = match state.search.search(&search_query).await { Ok(r) => r, Err(e) => { + // CLOSED, not EOSE: an empty EOSE would read as "no + // matches". Stop the whole REQ so later filters do not + // each wait out the search pool's acquire timeout too. warn!(sub_id = %sub_id, "NIP-50 search failed: {e}"); - break; + super::close::close_if_owner( + sub_id, + owner, + Some(SEARCH_FAILED_CLOSED), + conn, + state, + ) + .await; + return; } }; @@ -1614,6 +1625,10 @@ fn history_read_closed_reason(e: &buzz_db::DbError) -> &'static str { } } +/// CLOSED reason for a NIP-50 search the search service could not answer +/// (e.g. the read replica it queries is down). +pub(crate) const SEARCH_FAILED_CLOSED: &str = "error: search failed"; + /// CLOSED reason for a failed one-shot DB read (COUNT): a statement cancel gets /// the stable timeout reason; anything else keeps the raw error. pub(crate) fn db_read_closed_reason(e: &buzz_db::DbError) -> String { @@ -2051,6 +2066,63 @@ mod tests { assert!(closed_frames(&mut send_rx).is_empty()); } + /// A NIP-50 search the search pool cannot serve (its reader is down) ends + /// the REQ with a CLOSED error, never an EOSE that reads as "no matches", + /// and the REQ's later search filters never wait on the pool. + #[tokio::test] + async fn failed_search_closes_without_eose_or_trying_later_filters() { + const ACQUIRE: std::time::Duration = std::time::Duration::from_millis(500); + let state = crate::state::tests::test_state_with_database_url_and_pool_options( + "postgres://127.0.0.1:1/none", + sqlx::postgres::PgPoolOptions::new().acquire_timeout(ACQUIRE), + ) + .await; + let (conn, mut send_rx) = lifecycle_conn(); + let owner = claim_search_subscription("s", &conn, &state) + .await + .expect("claim"); + let filters: Vec = (0..4) + .map(|i| { + Filter::new() + .kind(nostr::Kind::TextNote) + .search(format!("needle{i}")) + }) + .collect(); + + let started = std::time::Instant::now(); + handle_search_req( + "s", + owner, + &filters, + &[], + &super::super::system_channels::NamedOnlyChannels::default(), + true, + &conn.tenant, + &[0; 32], + &conn, + &state, + None, + ) + .await; + let elapsed = started.elapsed(); + + assert_eq!( + closed_frames(&mut send_rx), + vec![format!(r#"["CLOSED","s","{SEARCH_FAILED_CLOSED}"]"#)], + "exactly one CLOSED error and no EOSE" + ); + assert!( + conn.subscriptions.lock().await.is_empty(), + "the failed search must release its ID" + ); + // Each filter that reaches the pool waits out ACQUIRE on the dead port. + assert!( + elapsed < ACQUIRE * 3, + "later filters must not try the pool: took {elapsed:?} for {} filters", + filters.len() + ); + } + /// Concurrent same-ID claims and stale teardowns: whichever claim lands /// last owns the ID, and every stale teardown is a silent no-op. #[tokio::test] diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 874e415bf0..44d385a8d3 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -51,6 +51,22 @@ async fn connect_audit_pool(config: &DbConfig) -> anyhow::Result { .map_err(Into::into) } +/// The search pool. Search is lag-tolerant, so it prefers the read replica +/// when one is configured. Like the main read pool it is lazy, so a reader +/// that is down at boot fails searches, not startup, and it waits no longer +/// than the writer pool for a connection, so a dead reader fails each search +/// within that bound instead of sqlx's 30s default. +fn connect_search_pool(config: &DbConfig) -> anyhow::Result { + let url = config + .read_database_url + .as_deref() + .unwrap_or(&config.database_url); + sqlx::postgres::PgPoolOptions::new() + .acquire_timeout(std::time::Duration::from_secs(config.acquire_timeout_secs)) + .connect_lazy(url) + .map_err(|e| anyhow::anyhow!("Search DB URL is invalid: {e}")) +} + fn relay_keypair_from_config(relay_private_key: Option<&str>) -> anyhow::Result { let hex = relay_private_key.ok_or_else(|| { anyhow::anyhow!( @@ -834,18 +850,8 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { // Postgres FTS: the searchable row IS the persisted event row (its // `tsvector` column is populated by the `insert_event` write), so there is // no external collection to provision — the search service just queries the - // same Postgres over its own pool. Search is lag-tolerant, so it prefers - // the read replica when one is configured. - let search_db_url = config - .read_database_url - .as_deref() - .unwrap_or(&config.database_url); - let step = StepTimer::start(StartupStep::SearchConnect); - let search_pool = sqlx::postgres::PgPoolOptions::new() - .connect(search_db_url) - .await - .map_err(|e| anyhow::anyhow!("Search DB connection failed: {e}"))?; - step.finish(); + // same Postgres over its own pool (see `connect_search_pool`). + let search_pool = connect_search_pool(&db_config)?; let search_metrics_pool = search_pool.clone(); let search = SearchService::new(search_pool); info!( @@ -3222,9 +3228,9 @@ mod tests { use super::{ apply_activity_refresh, apply_stock_refresh, buzz_auto_migrate_enabled, connect_audit_pool, - counted_partition_audit, dropped_in_memory_keys, emit_cached_fleet_usage_metrics, - idle_timeout_secs, jwks_next_retry_after_failed_refresh, nip_fi_jwks_refresh_loop, - record_legacy_usage_collection, refresh_due_fleet_usage, + connect_search_pool, counted_partition_audit, dropped_in_memory_keys, + emit_cached_fleet_usage_metrics, idle_timeout_secs, jwks_next_retry_after_failed_refresh, + nip_fi_jwks_refresh_loop, record_legacy_usage_collection, refresh_due_fleet_usage, refresh_legacy_active_gauge_recency, relay_keypair_from_config, run_jwks_refresh_supervisor, run_periodic_until_cancelled, startup_partition_audit, EmissionScope, FleetUsageSchedule, InMemoryMetricKey, PartitionAuditSchedule, @@ -3237,6 +3243,22 @@ mod tests { registry::{GenerationalAtomicStorage, Registry}, }; + #[tokio::test] + async fn search_pool_is_lazy_prefers_the_reader_and_bounds_acquire() { + // Unroutable per RFC 5737 TEST-NET-1: construction must not dial it. + let config = DbConfig { + read_database_url: Some("postgres://u:p@192.0.2.1:5432/none".to_owned()), // sadscan:disable np.postgres.1 + ..DbConfig::default() + }; + let pool = connect_search_pool(&config).expect("lazy construction must not dial"); + assert_eq!(pool.connect_options().get_host(), "192.0.2.1"); + assert_eq!( + pool.options().get_acquire_timeout(), + Duration::from_secs(config.acquire_timeout_secs), + "a dead reader must fail a search within the writer's acquire bound" + ); + } + #[tokio::test(start_paused = true)] async fn periodic_loop_exits_immediately_on_cancellation() { let cancel = CancellationToken::new(); diff --git a/crates/buzz-relay/src/startup_steps.rs b/crates/buzz-relay/src/startup_steps.rs index 55253bbf4e..dc6a6a5c28 100644 --- a/crates/buzz-relay/src/startup_steps.rs +++ b/crates/buzz-relay/src/startup_steps.rs @@ -59,8 +59,6 @@ pub enum StartupStep { AuditConnect, /// Connect Redis commands and pub/sub. RedisConnect, - /// Connect the search database pool. - SearchConnect, /// Warm NIP-FI JWKS snapshots (only when NIP-FI is configured). NipFiJwksWarm, /// Load the first NIP-FI binding snapshot (only for the `database` source). @@ -94,7 +92,6 @@ impl StartupStep { Self::MembershipBootstrap, Self::AuditConnect, Self::RedisConnect, - Self::SearchConnect, Self::NipFiJwksWarm, Self::NipFiSnapshotLoad, Self::MeshBoot, @@ -118,7 +115,6 @@ impl StartupStep { Self::MembershipBootstrap => "membership_bootstrap", Self::AuditConnect => "audit_connect", Self::RedisConnect => "redis_connect", - Self::SearchConnect => "search_connect", Self::NipFiJwksWarm => "nip_fi_jwks_warm", Self::NipFiSnapshotLoad => "nip_fi_snapshot_load", Self::MeshBoot => "mesh_boot", diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 0a63a83be8..86984983f4 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -886,6 +886,80 @@ mod postgres_tests { ); } + /// A read replica that is down at boot must not block startup: the main + /// read pool is lazy and degrades to the writer, and the search pool must + /// be lazy too. An eager search-pool connect waits out sqlx's 30s acquire + /// timeout and then fails boot, so this relay would never turn ready + /// inside `CHILD_TIMEOUT`. + #[test] + #[ignore = "requires PostgreSQL"] + fn unreachable_read_replica_does_not_block_boot() { + let database_url = std::env::var("DATABASE_URL") + .expect("postgres lane provides DATABASE_URL for each test process"); + let redis_url = std::env::var("REDIS_URL") + .expect("the postgres lane runs alongside Redis and exports REDIS_URL"); + let read_database_url = format!( + "postgres://buzz:buzz@127.0.0.1:{}/buzz", // sadscan:disable np.postgres.1 + reserve_closed_port() + ); + let metrics_port_value = reserve_closed_port().to_string(); + let health_port = reserve_closed_port(); + let health_port_value = health_port.to_string(); + let bind_addr = format!("127.0.0.1:{}", reserve_closed_port()); + + let mut process = RelayProcess::spawn(&[ + ("BUZZ_RELAY_PRIVATE_KEY", VALID_RELAY_PRIVATE_KEY), + ("BUZZ_METRICS_PORT", &metrics_port_value), + ("BUZZ_HEALTH_PORT", &health_port_value), + ("BUZZ_BIND_ADDR", &bind_addr), + ("DATABASE_URL", &database_url), + ("READ_DATABASE_URL", &read_database_url), + ("REDIS_URL", &redis_url), + ("BUZZ_GIT_CONFORMANCE_PROBE", "false"), + ]); + let ready = wait_for_readiness(&mut process, health_port); + let output = process.terminate(); + let logs = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + ready.starts_with("HTTP/1.1 200"), + "a dead read replica must not block boot: {ready}: {logs}" + ); + assert!( + logs.contains("Search service ready"), + "boot must have passed the search pool: {logs}" + ); + } + + /// Polls `/_readiness` until it answers 200, the relay exits, or + /// `CHILD_TIMEOUT` passes, and returns the last response. + fn wait_for_readiness(process: &mut RelayProcess, health_port: u16) -> String { + let deadline = Instant::now() + CHILD_TIMEOUT; + loop { + if process.try_wait().is_some() { + return "relay exited before turning ready".to_string(); + } + let response = TcpStream::connect(("127.0.0.1", health_port)) + .and_then(|mut stream| { + stream.set_read_timeout(Some(Duration::from_secs(1)))?; + stream.write_all( + b"GET /_readiness HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + )?; + let mut response = String::new(); + stream.read_to_string(&mut response)?; + Ok(response) + }) + .unwrap_or_default(); + if response.starts_with("HTTP/1.1 200") || Instant::now() >= deadline { + return response; + } + thread::sleep(Duration::from_millis(50)); + } + } + const NIP_FI_ISSUERS: &str = r#"[{ "issuer": "https://idp.example.com", "command_audiences": ["https://relay.example.com"], @@ -1030,29 +1104,7 @@ mod postgres_tests { fn database_source_snapshot_load_lets_readiness_turn_ready() { seed_bindings(1); let (mut process, health_port) = spawn_database_source(&[]); - let deadline = Instant::now() + CHILD_TIMEOUT; - let ready = loop { - assert!( - process.try_wait().is_none(), - "relay exited: {}", - String::from_utf8_lossy(&process.terminate().stdout) - ); - let response = TcpStream::connect(("127.0.0.1", health_port)) - .and_then(|mut stream| { - stream.set_read_timeout(Some(Duration::from_secs(1)))?; - stream.write_all( - b"GET /_readiness HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", - )?; - let mut response = String::new(); - stream.read_to_string(&mut response)?; - Ok(response) - }) - .unwrap_or_default(); - if response.starts_with("HTTP/1.1 200") || Instant::now() >= deadline { - break response; - } - thread::sleep(Duration::from_millis(50)); - }; + let ready = wait_for_readiness(&mut process, health_port); let output = process.terminate(); let logs = format!( "{}{}", diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 99d9565273..59debc65cd 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -275,6 +275,7 @@ run_unit_tests() { handlers::req::tests::timed_out_historical_read_deregisters_before_closed \ handlers::req::tests::superseded_timeout_leaves_replacement_intact \ handlers::req::tests::search_claim_retires_live_and_yields_to_replacement \ + handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters \ handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner \ handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim \ handlers::req::tests::revoke_then_replacement_keeps_replacement_whole \