jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 05:51:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

db: answer match-nothing event reads without taking a connection (#8663)

Browse Source 
## Summary

A NIP-01 `limit: 0` REQ asks for no stored events: it is a live-only
subscription. The relay still acquired a pool connection for it, then
ran a `LIMIT 0` query. Desktop is moving live subscriptions to `limit:
0`, so these acquires add up during a reconnect burst.

The existing match-nothing shortcuts in `query_events_on` (an empty
`kinds`, `authors`, `ids` or `#e` list) had the same cost, because they
ran after the caller had already acquired a connection.

This PR moves the query validation and the match-nothing checks into one
function, `answer_without_db`. It adds a zero page (`limit: 0`, after
the page clamp) to those checks, and every entry point now calls the
function before it acquires a connection:

- `query_events_with_operation`, which covers `Db::query_events` and its
writer variants
- `Db::query_events_routed`
- `Db::query_events_routed_bounded`
- `query_events_on`, for the session (aux) path that already holds a
connection

Invalid queries still return the same errors, in the same order, and
match-nothing queries still return an empty page. What changes besides
the connection:

- A routed match-nothing read now returns before `route_read`, so it
emits no `buzz_db_route_decision` and runs no replica proof. As desktop
moves live subscriptions to `limit: 0`, writer/replica route-decision
ratios will shift. Fewer reads are routed; no read changes destination.
- During a pool outage, an invalid query now returns its `InvalidData`
error instead of the pool error.
- A zero page skips checks that only SQL applies, such as rejecting a
negative `offset`. Current callers only build non-negative offsets.

**Simpler?** Yes. The four separate early returns become one predicate,
and the page-limit clamp is computed in one place (`page_limit`), shared
by the shortcut and the SQL builder.

### Related issue
N/A. This is a follow-up from https://github.com/block/buzz/pull/8651.

### Testing
- New unit test,
`runtime::postgres_tests::match_nothing_reads_take_no_connection`. Both
pools point at a closed port, replica routing is on, and the fence is
forced open, so a routed read that reached `route_read` would try the
reader first. Each match-nothing shape (`limit: 0`, `max_limit: 0`, and
an empty `kinds`, `authors`, `ids` or `#e` list) must return an empty
page through the writer, routed and routed-bounded entry points. Invalid
queries that would also match nothing must return `InvalidData` through
each entry point. None of these may record a
`buzz_db_pool_acquire_attempts_total`. A `limit: 1` routed control must
fail after trying both the reader and the writer. The test runs in the
`cargo nextest run -p buzz-db --lib` unit lane.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Luke Tornquist
2026-10-29 14:21:08 -04:00
committed by GitHub
parent cc28660c61
commit f6d31d13e4
2 changed files with 162 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
crates/buzz-db/src/runtime/tests.rs
View File

@@ -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::<std::collections::BTreeSet<_>>()
};
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");

77
crates/buzz-db/src/store/event.rs
View File

@@ -548,6 +548,9 @@ pub(crate) async fn query_events_with_operation(
q: &EventQuery,
operation: crate::observability::WriterOperation,
) -> Result<Vec<StoredEvent>> {
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<Vec<StoredEvent>> {
// 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<Result<Vec<StoredEvent>>> {
// 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<sqlx::Postgres> {
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<Vec<StoredEvent>> {
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<Vec<StoredEvent>> {
if let Some(answer) = crate::event::answer_without_db(q) {
return answer;
}
match self
.route_read(
path,