diff --git a/crates/buzz-relay/src/api/artifact.rs b/crates/buzz-relay/src/api/artifact.rs index c41ce5d15d..7284563be8 100644 --- a/crates/buzz-relay/src/api/artifact.rs +++ b/crates/buzz-relay/src/api/artifact.rs @@ -8,28 +8,39 @@ use serde_json::Value; use std::sync::Arc; /// Serve explicit artifact queries; `None` leaves the request to the generic -/// path (including existing extensions such as `#buzz-channel`). +/// path (including existing extensions such as `#buzz-channel`). `routed` +/// runs once the request is known to be an artifact read, before any of its +/// work awaits, so a request dropped mid-read is still described. pub(super) async fn query( state: &Arc, tenant: &TenantContext, reader: &nostr::PublicKey, filters: &[Value], count: bool, + routed: impl FnOnce(), ) -> Option, (StatusCode, Json)>> { - let mut artifact = false; + let mut artifact = Ok(false); for filter in filters { match route_filter(filter) { FilterRoute::Generic => {} - FilterRoute::Artifact => artifact = true, + FilterRoute::Artifact => artifact = Ok(true), FilterRoute::Rejected(reason) => { - return Some(Err(api_error(StatusCode::BAD_REQUEST, reason))) + artifact = Err(reason); + break; } } } - if !artifact { - return None; + match artifact { + Ok(false) => None, + Ok(true) => { + routed(); + Some(execute(state, tenant, reader, filters, count).await) + } + Err(reason) => { + routed(); + Some(Err(api_error(StatusCode::BAD_REQUEST, reason))) + } } - Some(execute(state, tenant, reader, filters, count).await) } async fn execute( diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs index 9246113259..d10b85e859 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs @@ -549,3 +549,106 @@ async fn incremental_unlisted_channels_require_membership_or_named_scope() { .await; assert_eq!(listed["changes"][0]["event"]["id"], event.id.to_hex()); } + +/// An incremental poll's access line counts its changes, removals included: +/// an empty poll is DEBUG, a poll with changes is INFO, and a failed poll is +/// WARN. Mutation: count only array responses in `query_events` → the +/// empty poll has no `result_count` and logs INFO → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn incremental_poll_access_level_follows_its_change_count() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + accept(&f, &f.revision(Uuid::new_v4(), "create", None)).await; + let poll = |types: &str| json!([{"artifact":"incremental","query":{"artifact":"current","#type":[types]}}]); + let line = |out: &crate::access::tests::Captured| { + let lines = out.lines(); + assert_eq!(lines.len(), 1, "one access line: {lines:?}"); + lines[0].clone() + }; + + let (_capture, out) = crate::access::tests::capture_scoped(); + let (status, body) = f.request("/query", poll("buzz.none")).await; + assert!(status.is_success(), "{status}: {body}"); + let empty = line(&out); + assert_eq!( + (&empty["level"], &empty["result_count"]), + (&"DEBUG".into(), &0.into()), + "{empty}" + ); + + let (_capture, out) = crate::access::tests::capture_scoped(); + let (status, body) = f.request("/query", poll("buzz.task")).await; + assert!(status.is_success(), "{status}: {body}"); + let served = line(&out); + assert_eq!( + (&served["level"], &served["result_count"]), + (&"INFO".into(), &1.into()), + "{served}" + ); + + let mut off = (*f.state).clone(); + Arc::make_mut(&mut off.config).artifact_incremental_enabled = false; + let f = Fixture { + state: Arc::new(off), + ..f + }; + let (_capture, out) = crate::access::tests::capture_scoped(); + let (status, _) = f.request("/query", poll("buzz.none")).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + let failed = line(&out); + assert_eq!(failed["level"], "WARN", "{failed}"); + assert!(failed.get("result_count").is_none(), "{failed}"); +} + +/// A wrapped incremental read dropped during its database work still +/// writes its one WARN aborted line with the standard filter summary, and +/// without its artifact predicates. Mutation: record the artifact summary +/// after `artifact::query` returns → the line has no `kinds` → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn dropped_incremental_read_keeps_its_filter_summary() { + let _guard = SERIAL.lock().await; + let f = fixture().await; + let marker = "private-predicate-marker"; + let body = json!([{"artifact":"incremental","query":{ + "artifact":"current","kinds":[45010],"#h":[f.home],"#project":[marker]}}]); + let mut lock = f.pool.begin().await.unwrap(); + sqlx::query("LOCK TABLE artifact_heads IN ACCESS EXCLUSIVE MODE") + .execute(&mut *lock) + .await + .unwrap(); + let (capture, out) = crate::access::tests::capture_access_only(tracing::Level::WARN); + let mut request = Box::pin(f.request("/query", body)); + // Poll the request until its candidate read waits on the lock. + loop { + tokio::select! { + _ = &mut request => panic!("the read must wait on the locked table"), + () = tokio::time::sleep(std::time::Duration::from_millis(20)) => {} + } + let waiting: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity WHERE wait_event_type='Lock' AND query LIKE '%artifact_heads%'", + ) + .fetch_one(&f.pool) + .await + .unwrap(); + if waiting > 0 { + break; + } + } + drop(request); + drop(capture); + lock.rollback().await.unwrap(); + let lines = out.lines(); + assert_eq!(lines.len(), 1, "exactly one access line: {lines:?}"); + let line = &lines[0]; + assert_eq!( + (&line["request"], &line["level"]), + (&"aborted".into(), &"WARN".into()), + "{line}" + ); + assert_eq!(line["route"], "/query", "{line}"); + assert_eq!(line["kinds"], "45010", "{line}"); + assert_eq!(line["channel_ids"], f.home.to_string(), "{line}"); + assert!(!out.raw().contains(marker), "{}", out.raw()); +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 12d82fe523..b6588a64a6 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1639,9 +1639,16 @@ pub async fn query_events( // helper; the caller's access line is the request's one line, covering // every outcome including a refused login. let result = query_events_authed(&state, &tenant, &headers, &body, &mut caller).await; - if let Ok(Json(Value::Array(events))) = &result { + // A generic or artifact read returns its events; an incremental read + // returns `{changes: [...]}`, removals included. + let returned = match &result { + Ok(Json(Value::Array(events))) => Some(events.len()), + Ok(Json(value)) => value.get("changes").and_then(Value::as_array).map(Vec::len), + Err(_) => None, + }; + if let Some(n) = returned { caller.complete(crate::access::Detail { - result_count: Some(events.len() as u64), + result_count: Some(n as u64), ..Default::default() }); } @@ -1699,9 +1706,10 @@ async fn query_member_events( // depth_limit, feed_types) that nostr::Filter silently drops. let raw_filters: Vec = serde_json::from_slice(body) .map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?; - if let Some(result) = super::artifact::query(state, tenant, &pubkey, &raw_filters, false).await + let routed = || caller.read_artifact(&raw_filters); + if let Some(result) = + super::artifact::query(state, tenant, &pubkey, &raw_filters, false, routed).await { - caller.read_artifact(&raw_filters); return result; } let thread_windows = thread_window::parse(&raw_filters)?; @@ -2376,8 +2384,8 @@ async fn count_member_events( let pubkey_bytes = pubkey.to_bytes().to_vec(); let raw: Vec = serde_json::from_slice(body) .map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?; - if let Some(result) = super::artifact::query(state, tenant, &pubkey, &raw, true).await { - caller.read_artifact(&raw); + let routed = || caller.read_artifact(&raw); + if let Some(result) = super::artifact::query(state, tenant, &pubkey, &raw, true, routed).await { return result; } let filters: Vec = serde_json::from_slice(body)