From 2f2f6e2774fb0899d2a927f5e5b1987603765f31 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Wed, 21 Oct 2026 19:35:03 -0400 Subject: [PATCH] fix(db): bound every statement and the #p mention scan (#8461) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem An agent `#p` + `#h` + `since` REQ (the `require_mention` subscription in `buzz-acp`) ran for over 45 minutes on a production writer and held `AccessShare` on every `events` partition. There were two causes: 1. **The mention side of the `#p` join was unbounded.** `build_query_events_sql` joined `event_mentions` on `(community_id, event_id)` only. The planner picked a nested loop whose outer side was every mention of the pubkey, ever, and whose inner side range-scanned the channel index from `since` on each partition. Cost ≈ total mentions × channel rows since `since`. 2. **Nothing cancelled it.** The writer `statement_timeout` defaulted to `0`. Only `#e` reads had a deadline (the 20s cap from #7854), and read-replica sessions had none at all. `pg_stat_statements` shows this shape has a multi-thousand-second max going back months, so it is chronic and was not introduced by a recent change. ## Change **`#p` mention bound** (`store/event.rs`, `push_mention_join`, shared by the page SELECT and COUNT) - Join on the full event key, `AND e.created_at = m.event_created_at`. Mention rows copy the event's `created_at` at insert (migration 0007 already relies on this equality), so the join can prune `events` partitions by time. - Repeat `since`/`until` on `m.event_created_at`, so the outer scan is a range on the existing `idx_event_mentions_pubkey_created`. `until` is a superset bound for the keyset cursor; the exact `(created_at, id)` cursor test stays on `e`. - Repeat `kinds` on `m.event_kind` when the filter has kinds. Every production mention writer binds the event kind (`insert_mentions` in `store/runtime/mod.rs`, and the replaceable-event path in `store/replaceable.rs`, back to the first Postgres schema), and no migration inserts mention rows, so this drops no rows written by the relay. Rows imported out of band with a NULL `event_kind` would no longer match a kinds-filtered `#p` read; a read-only production count on 2026-10-20 found 0 such rows on both production reader databases. The pushdown makes `idx_event_mentions_pubkey_kind_created` available to the planner for the `require_mention` kind-9 shape; its use has not been observed (see below). An E2E fixture with a hand-set NULL `event_kind` row confirmed the caveat: base returns 8 events, head 7. - The feed's mentions and needs-action queries (`store/feed.rs`, `build_mentions_query` / `build_needs_action_query`) join `event_mentions` the same way, so each joined row probes one `events` partition instead of all of them. Their `ORDER BY m.event_created_at ... LIMIT` already kept them from running away; this removes the per-row cross-partition probe. - Observed plan (E2E at `9cb6638b1`, before the `kinds` pushdown; 60,000 historical mentions + 2,500 recent decoys, stock planner, `EXPLAIN (ANALYZE, BUFFERS)` of the captured production SQL): the mention side is a bounded range on `idx_event_mentions_pubkey_created` (13 rows), then a hash join with a scan of the recent `events` partition. Base scanned 2,509 recent events and probed `event_mentions_pkey` 2,509 times. Head: **1.1 ms / 319 shared hits**; base: **8.5 ms / 10,049 shared hits**. Not measured against the production replica. A fresh `EXPLAIN (ANALYZE, BUFFERS)` replay at `a7065c367`, after the pushdown, ran on a smaller fixture where the planner chose a sequential mention scan, so it neither confirms nor replaces these numbers. **Reads: 20s deadline on every generic event read** - `query_events_on` and `count_events_on` always run under the transaction-local 20s deadline that `#e` reads used (`fetch_with_read_deadline`, formerly `fetch_with_e_tag_deadline`). It only tightens: shorter operator or caller caps are kept. - Read-replica sessions get a 20s session default through `after_connect`, using the same tighten-only SQL. - Cost: every generic REQ history read and COUNT now pays BEGIN + `set_config` + ROLLBACK (three extra round trips; before, only `#e` reads did). In the local E2E the end-to-end history read was still faster at head (median 4.1 ms vs 10.2 ms at base, 9 samples each, warm cache), so the round trips are dominated by the bounded plan; this does not isolate them or establish production p99. The round trips also lengthen how long each such read holds its writer connection; that hold time has not been measured against production pool sizing. - A replica read cancelled at its deadline (57014) is returned, not re-run on the writer. Usually the cancel means expensive work the writer would only repeat. It can also mean the replica waited on a lock only it holds (a replayed `ACCESS EXCLUSIVE` from primary DDL), when the writer would have answered quickly; that miss is the accepted price of keeping runaway reads off the primary. The rationale now lives once on `DbError::is_statement_cancelled` instead of in a dozen copied comments. The generic routed reads and `ReadSession` already did this; the eight routed reads with their own SQL (feed mentions, needs-action and activity; thread replies; channel window; relay membership; thread window and its auxiliary scan) now do too. A cancelled auxiliary scan also keeps its replica session instead of degrading it to the writer for the rest of the request. Other replica failures (e.g. hot-standby recovery conflicts) still fall back to the writer. - Paths with their own tighter `SET LOCAL` (artifact 2–5s, thread window 4s, personal reads, usage 5–15s, serving lease 1.5s) are unchanged. **Writes: every writer session is bounded** - The default `statement_timeout` changes from `0` to 120s (`DEFAULT_STATEMENT_TIMEOUT_MS`). - `BUZZ_DB_STATEMENT_TIMEOUT_MS` still overrides it. `0` or any value above 300s installs 300s (`MAX_STATEMENT_TIMEOUT_MS`), so a session can no longer disable it, and a clamped value logs a `warn!` when the pool is built. Only the operator-approved, fenced destruction paths below lift it, per transaction. **Long-running writer paths, audited** | Path | Behavior | |---|---| | Migrations / schema destruction (`with_exclusive_schema_destruction_lock`) | Unchanged: these lift lock and statement timeouts on their own detached connection | | Partition maintenance | Unchanged: own 5s `statement_timeout` | | Partition audit (`buzz-admin`) | Unchanged: 30s | | Community deletion inventory, purge and logical verification (`inventory_schema` whole-tenant counts, `purge_postgres` whole-community `DELETE` per table in one tx, `verify_postgres_logically_deleted` whole-tenant `EXISTS` per table right after the purge) | Lifted (`statement_timeout = 0`) transaction-locally by `lift_statement_timeout`, like migrations; lock and idle timeouts still bound them. Inventory counts each table in its own lifted transaction, so each table's lock and snapshot are released as soon as its count finishes, as the autocommit counts did. A cancelled purge would retry from scratch and eventually block, so a cap here would turn large-tenant deletion into an operator override | | Audit appends (`buzz-audit`, audit pool inherits the writer cap) | Waits on the audit-chain advisory lock. With `BUZZ_DB_LOCK_TIMEOUT_MS=0` (or at least the statement cap), `statement_timeout` (57014) cancels the wait instead of `lock_timeout` (55P03). The relay's audit worker now retries both with the same backoff instead of dropping the entry on 57014; admission refusal stays terminal | On HTTP routes a statement timeout already maps to the distinct retryable 503 (`api/mod.rs`, `statement_timeout_maps_to_distinct_503`). A cancelled REQ fails its history read instead of hanging. ## Tests - `p_tag_query_bounds_mention_side_by_time` (unit): page and COUNT SQL carry the `created_at` join and the `since`/`until`/`kinds` mention bounds, and the bounds are absent when the filter has none. - `p_tag_read_with_mention_bounds_keeps_results` (Postgres): through the production mention insert, unbounded, `since`, window and keyset-cursor reads return exactly the mentioned rows, and COUNT matches. Every bound sits exactly on a mentioned row's second and the cursor page includes a same-second sibling after the cursor id, so an off-by-one on either mention-side bound fails it. - `plain_read_is_cancelled_at_deadline_when_timeout_disabled` / `count_events_cancels_at_read_deadline_when_timeout_disabled` (Postgres): the production `#p` + `#h` + `since` shape and a plain COUNT are cancelled at 20s with the session timeout disabled. Previously these covered `#e` only. - `writer_statement_timeout_is_always_bounded` (unit) and `session_statement_timeouts_bound_writer_and_reader_sessions` (Postgres): through `Db::new`, writer sessions install `2min` / `5min` (for `0` and for 600s) / `45s`, and replica sessions install `20s`. - `deletion_whole_tenant_stages_outlive_the_writer_statement_timeout` (Postgres): on a scratch desired-state database with a 500ms writer cap, the real `inventory_schema`, `purge_postgres` and `verify_postgres_logically_deleted` each wait behind a held `ACCESS EXCLUSIVE` lock on `events` for over 1s and still complete, a plain statement is cancelled (57014), and pooled sessions still read `500ms` afterwards. A lifted transaction reads `statement_timeout = 0`, so a bounded lift (e.g. `'300000'`) fails it, as does removing any of the three `lift_statement_timeout` calls. - `mentions_query_is_tenant_scoped_and_joins_mentions_by_composite_key` / `needs_action_query_is_tenant_scoped_and_joins_mentions_by_composite_key` (unit): both feed queries carry the `e.created_at = m.event_created_at` join. - `replica_statement_cancel_propagates_without_writer_rerun` (Postgres): with the replica locked past its cap, each of the six routed reads above returns 57014 instead of the writer's copy. Removing the cancel arm in any of them fails it. - `thread_window_statement_cancel_propagates_without_writer_rerun` (Postgres): a thread-window read parked behind a replica lock and cancelled (57014; the window's own 1s `lock_timeout` would otherwise fire first) returns the cancel instead of the unlocked writer's window, and an auxiliary scan timed out on a replica session returns 57014 and leaves the session on the replica. Removing either guard fails it. - `audit_worker_retries_statement_timeout_until_original_entry_is_appended_once` (Postgres, `buzz-relay`): the production audit worker, with `lock_timeout` disabled and a 100ms statement cap, keeps the entry through repeated 57014 cancels while the chain lock is held and appends it exactly once after release. Treating only 55P03 as retryable fails it. - The existing writer and audit-pool GUC tests now expect `5min` for a disabled statement timeout. ## After deploy, watch - `buzz_db_route_decision{decision="replica",reason="statement_cancelled"}`: replica reads that now fail instead of falling back to the writer, including the replica-only lock case above. - `buzz_db_pool_acquire_wait_seconds` and `buzz_db_pool_waiters` on the writer pool: every generic REQ history read and COUNT now holds its connection for three extra round trips. ## Database checklist (`docs/database.md` § Review checklist) This PR changes schema-dependent `buzz-db` queries (the `#p` join and its `m.event_kind` pushdown). It does not change the schema. - **Phase:** none. No schema change, so this is not an expand, backfill or contract step. Nothing before or after it is required. - **Triggers, functions, procedures, rules, foreign keys:** none added. - **Row mutations / backfill:** no migration, and no row is mutated or backfilled. The one data-shape dependency is mention rows with a NULL `event_kind`: a kinds-filtered `#p` read no longer matches them (see the `kinds` bullet under Change). Relay writers always bind `event_kind`, so only rows imported out of band are affected. A read-only production count on 2026-10-20 found 0 such rows on both production reader databases. - **Contract migration:** N/A. - **Rule 4 locations:** none touched. No table, index or desired-state file changes, so no parity test applies. The new bounds use the existing `idx_event_mentions_pubkey_created` and `idx_event_mentions_pubkey_kind_created`. - **Deployed relay vs. new schema:** there is no new schema. The currently deployed relay and this relay run against the same tables. - **Lock and statement timeouts:** this PR is about them. Generic reads get a 20s deadline, replica sessions a 20s default, and writer sessions a 120s default capped at 300s. Community-deletion transactions lift `statement_timeout` per transaction; migrations and schema destruction are unchanged (see "Long-running writer paths, audited"). No index builds and no operator prebuild step. ## What got simpler - One read-deadline path instead of an `#e`-only branch in `query_events_on` and `count_events_on`. - The `#p` join SQL that was duplicated between SELECT and COUNT now lives in one `push_mention_join`. - The deletion tests share one scratch desired-state database helper instead of an inline copy. - "Is this query bounded?" now has one answer: yes, unless it runs in a migration, schema destruction, or community-deletion transaction. ## Identified debt (not in this PR) - `buzz-acp` re-sends the same filter after a timeout `CLOSED` (no backoff or narrowing). With the `#p` bound the `require_mention` read should finish well inside 20s, so this is a follow-up, not a blocker. --------- Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> --- .env.example | 19 +- crates/buzz-db/src/error.rs | 12 +- crates/buzz-db/src/runtime/mod.rs | 65 ++++- crates/buzz-db/src/runtime/tests.rs | 255 ++++++++++++++++- crates/buzz-db/src/store/deletion.rs | 297 ++++++++++++++++--- crates/buzz-db/src/store/event.rs | 329 +++++++++++++++++----- crates/buzz-db/src/store/feed.rs | 28 ++ crates/buzz-db/src/store/relay_members.rs | 6 + crates/buzz-db/src/store/thread.rs | 17 +- crates/buzz-db/src/store/thread_window.rs | 12 + crates/buzz-relay/src/main.rs | 5 +- crates/buzz-relay/src/state.rs | 71 ++++- 12 files changed, 960 insertions(+), 156 deletions(-) diff --git a/.env.example b/.env.example index a875154150..73d4d9f6ea 100644 --- a/.env.example +++ b/.env.example @@ -58,19 +58,24 @@ REDIS_URL=redis://localhost:6379 # BUZZ_PARTITION_MANAGER_ADVANCE_ENABLED=true # Writer-session Postgres timeouts for buzz-db-backed pools and the relay audit -# pool, all in milliseconds; 0 disables. The separately deployed push gateway -# owns its own database and session policy and does not consume these knobs. +# pool, all in milliseconds; 0 disables the lock and idle timeouts. The +# separately deployed push gateway owns its own database and session policy +# and does not consume these knobs. # lock_timeout: fail a statement that waits this long on any lock instead of # parking behind a wedged holder (default 5000). # BUZZ_DB_LOCK_TIMEOUT_MS=5000 # idle_in_transaction_session_timeout: reap sessions idle inside an open # transaction — bounds how long a wedged client can hold locks (default 60000). # BUZZ_DB_IDLE_TXN_TIMEOUT_MS=60000 -# statement_timeout: cap any single statement's runtime. Off by default — -# startup migrations/backfills legitimately run long statements. Warning: a -# pathologically low value (e.g. 1) also times out connection setup and can -# prevent any DB connection from establishing. -# BUZZ_DB_STATEMENT_TIMEOUT_MS=0 +# statement_timeout: cap any single statement's runtime (default 120000). A +# session cannot disable it: 0 or anything above 300000 installs 300000 (5 +# minutes) and logs a warning at startup. Migrations, schema destruction, and +# community-deletion inventory and purge transactions lift it for themselves. +# Generic event reads (REQ history, /query, /count) and read-replica sessions +# are further capped at 20s. Warning: a pathologically low value (e.g. 1) also +# times out connection setup and can prevent any DB connection from +# establishing. +# BUZZ_DB_STATEMENT_TIMEOUT_MS=120000 # ----------------------------------------------------------------------------- # Typesense (search) diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index e27b319bed..d0a2ffcb22 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -91,8 +91,16 @@ pub enum DbError { impl DbError { /// Whether Postgres cancelled the statement (SQLSTATE 57014: - /// `statement_timeout` or an explicit cancel). Re-running such a query on - /// the writer would only repeat the same expensive work. + /// `statement_timeout` or an explicit cancel). + /// + /// Replica-routed reads surface a cancel instead of re-running on the + /// writer. Usually the cancel means expensive work that the writer would + /// only repeat at the primary's cost. It can also mean the replica waited + /// on a lock that only it holds (a replayed `ACCESS EXCLUSIVE` from + /// primary DDL), when the writer would have answered quickly. That miss is + /// the accepted price of keeping runaway reads off the primary. Per-call + /// routed reads record it as the `statement_cancelled` route reason; + /// `ReadSession` reads return it to the relay, which logs the failed read. pub fn is_statement_cancelled(&self) -> bool { matches!( self, diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 4bda0d71a9..de8f6d3d1c 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -354,8 +354,8 @@ impl ReadSession { ReadSessionInner::Replica { tx, writer } => { match event::query_events_on(tx, q).await { Ok(rows) => return Ok(rows), - // A cancelled statement (timeout) would be just as slow on - // the writer; surface it instead of doubling the cost. + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. Err(e) if e.is_statement_cancelled() => return Err(e), Err(e) => { tracing::warn!( @@ -689,8 +689,12 @@ pub struct DbConfig { /// writer connections (env `BUZZ_DB_IDLE_TXN_TIMEOUT_MS`). `0` disables. pub idle_txn_timeout_ms: u64, /// Session `statement_timeout` in milliseconds for writer connections - /// (env `BUZZ_DB_STATEMENT_TIMEOUT_MS`). `0` disables it and is the - /// default because migrations and backfills may legitimately run long. + /// (env `BUZZ_DB_STATEMENT_TIMEOUT_MS`). Defaults to + /// [`DEFAULT_STATEMENT_TIMEOUT_MS`]. Every writer session is bounded: + /// `0` or a value above [`MAX_STATEMENT_TIMEOUT_MS`] installs the maximum + /// (see [`Self::effective_statement_timeout_ms`]). Migration and + /// schema-destruction sessions lift it on their own detached connection; + /// community-deletion inventory and purge lift it per transaction. pub statement_timeout_ms: u64, /// Whether every new writer-pool connection defaults all transactions to /// read-only. Intended for operator audit commands, never the relay pool. @@ -716,7 +720,7 @@ impl Default for DbConfig { usage_metrics_replica_max_age_ms: 30_000, lock_timeout_ms: DEFAULT_LOCK_TIMEOUT_MS, idle_txn_timeout_ms: DEFAULT_IDLE_TXN_TIMEOUT_MS, - statement_timeout_ms: 0, + statement_timeout_ms: DEFAULT_STATEMENT_TIMEOUT_MS, default_transaction_read_only: false, } } @@ -728,10 +732,31 @@ pub const DEFAULT_LOCK_TIMEOUT_MS: u64 = 5_000; /// Default writer `idle_in_transaction_session_timeout` in milliseconds. pub const DEFAULT_IDLE_TXN_TIMEOUT_MS: u64 = 60_000; +/// Default writer session `statement_timeout` in milliseconds (2 minutes). +/// Generic event reads tighten it further to +/// [`crate::event::READ_STATEMENT_TIMEOUT`] per statement. +pub const DEFAULT_STATEMENT_TIMEOUT_MS: u64 = 120_000; + +/// Upper bound on any writer session `statement_timeout`, in milliseconds +/// (5 minutes). Configuration cannot disable or exceed it. +pub const MAX_STATEMENT_TIMEOUT_MS: u64 = 300_000; + impl DbConfig { + /// The writer session `statement_timeout` actually installed: the + /// configured value, with `0` (disabled) and anything above + /// [`MAX_STATEMENT_TIMEOUT_MS`] clamped to the maximum. + pub fn effective_statement_timeout_ms(&self) -> u64 { + match self.statement_timeout_ms { + 0 => MAX_STATEMENT_TIMEOUT_MS, + ms => ms.min(MAX_STATEMENT_TIMEOUT_MS), + } + } + /// Overlay writer session timeouts from the shared `BUZZ_DB_*_TIMEOUT_MS` /// environment variables. Missing or invalid values retain the existing - /// configuration; explicit zeroes pass through to disable a timeout. + /// configuration; explicit zeroes pass through to disable the lock and + /// idle timeouts, while the statement timeout is still clamped at connect + /// time (see [`Self::effective_statement_timeout_ms`]). /// /// This belongs in `buzz-db` so relay, admin, deletion, and audit writers /// share one policy. The separately deployed push gateway owns its own @@ -810,7 +835,15 @@ impl Db { let push_enabled = config.push_enabled; let lock_timeout_ms = config.lock_timeout_ms; let idle_txn_timeout_ms = config.idle_txn_timeout_ms; - let statement_timeout_ms = config.statement_timeout_ms; + let statement_timeout_ms = config.effective_statement_timeout_ms(); + if statement_timeout_ms != config.statement_timeout_ms { + tracing::warn!( + configured_ms = config.statement_timeout_ms, + installed_ms = statement_timeout_ms, + "writer statement_timeout cannot be disabled or exceed the maximum; \ + installing the maximum (BUZZ_DB_STATEMENT_TIMEOUT_MS)" + ); + } let default_transaction_read_only = config.default_transaction_read_only; let options = PgPoolOptions::new() .max_connections(config.max_connections) @@ -845,7 +878,8 @@ impl Db { // cancel the holder. `idle_in_transaction_session_timeout` // reaps only holders idling inside an open transaction, // while actively executing holders are bounded only by - // `statement_timeout` (off by default). Bare values are + // `statement_timeout` (always on; see + // `effective_statement_timeout_ms`). Bare values are // milliseconds. Migration/schema-destruction connections // reset lock and statement timeouts before their intentional // long wait (see `with_exclusive_schema_destruction_lock`). @@ -950,6 +984,11 @@ impl Db { /// No floor guard or writer-isolation assertion: replica sessions are /// read-only, so the commit-time trigger from migration 0021 never fires /// here and the write fence that depends on READ COMMITTED is never reached. + /// + /// Every replica session's `statement_timeout` is lowered to the + /// event-read deadline ([`crate::event::READ_STATEMENT_TIMEOUT`]) unless + /// the server/role default is already tighter, so no replica read is + /// unbounded; per-call `SET LOCAL` caps still apply on top. fn connect_read_pool(config: &DbConfig, url: &str, max_connections: u32) -> Result { Ok(PgPoolOptions::new() .max_connections(max_connections) @@ -957,6 +996,16 @@ impl Db { .acquire_timeout(Self::READER_ACQUIRE_TIMEOUT) .max_lifetime(Duration::from_secs(config.max_lifetime_secs)) .idle_timeout(Duration::from_secs(config.idle_timeout_secs)) + .after_connect(|conn, _meta| { + Box::pin(async move { + sqlx::query(crate::event::TIGHTEN_STATEMENT_TIMEOUT_SQL) + .bind(crate::event::READ_STATEMENT_TIMEOUT) + .bind(false) + .execute(&mut *conn) + .await + .map(|_| ()) + }) + }) .connect_lazy(url)?) } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index 39a6b1e860..85e9722280 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -2950,11 +2950,13 @@ async fn writer_pool_metrics_stop_after_session_timeout_setup_failure() { max_connections: 1, min_connections: 1, acquire_timeout_secs: 1, - statement_timeout_ms: u64::MAX, + // The statement timeout is clamped before install; the lock timeout + // is passed through, so it can still drive the setup failure. + lock_timeout_ms: u64::MAX, ..DbConfig::default() }) .await - .expect_err("Postgres must reject an out-of-range statement timeout"); + .expect_err("Postgres must reject an out-of-range lock timeout"); assert!( error.to_string().contains("pool timed out") || error.to_string().contains("invalid value for parameter") @@ -3039,13 +3041,85 @@ fn session_timeout_env_overlay_zero_passthrough_and_invalid_fallback() { } } - let defaults = (DEFAULT_LOCK_TIMEOUT_MS, DEFAULT_IDLE_TXN_TIMEOUT_MS, 0); + let defaults = ( + DEFAULT_LOCK_TIMEOUT_MS, + DEFAULT_IDLE_TXN_TIMEOUT_MS, + DEFAULT_STATEMENT_TIMEOUT_MS, + ); assert_eq!(unset, defaults, "unset env must keep the defaults"); assert_eq!(overridden, (2000, 30000, 10000)); - assert_eq!(zero, (0, 0, 0), "explicit 0 must disable each timeout"); + // The overlay passes an explicit 0 through; the writer constructor then + // installs the maximum statement timeout for it (see + // `writer_statement_timeout_is_always_bounded`). + assert_eq!(zero, (0, 0, 0), "explicit 0 must pass through the overlay"); assert_eq!(junk, defaults, "junk env must keep the defaults"); } +/// No writer session is unbounded: `0` and values above the maximum clamp to +/// [`MAX_STATEMENT_TIMEOUT_MS`]; in-range values pass through unchanged. +#[test] +fn writer_statement_timeout_is_always_bounded() { + let effective = |statement_timeout_ms| { + DbConfig { + statement_timeout_ms, + ..DbConfig::default() + } + .effective_statement_timeout_ms() + }; + assert_eq!( + DbConfig::default().effective_statement_timeout_ms(), + 120_000 + ); + assert_eq!(effective(0), MAX_STATEMENT_TIMEOUT_MS); + assert_eq!( + effective(MAX_STATEMENT_TIMEOUT_MS + 1), + MAX_STATEMENT_TIMEOUT_MS + ); + assert_eq!(effective(u64::MAX), MAX_STATEMENT_TIMEOUT_MS); + assert_eq!(effective(30_000), 30_000); + assert_eq!(MAX_STATEMENT_TIMEOUT_MS, 300_000); +} + +/// Through the production constructors, every writer session carries the +/// clamped statement timeout and every replica session the 20 s read +/// deadline. Mutation oracles: dropping the clamp leaves `0`/`10min` on the +/// writer; dropping the reader `after_connect` leaves the server default (`0`). +#[tokio::test] +#[ignore = "requires Postgres"] +async fn session_statement_timeouts_bound_writer_and_reader_sessions() { + let url = crate::test_support::database_url(); + for (configured, writer_expected) in [ + (DEFAULT_STATEMENT_TIMEOUT_MS, "2min"), + (0, "5min"), + (600_000, "5min"), + (45_000, "45s"), + ] { + let db = Db::new(&DbConfig { + database_url: url.clone(), + read_database_url: Some(url.clone()), + max_connections: 1, + min_connections: 0, + statement_timeout_ms: configured, + ..DbConfig::default() + }) + .await + .expect("connect Db with reader"); + let writer: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&db.pool) + .await + .expect("writer timeout"); + assert_eq!(writer, writer_expected, "writer, configured {configured}"); + let reader_pool = db.read_pool.clone().expect("reader configured"); + let reader: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&reader_pool) + .await + .expect("reader timeout"); + assert_eq!(reader, crate::event::READ_STATEMENT_TIMEOUT, "reader"); + db.pool.close().await; + reader_pool.close().await; + } +} + /// The production writer constructor installs all three timeout GUCs, bounds /// ordinary lock waits, and exempts the intentional migration lock wait. #[tokio::test] @@ -3081,7 +3155,10 @@ async fn session_timeouts_install_through_db_new_and_bound_lock_waits() { .expect("read effective GUCs"); assert_eq!(lock, "500ms"); assert_eq!(idle, "1min"); - assert_eq!(statement, "0"); + assert_eq!( + statement, "5min", + "a disabled statement timeout installs the maximum" + ); let mut holder = db.pool.acquire().await.expect("holder connection"); sqlx::raw_sql("BEGIN; LOCK TABLE events IN ACCESS EXCLUSIVE MODE") @@ -3565,9 +3642,10 @@ async fn floor_guard_blocks_updates_that_move_rows_below_the_fence() { // ---- Statement-deadline cancellation vs. replica fallback ---------------- // // A replica read cancelled by `statement_timeout` (SQLSTATE 57014) must be -// surfaced, not re-run on the writer: the writer would be just as slow and -// the relay maps it to a distinct non-retried "query timed out". Every other -// replica failure keeps failing closed to the writer. +// surfaced, not re-run on the writer (policy on +// `DbError::is_statement_cancelled`); the relay maps it to a distinct +// non-retried "query timed out". Every other replica failure keeps failing +// closed to the writer. /// Writer and replica scratch DBs holding one root plus an `e`-tagged reply /// that mentions `mentioned` — the writer copy says "writer-reply" so a @@ -3579,6 +3657,7 @@ struct EtagFixture { replica: PgPool, rname: String, cid: CommunityId, + channel: Uuid, root_hex: String, mentioned_hex: String, } @@ -3623,6 +3702,7 @@ async fn e_tag_fixture(prefix: &str) -> EtagFixture { replica, rname, cid: CommunityId::from_uuid(community), + channel, root_hex: root.id.to_hex(), mentioned_hex: mentioned.public_key().to_hex(), } @@ -3683,10 +3763,10 @@ async fn replica_statement_cancel_propagates_without_writer_rerun() { ); let mut locker = fx.replica.begin().await.expect("begin locker"); - sqlx::query("LOCK TABLE events IN ACCESS EXCLUSIVE MODE") + sqlx::query("LOCK TABLE events, relay_members IN ACCESS EXCLUSIVE MODE") .execute(&mut *locker) .await - .expect("lock replica events"); + .expect("lock replica events and relay_members"); let cancelled = |r: crate::Result<()>, what: &str| match r { Err(e) => assert!(e.is_statement_cancelled(), "{what}: want 57014, got {e}"), @@ -3711,6 +3791,47 @@ async fn replica_statement_cancel_propagates_without_writer_rerun() { .map(drop), "get_events_by_ids_routed", ); + // The routed reads with their own SQL (feed, thread, channel window, + // relay membership) fall back to the writer on replica failure, but not + // on a cancel: the writer would repeat the same expensive work. + let mentioned = hex::decode(&fx.mentioned_hex).expect("hex"); + let channels = [fx.channel]; + cancelled( + db.query_feed_mentions_routed("t_cxl", fx.cid, &mentioned, &channels, None, 10) + .await + .map(drop), + "query_feed_mentions_routed", + ); + cancelled( + db.query_feed_needs_action_routed("t_cxl", fx.cid, &mentioned, &channels, None, 10) + .await + .map(drop), + "query_feed_needs_action_routed", + ); + cancelled( + db.query_feed_activity_routed("t_cxl", fx.cid, &channels, None, 10) + .await + .map(drop), + "query_feed_activity_routed", + ); + cancelled( + db.get_thread_replies(fx.cid, &root_id, None, 10, None) + .await + .map(drop), + "get_thread_replies", + ); + cancelled( + db.get_channel_window_with_session(fx.cid, fx.channel, 10, None, None) + .await + .map(drop), + "get_channel_window_with_session", + ); + cancelled( + db.is_relay_member(fx.cid, &fx.mentioned_hex) + .await + .map(drop), + "is_relay_member", + ); locker.rollback().await.expect("unlock"); drop(db); @@ -3773,6 +3894,120 @@ async fn read_session_statement_cancel_propagates_without_writer_rerun() { fx.drop().await; } +/// Cancels (57014) the backend on `db_name` parked behind a lock. The thread +/// window sets its own 1s `lock_timeout`, which would otherwise fire (55P03) +/// before any statement cap. +async fn cancel_lock_waiter(admin: &PgPool, db_name: &str) { + for _ in 0..150 { + let pid: Option = sqlx::query_scalar( + "SELECT pid FROM pg_stat_activity WHERE datname = $1 AND wait_event_type = 'Lock' LIMIT 1", + ) + .bind(db_name) + .fetch_optional(admin) + .await + .expect("poll lock waiters"); + if let Some(pid) = pid { + let cancelled: bool = sqlx::query_scalar("SELECT pg_cancel_backend($1)") + .bind(pid) + .fetch_one(admin) + .await + .expect("cancel lock waiter"); + assert!(cancelled, "pg_cancel_backend({pid}) refused"); + return; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + panic!("no lock waiter appeared on {db_name}"); +} + +/// The thread-window reads keep the same rule as the routed reads above: a +/// replica cancel surfaces 57014. The writer copy is unlocked, so a writer +/// re-run would succeed and fail the test. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn thread_window_statement_cancel_propagates_without_writer_rerun() { + use crate::thread_window::{AuxQuery, ScanBudget}; + let fx = e_tag_fixture("twcxl").await; + let db = fx.routed_db().await; + let request = buzz_core::thread_window::Request::parse(&serde_json::json!({ + "thread_window": true, "#h": [fx.channel], "#e": [fx.root_hex], + "kinds": [9], "limit": 10, + })) + .expect("thread window request"); + + // Healthy control: the window is served from the replica. + let (_, session) = db + .get_thread_window_with_session(fx.cid, &request, &mut ScanBudget::default()) + .await + .expect("healthy window"); + assert!(session.is_replica(), "fixture must route to replica"); + drop(session); + + let mut locker = fx.replica.begin().await.expect("begin locker"); + sqlx::query("LOCK TABLE events IN ACCESS EXCLUSIVE MODE") + .execute(&mut *locker) + .await + .expect("lock replica events"); + + let mut budget = ScanBudget::default(); + let (window, ()) = tokio::join!( + db.get_thread_window_with_session(fx.cid, &request, &mut budget), + cancel_lock_waiter(&fx.admin, &fx.rname), + ); + match window { + Err(e) => assert!(e.is_statement_cancelled(), "window: want 57014, got {e}"), + Ok(_) => panic!("window: replica cancel was re-run on the writer"), + } + + // Auxiliary scan on a proved replica session: a cancel neither re-runs on + // the writer nor degrades the session for the rest of the request. + let reader_tx = db + .read_pool + .as_ref() + .expect("reader pool") + .begin() + .await + .expect("begin reader tx"); + let mut session = ReadSession { + inner: ReadSessionInner::Replica { + tx: reader_tx, + writer: fx.writer.clone(), + }, + }; + let ReadSessionInner::Replica { tx, .. } = &mut session.inner else { + unreachable!() + }; + sqlx::query("SET LOCAL statement_timeout = '50ms'") + .execute(&mut **tx) + .await + .expect("cap session"); + let targets = [fx.root_hex.clone()]; + let accessible = [fx.channel]; + let aux = AuxQuery { + community: fx.cid, + targets: &targets, + kinds: &[7], + accessible: &accessible, + cursor: None, + }; + match session + .thread_window_aux(&aux, &mut ScanBudget::default()) + .await + { + Err(e) => assert!(e.is_statement_cancelled(), "aux: want 57014, got {e}"), + Ok(_) => panic!("aux: replica cancel was re-run on the writer"), + } + assert!( + session.is_replica(), + "aux: a cancel must not degrade the session" + ); + + locker.rollback().await.expect("unlock"); + drop(session); + drop(db); + fx.drop().await; +} + #[tokio::test] #[ignore = "requires Postgres"] async fn replica_non_cancel_failure_still_falls_back_to_writer() { diff --git a/crates/buzz-db/src/store/deletion.rs b/crates/buzz-db/src/store/deletion.rs index 2120df5a16..c5d6245127 100644 --- a/crates/buzz-db/src/store/deletion.rs +++ b/crates/buzz-db/src/store/deletion.rs @@ -1234,19 +1234,14 @@ impl DeletionStore { let mut row_counts = BTreeMap::new(); for table in &live_tables { let sql = format!("SELECT count(*)::BIGINT FROM {table} WHERE community_id = $1"); - let count: i64 = sqlx::query_scalar(AssertSqlSafe(sql)) - .bind(community.as_uuid()) - .fetch_one(&self.pool) - .await?; - row_counts.insert(table.clone(), count); + row_counts.insert(table.clone(), self.count_tenant_rows(sql, community).await?); } for (table, rows) in RELAY_ADMIN_TABLES { let sql = format!("SELECT count(*)::BIGINT {rows}"); - let count: i64 = sqlx::query_scalar(AssertSqlSafe(sql)) - .bind(community.as_uuid()) - .fetch_one(&self.pool) - .await?; - row_counts.insert((*table).to_owned(), count); + row_counts.insert( + (*table).to_owned(), + self.count_tenant_rows(sql, community).await?, + ); } Ok(SchemaManifest { scoped_tables: live_tables.into_iter().collect(), @@ -1255,6 +1250,24 @@ impl DeletionStore { }) } + /// One inventory count, in its own read-only transaction. + /// + /// A whole-tenant count on a large community can outrun the writer + /// statement timeout, so the transaction lifts it. One transaction per + /// count releases each table's `AccessShare` lock (and snapshot) as soon + /// as its count finishes, as autocommit counts did, instead of holding + /// every table until the last count. + async fn count_tenant_rows(&self, sql: String, community: CommunityId) -> Result { + let mut tx = self.pool.begin().await?; + lift_statement_timeout(&mut tx).await?; + let count = sqlx::query_scalar(AssertSqlSafe(sql)) + .bind(community.as_uuid()) + .fetch_one(&mut *tx) + .await?; + tx.rollback().await?; + Ok(count) + } + /// Freeze inventory and move submitted → inventoried atomically. pub async fn freeze_inventory( &self, @@ -2242,6 +2255,7 @@ impl DeletionStore { pub async fn purge_postgres(&self, token: &LeaseToken) -> Result> { let generation = require_fence_generation(token)?; let mut tx = self.pool.begin().await?; + lift_statement_timeout(&mut tx).await?; // Revalidate the exact catalog inside the purge transaction under the // shared schema/destruction lock. Migrations hold the exclusive // counterpart for their entire run, so no migration can commit a new @@ -2378,6 +2392,9 @@ impl DeletionStore { pub async fn verify_postgres_logically_deleted(&self, token: &LeaseToken) -> Result<()> { let generation = require_fence_generation(token)?; let mut tx = self.pool.begin().await?; + // One whole-tenant `EXISTS` per table, run right after the purge while + // the `community_id` index still points at dead tuples. + lift_statement_timeout(&mut tx).await?; // The absence proof is only as strong as the surface it iterates: // validate the live catalog under the shared schema/destruction lock // so a scoped table committed after the purge fails this stage closed @@ -3700,6 +3717,24 @@ async fn verify_lease_and_fence( } } +/// Disable `statement_timeout` for the rest of `tx`. +/// +/// Community deletion runs whole-tenant statements (per-table counts at +/// inventory, one `DELETE ... WHERE community_id` per table at purge, one +/// `EXISTS ... WHERE community_id` per table at logical verification) whose +/// runtime scales with the tenant. Under the always-on writer cap a large +/// tenant's purge would be cancelled (57014, transient), retried from +/// scratch, and eventually blocked. Like migrations and schema destruction, +/// these paths are operator-approved and fenced; `lock_timeout` and +/// `idle_in_transaction_session_timeout` still bound them. Transaction-local, +/// so the pooled session's cap returns at commit or rollback. +async fn lift_statement_timeout(tx: &mut Transaction<'_, Postgres>) -> Result<()> { + sqlx::query("SELECT set_config('statement_timeout', '0', true)") + .execute(&mut **tx) + .await?; + Ok(()) +} + async fn set_executor_gucs( tx: &mut Transaction<'_, Postgres>, community: CommunityId, @@ -7402,6 +7437,216 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); ); } + /// A scratch database bootstrapped from the desired-state schema, for + /// tests that alter shared objects (locks, session caps) and must not + /// disturb the shared test database. Returns the admin pool, the scratch + /// database name and its URL; drop with [`drop_scratch_database`]. + async fn scratch_desired_state_database(prefix: &str) -> (PgPool, String, String) { + let base_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); + let admin = PgPool::connect(&base_url) + .await + .expect("connect admin database"); + let probe_db = format!("{prefix}_{}", Uuid::new_v4().simple()); + sqlx::query(AssertSqlSafe(format!("CREATE DATABASE {probe_db}"))) + .execute(&admin) + .await + .expect("create probe database"); + let (base_prefix, _) = base_url.rsplit_once('/').expect("database url has a path"); + let probe_url = format!("{base_prefix}/{probe_db}"); + + let schema_sql = crate::test_support::desired_state_schema_sql(); + let bootstrap = PgPool::connect(&probe_url) + .await + .expect("connect probe database"); + sqlx::raw_sql(AssertSqlSafe(schema_sql)) + .execute(&bootstrap) + .await + .expect("apply desired-state schema"); + bootstrap.close().await; + (admin, probe_db, probe_url) + } + + async fn drop_scratch_database(admin: &PgPool, probe_db: &str) { + sqlx::query(AssertSqlSafe(format!( + "DROP DATABASE {probe_db} WITH (FORCE)" + ))) + .execute(admin) + .await + .expect("drop probe database"); + } + + /// Run `op` while another session holds `ACCESS EXCLUSIVE` on `events` + /// until `op` has waited on it past a 500ms writer statement cap. Under + /// the cap the wait itself is cancelled (57014); `lock_timeout` (5s by + /// default) is far enough away not to fire first. + async fn run_behind_events_lock_past_cap( + db: &Db, + admin: &PgPool, + probe_db: &str, + op: impl std::future::Future>, + ) -> Result { + let mut holder = db.pool.begin().await.expect("begin lock holder"); + sqlx::query("LOCK TABLE events IN ACCESS EXCLUSIVE MODE") + .execute(&mut *holder) + .await + .expect("lock events"); + let release = async { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let parked: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity \ + WHERE datname = $1 AND wait_event_type = 'Lock')", + ) + .bind(probe_db) + .fetch_one(admin) + .await + .expect("inspect lock waiters"); + if parked { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("deletion statement never waited on the events lock"); + // Keep it waiting past the 500ms cap before letting it through. + tokio::time::sleep(Duration::from_millis(1_000)).await; + holder.rollback().await.expect("release events lock"); + }; + let (result, ()) = tokio::join!(op, release); + result + } + + /// Community-deletion inventory, purge and logical-verification statements + /// outlive the always-on writer statement cap, through the production + /// paths, and only for their own transactions. Mutation oracles: drop the + /// `lift_statement_timeout` call in `count_tenant_rows` → inventory is + /// cancelled (57014) → RED; drop it in `purge_postgres` → purge is + /// cancelled → RED; drop it in `verify_postgres_logically_deleted` → + /// verification is cancelled → RED; lift with session scope → `SHOW` + /// after is not `500ms` → RED; lift to a bounded value (e.g. `'300000'`) + /// → `SHOW` inside the lifted transaction is not `0` → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deletion_whole_tenant_stages_outlive_the_writer_statement_timeout() { + let (admin, probe_db, probe_url) = + scratch_desired_state_database("buzz_deletion_cap").await; + let db = Db::new(&DbConfig { + database_url: probe_url, + max_connections: 5, + min_connections: 0, + statement_timeout_ms: 500, + ..DbConfig::default() + }) + .await + .expect("connect capped writer"); + let store = db.deletion_store(); + + // Control: the writer cap is live on an ordinary statement. + let error = sqlx::query("SELECT pg_sleep(1)") + .execute(&db.pool) + .await + .expect_err("writer cap must cancel the plain sleep"); + assert_eq!( + error.as_database_error().and_then(|e| e.code()).as_deref(), + Some("57014"), + "unexpected error: {error:?}" + ); + + // The lift is unbounded, not merely longer than the cap: a bounded + // ceiling would bring back the large-tenant purge block. + let mut lifted = db.pool.begin().await.expect("begin lifted tx"); + lift_statement_timeout(&mut lifted).await.expect("lift"); + let inside: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&mut *lifted) + .await + .expect("show lifted"); + assert_eq!(inside, "0", "the lift must disable the cap, not raise it"); + lifted.rollback().await.expect("rollback lifted tx"); + + let host = format!("deletion-cap-{}.example", Uuid::new_v4().simple()); + let community = db + .ensure_configured_community(&host) + .await + .expect("create community"); + let submitted = store + .submit(&host, "test-operator", Some("test deletion")) + .await + .expect("submit"); + let schema = run_behind_events_lock_past_cap( + &db, + &admin, + &probe_db, + store.inventory_schema(community.id), + ) + .await + .expect("inventory outlives the writer cap"); + let inventory = FrozenInventory { + schema, + storage: empty_storage_manifest(community.id), + }; + let request = store + .freeze_inventory(submitted.id, &inventory) + .await + .expect("freeze inventory"); + store + .approve(request.id, "approver", None) + .await + .expect("approve"); + let claim = store + .claim_specific(request.id, "executor", DEFAULT_LEASE_DURATION) + .await + .expect("claim") + .expect("won claim"); + store.begin_quiescing(&claim.lease).await.expect("quiesce"); + let generation = store.fence(&claim.lease).await.expect("fence"); + let token = LeaseToken { + fence_generation: Some(generation), + ..claim.lease + }; + store + .freeze_destructive_storage_manifest(&token, &inventory.storage) + .await + .expect("freeze destructive storage"); + store.mark_drained(&token).await.expect("drain"); + store + .mark_bindings_removed(&token, serde_json::json!({"keys": 0})) + .await + .expect("bindings"); + run_behind_events_lock_past_cap(&db, &admin, &probe_db, store.purge_postgres(&token)) + .await + .expect("purge outlives the writer cap"); + assert_eq!( + store.get(request.id).await.expect("request").stage, + DeletionStage::PostgresPurged + ); + store + .mark_cache_purged(&token, serde_json::json!({"keys": 0})) + .await + .expect("cache purged"); + run_behind_events_lock_past_cap( + &db, + &admin, + &probe_db, + store.verify_postgres_logically_deleted(&token), + ) + .await + .expect("logical verification outlives the writer cap"); + + for _ in 0..5 { + let restored: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&db.pool) + .await + .expect("show restored"); + assert_eq!(restored, "500ms", "the lift is transaction-local"); + } + + db.pool.close().await; + drop_scratch_database(&admin, &probe_db).await; + } + #[tokio::test] #[ignore = "requires Postgres"] async fn serving_lease_acquire_waits_for_fence_lock_then_refuses() { @@ -8350,29 +8595,8 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); #[tokio::test] #[ignore = "requires Postgres"] async fn desired_state_schema_bootstrap_progresses_beyond_fencing() { - let base_url = std::env::var("BUZZ_TEST_DATABASE_URL") - .or_else(|_| std::env::var("DATABASE_URL")) - .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); - let admin = PgPool::connect(&base_url) - .await - .expect("connect admin database"); - let probe_db = format!("buzz_desired_state_{}", Uuid::new_v4().simple()); - sqlx::query(AssertSqlSafe(format!("CREATE DATABASE {probe_db}"))) - .execute(&admin) - .await - .expect("create probe database"); - let (base_prefix, _) = base_url.rsplit_once('/').expect("database url has a path"); - let probe_url = format!("{base_prefix}/{probe_db}"); - - let schema_sql = crate::test_support::desired_state_schema_sql(); - let bootstrap = PgPool::connect(&probe_url) - .await - .expect("connect probe database"); - sqlx::raw_sql(AssertSqlSafe(schema_sql)) - .execute(&bootstrap) - .await - .expect("apply desired-state schema"); - bootstrap.close().await; + let (admin, probe_db, probe_url) = + scratch_desired_state_database("buzz_desired_state").await; let db = Db::new(&DbConfig { database_url: probe_url, @@ -8452,11 +8676,6 @@ DROP FUNCTION community_write_fence_excluded_table(NAME); assert_eq!(terminal.stage, DeletionStage::RetentionPending); db.pool.close().await; - sqlx::query(AssertSqlSafe(format!( - "DROP DATABASE {probe_db} WITH (FORCE)" - ))) - .execute(&admin) - .await - .expect("drop probe database"); + drop_scratch_database(&admin, &probe_db).await; } } diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index a53b82944b..c279e4831d 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -588,14 +588,7 @@ pub(crate) async fn query_events_on( } let mut qb = build_query_events_sql(q); - let rows = if q.e_tags.is_some() { - fetch_with_e_tag_deadline(conn, &mut qb).await? - } else { - qb.build() - .persistent(!has_tag_values(q)) - .fetch_all(&mut *conn) - .await? - }; + let rows = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q)).await?; let mut out = Vec::with_capacity(rows.len()); for row in rows { @@ -629,15 +622,9 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { "SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, \ e.sig, e.received_at, e.channel_id \ FROM events e \ - INNER JOIN event_mentions m \ - ON e.community_id = m.community_id AND e.id = m.event_id \ - WHERE e.community_id = ", + INNER JOIN event_mentions m ON ", ); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND m.community_id = "); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = "); - b.push_bind(p_hex.to_ascii_lowercase()); + push_mention_join(&mut b, q, p_hex); b } else { let mut b = QueryBuilder::new(if e_tag_fence { @@ -836,31 +823,84 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { qb } -/// Server-side deadline for e-tag (aux-closure / `#e`) reads, kept under the -/// desktop's 30 s request timeout so a runaway query is cancelled once no -/// client can still be waiting for it. It only ever tightens the effective -/// `statement_timeout`: a shorter operator cap (`BUZZ_DB_STATEMENT_TIMEOUT_MS`) -/// is kept, and a disabled (`0`) or longer one is lowered to 20 s. The limit is -/// per statement, not per request. -const E_TAG_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('statement_timeout', \ - CASE WHEN current_setting('statement_timeout')::interval = interval '0' \ - OR current_setting('statement_timeout')::interval > interval '20s' \ - THEN '20s' ELSE current_setting('statement_timeout') END, true)"; +/// `#p` reads join `event_mentions` by the full event key, including +/// `created_at`: mention rows copy the event's `created_at` at insert, so the +/// equality is exact and turns each joined row into one primary-key probe on +/// the right partition instead of a channel-index range scan. +/// +/// `since`/`until` are repeated on the mention side so the outer scan of +/// `idx_event_mentions_pubkey_created` is a range, not every mention of the +/// pubkey ever. Without them a busy agent reconnecting with an old `since` +/// produced (all of its mentions) × (channel rows since `since`) work; on +/// bb-public one such REQ ran for over 45 minutes. `until` is a superset bound +/// for the keyset cursor, whose exact `(created_at, id)` test stays on `e`. +fn push_mention_join(b: &mut QueryBuilder, q: &EventQuery, p_hex: &str) { + b.push( + "e.community_id = m.community_id AND e.id = m.event_id \ + AND e.created_at = m.event_created_at \ + WHERE e.community_id = ", + ); + b.push_bind(q.community_id.as_uuid()); + b.push(" AND m.community_id = "); + b.push_bind(q.community_id.as_uuid()); + b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = "); + b.push_bind(p_hex.to_ascii_lowercase()); + if let Some(s) = q.since { + b.push(" AND m.event_created_at >= ").push_bind(s); + } + if let Some(u) = q.until { + b.push(" AND m.event_created_at <= ").push_bind(u); + } + // Every mention insert copies the event's kind, so repeating the `kinds` + // filter on `m` drops no rows and lets the planner range-scan + // `idx_event_mentions_pubkey_kind_created`. + if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) { + b.push(" AND m.event_kind IN ("); + let mut sep = b.separated(", "); + for k in ks { + sep.push_bind(*k); + } + b.push(")"); + } +} -/// Run an e-tag read under [`E_TAG_STATEMENT_TIMEOUT_SQL`] without changing the +/// Server-side deadline for every generic event read ([`query_events_on`] and +/// [`count_events_on`]: REQ history, `/query`, `/count`, aux closures), kept +/// under the desktop's 30 s request timeout so a runaway query is cancelled +/// once no client can still be waiting for it. It only ever tightens the +/// effective `statement_timeout`: a shorter operator or caller cap is kept, +/// and a disabled (`0`) or longer one is lowered to 20 s. The limit is per +/// statement, not per request. +pub const READ_STATEMENT_TIMEOUT: &str = "20s"; + +/// Lower `statement_timeout` to `$1` unless it is already tighter; `$2` +/// selects transaction-local (`true`) or session (`false`) scope. +pub(crate) const TIGHTEN_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('statement_timeout', \ + CASE WHEN current_setting('statement_timeout')::interval = interval '0' \ + OR current_setting('statement_timeout')::interval > $1::text::interval \ + THEN $1::text ELSE current_setting('statement_timeout') END, $2)"; + +/// Run an event read under [`READ_STATEMENT_TIMEOUT`] without changing the /// deadline for anything else on `conn`. `begin()` opens a transaction on an /// autocommit writer connection or a savepoint inside the replica's read /// transaction; rolling it back (the read has no writes to keep) reverts the /// transaction-local `set_config` in both cases. -async fn fetch_with_e_tag_deadline( +async fn fetch_with_read_deadline( conn: &mut sqlx::PgConnection, qb: &mut QueryBuilder, + persistent: bool, ) -> Result> { let mut tx = sqlx::Connection::begin(&mut *conn).await?; - sqlx::query(E_TAG_STATEMENT_TIMEOUT_SQL) + sqlx::query(TIGHTEN_STATEMENT_TIMEOUT_SQL) + .bind(READ_STATEMENT_TIMEOUT) + .bind(true) .execute(&mut *tx) .await?; - let rows = qb.build().fetch_all(&mut *tx).await?; + let rows = qb + .build() + .persistent(persistent) + .fetch_all(&mut *tx) + .await?; tx.rollback().await?; Ok(rows) } @@ -1059,18 +1099,27 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer return Ok(0); } + let mut qb = count_events_sql(q); + // Same transaction-local deadline as `query_events_on`, so a COUNT cannot + // stall indefinitely either. + let row = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q)) + .await? + .into_iter() + .next() + .ok_or_else(|| DbError::InvalidData("COUNT returned no rows".to_string()))?; + let cnt: i64 = row.try_get("cnt")?; + + Ok(cnt) +} + +/// SQL for [`count_events_on`], after its match-nothing early returns. +fn count_events_sql(q: &EventQuery) -> QueryBuilder { let mut qb: QueryBuilder = if let Some(ref p_hex) = q.p_tag_hex { let mut b = QueryBuilder::new( "SELECT COUNT(*) as cnt FROM events e \ - INNER JOIN event_mentions m \ - ON e.community_id = m.community_id AND e.id = m.event_id \ - WHERE e.community_id = ", + INNER JOIN event_mentions m ON ", ); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND m.community_id = "); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = "); - b.push_bind(p_hex.to_ascii_lowercase()); + push_mention_join(&mut b, q, p_hex); b } else { let mut b = QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE community_id = "); @@ -1195,22 +1244,7 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer } } - let row = if q.e_tags.as_deref().is_some_and(|e| !e.is_empty()) { - // Run under the same transaction-local deadline as `query_events_on` - // so a COUNT over a long thread cannot stall indefinitely either. - let rows = fetch_with_e_tag_deadline(conn, &mut qb).await?; - rows.into_iter() - .next() - .ok_or_else(|| DbError::InvalidData("COUNT returned no rows".to_string()))? - } else { - qb.build() - .persistent(!has_tag_values(q)) - .fetch_one(&mut *conn) - .await? - }; - let cnt: i64 = row.try_get("cnt")?; - - Ok(cnt) + qb } /// SQL predicate over an `events` row for coordinates that carry no @@ -2332,8 +2366,8 @@ impl Db { Self::record_route(path, "replica", reason); Ok(events) } - // A cancelled statement (timeout) would be just as slow on the - // writer; surface it instead of doubling the cost. + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. Err(e) if e.is_statement_cancelled() => { Self::record_route(path, "replica", "statement_cancelled"); Err(e) @@ -2391,8 +2425,8 @@ impl Db { Self::record_route(path, "replica", reason); Ok(events) } - // A cancelled statement (timeout) would be just as slow on the - // writer; surface it instead of doubling the cost. + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. Err(e) if e.is_statement_cancelled() => { Self::record_route(path, "replica", "statement_cancelled"); Err(e) @@ -2455,8 +2489,8 @@ impl Db { Self::record_route(path, "replica", reason); Ok(count) } - // A cancelled statement (timeout) would be just as slow on the - // writer; surface it instead of doubling the cost. + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. Err(e) if e.is_statement_cancelled() => { Self::record_route(path, "replica", "statement_cancelled"); Err(e) @@ -2714,8 +2748,8 @@ impl Db { Self::record_route(path, "replica", reason); Ok(events) } - // A cancelled statement (timeout) would be just as slow on the - // writer; surface it instead of doubling the cost. + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. Err(e) if e.is_statement_cancelled() => { Self::record_route(path, "replica", "statement_cancelled"); Err(e) @@ -2856,6 +2890,44 @@ mod e_tag_filter_shape_tests { let qb = build_query_events_sql(&q); assert!(!qb.sql().as_str().contains("MATERIALIZED")); } + + /// `#p` reads bound the mention side (bb-public runaway REQ): the join + /// carries the event's `created_at`, and `since`/`until` also bound + /// `m.event_created_at`, for both the page SELECT and the COUNT. + #[test] + fn p_tag_query_bounds_mention_side_by_time() { + let mut q = EventQuery::for_community(CommunityId::from_uuid(uuid::Uuid::nil())); + q.p_tag_hex = Some("ab".repeat(32)); + q.channel_id = Some(uuid::Uuid::nil()); + q.since = Some(Utc::now()); + q.until = Some(Utc::now()); + q.kinds = Some(vec![9]); + let select = build_query_events_sql(&q).sql().as_str().to_owned(); + let count = count_events_sql(&q).sql().as_str().to_owned(); + for (label, sql) in [("select", select), ("count", count)] { + assert!( + sql.contains("AND e.created_at = m.event_created_at"), + "{label}: {sql}" + ); + assert!( + sql.contains("AND m.event_created_at >= $"), + "{label}: {sql}" + ); + assert!( + sql.contains("AND m.event_created_at <= $"), + "{label}: {sql}" + ); + assert!(sql.contains("AND m.event_kind IN ($"), "{label}: {sql}"); + } + + q.since = None; + q.until = None; + q.kinds = None; + let unbounded = build_query_events_sql(&q).sql().as_str().to_owned(); + assert!(!unbounded.contains("m.event_created_at >=")); + assert!(!unbounded.contains("m.event_created_at <=")); + assert!(!unbounded.contains("m.event_kind")); + } } #[cfg(test)] @@ -2949,11 +3021,11 @@ mod postgres_tests { assert!(prepared_like(&mut conn, "%FROM events%").await > 0); } - /// Effective `statement_timeout` observed *inside* the e-tag deadline, and + /// Effective `statement_timeout` observed *inside* the read deadline, and /// the value restored on `conn` afterwards. async fn deadline_inside_and_after(conn: &mut sqlx::PgConnection) -> (String, String) { let mut qb = QueryBuilder::new("SELECT current_setting('statement_timeout')"); - let rows = fetch_with_e_tag_deadline(conn, &mut qb) + let rows = fetch_with_read_deadline(conn, &mut qb, true) .await .expect("deadline read"); let inside: String = rows[0].get(0); @@ -3105,7 +3177,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn e_tag_deadline_only_tightens_and_restores_timeout() { + async fn read_deadline_only_tightens_and_restores_timeout() { let pool = setup_pool().await; // (operator setting, expected effective value inside the read) let cases = [("0", "20s"), ("5s", "5s"), ("1min", "20s")]; @@ -3141,7 +3213,7 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] - async fn e_tag_deadline_restores_timeout_after_error() { + async fn read_deadline_restores_timeout_after_error() { let pool = setup_pool().await; let mut conn = pool.acquire().await.expect("acquire"); sqlx::query("SET statement_timeout = '1min'") @@ -3150,7 +3222,7 @@ mod postgres_tests { .expect("set session timeout"); // Autocommit connection (writer path). let mut qb = QueryBuilder::new("SELECT 1/0"); - let err = fetch_with_e_tag_deadline(&mut conn, &mut qb).await; + let err = fetch_with_read_deadline(&mut conn, &mut qb, true).await; assert!(err.is_err(), "division by zero must fail"); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *conn) @@ -3162,7 +3234,7 @@ mod postgres_tests { // the failed savepoint must roll back and leave the outer tx usable. let mut tx = sqlx::Connection::begin(&mut *conn).await.expect("begin"); let mut qb = QueryBuilder::new("SELECT 1/0"); - let err = fetch_with_e_tag_deadline(&mut tx, &mut qb).await; + let err = fetch_with_read_deadline(&mut tx, &mut qb, true).await; assert!(err.is_err(), "division by zero must fail"); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *tx) @@ -3176,12 +3248,13 @@ mod postgres_tests { .expect("reset"); } - /// Pins the dispatch: an e-tag read through the production + /// Pins the dispatch: a plain (non-`#e`) read — the agent `#p` + `#h` + + /// `since` REQ shape that ran away on bb-public — through the production /// `query_events_on` is cancelled at the 20 s deadline even with the /// session timeout disabled, while `events` is locked. (Slow by design.) #[tokio::test] #[ignore = "requires Postgres"] - async fn e_tag_read_is_cancelled_at_deadline_when_timeout_disabled() { + async fn plain_read_is_cancelled_at_deadline_when_timeout_disabled() { let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let mut locker = pool.begin().await.expect("begin locker"); @@ -3196,7 +3269,10 @@ mod postgres_tests { .await .expect("disable timeout"); let mut q = EventQuery::for_community(community); - q.e_tags = Some(vec!["00".repeat(32)]); + q.p_tag_hex = Some("ab".repeat(32)); + q.channel_id = Some(Uuid::new_v4()); + q.kinds = Some(vec![9]); + q.since = Some(Utc::now() - chrono::Duration::days(30)); let started = std::time::Instant::now(); let result = tokio::time::timeout( std::time::Duration::from_secs(40), @@ -3215,13 +3291,13 @@ mod postgres_tests { .expect("reset"); } - /// Pins the dispatch: a COUNT with an e-tag filter through the production + /// Pins the dispatch: a plain (non-`#e`) COUNT through the production /// `count_events_on` is cancelled at the 20 s deadline even with the /// session timeout disabled, while `events` is locked. Connection and /// setting are restored afterwards. (Slow by design.) #[tokio::test] #[ignore = "requires Postgres"] - async fn count_events_cancels_at_e_tag_deadline_when_timeout_disabled() { + async fn count_events_cancels_at_read_deadline_when_timeout_disabled() { let pool = setup_pool().await; let community = CommunityId::from_uuid(make_test_community(&pool).await); let mut locker = pool.begin().await.expect("begin locker"); @@ -3236,7 +3312,7 @@ mod postgres_tests { .await .expect("disable timeout"); let mut q = EventQuery::for_community(community); - q.e_tags = Some(vec!["00".repeat(32)]); + q.kinds = Some(vec![9]); let started = std::time::Instant::now(); let result = tokio::time::timeout( std::time::Duration::from_secs(40), @@ -3262,6 +3338,111 @@ mod postgres_tests { .expect("reset"); } + /// The time-bounded `#p` join returns exactly the mentioned events in + /// the window, through the production mention insert, for both the page + /// and the COUNT, including the keyset cursor (`until` + `before_id`). + /// + /// Every bound sits exactly on a mentioned row's second, so an + /// off-by-one on the mention side (`>=`→`>` or `<=`→`<` in + /// `push_mention_join`) drops a row and fails an assertion. The cursor + /// page includes a second mentioned event in the cursor's own second + /// whose id sorts after the cursor id: `until` must be an inclusive + /// superset bound on `m`, leaving the exact cursor test to `e`. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn p_tag_read_with_mention_bounds_keeps_results() { + let pool = setup_pool().await; + let community_uuid = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_uuid); + let channel = make_test_channel(&pool, community_uuid, None).await; + let author = Keys::generate(); + let mentioned = Keys::generate().public_key().to_hex(); + let now = nostr::Timestamp::now().as_secs(); + let at = |secs: u64| DateTime::from_timestamp(secs as i64, 0).unwrap(); + let mut events = Vec::new(); + for (i, created) in [now - 300, now - 200, now - 100, now - 100] + .into_iter() + .enumerate() + { + let event = EventBuilder::new(Kind::Custom(9), format!("mention {i}")) + .tags([Tag::parse(["p", &mentioned]).unwrap()]) + .custom_created_at(nostr::Timestamp::from(created)) + .sign_with_keys(&author) + .unwrap(); + insert_event(&pool, community, &event, Some(channel)) + .await + .unwrap(); + crate::insert_mentions(&pool, community, &event, Some(channel)) + .await + .unwrap(); + events.push((created, event.id.as_bytes().to_vec())); + } + // Production order: created_at DESC, id ASC. `newest` is the + // same-second pair at now-100; `newest[0]` becomes the cursor. + events.sort_by(|a, b| b.0.cmp(&a.0).then_with(|| a.1.cmp(&b.1))); + let ids: Vec> = events.into_iter().map(|(_, id)| id).collect(); + let (newest, mid, oldest) = (&ids[..2], &ids[2], &ids[3]); + let unmentioned = EventBuilder::new(Kind::Custom(9), "no mention") + .custom_created_at(nostr::Timestamp::from(now - 150)) + .sign_with_keys(&author) + .unwrap(); + insert_event(&pool, community, &unmentioned, Some(channel)) + .await + .unwrap(); + + let read = |q: EventQuery| { + let pool = pool.clone(); + async move { + let rows = query_events(&pool, &q).await.unwrap(); + let count = count_events(&pool, &q).await.unwrap(); + let ids: Vec> = rows + .iter() + .map(|row| row.event.id.as_bytes().to_vec()) + .collect(); + (ids, count) + } + }; + let mut q = EventQuery::for_community(community); + q.p_tag_hex = Some(mentioned.to_ascii_uppercase()); + q.channel_id = Some(channel); + q.kinds = Some(vec![9]); + + assert_eq!(read(q.clone()).await, (ids.clone(), 4), "unbounded"); + + // `since` exactly on `mid`'s second keeps it (inclusive). + q.since = Some(at(now - 200)); + assert_eq!( + read(q.clone()).await, + (vec![newest[0].clone(), newest[1].clone(), mid.clone()], 3), + "since on a row's second" + ); + + // `since` == `until` on `mid`'s second keeps exactly `mid`. + q.until = Some(at(now - 200)); + assert_eq!( + read(q.clone()).await, + (vec![mid.clone()], 1), + "window collapsed onto one row's second" + ); + + // Keyset cursor at the newest second: the same-second sibling after + // the cursor id must survive the mention-side `until` bound. + q.since = None; + q.until = Some(at(now - 100)); + q.before_id = Some(newest[0].clone()); + q.limit = Some(10); + let rows = query_events(&pool, &q).await.unwrap(); + let page: Vec> = rows + .iter() + .map(|row| row.event.id.as_bytes().to_vec()) + .collect(); + assert_eq!( + page, + vec![newest[1].clone(), mid.clone(), oldest.clone()], + "cursor page keeps the same-second sibling" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn event_insert_in_existing_transaction_rolls_back_with_caller() { diff --git a/crates/buzz-db/src/store/feed.rs b/crates/buzz-db/src/store/feed.rs index 5819136fe6..1094c948e8 100644 --- a/crates/buzz-db/src/store/feed.rs +++ b/crates/buzz-db/src/store/feed.rs @@ -96,6 +96,7 @@ fn build_mentions_query( let mut qb: QueryBuilder = QueryBuilder::new(format!( "SELECT {EVENT_COLS} FROM events e \ INNER JOIN event_mentions m ON e.community_id = m.community_id AND e.id = m.event_id \ + AND e.created_at = m.event_created_at \ WHERE e.community_id = " )); qb.push_bind(*community.as_uuid()); @@ -185,6 +186,7 @@ fn build_needs_action_query( let mut qb: QueryBuilder = QueryBuilder::new(format!( "SELECT {EVENT_COLS} FROM events e \ INNER JOIN event_mentions m ON e.community_id = m.community_id AND e.id = m.event_id \ + AND e.created_at = m.event_created_at \ WHERE e.community_id = " )); qb.push_bind(*community.as_uuid()); @@ -379,6 +381,12 @@ impl Db { Self::record_route(path, "replica", reason); Ok(events) } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + Err(e) + } Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); @@ -464,6 +472,12 @@ impl Db { Self::record_route(path, "replica", reason); Ok(events) } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + Err(e) + } Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); @@ -539,6 +553,12 @@ impl Db { Self::record_route(path, "replica", reason); Ok(events) } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + Err(e) + } Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); @@ -1091,6 +1111,10 @@ mod postgres_tests { sql.contains("INNER JOIN event_mentions m ON e.community_id = m.community_id AND e.id = m.event_id"), "mentions must join event_mentions on the composite tenant/event key: {sql}" ); + assert!( + sql.contains("AND e.id = m.event_id AND e.created_at = m.event_created_at"), + "mentions must carry the mention's created_at so each probe prunes to one events partition: {sql}" + ); assert!( sql.contains("WHERE e.community_id = "), "mentions feed must scope events to the tenant community: {sql}" @@ -1121,6 +1145,10 @@ mod postgres_tests { sql.contains("INNER JOIN event_mentions m ON e.community_id = m.community_id AND e.id = m.event_id"), "needs_action must join event_mentions on the composite tenant/event key: {sql}" ); + assert!( + sql.contains("AND e.id = m.event_id AND e.created_at = m.event_created_at"), + "needs_action must carry the mention's created_at so each probe prunes to one events partition: {sql}" + ); assert!( sql.contains("WHERE e.community_id = "), "needs_action feed must scope events to the tenant community: {sql}" diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 0b1e465712..6bc4de1a95 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -1232,6 +1232,12 @@ impl Db { Self::record_route(path, "replica", reason); Ok(is_member) } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + Err(e) + } Err(e) => { tracing::warn!(path, "replica read failed; re-running on writer: {e}"); Self::record_route(path, "writer", "replica_error"); diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 9a18554efe..0e4c9d9375 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -929,6 +929,12 @@ impl Db { // discarded, so counting it would overstate offload. Self::record_route("thread_eof", "writer", "stale"); } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + return Err(e); + } Err(e) => { // Mid-request replica failure (e.g. a hot-standby // recovery conflict) fails closed to the writer. @@ -1059,8 +1065,9 @@ impl Db { /// `BUZZ_REPLICA_HEAD_MAX_AGE_SECS` until it has, proven by a /// post-then-immediately-refetch test. /// - /// Every failure fails closed to the writer and is recorded in - /// `buzz_db_route_decision`. + /// 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`. #[datastore_span(name = "get_channel_window", system = "postgresql")] pub async fn get_channel_window_with_session( &self, @@ -1106,6 +1113,12 @@ impl Db { }, )); } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(e) if e.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + return Err(e); + } Err(e) => { // A mid-request replica failure (e.g. a hot-standby // recovery conflict cancelling the held snapshot) diff --git a/crates/buzz-db/src/store/thread_window.rs b/crates/buzz-db/src/store/thread_window.rs index 659de340a6..926ffae8df 100644 --- a/crates/buzz-db/src/store/thread_window.rs +++ b/crates/buzz-db/src/store/thread_window.rs @@ -251,6 +251,12 @@ impl Db { Err(error @ (DbError::InvalidData(_) | DbError::ThreadWindowBudgetExceeded(_))) => { return Err(error); } + // Surface a cancel instead of re-running it on the writer; see + // `DbError::is_statement_cancelled` for the trade-off. + Err(error) if error.is_statement_cancelled() => { + Self::record_route(path, "replica", "statement_cancelled"); + return Err(error); + } Err(error) => { tracing::warn!(%error, path, "replica thread window failed; re-running on writer"); Self::record_route(path, "writer", "replica_error"); @@ -436,6 +442,8 @@ async fn writer_aux( impl ReadSession { /// Strict NIP-CW thread-mode auxiliary scan on the same proved snapshot, with the /// existing permanent writer degradation on mid-request replica failure. + /// Exception: a statement cancel (57014) is returned and the session stays + /// on the replica. #[datastore_span(name = "thread_window_aux", system = "postgresql")] pub async fn thread_window_aux( &mut self, @@ -448,6 +456,10 @@ impl ReadSession { Err(error @ (DbError::InvalidData(_) | DbError::ThreadWindowBudgetExceeded(_))) => { return Err(error); } + // Surface a cancel instead of re-running it on the writer (see + // `DbError::is_statement_cancelled`) and keep the proved + // replica session. + Err(error) if error.is_statement_cancelled() => return Err(error), Err(error) => { tracing::warn!(%error, "thread auxiliary read failed; degrading to writer"); metrics::counter!("buzz_db_read_session_degraded").increment(1); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 2a0ab1c7de..fa181f2a47 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -3181,7 +3181,10 @@ mod tests { .expect("read effective audit writer GUCs"); assert_eq!(lock, "500ms"); assert_eq!(idle, "1min"); - assert_eq!(statement, "0"); + assert_eq!( + statement, "5min", + "a disabled statement timeout installs the maximum" + ); let push: String = sqlx::query_scalar("SELECT current_setting('buzz.push_enabled')") .fetch_one(&pool) .await diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 1a20aa6bc4..b272fb40f0 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -2417,13 +2417,13 @@ async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::Ne metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64()); return; } - Err(error) if is_lock_timeout(&error) => { + Err(error) if is_retryable_append_failure(&error) => { retries += 1; metrics::counter!("buzz_audit_log_lock_retries_total").increment(1); tracing::warn!( retries, retry_delay_ms, - "Audit advisory lock timed out; preserving entry for retry" + "Audit append timed out; preserving entry for retry: {error}" ); tokio::time::sleep(std::time::Duration::from_millis(retry_delay_ms)).await; retry_delay_ms = (retry_delay_ms * 2).min(1_000); @@ -2437,18 +2437,23 @@ async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::Ne } } -/// Whether an audit append failed on `lock_timeout` (`55P03`), either on the -/// audit-chain lock or on the community admission lock taken before it. Both -/// are transient; the entry is kept and retried. Admission refusal -/// (`AccessDenied`) is terminal: a quiescing or fenced community refuses for -/// good. -fn is_lock_timeout(error: &buzz_audit::AuditError) -> bool { +/// Whether an audit append was rolled back by a timeout: `lock_timeout` +/// (`55P03`) while waiting on the audit-chain lock or the community admission +/// lock taken before it, or `statement_timeout` (`57014`) on any statement of +/// the append transaction — a lock wait when the lock timeout is disabled or +/// longer than the writer cap, or a slow head read or insert. The whole +/// append is one transaction, so either code means nothing was written; the +/// entry is kept and retried. `buzz_audit_log_lock_retries_total` keeps its +/// name for dashboard continuity and counts both codes. +/// Admission refusal (`AccessDenied`) is terminal: a quiescing or fenced +/// community refuses for good. +fn is_retryable_append_failure(error: &buzz_audit::AuditError) -> bool { let database_error = match error { buzz_audit::AuditError::Database(sqlx::Error::Database(e)) | buzz_audit::AuditError::Admission(buzz_db::DbError::Sqlx(sqlx::Error::Database(e))) => e, _ => return false, }; - database_error.code().as_deref() == Some("55P03") + matches!(database_error.code().as_deref(), Some("55P03" | "57014")) } impl std::fmt::Debug for AppState { @@ -2967,15 +2972,41 @@ pub(crate) mod tests { CommunityDeletion, } + /// Which timeout cancels the audit append while it waits on the lock. + #[derive(Clone, Copy)] + enum AuditWaitTimeout { + /// `lock_timeout` (55P03). + Lock, + /// `statement_timeout` with `lock_timeout` disabled (57014), as under + /// `BUZZ_DB_LOCK_TIMEOUT_MS=0` with the writer statement cap. + Statement, + } + async fn audit_worker_retries_lock_timeout_until_original_entry_is_appended_once() { - assert_audit_worker_retries_lock_timeout(HeldAuditLock::AuditChain).await; + assert_audit_worker_retries_lock_timeout(HeldAuditLock::AuditChain, AuditWaitTimeout::Lock) + .await; } async fn audit_worker_retries_admission_lock_timeout_until_original_entry_is_appended_once() { - assert_audit_worker_retries_lock_timeout(HeldAuditLock::CommunityDeletion).await; + assert_audit_worker_retries_lock_timeout( + HeldAuditLock::CommunityDeletion, + AuditWaitTimeout::Lock, + ) + .await; } - async fn assert_audit_worker_retries_lock_timeout(held: HeldAuditLock) { + async fn audit_worker_retries_statement_timeout_until_original_entry_is_appended_once() { + assert_audit_worker_retries_lock_timeout( + HeldAuditLock::AuditChain, + AuditWaitTimeout::Statement, + ) + .await; + } + + async fn assert_audit_worker_retries_lock_timeout( + held: HeldAuditLock, + timeout: AuditWaitTimeout, + ) { let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL"); let observer = sqlx::PgPool::connect(&database_url) .await @@ -2987,11 +3018,18 @@ pub(crate) mod tests { .after_connect(move |conn, _meta| { let application_name = hook_application_name.clone(); Box::pin(async move { + let (lock_timeout, statement_timeout) = match timeout { + AuditWaitTimeout::Lock => ("100", "0"), + AuditWaitTimeout::Statement => ("0", "100"), + }; sqlx::query( "SELECT set_config('application_name', $1, false), \ - set_config('lock_timeout', '100', false)", + set_config('lock_timeout', $2, false), \ + set_config('statement_timeout', $3, false)", ) .bind(application_name) + .bind(lock_timeout) + .bind(statement_timeout) .execute(&mut *conn) .await?; Ok(()) @@ -3132,6 +3170,13 @@ pub(crate) mod tests { super::audit_worker_retries_admission_lock_timeout_until_original_entry_is_appended_once() .await; } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn audit_worker_retries_statement_timeout_until_original_entry_is_appended_once() { + super::audit_worker_retries_statement_timeout_until_original_entry_is_appended_once() + .await; + } } #[test]