fix(relay): keep artifact read summaries on abort and count incremental changes

An artifact read now records its summary as soon as it routes as an
artifact read, before its database work awaits, so a dropped read keeps
its filter summary. /query records an incremental response's change
count, removals included, so an empty poll logs at DEBUG like other
empty reads.

Co-authored-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>

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<AppState>,
     tenant: &TenantContext,
     reader: &nostr::PublicKey,
     filters: &[Value],
     count: bool,
+    routed: impl FnOnce(),
 ) -> Option<Result<Json<Value>, (StatusCode, Json<Value>)>> {
-    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(

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());
+}

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<Value> = 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<Value> = 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<nostr::Filter> = serde_json::from_slice(body)