From 49d49bf0b80ad3cb9d8204caa95283b284fc1829 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 23 Oct 2026 11:49:13 -0400 Subject: [PATCH] fix(nip-ar): keep sync IDs, scope and retirement exact Four gaps from review, each a way for a sync's copy to differ from the current-state query without a resync: - A live REQ that reused a sync's subscription ID left the sync in the divert map, so the new subscription silently lost its events. The live claim now stops that sync. - A refused sync REQ (bad filter, scope, auth) sent CLOSED but left an older subscription with the same ID running. Every refusal now retires the old holder under the lifecycle lock before CLOSED. - Live heads were checked only against d/type, so a #status=open sync got a new #status=done artifact. The sync now tracks the d of each artifact in the client's copy, sends a head that matches every predicate, sends a non-matching head (or deletion) only for an artifact the client holds, and drops it from the copy. The head read now returns its deletion flag. - An admin direct or report-resolution delete soft-deleted a 45010 head with no live signal. Admin deletes now refuse artifact revisions, like kind 5 does; op=delete and kind 9005 remain, and syncs observe both. NIP-AR rules 3 and 6 state the "before" set and the retirement rule. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-core/src/artifact_sync.rs | 59 +++++---- .../src/store/artifact_postgres_tests.rs | 10 +- crates/buzz-db/src/store/artifact_query.rs | 20 ++- .../buzz-db/src/store/relay_admin_actions.rs | 16 +++ crates/buzz-relay/src/api/admin/mod.rs | 37 ++++++ .../buzz-relay/src/handlers/artifact_sync.rs | 114 +++++++++++------- .../handlers/artifact_sync_postgres_tests.rs | 87 +++++++++++++ crates/buzz-relay/src/handlers/req.rs | 75 ++++++++++++ docs/nips/NIP-AR.md | 4 +- 9 files changed, 340 insertions(+), 82 deletions(-) diff --git a/crates/buzz-core/src/artifact_sync.rs b/crates/buzz-core/src/artifact_sync.rs index 769f624d95..87f63d417e 100644 --- a/crates/buzz-core/src/artifact_sync.rs +++ b/crates/buzz-core/src/artifact_sync.rs @@ -18,25 +18,17 @@ pub struct ArtifactSyncFilter { pub tags: Vec<(String, Vec)>, } -/// Tags NIP-AR fixes at creation. A head that fails a predicate on one of -/// them was never in scope and never will be. -const IMMUTABLE_TAGS: [&str; 2] = ["d", "type"]; - impl ArtifactSyncFilter { - /// Whether `head` fails a predicate on an immutable tag, so no change to - /// it can be in scope (NIP-AR sync rule 3). Mutable tags are left to the - /// client's match rule: a head that stopped matching must still be sent. - pub fn never_in_scope(&self, head: &nostr::Event) -> bool { - self.tags - .iter() - .filter(|(name, _)| IMMUTABLE_TAGS.contains(&name.as_str())) - .any(|(name, values)| { - let first = head.tags.iter().find_map(|tag| match tag.as_slice() { - [tag_name, value, ..] if tag_name == name => Some(value), - _ => None, - }); - !first.is_some_and(|value| values.contains(value)) + /// Whether `head` matches every predicate with the NIP-AR match rule: a + /// tag with that name whose first value is one of the requested values. + /// The same rule the current-state query applies (NIP-AR sync rule 3). + pub fn matches(&self, head: &nostr::Event) -> bool { + self.tags.iter().all(|(name, values)| { + head.tags.iter().any(|tag| match tag.as_slice() { + [tag_name, value, ..] => tag_name == name && values.contains(value), + _ => false, }) + }) } } @@ -101,7 +93,7 @@ mod tests { } #[test] - fn never_in_scope_checks_only_immutable_tags() { + fn matches_applies_every_predicate() { let h = Uuid::new_v4().to_string(); let filter = parse_sync( &json!({"artifact":"current","#h":[h],"#type":["work.set"],"#status":["open"]}), @@ -114,20 +106,27 @@ mod tests { .sign_with_keys(&keys) .unwrap() }; - // Another type: never in scope, whatever its other tags. - assert!(filter.never_in_scope(&head(&[ - ["h", &h], - ["type", "work.goals"], - ["status", "open"] - ]))); - assert!(filter.never_in_scope(&head(&[["h", &h]]))); - // Same type but a mutable tag no longer matches: still sent, so the - // client can drop it. - assert!(!filter.never_in_scope(&head(&[ + assert!(filter.matches(&head(&[ ["h", &h], ["type", "work.set"], - ["status", "done"] + ["status", "open"] + ]))); + // A mutable tag that does not match, a missing tag, another type or + // another channel: not in scope. + for tags in [ + &[["h", &h], ["type", "work.set"], ["status", "done"]][..], + &[["h", &h], ["type", "work.set"]][..], + &[["h", &h], ["type", "work.goals"], ["status", "open"]][..], + &[["h", "other"], ["type", "work.set"], ["status", "open"]][..], + ] { + assert!(!filter.matches(&head(tags)), "{tags:?}"); + } + // Any tag of the name can match, as in the query. + assert!(filter.matches(&head(&[ + ["h", &h], + ["type", "work.set"], + ["status", "done"], + ["status", "open"] ]))); - assert!(!filter.never_in_scope(&head(&[["h", &h], ["type", "work.set"]]))); } } diff --git a/crates/buzz-db/src/store/artifact_postgres_tests.rs b/crates/buzz-db/src/store/artifact_postgres_tests.rs index 26ebf9a4ec..2f8a2e5f7a 100644 --- a/crates/buzz-db/src/store/artifact_postgres_tests.rs +++ b/crates/buzz-db/src/store/artifact_postgres_tests.rs @@ -782,12 +782,20 @@ async fn sync_snapshot_pages_are_one_cut_and_head_includes_deletions() { Err(super::artifact_query::SnapshotDenied::Restricted) )); - let head = + let (head, is_deletion) = f.db.artifact_head(f.community, reader.as_bytes(), ids[4], &[]) .await .unwrap() .unwrap(); assert_eq!(head.event.id, deleted.id); + assert!(is_deletion); + let (head, is_deletion) = + f.db.artifact_head(f.community, reader.as_bytes(), ids[0], &[]) + .await + .unwrap() + .unwrap(); + assert_eq!(head.event.id, heads[0].id); + assert!(!is_deletion); let (view, _) = f .query( &f.owner, diff --git a/crates/buzz-db/src/store/artifact_query.rs b/crates/buzz-db/src/store/artifact_query.rs index e74669c7f2..c63f12afd2 100644 --- a/crates/buzz-db/src/store/artifact_query.rs +++ b/crates/buzz-db/src/store/artifact_query.rs @@ -194,23 +194,31 @@ impl Db { } /// The current head of one artifact, including a deletion, when the - /// reader can read the head's channel. `named_channels` are the system - /// channels the caller's request names. + /// reader can read the head's channel, with whether it is a deletion. + /// `named_channels` are the system channels the caller's request names. pub async fn artifact_head( &self, community: CommunityId, reader: &[u8], artifact: Uuid, named_channels: &[Uuid], - ) -> Result> { + ) -> Result> { let mut connection = read_connection(self).await?; - let mut qb = QueryBuilder::::new(EVENT_COLUMNS); + let mut qb = QueryBuilder::::new( + "SELECT e.id,e.pubkey,e.created_at,e.kind,e.tags,e.content,e.sig,e.received_at,e.channel_id,a.deleted FROM events e ", + ); qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND a.artifact_id=") .push_bind(artifact) .push(" "); push_scope(&mut qb, community, reader, &[], named_channels.to_vec()); - let rows = qb.build().fetch_all(&mut *connection).await?; - Ok(rows_to_events(rows)?.into_iter().next()) + let Some(row) = qb.build().fetch_optional(&mut *connection).await? else { + return Ok(None); + }; + let deleted: bool = sqlx::Row::try_get(&row, "deleted")?; + Ok(rows_to_events(vec![row])? + .into_iter() + .next() + .map(|head| (head, deleted))) } /// Open a sync snapshot over `channels`, after confirming in the same diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 6c819d3d24..53f5149953 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -803,6 +803,22 @@ pub async fn execute_delete_with_marker( .fetch_optional(tx.conn()) .await?; + // NIP-AR: an artifact revision leaves current state only through its own + // `op=delete` or a kind-9005 redaction. Both are fanned out, so open syncs + // see them; this delete would retire a head with no live signal. + let kind: Option = + sqlx::query_scalar("SELECT kind FROM events WHERE community_id = $1 AND id = $2") + .bind(community_id.as_uuid()) + .bind(target_event_id) + .fetch_optional(tx.conn()) + .await?; + if kind == Some(45010) { + return Err(crate::error::DbError::InvalidData( + "artifact revisions cannot be deleted by admin action; use op=delete or kind 9005 redaction" + .into(), + )); + } + // Canonical delete + thread_metadata counters, fenced by this transaction. // Counters move only when the row transitions to deleted, so a second // action against an already-deleted target leaves them unchanged. diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index b5cfa1ef49..15ad690bc7 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -11686,6 +11686,43 @@ mod postgres_tests { assert_eq!(live, 0); } + /// An artifact revision is not admin-deletable: open NIP-AR syncs would + /// get no live signal (they see `op=delete` and kind 9005). The action + /// fails terminally and the head stays. Mutation: drop the kind check in + /// `execute_delete_with_marker` → 200 and the head is gone → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn direct_delete_refuses_artifact_revisions() { + let (pool, community, host, state) = direct_fixture().await; + let id = seed_signed_event(&pool, community, &test_operator_keys()).await; + sqlx::query("UPDATE events SET kind = 45010 WHERE community_id = $1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .unwrap(); + let path = format!("/events/{id}/delete?communityHost={host}"); + let (status, body) = direct_post( + &state, + &path, + serde_json::json!({ "requestId": Uuid::new_v4() }), + ) + .await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}"); + assert!( + body.to_string() + .contains("artifact revisions cannot be deleted"), + "{body}" + ); + let live: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM events WHERE community_id = $1 AND deleted_at IS NULL", + ) + .bind(community.as_uuid()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(live, 1); + } + /// F1: attempt B passes the stored-action lookup, attempt A (same request) /// accepts and deletes, then B validates against a gone event. B must /// replay A (200), and a B carrying a different intent must get 409 — diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index 55c35e1183..22f9ef1ab3 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync.rs @@ -4,7 +4,7 @@ //! says that one `d` changed; the sync task then reads that artifact's current //! head and sends it (NIP-AR Sync rules 3 and 4). Any possible loss of a live //! entry ends the sync with `CLOSED "resync: ..."` (rule 5). -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex as StdMutex, PoisonError}; use std::time::Duration; @@ -92,6 +92,15 @@ impl Pending { } } +/// The artifact identity (`d`) of a revision or removal. +fn artifact_id(event: &Event) -> Option { + event + .tags + .iter() + .find(|tag| tag.as_slice().first().map(String::as_str) == Some("d")) + .and_then(|tag| tag.as_slice().get(1)?.parse().ok()) +} + /// One live sync's signal inbox. Fan-out and loss detection write to it; the /// sync's own task is its only reader. pub struct SyncHandle { @@ -126,13 +135,7 @@ impl SyncHandle { if ACCESS_KINDS.contains(&kind) { return self.resync("resync: channel access changed"); } - let Some(d) = stored - .event - .tags - .iter() - .find(|tag| tag.as_slice().first().map(String::as_str) == Some("d")) - .and_then(|tag| tag.as_slice().get(1)?.parse::().ok()) - else { + let Some(d) = artifact_id(&stored.event) else { return; }; let removal = (kind == 45011).then(|| stored.event.clone()); @@ -236,6 +239,14 @@ impl ArtifactSyncs { } } +#[cfg(test)] +impl ArtifactSyncs { + /// Install an idle sync on `sub`, as a sync claim does. + pub(crate) fn insert_idle(&self, conn: ConnId, sub: &str) { + self.insert(conn, sub, Arc::new(SyncHandle::new())); + } +} + /// Resync every sync on this node whenever its Redis link may have lost a /// message (rule 5). A sync records the epoch its interest was confirmed on, /// so any later change reaches it here. Runs until the link owner stops. @@ -329,44 +340,35 @@ pub async fn handle_sync( let filter = match filter { Ok(filter) => filter, Err(reason) => { - conn.send(RelayMessage::closed(&sub_id, &format!("invalid: {reason}"))); + refuse(&sub_id, &format!("invalid: {reason}"), &conn, &state).await; return; } }; let reader = match conn.auth_state_snapshot() { AuthState::Authenticated(ctx) => { if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) { - conn.send(RelayMessage::closed( - &sub_id, - "restricted: insufficient scope", - )); + refuse(&sub_id, "restricted: insufficient scope", &conn, &state).await; return; } // A channel-scoped token reaches only its own channels; outside // them the answer matches an unreadable channel. if let Some(allowed) = ctx.channel_ids.as_deref() { if filter.channels.iter().any(|id| !allowed.contains(id)) { - conn.send(RelayMessage::closed(&sub_id, RESTRICTED)); + refuse(&sub_id, RESTRICTED, &conn, &state).await; return; } } ctx.pubkey.to_bytes().to_vec() } _ => { - conn.send(RelayMessage::closed( - &sub_id, - "auth-required: not authenticated", - )); + refuse(&sub_id, "auth-required: not authenticated", &conn, &state).await; return; } }; let _effect = match conn.nip_fi_gate.acquire_effect().await { Ok(permit) => permit, Err(crate::nip_fi_gate::SessionExpired) => { - conn.send(RelayMessage::closed( - &sub_id, - "restricted: authorization denied", - )); + refuse(&sub_id, "restricted: authorization denied", &conn, &state).await; return; } }; @@ -393,9 +395,10 @@ pub async fn handle_sync( // time budget and the cancellation race, so a slow reader cannot hold // the handler slot past the budget or get EOSE after retirement. let initial = async { - send_initial_set(&sub_id, owner, &filter, &reader, &conn, &state).await?; + let sent = send_initial_set(&sub_id, owner, &filter, &reader, &conn, &state).await?; // A resync requested before EOSE is queued ends the sync without it. - send_eose(&sub_id, owner, &handle, &conn).await + send_eose(&sub_id, owner, &handle, &conn).await?; + Ok::<_, Stop>(sent) }; let outcome = tokio::select! { biased; @@ -403,7 +406,7 @@ pub async fn handle_sync( () = handle.cancel.cancelled() => return, out = tokio::time::timeout(SNAPSHOT_BUDGET, initial) => Some(out), }; - match outcome { + let in_set = match outcome { None => { super::close::close_if_owner(&sub_id, owner, None, &conn, &state).await; return; @@ -418,15 +421,28 @@ pub async fn handle_sync( close(&sub_id, owner, reason, &conn, &state).await; return; } - Some(Ok(Ok(()))) => {} - } + Some(Ok(Ok(sent))) => sent, + }; drop(_effect); drop(permit); - run_live(&sub_id, owner, &filter, &reader, &handle, &conn, &state).await; + run_live( + &sub_id, owner, &filter, &reader, in_set, &handle, &conn, &state, + ) + .await; } const RESTRICTED: &str = "restricted: channel not readable"; +const OVER_MAX_EVENTS: &str = "error: sync set exceeds max_events"; + +/// Refuse a sync `REQ` with `CLOSED reason`. A `REQ` replaces whatever held +/// its ID (NIP-01), and `CLOSED` tells the client the ID is closed, so the +/// old holder is retired before the frame, under the lifecycle lock. +async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, state: &AppState) { + let mut subs = conn.subscriptions.lock().await; + super::close::retire_locked(&mut subs, sub_id, conn, state).await; + conn.send(RelayMessage::closed(sub_id, reason)); +} /// Why a sync stops sending. enum Stop { @@ -499,8 +515,8 @@ async fn confirm(channels: &[Uuid], epoch: u64, conn: &ConnectionState, state: & .unwrap_or(false) } -/// Read and send every head in scope from one snapshot. After an `Err` -/// nothing more is sent. +/// Read and send every head in scope from one snapshot, and return their +/// `d`. After an `Err` nothing more is sent. async fn send_initial_set( sub_id: &str, owner: u64, @@ -508,7 +524,7 @@ async fn send_initial_set( reader: &[u8], conn: &ConnectionState, state: &AppState, -) -> Result<(), Stop> { +) -> Result, Stop> { let mut snapshot = match state .db .open_artifact_snapshot(conn.tenant.community(), reader, &filter.channels) @@ -521,7 +537,7 @@ async fn send_initial_set( return Err(Stop::Close("error: sync read failed")); } }; - let mut sent = 0usize; + let mut sent = HashSet::new(); loop { let page = snapshot .next_page(&filter.tags, SNAPSHOT_PAGE) @@ -533,11 +549,11 @@ async fn send_initial_set( if page.is_empty() { break; } - sent += page.len(); - if sent > MAX_SYNC_EVENTS { - return Err(Stop::Close("error: sync set exceeds max_events")); + if sent.len() + page.len() > MAX_SYNC_EVENTS { + return Err(Stop::Close(OVER_MAX_EVENTS)); } for stored in page { + sent.extend(artifact_id(&stored.event)); send_owned( sub_id, owner, @@ -551,7 +567,7 @@ async fn send_initial_set( .finish() .await .map_err(|_| Stop::Close("error: sync read failed"))?; - Ok(()) + Ok(sent) } /// Queue a frame while this request owns `sub_id`, waiting for room: the @@ -635,12 +651,15 @@ async fn try_send_owned( } /// Rules 3–5 after `EOSE`: read each changed head, one read at a time, and -/// send it. A dropped frame or a resync request ends the sync. +/// send it. A dropped frame or a resync request ends the sync. `in_set` is +/// the client's copy as this sync sent it: the `d` of each artifact in scope. +#[allow(clippy::too_many_arguments)] async fn run_live( sub_id: &str, owner: u64, filter: &ArtifactSyncFilter, reader: &[u8], + mut in_set: HashSet, handle: &SyncHandle, conn: &ConnectionState, state: &AppState, @@ -668,16 +687,18 @@ async fn run_live( return; } }; - // A head of another type (or `d`) was never in scope; the change - // signal is per channel, so it reaches every sync there. - if head + // Rule 3: send a change only for an artifact in scope before it + // (in the client's copy) or after it (a live head that matches + // every predicate). The change signal is per channel, so it + // reaches syncs that never matched this artifact. + let in_scope = head .as_ref() - .is_some_and(|head| filter.never_in_scope(&head.event)) - { + .is_some_and(|(head, deleted)| !deleted && filter.matches(&head.event)); + if !in_scope && !in_set.contains(&d) { continue; } // A head the reader cannot read leaves only the source removal. - let Some(event) = head.map(|head| head.event).or(removal) else { + let Some(event) = head.map(|(head, _)| head.event).or(removal) else { continue; }; #[cfg(test)] @@ -693,6 +714,13 @@ async fn run_live( return; } } + if !in_scope { + in_set.remove(&d); + } else if in_set.insert(d) && in_set.len() > MAX_SYNC_EVENTS { + // A new sync of this set would be refused the same way. + close(sub_id, owner, OVER_MAX_EVENTS, conn, state).await; + return; + } } tokio::select! { () = handle.cancel.cancelled() => return, diff --git a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs index da5ec05403..b54a5a3125 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs @@ -199,6 +199,93 @@ fn sync_filter(channels: &[Uuid]) -> Value { json!({"artifact": "current", "#h": channels, "#type": ["buzz.task"]}) } +/// Accept one `buzz.task` revision with a `status` tag and return it. +async fn status_revision( + f: &Fixture, + keys: &Keys, + home: Uuid, + d: Uuid, + status: &str, + prev: Option<&Event>, +) -> Event { + let op = if prev.is_some() { "update" } else { "create" }; + let mut tags = vec![ + vec!["ar".to_owned(), "1".into()], + vec!["d".into(), d.to_string()], + vec!["h".into(), home.to_string()], + vec!["type".into(), "buzz.task".into()], + vec!["op".into(), op.into()], + vec!["title".into(), "Sync task".into()], + vec!["status".into(), status.into()], + ]; + if let Some(prev) = prev { + tags.push(vec!["prev".into(), prev.id.to_hex()]); + } + let event = EventBuilder::new(Kind::Custom(45010), "body") + .tags(tags.into_iter().map(|t| Tag::parse(t).unwrap())) + .sign_with_keys(keys) + .unwrap(); + ingest(f, keys, event.clone()).await; + event +} + +async fn live_event(rx: &mut mpsc::Receiver) -> String { + let message = frame(rx).await; + assert_eq!(message[0].as_str(), Some("EVENT"), "{message}"); + message[2]["id"].as_str().unwrap().to_owned() +} + +/// Rule 3 with a mutable predicate: a live head goes out only for an +/// artifact in scope before the change (in the client's copy) or after it. +/// A head that never matched never crosses the socket; one that stops +/// matching is sent once, so the client can drop it. Mutation: send every +/// same-type head (the old `never_in_scope` check) → the `done` task is +/// sent → RED. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn live_heads_follow_every_predicate() { + let f = fixture().await; + let owner = Keys::generate(); + let home = create_channel(&f.state, &f.tenant, &owner, &[]).await; + let open = status_revision(&f, &owner, home, Uuid::new_v4(), "open", None).await; + + let (conn, mut rx) = connect(&f, &owner); + let filter = + json!({"artifact": "current", "#h": [home], "#type": ["buzz.task"], "#status": ["open"]}); + let task = start_sync(&f, &conn, filter); + assert_eq!(initial_set(&mut rx).await, vec![open.id.to_hex()]); + + // Never in scope: not sent. + let other = Uuid::new_v4(); + let done = status_revision(&f, &owner, home, other, "done", None).await; + no_frame(&mut rx).await; + // Leaves the set: sent once. + let d = open + .tags + .iter() + .find_map(|t| match t.as_slice() { + [name, value, ..] if name == "d" => value.parse().ok(), + _ => None, + }) + .unwrap(); + let closed = status_revision(&f, &owner, home, d, "done", Some(&open)).await; + assert_eq!(live_event(&mut rx).await, closed.id.to_hex()); + let again = status_revision(&f, &owner, home, d, "done", Some(&closed)).await; + no_frame(&mut rx).await; + // Enters the set: sent. + let reopened = status_revision(&f, &owner, home, other, "open", Some(&done)).await; + assert_eq!(live_event(&mut rx).await, reopened.id.to_hex()); + let back = status_revision(&f, &owner, home, d, "open", Some(&again)).await; + assert_eq!(live_event(&mut rx).await, back.id.to_hex()); + + super::close::handle_close("s".into(), Arc::clone(&conn), Arc::clone(&f.state)).await; + assert_eq!(frame(&mut rx).await, json!(["CLOSED", "s", ""])); + tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("sync task ends on CLOSE") + .unwrap(); +} + /// Rules 2–4 over the real pipeline: the initial set in `d` order, `EOSE`, /// then the current head of each later change (including a deletion), and /// no live entry ever arrives as the published event before `EOSE`. diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 9828c18220..94e84d279e 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -1524,6 +1524,9 @@ async fn claim_live_subscription( } let owner = super::close::next_owner(); subs.insert(sub_id.to_string(), owner); + // A sync that held the ID stops; the registry entry and topics below are + // replaced in place. + state.artifact_syncs.remove(conn.conn_id, sub_id); let community = conn.tenant.community(); let replaced = match channel_ids { Some(ids) => state.sub_registry.register_channels_scoped( @@ -1933,6 +1936,78 @@ mod tests { .collect() } + /// A live REQ that reuses a sync's ID retires the sync, so fan-out + /// delivers to the new subscription instead of the stale sync inbox. + /// Mutation: drop the `artifact_syncs.remove` in + /// `claim_live_subscription` → the event is diverted → RED. + #[tokio::test] + async fn live_req_replacing_a_sync_retires_it() { + let state = crate::state::tests::test_state().await; + let (conn, _send_rx) = lifecycle_conn(); + state.artifact_syncs.insert_idle(conn.conn_id, "s"); + claim_live_subscription( + "s", + &text_filters(), + Some(&[uuid::Uuid::new_v4()]), + &conn, + &state, + ) + .await + .expect("claim"); + assert!(!state.artifact_syncs.contains(conn.conn_id, "s")); + let event = nostr::EventBuilder::text_note("x") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign"); + let matches = vec![(conn.conn_id, "s".to_string())]; + let delivered = state + .artifact_syncs + .divert(&buzz_core::StoredEvent::new(event, None), matches.clone()); + assert_eq!(delivered, matches); + } + + /// A refused sync REQ answers `CLOSED`, so whatever held its ID must stop + /// first (NIP-01 replacement): no later frame may arrive under a closed + /// ID. Covers a filter refusal and an auth refusal. Mutation: send the + /// refusal without `retire_locked` → the old subscription stays → RED. + #[tokio::test] + async fn refused_sync_req_retires_the_old_holder() { + let state = crate::state::tests::test_state().await; + let (conn, mut send_rx) = lifecycle_conn(); + let conn = Arc::new(conn); + let channel = uuid::Uuid::new_v4(); + let topic = EventTopic::Channel(channel); + let valid = buzz_core::artifact_sync::parse_sync( + &serde_json::json!({"artifact":"current","#h":[channel.to_string()]}), + ); + for (filter, reason) in [ + (Err("sync requires #h"), "invalid: sync requires #h"), + (valid, "auth-required: not authenticated"), + ] { + claim_live_subscription("s", &text_filters(), Some(&[channel]), &conn, &state) + .await + .expect("claim"); + let permit = Arc::new(tokio::sync::Semaphore::new(1)) + .acquire_owned() + .await + .expect("permit"); + super::super::artifact_sync::handle_sync( + "s".into(), + filter, + Arc::clone(&conn), + Arc::clone(&state), + permit, + ) + .await; + assert!(conn.subscriptions.lock().await.is_empty(), "{reason}"); + assert!(!registered(&state, &conn, "s"), "{reason}"); + assert_eq!(state.pubsub.topic_refcount(&conn.tenant, topic).await, 0); + assert_eq!( + closed_frames(&mut send_rx), + vec![serde_json::json!(["CLOSED", "s", reason]).to_string()] + ); + } + } + #[tokio::test] async fn timed_out_historical_read_deregisters_before_closed() { let state = crate::state::tests::test_state().await; diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index f16e14152b..5b204b0da5 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -149,7 +149,7 @@ Rules: 1. **Live before stored.** The relay MUST register live interest, and wait until it is active, before it opens the snapshot of rule 2. Live interest is active only when the pub/sub server has confirmed the subscription (for Redis, the reply to `SUBSCRIBE`) and the node's pub/sub connection is still up. While that connection is down or reconnecting, a new sync waits, or gets `CLOSED "error: ..."`. Then every change that commits after the snapshot causes a live entry, or a resync when the relay detects that the entry was lost (rule 5). This is the order that `REQ` already uses. The relay MUST NOT send a live entry before `EOSE`. During the read it holds the `d` of each changed artifact. After `EOSE` it reads and sends their heads as in rule 3. A held change can already be in the snapshot; then the relay sends the same head again, a newer one, or nothing (rule 4). The held list is bounded: if it grows past `max_events` distinct `d`, the relay sends `CLOSED "resync: ..."` and no `EOSE`. 2. **Initial set: one consistent cut, read in pages.** The relay reads the set inside one read-only transaction that has one database snapshot (Postgres `REPEATABLE READ`). It opens the transaction after it receives the `REQ` and after rule 1, on the writer database or on a replica proven at least as fresh. So every change that the relay accepted with `OK` before it received the `REQ` is in the set. Inside the transaction it reads pages ordered by `d`. Each page starts after the last `d` of the page before it (keyset pages, not offset pages). All pages see the same snapshot. So the set at `EOSE` is the exact set at one point in time: a combination of heads that existed together, never a mix of earlier and later states. This is the complete-set read. The relay sends one `EVENT` for each artifact in scope as it reads it, then `EOSE`. The client sees one stream; it never joins pages. A snapshot does not need the whole set in one result, so there is no byte limit on the whole set. The relay bounds how long it holds a snapshot open, with a budget that it defines. It MAY copy pages into a bounded buffer so that it can close the snapshot before a slow reader has received them. If the read fails, the snapshot is lost, or the budget runs out (for example, the reader is too slow), the relay sends `CLOSED "error: ..."` and no `EOSE` (rule 8). It MUST NOT continue the read on another snapshot. If the reader cannot read a channel that `#h` names, the relay sends `CLOSED "restricted: ..."` and no `EOSE`, whether or not the channel exists. -3. **Live changes.** After `EOSE`, when an artifact in scope changes, the relay reads its current head. If the reader can read the head's channel, it sends the head, even when the head is a deletion or no longer matches. If the artifact moved to a channel the reader cannot read, it sends the kind 45011 removal stored in the source. It never sends anything about an artifact the reader cannot read. A change is an accepted revision (any `op`), or a move's removal from the source. An artifact is in scope for a change when it was in scope before the change or is in scope after it. The client decides "no longer matches" with the NIP-AR match rule: the tag name and its first value in the same tag. +3. **Live changes.** After `EOSE`, when an artifact in scope changes, the relay reads its current head. If the reader can read the head's channel, it sends the head, even when the head is a deletion or no longer matches. If the artifact moved to a channel the reader cannot read, it sends the kind 45011 removal stored in the source. It never sends anything about an artifact the reader cannot read. A change is an accepted revision (any `op`), or a move's removal from the source. An artifact is in scope for a change when it was in scope before the change or is in scope after it. "Before" is the client's copy as this sync sent it: the initial set plus later heads that matched. "After" is a current head, not a deletion, that matches every predicate of the filter with the NIP-AR match rule (the tag name and its first value in the same tag). The relay sends nothing for an artifact that is in scope neither before nor after, so a head that never matched the filter never reaches the client. 4. **Heads only, never older.** The relay sends the current head, never an older revision. It MAY skip intermediate revisions and MAY repeat a head the reader already has. For any one `d`, it MUST NOT send a head that it read before the head it last sent for that `d`. To keep this, for each sync the relay reads the head of one `d` one read at a time, starts each read only after the result of the read before it (a head, a kind 45011 removal or nothing) is in the outbound queue, and reads from the writer database or a replica proven at least as fresh. Clients that need every revision use the NIP-AR history view. 5. **No silent loss.** Live fan-out can lose entries: a publish can fail, a node's pub/sub connection can drop for seconds, and a node's local fan-out can lag. Pub/sub does not replay any of these. A sync MUST NOT lose a change silently. So: - A node MUST send `CLOSED "resync: ..."` to every sync it serves that a loss on that node can affect: its pub/sub connection dropped or reconnected; its local fan-out lagged or dropped entries, including a message that it could not decode; or the connection's own send queue was full and dropped an entry for that sync. One drop is enough: the slow-client grace limit does not apply to a sync. It MAY resync every sync it serves. @@ -157,7 +157,7 @@ Rules: - A node MUST retry a failed publish until it succeeds or the node stops. Delivery order does not matter. For a sync, a live entry is only a signal that one `d` changed. The reader's node never sends the published event itself to a sync; it reads the current head (rule 3). So a late retry of an older change sends the current head or nothing, never an older one (rule 4). So a sync that is open and has had no resync has received every change, except when a node crashes after it commits a change and before it publishes it. A client that must bound that window can sync again on its own schedule. The relay does not read the set again on a timer. -6. **Retirement.** A moderation redaction that retires the current revision of an artifact in scope makes the relay send `CLOSED "resync: ..."`. The relay withholds the redacted head, so a new sync is the only way the client learns that the artifact left the set. +6. **Retirement.** A moderation redaction that retires the current revision of an artifact in scope makes the relay send `CLOSED "resync: ..."`. The relay withholds the redacted head, so a new sync is the only way the client learns that the artifact left the set. A relay MUST NOT retire a head in any other way that open syncs cannot observe, such as an operator delete outside NIP-29; it refuses such deletes or ends the affected syncs with `resync`. 7. **Commit point.** A client MUST replace its local set only when it receives `EOSE`. If the connection drops before `EOSE`, it discards what it received. After `EOSE`, it applies each live entry as it arrives. On any disconnect it starts a new sync; there is no resume. Before a new sync after a resync, an error or a dropped connection, the client waits a random delay that grows after each failure in a row (backoff with jitter). 8. **No partial success.** If the relay cannot deliver the full set within its limits or budget, it MUST send `CLOSED` and MUST NOT send `EOSE`. A closed connection without `EOSE` is never proof of an empty set. The relay bounds its buffers for a sync in bytes as well as in count, and it ends the snapshot and its reads as soon as the client disconnects or sends `CLOSE`. 9. **Resync.** The relay MUST send `CLOSED "resync: ..."` when the reader's access to a channel that `#h` names changes: membership, channel visibility or channel deletion. This closes the whole sync, also when `#h` names other channels. An ordinary `REQ` can keep its other channels; a sync cannot, because its set is no longer exact. Archive does not change reads (see Home channel lifecycle), so it does not cause a resync.