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]