From 47be2e2d3f2a438b3a5fd7df0c710b81d8c0d87e Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Thu, 5 Nov 2026 17:12:30 -0500 Subject: [PATCH] fix(db): keep busy-reader history reads off the writer and cap them per pod (#9132) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem A routed read that misses the reader's 150ms acquire budget (`reader_acquire_timeout`) falls back to the writer, whatever the read is. Under a burst of subscription-history reads (REQ history, channel and thread windows, feeds, search hydration, COUNT), the reader pool fills, and every miss moves a heavy history query onto the writer. There it competes with the authentication, tenant-resolution and moderation lookups that only the writer serves, and their latency climbs with it. Nothing bounds how many history reads a pod sends to the reader at once, so the reader queue grows without limit as well. Separately, the compatibility families `buzz_db_pool_acquire_wait_seconds` and `buzz_db_pool_acquisitions_total` have no `operation` label, so a breakdown by operation reports every sample as unlabelled. ## Changes 1. **History reads no longer spill onto the writer when the reader is busy** (`Db::route_read`). `route_read` now returns `Result`. For `ReaderOperation::SubscriptionHistory`, a `reader_acquire_timeout` on a saturated reader pool rejects the read with `sqlx::Error::PoolTimedOut`. Saturated means every connection the pool may hold exists and none is idle (`Db::reader_saturated`). The check is needed because sqlx also reports a reader that refuses dials as `PoolTimedOut`, and that reader is down, not busy. The relay already answers that with its generic retryable `CLOSED` `error: database error`, so there is no new client contract. A reader that is wrong rather than busy (unreachable, validation failure, proof timeout, stale token) and prechecks that never reach the reader still fall back to the writer. Authorization reads (`is_relay_member`) keep the writer fallback unchanged. 2. **Per-pod cap on reader-served history reads** (`runtime/history_reads.rs`). A FIFO `tokio::sync::Semaphore` admits at most the reader pool size less one fifth (at least one) history reads at once. For example, the default pool of 20 admits 16. The remaining connections are normally free for authorization reads. A read waits at most 1s for a permit and is otherwise rejected the same way. The permit is owned by the reader transaction (`ReaderTx`) and released when the transaction is dropped; routed reads drop it before re-running a failed replica read on the writer. That holds whether the transaction is a per-call routed read or a `ReadSession` the relay keeps for follow-up queries. The cap bounds history transactions, not connections: sqlx returns a dropped transaction's connection asynchronously through the reader release check. Released connections can therefore briefly occupy the free slots, for one round trip normally and at most the 5s release timeout when sockets stall. In that window, authorization reads fall back to the writer as before, and history reads on the saturated pool are rejected, not spilled. 3. **Rejections are observable**: `buzz_db_route_decision{decision="rejected",reason="history_read_cap"|"reader_acquire_timeout"}`. 4. **Legacy pool families gain `operation`.** The population is unchanged, and summing over `operation` gives the original series. 5. **HTTP reads answer 503 `database unavailable` when the database can't be reached** (`api::db_read_error`). This covers a pool timeout, closed pool or socket failure on `POST /query`, `/count` and every bridge `/query` branch that serves a history read (channel window, the three feeds, thread replies), including a shed history read. These used to return a generic 500. The thread-window path already answered 503 for pool timeouts; this applies the same mapping to the shared helper. Pool sizes are unchanged on purpose. A larger pool adds lock contention on the writer rather than relieving it. Docs: `deploy/charts/buzz/README.md` (operator contract) and `PLANS/REPLICA_FULL_READ_ROUTING_DESIGN.md` (fail-closed exceptions). ## What got simpler The concept count is roughly flat. Instead of a caller-side permit, the permit lives in the one type every replica transaction already flows through (`RouteDecision::Replica` / `ReadSessionInner::Replica`). The 11 routed call sites change only by a `?`, and the compiler enforces that change: a call site cannot silently fall through to the writer on a rejection. No new error variant, CLOSED reason or env var. ## Tests - `postgres_tests::saturated_reader_spends_one_acquire_budget_and_rejects_history_reads` (rewrites the former `routed_fallback_spends_one_acquire_budget_when_aurora_cache_is_cold`). A size-1 reader is held saturated with a cold Aurora cache. A history read (`count_events_routed`) is rejected with `PoolTimedOut` within one acquire budget, records `rejected/reader_acquire_timeout`, and releases its permit. An authorization read on the same reader still falls back as `writer/reader_acquire_timeout`. - `postgres_tests::match_nothing_reads_take_no_connection` (unchanged): its control still requires a history read against an unreachable reader to fall back to the writer. It caught the first cut of this change, which rejected that read. - `postgres_tests::history_permit_follows_the_read_session_and_a_full_gate_rejects`: a held replica `ReadSession` keeps its permit. With the gate full, the next history read waits the full permit budget and is then rejected as `rejected/history_read_cap`, with no writer route recorded. Dropping the session frees the permit and the next read routes to the replica. - `history_reads::tests`: a cap sizing table, plus bounded-wait rejection and reopen under paused time. - `observability` label tests now require `operation` on both legacy families. - `api::db_read_error_tests::unavailable_database_maps_to_retryable_503`: pool timeout, closed pool and I/O map to 503 `database unavailable`. Other errors stay 500. - `postgres_tests::shed_history_read_maps_to_503_on_every_bridge_branch`: with the reader saturated, `/query` channel-window, mentions, needs_action, activity and thread-reply requests each answer 503 `database unavailable` (each serves 200 beforehand). - Mutation-checked on Blox: re-enabling the writer spill, forcing the saturation check either way, skipping the gate, dropping the permit early and removing the legacy label each fail a test. ## Residual - During a reader failover whose old sockets stall, both slots sit in the release check, so the pool reads as saturated. History reads are shed as retryable for up to the release bound (5s) instead of loading the writer. `stalled_reader_sessions_release_their_slots_for_new_connections` now retries through that window and still requires recovery onto the reader. - A reader that accepts no TCP at all can fill the pool with in-flight dials and look saturated. Its history reads are then rejected rather than moved to the writer, which is the same outcome as a full gate. - Shedding relies on clients backing off on a retryable `CLOSED` (RPR #9). WebSocket clients already treat `error: database error` as retryable. 🤖 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 --- Justfile | 4 +- PLANS/REPLICA_FULL_READ_ROUTING_DESIGN.md | 8 +- crates/buzz-db/src/runtime/history_reads.rs | 131 ++++++++++++++ crates/buzz-db/src/runtime/mod.rs | 103 +++++++++-- crates/buzz-db/src/runtime/observability.rs | 42 +++-- crates/buzz-db/src/runtime/tests.rs | 164 ++++++++++++++++-- .../tests/reader_pool_postgres_tests.rs | 15 +- crates/buzz-db/src/store/event.rs | 23 ++- crates/buzz-db/src/store/feed.rs | 15 +- crates/buzz-db/src/store/relay_members.rs | 8 +- crates/buzz-db/src/store/thread.rs | 7 +- crates/buzz-db/src/store/thread_window.rs | 2 +- crates/buzz-relay/src/api/bridge.rs | 80 ++++++++- crates/buzz-relay/src/api/mod.rs | 25 ++- deploy/charts/buzz/README.md | 17 ++ scripts/run-tests.sh | 3 + 16 files changed, 569 insertions(+), 78 deletions(-) create mode 100644 crates/buzz-db/src/runtime/history_reads.rs 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