diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index c9e1c6dc36..c5d2f10243 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -1008,6 +1008,120 @@ async fn read_falls_back_to_writer_when_no_replica_configured() { assert!(db.read_pool_stats().is_none()); } +/// A match-nothing read is answered before any pool acquire, on every query +/// entry point, and an invalid query that would also match nothing still +/// returns its validation error. Replica routing is on and the fence is open, +/// so a routed read that reached `route_read` would try the reader first. Both +/// pools point at a closed port, and the `limit: 1` control proves that a real +/// page does try to acquire, from the reader and then the writer. +#[tokio::test] +async fn match_nothing_reads_take_no_connection() { + use metrics_util::debugging::DebuggingRecorder; + + let dead_pool = || { + sqlx::postgres::PgPoolOptions::new() + .acquire_timeout(std::time::Duration::from_millis(200)) + .connect_lazy("postgres://unused@127.0.0.1:1/unused") // sadscan:disable np.postgres.1 + .expect("lazy pool") + }; + let mut db = Db::from_pools(dead_pool(), dead_pool()); + db.set_replica_read_max_age_for_tests(Some(std::time::Duration::from_secs(3600))); + db.fence().force_open_for_tests(chrono::Utc::now()); + let community = CommunityId::from_uuid(Uuid::new_v4()); + let page = |limit| { + let mut q = EventQuery::for_community(community); + q.limit = Some(limit); + q + }; + let with = |edit: fn(&mut EventQuery)| { + let mut q = page(10); + edit(&mut q); + q + }; + let empty_pages = [ + ("limit 0", page(0)), + ("max_limit 0", with(|q| q.max_limit = Some(0))), + ("empty kinds", with(|q| q.kinds = Some(vec![]))), + ("empty authors", with(|q| q.authors = Some(vec![]))), + ("empty ids", with(|q| q.ids = Some(vec![]))), + ("empty #e", with(|q| q.e_tags = Some(vec![]))), + ]; + // Validation runs before the match-nothing check, so the error wins. + let invalid = [ + ( + "before_id without until", + with(|q| { + q.limit = Some(0); + q.before_id = Some(vec![0; 32]); + }), + ), + ( + "global_only with channel_id", + with(|q| { + q.kinds = Some(vec![]); + q.global_only = true; + q.channel_id = Some(Uuid::nil()); + }), + ), + ]; + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let _guard = metrics::set_default_local_recorder(&recorder); + let acquiring_roles = || { + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(key, _, _, _)| key.key().name() == "buzz_db_pool_acquire_attempts_total") + .filter_map(|(key, _, _, _)| { + key.key() + .labels() + .find(|label| label.key() == "pool_role") + .map(|label| label.value().to_owned()) + }) + .collect::>() + }; + + for (case, q) in &empty_pages { + let results = [ + db.query_events(q).await, + db.query_events_routed("test", q).await, + db.query_events_routed_bounded("test", q).await, + ]; + for (entry, result) in ["writer", "routed", "routed bounded"].iter().zip(results) { + let events = result.unwrap_or_else(|e| panic!("{case} via {entry}: {e}")); + assert!(events.is_empty(), "{case} via {entry} must match nothing"); + } + } + for (case, q) in &invalid { + let results = [ + db.query_events(q).await, + db.query_events_routed("test", q).await, + db.query_events_routed_bounded("test", q).await, + ]; + for (entry, result) in ["writer", "routed", "routed bounded"].iter().zip(results) { + assert!( + matches!(result, Err(DbError::InvalidData(_))), + "{case} via {entry} must fail validation, got {result:?}" + ); + } + } + assert!( + acquiring_roles().is_empty(), + "a match-nothing read must not try to acquire: {:?}", + acquiring_roles() + ); + + db.query_events_routed("test", &page(1)) + .await + .expect_err("a real page needs the (unreachable) pools"); + assert_eq!( + acquiring_roles(), + ["reader", "writer"].map(str::to_owned).into(), + "the control must have tried the reader, then the writer" + ); +} + #[test] fn read_budget_zero_disables_and_large_values_clamp_to_staleness() { assert_eq!(read_budget_from_ms(0), None, "0 = bounded routing off"); diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index c279e4831d..151151f738 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -548,6 +548,9 @@ pub(crate) async fn query_events_with_operation( q: &EventQuery, operation: crate::observability::WriterOperation, ) -> Result> { + if let Some(answer) = answer_without_db(q) { + return answer; + } let mut conn = crate::observability::acquire_writer(pool, operation).await?; query_events_on(&mut conn, q).await } @@ -559,32 +562,8 @@ pub(crate) async fn query_events_on( conn: &mut sqlx::PgConnection, q: &EventQuery, ) -> Result> { - // Composite cursor requires both halves. - if q.before_id.is_some() && q.until.is_none() { - return Err(DbError::InvalidData( - "before_id requires until to be set".to_string(), - )); - } - - // global_only and channel_id are mutually exclusive. - if q.global_only && q.channel_id.is_some() { - return Err(DbError::InvalidData( - "global_only and channel_id are mutually exclusive".to_string(), - )); - } - - // Empty list means "match nothing" — return empty immediately. - if q.kinds.as_deref().is_some_and(|k| k.is_empty()) { - return Ok(vec![]); - } - if q.authors.as_deref().is_some_and(|a| a.is_empty()) { - return Ok(vec![]); - } - if q.ids.as_deref().is_some_and(|i| i.is_empty()) { - return Ok(vec![]); - } - if q.e_tags.as_deref().is_some_and(|e| e.is_empty()) { - return Ok(vec![]); + if let Some(answer) = answer_without_db(q) { + return answer; } let mut qb = build_query_events_sql(q); @@ -599,10 +578,44 @@ pub(crate) async fn query_events_on( Ok(out) } -/// SQL for [`query_events_on`], after its match-nothing early returns. +/// The result of `q` when its shape alone decides it: an invalid query, or one +/// that matches nothing. Every entry point checks this before it takes a +/// connection, so a match-nothing read (a NIP-01 `limit: 0` live REQ, say) +/// costs no pool acquire. +pub(crate) fn answer_without_db(q: &EventQuery) -> Option>> { + // Composite cursor requires both halves. + if q.before_id.is_some() && q.until.is_none() { + return Some(Err(DbError::InvalidData( + "before_id requires until to be set".to_string(), + ))); + } + + // global_only and channel_id are mutually exclusive. + if q.global_only && q.channel_id.is_some() { + return Some(Err(DbError::InvalidData( + "global_only and channel_id are mutually exclusive".to_string(), + ))); + } + + // A zero page, or an empty list, means "match nothing". + let matches_nothing = page_limit(q) == 0 + || q.kinds.as_deref().is_some_and(|k| k.is_empty()) + || q.authors.as_deref().is_some_and(|a| a.is_empty()) + || q.ids.as_deref().is_some_and(|i| i.is_empty()) + || q.e_tags.as_deref().is_some_and(|e| e.is_empty()); + matches_nothing.then(|| Ok(vec![])) +} + +/// The `LIMIT` [`query_events_on`] applies: the requested page, clamped. +fn page_limit(q: &EventQuery) -> i64 { + q.limit + .unwrap_or(100) + .min(q.max_limit.unwrap_or(DEFAULT_MAX_PAGE_LIMIT)) +} + +/// SQL for [`query_events_on`], after [`answer_without_db`]. fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { - let clamp = q.max_limit.unwrap_or(DEFAULT_MAX_PAGE_LIMIT); - let limit_val = q.limit.unwrap_or(100).min(clamp); + let limit_val = page_limit(q); let offset_val = q.offset.unwrap_or(0); // e-tag reads filter inside a MATERIALIZED CTE and sort/limit outside it. @@ -2351,6 +2364,9 @@ impl Db { path: &'static str, q: &EventQuery, ) -> Result> { + if let Some(answer) = crate::event::answer_without_db(q) { + return answer; + } let predicate = crate::RoutePredicate::for_query(q, self.replica_read_max_age.is_some()); match self .route_read( @@ -2411,6 +2427,9 @@ impl Db { path: &'static str, q: &EventQuery, ) -> Result> { + if let Some(answer) = crate::event::answer_without_db(q) { + return answer; + } match self .route_read( path,