diff --git a/Justfile b/Justfile index 5f879caf55..86c2f4d05b 100644 --- a/Justfile +++ b/Justfile @@ -535,7 +535,7 @@ test-unit: # exact name for the same reason: the rest of handlers::req needs a DB. # The third clause adds the NIP-FI HTTP ingress and its router/config # neighbours: nip_fi_http, nip_fi_config, nip_fi_resolver (snapshot - # resolver and its WS/HTTP/audio router tests), router, api::parse_query_tests, + # resolver and its WS/HTTP/audio router tests), router, api::parse_query_tests, api::db_read_error_tests, # the profile search request bounds (api::profiles), the per-user # search cap (api::bridge::search_slots), the Git transport # off_mode_precedence_tests, and the git object-store @@ -557,7 +557,7 @@ test-unit: # subscription::tests covers the in-memory registry and its fan-out # indexes; it is infra-free. 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::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::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(/^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::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/PLANS/REPLICA_FULL_READ_ROUTING_DESIGN.md b/PLANS/REPLICA_FULL_READ_ROUTING_DESIGN.md index c64d7e048b..34bd874fb9 100644 --- a/PLANS/REPLICA_FULL_READ_ROUTING_DESIGN.md +++ b/PLANS/REPLICA_FULL_READ_ROUTING_DESIGN.md @@ -14,8 +14,12 @@ routing**: the read may be served from a read replica when `RoutePredicate::for_query` derives from the query shape. The seam fails closed to the writer on replica errors, with one deliberate exception: a statement cancelled by `statement_timeout` (SQLSTATE `57014`) is surfaced rather than -re-run, because the writer would run the same slow statement. It is a genuine -no-op until the budget is configured (`crates/buzz-db/src/lib.rs`). +re-run, because the writer would run the same slow statement. A busy reader +is the other exception: a subscription-history read that cannot get a reader +connection within its bounded wait is rejected as retryable instead of moving +to the writer, and history reads are capped per pod below the reader pool +size (`Db::route_read`). It is a genuine no-op until the budget is configured +(`crates/buzz-db/src/lib.rs`). ## Routing rule diff --git a/crates/buzz-db/src/runtime/history_reads.rs b/crates/buzz-db/src/runtime/history_reads.rs new file mode 100644 index 0000000000..ec2a59c91d --- /dev/null +++ b/crates/buzz-db/src/runtime/history_reads.rs @@ -0,0 +1,131 @@ +//! Per-pod admission for replica-served subscription-history reads. +//! +//! History pages are the heaviest routed reads. Unbounded, a burst of them +//! fills the reader pool, every further routed read misses its reader acquire +//! budget, and before this gate those misses fell back to the writer, where +//! they starved the authentication, tenant and moderation lookups that only +//! the writer can serve. The gate holds running history transactions below the +//! reader pool's size, and a history read that cannot get a reader is rejected +//! as retryable instead of moved to the writer (see [`super::Db::route_read`]). +//! +//! The cap bounds history transactions, not reader connections. A permit is +//! released when its transaction is dropped, while sqlx still returns the +//! connection asynchronously through the pool's release check. Released +//! connections can therefore briefly hold the slots the cap leaves free for +//! authorization reads: usually one round trip, at most the release check's +//! timeout when reader sockets stall. Authorization reads that miss a reader +//! in that window fall back to the writer as before; history reads on the +//! saturated pool are rejected, not moved to the writer. + +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; + +/// Concurrency cap on history reads holding a reader connection, one per +/// `Db` (shared by its clones, so one per relay process). +#[derive(Clone, Debug)] +pub(crate) struct HistoryReadGate { + permits: Arc, +} + +impl HistoryReadGate { + /// How long a history read waits for a permit before it is rejected. The + /// semaphore is FIFO, so this bounds queueing delay under a burst; a read + /// that would wait longer is shed and the client retries. + pub(crate) const PERMIT_WAIT: Duration = Duration::from_secs(1); + + /// A gate for a reader pool of `read_max_connections`. + pub(crate) fn for_read_pool(read_max_connections: u32) -> Self { + Self { + permits: Arc::new(Semaphore::new(Self::cap(read_max_connections) as usize)), + } + } + + /// History transactions allowed at once: the pool less one fifth (at + /// least one) left to authorization and bootstrap reads, and never below + /// one. 25 → 20, 5 → 4, 2 → 1, 1 → 1. + pub(crate) fn cap(read_max_connections: u32) -> u32 { + read_max_connections + .saturating_sub((read_max_connections / 5).max(1)) + .max(1) + } + + /// Wait up to [`Self::PERMIT_WAIT`] for a permit. `None` means the read + /// must be rejected. + pub(crate) async fn admit(&self) -> Option { + tokio::time::timeout(Self::PERMIT_WAIT, self.permits.clone().acquire_owned()) + .await + .ok()? + .ok() + } + + /// Permits not currently held (tests). + #[cfg(test)] + pub(crate) fn available(&self) -> usize { + self.permits.available_permits() + } +} + +/// A proved reader transaction plus the history permit it was admitted +/// under, if any. The permit is released when the transaction is dropped, +/// wherever the transaction travels: a per-call routed read, or a +/// [`super::ReadSession`] the relay keeps for follow-up queries. The reader +/// connection itself returns to the pool asynchronously after that (see the +/// module docs). +pub(crate) struct ReaderTx { + tx: sqlx::Transaction<'static, sqlx::Postgres>, + _permit: Option, +} + +impl ReaderTx { + pub(crate) fn new( + tx: sqlx::Transaction<'static, sqlx::Postgres>, + permit: Option, + ) -> Self { + Self { + tx, + _permit: permit, + } + } +} + +impl std::ops::Deref for ReaderTx { + type Target = sqlx::Transaction<'static, sqlx::Postgres>; + + fn deref(&self) -> &Self::Target { + &self.tx + } +} + +impl std::ops::DerefMut for ReaderTx { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.tx + } +} + +#[cfg(test)] +mod tests { + use super::HistoryReadGate; + + #[test] + fn cap_leaves_a_fifth_of_the_reader_pool_and_never_reaches_zero() { + for (pool, cap) in [(25, 20), (10, 8), (5, 4), (4, 3), (2, 1), (1, 1), (0, 1)] { + assert_eq!(HistoryReadGate::cap(pool), cap, "pool={pool}"); + } + } + + #[tokio::test(start_paused = true)] + async fn admit_rejects_after_the_bounded_wait_and_reopens_on_release() { + let gate = HistoryReadGate::for_read_pool(1); + let held = gate.admit().await.expect("first read is admitted"); + let started = tokio::time::Instant::now(); + assert!(gate.admit().await.is_none(), "a full gate rejects"); + assert_eq!(started.elapsed(), HistoryReadGate::PERMIT_WAIT); + drop(held); + assert!( + gate.admit().await.is_some(), + "a released permit admits again" + ); + } +} diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 7c085dcc0f..2c0d045263 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -1,6 +1,7 @@ mod admitted_tx; mod cold_start; mod connection_observability; +mod history_reads; pub mod migration; pub(crate) mod observability; pub(crate) mod push_enqueue; @@ -302,6 +303,10 @@ pub struct Db { /// not yet probed (or the probe hit a transient error and will retry). /// Shared across `Db` clones. pub(crate) reader_aurora_identity: std::sync::Arc>, + /// Concurrency cap on replica-served subscription-history reads, sized + /// from [`Db::read_max_connections`] (see [`history_reads`]). Shared + /// across `Db` clones. + pub(crate) history_reads: history_reads::HistoryReadGate, } /// The session that served (or will serve) a routed read, so follow-up @@ -329,7 +334,7 @@ pub(crate) enum ReadSessionInner { /// to the writer instead of surfacing an error: degraded capacity, /// never holes — and never a 500 the writer could have served. Replica { - tx: sqlx::Transaction<'static, sqlx::Postgres>, + tx: history_reads::ReaderTx, writer: PgPool, }, /// The writer pool (cheap clone; Arc-backed). @@ -392,8 +397,11 @@ pub(crate) enum RouteDecision { /// replica, so a post-verification writer re-run or a mid-query replica /// failure emits exactly one `buzz_db_route_decision` event per request /// (the offload percentage is read straight off `decision="replica"`). + /// + /// A history read's transaction carries its [`history_reads`] permit, so + /// the permit is released when the transaction is dropped. Replica( - sqlx::Transaction<'static, sqlx::Postgres>, + history_reads::ReaderTx, replica_fence::TokenEntry, &'static str, ), @@ -819,6 +827,7 @@ impl Db { fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), replica_read_max_age, reader_aurora_identity: std::sync::Arc::new(std::sync::OnceLock::new()), + history_reads: history_reads::HistoryReadGate::for_read_pool(read_max_connections), }) } @@ -964,12 +973,14 @@ impl Db { } /// Reader acquire timeout — deliberately far below the writer's - /// (seconds-denominated) timeout. Failing closed to the writer must be - /// fast: a saturated reader pool that made routed reads wait the full - /// writer-style timeout would add dead latency during exactly the load - /// spike the offload exists for. A miss here surfaces as - /// `writer/reader_acquire_timeout` (see [`Db::proved_reader`] for why - /// the reason names the mechanism rather than a diagnosis). + /// (seconds-denominated) timeout. A miss must resolve fast: a saturated + /// reader pool that made routed reads wait the full writer-style timeout + /// would add dead latency during exactly the load spike the offload + /// exists for. A miss surfaces as `writer/reader_acquire_timeout`, or as + /// `rejected/reader_acquire_timeout` for a history read, which is not + /// moved to the writer (see [`Db::route_read`], and + /// [`Db::proved_reader`] for why the reason names the mechanism rather + /// than a diagnosis). const READER_ACQUIRE_TIMEOUT: Duration = Duration::from_millis(150); /// Client-side bound on a routed read's fence proof (Aurora capability @@ -1369,6 +1380,9 @@ impl Db { push_enqueue: None, max_connections: pool.options().get_max_connections(), read_max_connections: pool.options().get_max_connections(), + history_reads: history_reads::HistoryReadGate::for_read_pool( + pool.options().get_max_connections(), + ), pool, read_pool: None, fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), @@ -1390,6 +1404,9 @@ impl Db { max_connections: pool.options().get_max_connections(), read_max_connections: read_pool.options().get_max_connections(), pool, + history_reads: history_reads::HistoryReadGate::for_read_pool( + read_pool.options().get_max_connections(), + ), read_pool: Some(read_pool), fence: std::sync::Arc::new(replica_fence::ReplicaFence::new()), replica_read_max_age: None, @@ -1839,22 +1856,38 @@ impl Db { /// Shared route decision for one read: evaluate the predicate against a /// proved reader session and record the decision. Fail closed to the - /// writer everywhere. + /// writer, with one exception for subscription-history reads. + /// + /// History reads are the heavy, high-volume routed reads. They hold a + /// [`history_reads`] permit for as long as their reader transaction, so + /// they never fill the reader pool, and they are never moved to the + /// writer merely because the reader is busy: a history read that waits + /// out its permit ([`history_reads::HistoryReadGate::PERMIT_WAIT`]) or + /// its reader acquire ([`Db::READER_ACQUIRE_TIMEOUT`]) on a saturated + /// reader pool ([`Db::reader_saturated`]) is rejected with + /// [`sqlx::Error::PoolTimedOut`], which the relay answers with its + /// retryable `CLOSED`. Spilling that load onto the writer starved the + /// authentication, tenant and moderation reads only the writer serves. + /// A reader that is wrong rather than busy (unreachable, validation + /// failure, proof timeout, stale token) still falls back to the writer, + /// as do prechecks that never reach the reader. Recorded as + /// `decision="rejected"` with `reason="history_read_cap"` or + /// `"reader_acquire_timeout"`. pub(crate) async fn route_read( &self, path: &'static str, predicate: RoutePredicate, operation: observability::ReaderOperation, - ) -> RouteDecision { + ) -> Result { let Some(read_pool) = &self.read_pool else { Self::record_route(path, "writer", "disabled"); - return RouteDecision::Writer; + return Ok(RouteDecision::Writer); }; // Cheap prechecks on the shared ring before spending a reader // checkout; the connection-local observation still has to prove it. let Some(newest) = self.fence.newest() else { Self::record_route(path, "writer", "uninitialized"); - return RouteDecision::Writer; + return Ok(RouteDecision::Writer); }; // Precheck helpers against the newest shared entry: if the newest // cannot satisfy an arm, no proved (older-or-equal) entry can. @@ -1885,8 +1918,17 @@ impl Db { }; if let Err(reason) = precheck { Self::record_route(path, "writer", reason); - return RouteDecision::Writer; + return Ok(RouteDecision::Writer); } + let history = operation == observability::ReaderOperation::SubscriptionHistory; + let permit = if history { + let Some(permit) = self.history_reads.admit().await else { + return Err(Self::reject_history_read(path, "history_read_cap")); + }; + Some(permit) + } else { + None + }; match self.proved_reader(read_pool, operation).await { Ok((tx, entry)) => { // Re-evaluate against the entry the session actually proved @@ -1912,21 +1954,50 @@ impl Db { } }; match verdict { - Some(reason) => RouteDecision::Replica(tx, entry, reason), + Some(reason) => Ok(RouteDecision::Replica( + history_reads::ReaderTx::new(tx, permit), + entry, + reason, + )), None => { // The session proves an older entry than the // predicate needs (replication lag) — fail closed. Self::record_route(path, "writer", "stale"); - RouteDecision::Writer + Ok(RouteDecision::Writer) } } } + Err(reason @ "reader_acquire_timeout") + if history && Self::reader_saturated(read_pool) => + { + Err(Self::reject_history_read(path, reason)) + } Err(reason) => { Self::record_route(path, "writer", reason); - RouteDecision::Writer + Ok(RouteDecision::Writer) } } } + + /// Whether a reader acquire timeout means the reader is busy rather than + /// unreachable: every connection the pool may hold exists and none is + /// idle. `PoolTimedOut` alone cannot tell them apart, because sqlx keeps + /// retrying a refused or failing dial until the acquire budget runs out. + /// A reader that refuses or drops dials leaves the pool below its ceiling, + /// so its history reads still fall back to the writer. A reader that + /// accepts no TCP at all can fill the ceiling with in-flight dials and + /// read as busy; those reads are rejected as retryable, the same outcome + /// as a full gate. + fn reader_saturated(read_pool: &PgPool) -> bool { + read_pool.size() >= read_pool.options().get_max_connections() && read_pool.num_idle() == 0 + } + + /// Record a rejected history read (see [`Db::route_read`]) and build its + /// retryable error. + fn reject_history_read(path: &'static str, reason: &'static str) -> DbError { + Self::record_route(path, "rejected", reason); + DbError::Sqlx(sqlx::Error::PoolTimedOut) + } } #[cfg(test)] diff --git a/crates/buzz-db/src/runtime/observability.rs b/crates/buzz-db/src/runtime/observability.rs index 7e1ae4c1cf..e115b4d812 100644 --- a/crates/buzz-db/src/runtime/observability.rs +++ b/crates/buzz-db/src/runtime/observability.rs @@ -292,17 +292,21 @@ fn record_pool_acquire( ) { // Preserve the original observed population for existing dashboards. // Newly instrumented raw-pool seams must not create a deployment-time - // discontinuity in these compatibility families. + // discontinuity in these compatibility families. They carry `operation` + // so a breakdown by it attributes every sample instead of reporting it + // as unlabelled; summing over `operation` gives the original series. if emit_legacy && outcome != Outcome::Cancelled { metrics::histogram!( "buzz_db_pool_acquire_wait_seconds", "pool_role" => pair.pool_role(), + "operation" => pair.operation(), "outcome" => outcome.as_str(), ) .record(elapsed.as_secs_f64()); metrics::counter!( "buzz_db_pool_acquisitions_total", "pool_role" => pair.pool_role(), + "operation" => pair.operation(), "outcome" => outcome.as_str(), ) .increment(1); @@ -764,15 +768,24 @@ mod tests { }) .collect::>(); + for name in [ + "buzz_db_pool_acquire_wait_seconds", + "buzz_db_pool_acquisitions_total", + ] { + let labels = [ + ("operation", "subscription_history"), + ("outcome", "timeout"), + ("pool_role", "reader"), + ] + .into_iter() + .map(|(key, value)| (key.to_owned(), value.to_owned())) + .collect::>(); + assert!( + keys.contains(&(name.to_owned(), labels)), + "legacy family {name} must carry the operation label; got {keys:?}" + ); + } for expected in [ - ( - "buzz_db_pool_acquire_wait_seconds", - [("outcome", "timeout"), ("pool_role", "reader")], - ), - ( - "buzz_db_pool_acquisitions_total", - [("outcome", "timeout"), ("pool_role", "reader")], - ), ( "buzz_db_advisory_lock_wait_seconds", [("lock_type", "replacement"), ("outcome", "success")], @@ -819,15 +832,8 @@ mod tests { "buzz_db_pool_acquisitions_total", ] { assert!( - !keys.contains(&( - name.to_owned(), - [ - ("outcome".to_owned(), "success".to_owned()), - ("pool_role".to_owned(), "writer".to_owned()), - ] - .into_iter() - .collect(), - )), + !keys.iter().any(|(key, labels)| key == name + && labels.get("pool_role").map(String::as_str) == Some("writer")), "newly instrumented seams must not expand legacy metric population" ); } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index b7c9d5fe2a..7ef4cc6f95 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -170,6 +170,7 @@ async fn event_write_transaction_preserves_legacy_acquisition_metrics() { ( "buzz_db_pool_acquisitions_total".to_owned(), [ + ("operation".to_owned(), "event_write".to_owned()), ("outcome".to_owned(), "error".to_owned()), ("pool_role".to_owned(), "writer".to_owned()), ] @@ -1581,6 +1582,117 @@ async fn read_session_degrades_to_writer_when_replica_connection_dies() { drop(session); } +/// The history-read permit lives exactly as long as the reader transaction +/// it admitted, including a [`ReadSession`] held for follow-up queries; a +/// history read that cannot get a permit within the bounded wait is +/// rejected as retryable and never moves to the writer. +#[tokio::test(flavor = "current_thread")] +#[ignore = "requires Postgres"] +async fn history_permit_follows_the_read_session_and_a_full_gate_rejects() { + let (writer, _writer_db) = ScratchDb::migrated(&admin_url().await, "cap_w").await; + let (replica, _replica_db) = ScratchDb::migrated(&admin_url().await, "cap_r").await; + + let author = nostr::Keys::generate(); + let community = Uuid::new_v4(); + let channel = Uuid::new_v4(); + seed_community_channel(&writer, community, channel, &author).await; + seed_community_channel(&replica, community, channel, &author).await; + let base = 1_700_000_000u64; + for pool in [&writer, &replica] { + for (content, at) in [("m1", base), ("m2", base + 10)] { + insert_top_level( + pool, + community, + channel, + &signed_event_at(&author, content, at), + ) + .await; + } + } + + let db = Db::from_pools(writer.clone(), replica.clone()); + db.fence().force_open_for_tests(chrono::Utc::now()); + let cid = CommunityId::from_uuid(community); + let cap = db.history_reads.available(); + assert_eq!( + cap, + super::history_reads::HistoryReadGate::cap(db.read_max_connections) as usize + ); + + let head = db + .get_channel_window(cid, channel, 1, None, None) + .await + .expect("head window"); + let cursor = head.next_cursor.expect("has_more implies next_cursor"); + let (_window, session) = db + .get_channel_window_with_session(cid, channel, 10, Some(cursor.clone()), None) + .await + .expect("routed cursor window"); + assert!(session.is_replica(), "fixture must route to the replica"); + assert_eq!( + db.history_reads.available(), + cap - 1, + "the held session must keep its permit" + ); + + // Fill the gate: every remaining permit is held elsewhere. + let mut others = Vec::new(); + for _ in 1..cap { + others.push(db.history_reads.admit().await.expect("free permit")); + } + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let started = Instant::now(); + let rejected = { + let _guard = metrics::set_default_local_recorder(&recorder); + db.get_channel_window_with_session(cid, channel, 10, Some(cursor.clone()), None) + .await + } + .err() + .expect("a full gate must reject the history read"); + assert!( + matches!(rejected, DbError::Sqlx(sqlx::Error::PoolTimedOut)), + "the rejection must be the retryable pool timeout; got {rejected:?}" + ); + assert!( + started.elapsed() >= super::history_reads::HistoryReadGate::PERMIT_WAIT, + "the read must wait out the bounded permit wait before rejecting" + ); + let decisions: Vec<(String, String)> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, ..)| key.key().name() == "buzz_db_route_decision") + .map(|(key, ..)| { + let label = |name: &str| { + key.key() + .labels() + .find(|l| l.key() == name) + .map(|l| l.value().to_owned()) + .unwrap_or_default() + }; + (label("decision"), label("reason")) + }) + .collect(); + assert_eq!( + decisions, + vec![("rejected".to_owned(), "history_read_cap".to_owned())], + "a capped history read is rejected once and never served by the writer" + ); + + // Dropping the session releases its permit, and the next read routes. + drop(session); + assert_eq!(db.history_reads.available(), 1); + let (_window, session) = db + .get_channel_window_with_session(cid, channel, 10, Some(cursor), None) + .await + .expect("a released permit admits the next read"); + assert!(session.is_replica()); + drop(session); + drop(others); + assert_eq!(db.history_reads.available(), cap); +} + /// Snapshot continuity (Wren, review of 17ea2ff6a): the routed request /// runs inside ONE `REPEATABLE READ, READ ONLY` transaction whose first /// statement was the heartbeat observation — so a row committed on the @@ -1994,8 +2106,10 @@ async fn is_relay_member_is_bounded_routed_and_fails_closed() { /// the replica (not the writer) served the read. Every assertion /// requests A and demands B's rows never appear, including B's /// `replica-only` row, which is the one a leaky predicate would surface. -/// The routed fallback must cost ONE reader acquire budget, even when the -/// Aurora capability cache is cold. +/// A saturated reader must cost a routed read ONE reader acquire budget, +/// even when the Aurora capability cache is cold. A history read is then +/// rejected as retryable instead of moving to the writer; an authorization +/// read still falls back to the writer. /// /// Regression test for a stacked-budget bug found at `9fa3c9c0b`: the /// capability probe used to `acquire()` from the pool itself and return @@ -2020,7 +2134,7 @@ async fn is_relay_member_is_bounded_routed_and_fails_closed() { /// already covers. #[tokio::test(flavor = "current_thread")] #[ignore = "requires Postgres"] -async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() { +async fn saturated_reader_spends_one_acquire_budget_and_rejects_history_reads() { let (seed, scratch) = ScratchDb::migrated(&admin_url().await, "one_budget").await; seed.close().await; let scratch_url = scratch.url().to_owned(); @@ -2085,20 +2199,41 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() { // local recorder is installed and the label assertions would vacuously // see an empty snapshot. let start = std::time::Instant::now(); - let count = { + let rejected = { let _guard = metrics::set_default_local_recorder(&recorder); db.count_events_routed("one_budget_probe", &query).await } - .expect("writer fallback still answers the read"); + .expect_err("a history read must not spill onto the writer"); let elapsed = start.elapsed(); - assert_eq!(count, 0, "writer answered on an empty scratch database"); + assert!( + matches!(rejected, DbError::Sqlx(sqlx::Error::PoolTimedOut)), + "the rejection must be the retryable pool timeout; got {rejected:?}" + ); assert!( elapsed < Duration::from_millis(250), - "routed fallback must spend ONE {}ms acquire budget, not two; took {}ms", + "routed read must spend ONE {}ms acquire budget, not two; took {}ms", Db::READER_ACQUIRE_TIMEOUT.as_millis(), elapsed.as_millis() ); + assert_eq!( + db.history_reads.available(), + 1, + "a rejected history read must release its permit" + ); + + // An authorization read on the same saturated reader keeps the writer + // fallback: only history reads are rejected. + let is_member = { + let _guard = metrics::set_default_local_recorder(&recorder); + db.is_relay_member( + CommunityId::from_uuid(Uuid::new_v4()), + &nostr::Keys::generate().public_key().to_hex(), + ) + .await + } + .expect("authorization read falls back to the writer"); + assert!(!is_member, "writer answered on an empty scratch database"); let reasons: std::collections::HashMap<(String, String), u64> = snapshotter .snapshot() @@ -2121,10 +2256,15 @@ async fn routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold() { }) .collect(); + assert_eq!( + reasons.get(&("rejected".to_owned(), "reader_acquire_timeout".to_owned())), + Some(&1), + "the history read must be rejected as reader_acquire_timeout; got {reasons:?}" + ); assert_eq!( reasons.get(&("writer".to_owned(), "reader_acquire_timeout".to_owned())), Some(&1), - "saturated reader must fall back as writer/reader_acquire_timeout; got {reasons:?}" + "only the authorization read may fall back to the writer; got {reasons:?}" ); // `reader_validation_error` would mean we misclassified a timeout as a // broken reader, and `pool_busy` is the retired name — neither may @@ -3832,7 +3972,7 @@ async fn read_session_statement_cancel_propagates_without_writer_rerun() { .expect("begin reader tx"); let mut session = ReadSession { inner: ReadSessionInner::Replica { - tx: reader_tx, + tx: super::history_reads::ReaderTx::new(reader_tx, None), writer: fx.writer.clone(), }, }; @@ -3847,7 +3987,7 @@ async fn read_session_statement_cancel_propagates_without_writer_rerun() { unreachable!() }; sqlx::query("SET LOCAL statement_timeout = '50ms'") - .execute(&mut **tx) + .execute(&mut ***tx) .await .expect("cap session"); let mut locker = fx.replica.begin().await.expect("begin locker"); @@ -3946,7 +4086,7 @@ async fn thread_window_statement_cancel_propagates_without_writer_rerun() { .expect("begin reader tx"); let mut session = ReadSession { inner: ReadSessionInner::Replica { - tx: reader_tx, + tx: super::history_reads::ReaderTx::new(reader_tx, None), writer: fx.writer.clone(), }, }; @@ -3954,7 +4094,7 @@ async fn thread_window_statement_cancel_propagates_without_writer_rerun() { unreachable!() }; sqlx::query("SET LOCAL statement_timeout = '50ms'") - .execute(&mut **tx) + .execute(&mut ***tx) .await .expect("cap session"); let targets = [fx.root_hex.clone()]; diff --git a/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs index f53fed7c76..41e84cd2d4 100644 --- a/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs +++ b/crates/buzz-db/src/runtime/tests/reader_pool_postgres_tests.rs @@ -390,8 +390,9 @@ async fn proof_timeout_mid_begin_leaves_no_reader_transaction_open() { /// not strand the reader pool. Each stalled warm connection costs one routed /// read a proof timeout; once the release check gives up on those sockets, /// their slots are freed and routed reads reach the reader again through -/// new connections, instead of falling back to the writer until the kernel -/// gives up on the old sockets. +/// new connections, instead of falling back to the writer (or, for history +/// reads on the saturated pool, being rejected) until the kernel gives up on +/// the old sockets. #[tokio::test(flavor = "current_thread")] #[ignore = "requires Postgres"] async fn stalled_reader_sessions_release_their_slots_for_new_connections() { @@ -463,9 +464,13 @@ async fn stalled_reader_sessions_release_their_slots_for_new_connections() { .sum::() }; loop { - db.count_events_routed("stalled_reader_probe", &query) - .await - .expect("routed read answers"); + // While both slots sit in the release check the reader pool is + // saturated, so a history read is rejected as retryable rather than + // moved to the writer; the caller retries, as a client would. + match db.count_events_routed("stalled_reader_probe", &query).await { + Ok(_) | Err(crate::error::DbError::Sqlx(sqlx::Error::PoolTimedOut)) => {} + Err(e) => panic!("routed read answers or is rejected as retryable: {e:?}"), + } for (key, n) in drain_counters(&snapshotter) { *totals.entry(key).or_default() += n; } diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 011ccc02b7..7913ff46d2 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -2702,7 +2702,8 @@ impl Db { /// unset, even covered-eligible queries stay on the writer, so merging /// this seam is a true no-op until the budget is configured. Every /// failure fails closed to the writer, except a statement-timeout cancel - /// (57014), which is surfaced rather than re-run. + /// (57014), which is surfaced rather than re-run, and a busy reader, + /// which rejects the read as retryable (see [`Db::route_read`]). #[datastore_span(name = "query_events_routed", system = "postgresql")] pub async fn query_events_routed( &self, @@ -2719,7 +2720,7 @@ impl Db { predicate, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { crate::RouteDecision::Replica(mut tx, _entry, reason) => { match crate::event::query_events_on(&mut tx, q, crate::event::ReadDeadline::Session) @@ -2740,6 +2741,9 @@ impl Db { // writer rather than surfacing a routed error. tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::event::query_events_with_operation( &self.pool, q, @@ -2783,7 +2787,7 @@ impl Db { crate::RoutePredicate::Bounded, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { crate::RouteDecision::Replica(mut tx, _entry, reason) => { match crate::event::query_events_on(&mut tx, q, crate::event::ReadDeadline::Session) @@ -2802,6 +2806,9 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::event::query_events_with_operation( &self.pool, q, @@ -2849,7 +2856,7 @@ impl Db { crate::RoutePredicate::Bounded, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { crate::RouteDecision::Replica(mut tx, _entry, reason) => { match crate::event::count_events_on(&mut tx, q, crate::event::ReadDeadline::Session) @@ -2868,6 +2875,9 @@ impl Db { Err(e) => { tracing::warn!(path, "replica count failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::event::count_events(&self.pool, q).await } } @@ -3110,7 +3120,7 @@ impl Db { crate::RoutePredicate::Bounded, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { crate::RouteDecision::Replica(mut tx, _entry, reason) => { match crate::event::get_events_by_ids_on(&mut tx, community_id, ids).await { @@ -3127,6 +3137,9 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::event::get_events_by_ids_with_operation( &self.pool, community_id, diff --git a/crates/buzz-db/src/store/feed.rs b/crates/buzz-db/src/store/feed.rs index 1094c948e8..598053eaa4 100644 --- a/crates/buzz-db/src/store/feed.rs +++ b/crates/buzz-db/src/store/feed.rs @@ -365,7 +365,7 @@ impl Db { RoutePredicate::Bounded, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { RouteDecision::Replica(mut tx, _entry, reason) => match crate::feed::query_mentions_on( &mut tx, @@ -390,6 +390,9 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::feed::query_mentions( &self.pool, community, @@ -455,7 +458,7 @@ impl Db { RoutePredicate::Bounded, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { RouteDecision::Replica(mut tx, _entry, reason) => { match crate::feed::query_needs_action_on( @@ -481,6 +484,9 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::feed::query_needs_action( &self.pool, community, @@ -538,7 +544,7 @@ impl Db { RoutePredicate::Bounded, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { RouteDecision::Replica(mut tx, _entry, reason) => match crate::feed::query_activity_on( &mut tx, @@ -562,6 +568,9 @@ impl Db { Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); + // Release the reader and its history permit before the + // writer re-run, so a replica-error burst can't fill the cap. + drop(tx); crate::feed::query_activity( &self.pool, community, diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index a9f48fd8a1..a95f3f69ea 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1194,9 +1194,9 @@ impl Db { /// Replica-routed on the bounded arm — the one PERMISSION read routed by /// explicit product decision (bounded-stale membership beats the 10s /// cache it replaced). Admits and revokes may lag by at most the budget - /// `B`; everything else fails closed to the writer, exactly like - /// [`Db::query_events_routed_bounded`]. Not precedent for routing other - /// permission reads. + /// `B`; everything else fails closed to the writer, including a busy + /// reader that history reads would be rejected for (see + /// [`Db::route_read`]). Not precedent for routing other permission reads. #[datastore_span(name = "is_relay_member", system = "postgresql")] pub async fn is_relay_member(&self, community: CommunityId, pubkey: &str) -> Result { let path = "relay_membership"; @@ -1206,7 +1206,7 @@ impl Db { RoutePredicate::Bounded, crate::observability::ReaderOperation::Authorization, ) - .await + .await? { RouteDecision::Replica(mut tx, _entry, reason) => { match is_relay_member_on(&mut tx, community, pubkey).await { diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index d5407cbf75..f78786d758 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -942,7 +942,7 @@ impl Db { predicate, crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { match crate::thread::get_thread_replies_on( &mut tx, @@ -1104,7 +1104,8 @@ impl Db { /// /// Every replica failure except a cancelled statement (57014, which the /// writer would only repeat) fails closed to the writer and is recorded - /// in `buzz_db_route_decision`. + /// in `buzz_db_route_decision`. A busy reader rejects the read as + /// retryable instead (see [`Db::route_read`]). #[datastore_span(name = "get_channel_window", system = "postgresql")] pub async fn get_channel_window_with_session( &self, @@ -1125,7 +1126,7 @@ impl Db { RoutePredicate::from_channel_cursor(channel_id, &cursor), crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { RouteDecision::Replica(mut tx, _entry, reason) => { match crate::thread::get_channel_window_on( diff --git a/crates/buzz-db/src/store/thread_window.rs b/crates/buzz-db/src/store/thread_window.rs index 926ffae8df..daf37f7e94 100644 --- a/crates/buzz-db/src/store/thread_window.rs +++ b/crates/buzz-db/src/store/thread_window.rs @@ -223,7 +223,7 @@ impl Db { RoutePredicate::from_channel_cursor(request.channel, &cursor), crate::observability::ReaderOperation::SubscriptionHistory, ) - .await + .await? { let result = async { set_deadline(&mut tx).await?; diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index d2fb8fe2cb..a241e1cb58 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1083,7 +1083,7 @@ async fn handle_channel_window_filter( kind_filter.as_deref(), ) .await - .map_err(|e| internal_error(&format!("channel window error: {e}")))?; + .map_err(|e| db_read_error("channel window error", &e))?; // 1. Rows, in keyset order. let mut row_ids_hex = Vec::with_capacity(window.rows.len()); @@ -1930,7 +1930,7 @@ async fn query_member_events( remaining, ) .await - .map_err(|e| internal_error(&format!("feed mentions error: {e}")))?, + .map_err(|e| db_read_error("feed mentions error", &e))?, "needs_action" => state .db .query_feed_needs_action_routed( @@ -1942,7 +1942,7 @@ async fn query_member_events( remaining, ) .await - .map_err(|e| internal_error(&format!("feed needs_action error: {e}")))?, + .map_err(|e| db_read_error("feed needs_action error", &e))?, "activity" => state .db .query_feed_activity_routed( @@ -1953,7 +1953,7 @@ async fn query_member_events( remaining, ) .await - .map_err(|e| internal_error(&format!("feed activity error: {e}")))?, + .map_err(|e| db_read_error("feed activity error", &e))?, _ => continue, }; for se in type_events { @@ -2027,7 +2027,7 @@ async fn query_member_events( thread_cursor.as_deref(), ) .await - .map_err(|e| internal_error(&format!("thread query error: {e}")))?; + .map_err(|e| db_read_error("thread query error", &e))?; let mut thread_row_ids = Vec::with_capacity(thread_replies.len() + 1); thread_row_ids.push(root_hex.to_string()); @@ -10195,7 +10195,17 @@ pub(crate) mod postgres_tests { .await .expect("replica statement_timeout"); } - let pool = sqlx::PgPool::connect(&format!("{base}/{name}")) + // The replica gets a short acquire timeout, like production's + // reader pool, so a shed read against a held-full replica fails + // fast instead of waiting out sqlx's 30s default. + let options = sqlx::postgres::PgPoolOptions::new(); + let options = if role == "r" { + options.acquire_timeout(std::time::Duration::from_secs(1)) + } else { + options + }; + let pool = options + .connect(&format!("{base}/{name}")) .await .expect("connect scratch"); buzz_db::migration::run_migrations(&pool) @@ -10564,6 +10574,64 @@ pub(crate) mod postgres_tests { fx.drop().await; } + /// A shed history read answers 503 `database unavailable` on every + /// bridge `/query` branch that serves one: channel window, the three + /// feeds and legacy thread replies. A saturated reader (every connection + /// checked out, none idle) rejects history reads instead of moving them + /// to the writer, the same `PoolTimedOut` as a full history gate. + /// Mutation: map any of these branches through `internal_error` → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn shed_history_read_maps_to_503_on_every_bridge_branch() { + let fx = cancel_fixture().await; + let h = [&fx.channel]; + let cases = [ + ( + "channel window", + serde_json::json!({"kinds": [9], "#h": h, "top_level": true}), + ), + ( + "feed mentions", + serde_json::json!({"kinds": [9], "feed_types": ["mentions"]}), + ), + ( + "feed needs_action", + serde_json::json!({"kinds": [9], "feed_types": ["needs_action"]}), + ), + ( + "feed activity", + serde_json::json!({"kinds": [9], "feed_types": ["activity"]}), + ), + ( + "thread replies", + serde_json::json!({"kinds": [9], "#e": [&fx.root], "depth_limit": 1}), + ), + ]; + for (branch, filter) in &cases { + let (status, body) = fx.http("/query", filter).await; + assert_eq!(status, StatusCode::OK, "healthy {branch}: {body}"); + } + + let mut held = Vec::new(); + for _ in 0..fx.replica.options().get_max_connections() { + held.push(fx.replica.acquire().await.expect("hold reader connection")); + } + for (branch, filter) in &cases { + let (status, body) = fx.http("/query", filter).await; + assert_eq!( + (status, body["error"].as_str()), + ( + StatusCode::SERVICE_UNAVAILABLE, + Some(super::super::DATABASE_UNAVAILABLE) + ), + "shed {branch}: {body}" + ); + } + + drop(held); + fx.drop().await; + } + /// The search itself (not its hydrate) cancelled by the search pool's /// statement timeout answers with the timeout contract on every search /// caller: HTTP `/query` and `/profiles/search` 503 `query timed out`, WS diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index f38ad8f81b..e7ab1622b6 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -35,12 +35,22 @@ pub(crate) fn internal_error(msg: &str) -> (StatusCode, Json) /// would re-run the same expensive query. pub(crate) const QUERY_TIMED_OUT: &str = "query timed out"; +/// Stable client-visible body for a read the relay could not get a database +/// connection for (pool timeout, closed pool, socket failure), including a +/// history read shed because the reader is busy. Transient: retry later. +pub(crate) const DATABASE_UNAVAILABLE: &str = "database unavailable"; + /// Map a DB read failure: a statement-deadline cancel becomes a distinct -/// 503 `query timed out`; anything else stays a generic 500. +/// 503 `query timed out`, an unreachable or saturated database a 503 +/// `database unavailable`; anything else stays a generic 500. pub(crate) fn db_read_error( context: &str, e: &buzz_db::DbError, ) -> (StatusCode, Json) { + if e.is_connection_unavailable() { + tracing::warn!(error_kind = %e.kind(), "{context}"); + return api_error(StatusCode::SERVICE_UNAVAILABLE, DATABASE_UNAVAILABLE); + } read_error(context, e.is_statement_cancelled(), e.kind()) } @@ -811,6 +821,19 @@ mod db_read_error_tests { assert_eq!(body.0["error"], "internal server error"); } + #[test] + fn unavailable_database_maps_to_retryable_503() { + for error in [ + sqlx::Error::PoolTimedOut, + sqlx::Error::PoolClosed, + sqlx::Error::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset)), + ] { + let (status, body) = db_read_error("ctx", &buzz_db::DbError::Sqlx(error)); + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(body.0["error"], DATABASE_UNAVAILABLE); + } + } + #[test] fn non_cancel_search_error_stays_generic_500() { let (status, body) = search_read_error( diff --git a/deploy/charts/buzz/README.md b/deploy/charts/buzz/README.md index f00ac8aa08..328307ec61 100644 --- a/deploy/charts/buzz/README.md +++ b/deploy/charts/buzz/README.md @@ -432,6 +432,23 @@ open (`open_transaction`). Each close increments `buzz_db_reader_release_closed_total{reason}`, a counter with those three series. A rise in `timeout` means reader sockets stopped answering. +Subscription-history reads are the exception to the writer fallback when the +reader is merely busy. Each pod admits at most its reader pool size less one +fifth (16 at the default pool size of 20) history reads to the reader at +once, which normally leaves authorization reads a free reader connection +(connections still finishing their release can briefly take it). A history read +that waits more than 1s for admission, or misses the reader's 150ms acquire +budget while every reader connection is busy, is rejected with the retryable +`CLOSED` `error: database error` instead of moving to the writer, where it +would compete with authentication and tenant lookups. Each rejection +increments `buzz_db_route_decision{decision="rejected"}`, with +`reason="history_read_cap"` or `"reader_acquire_timeout"`. A history read +the reader cannot serve for other reasons (unreachable, validation failure, +proof timeout, stale replica) still falls back to the writer. The +compatibility families `buzz_db_pool_acquire_wait_seconds` and +`buzz_db_pool_acquisitions_total` carry `operation` as well as `pool_role` +and `outcome`. + ## Relay Pod extensions The chart exposes narrow extension points for init containers, volumes, relay diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 92f8ef81f2..c822a33064 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -401,6 +401,9 @@ run_unit_tests() { run_test_step "buzz-relay API query parsing tests" \ cargo test -p buzz-relay --lib api::parse_query_tests:: -- --nocapture + run_test_step "buzz-relay API database error mapping tests" \ + cargo test -p buzz-relay --lib api::db_read_error_tests:: -- --nocapture + run_test_step "buzz-relay profile search request tests" \ cargo test -p buzz-relay --lib api::profiles:: -- --nocapture