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.