diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index 182b12ab92..ec69fa586f 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync.rs @@ -217,6 +217,19 @@ impl ArtifactSyncs { } } + /// End the syncs among `matches` with `reason`: a change reached them, + /// but this node cannot deliver its signal (rule 5). + pub fn resync_matches(&self, matches: &[(ConnId, SubId)], reason: &'static str) { + if self.syncs.is_empty() { + return; + } + for key in matches { + if let Some(handle) = self.syncs.get(key) { + handle.resync(reason); + } + } + } + /// Divert fan-out matches that are syncs into their signal inboxes, and /// return the matches that receive the event itself. pub fn divert( @@ -246,6 +259,16 @@ impl ArtifactSyncs { pub(crate) fn insert_idle(&self, conn: ConnId, sub: &str) { self.insert(conn, sub, Arc::new(SyncHandle::new())); } + + /// The resync reason recorded for the sync on `sub`, if any. + pub(crate) fn pending_resync(&self, conn: ConnId, sub: &str) -> Option<&'static str> { + let handle = self.syncs.get(&(conn, sub.to_owned()))?; + let pending = handle + .pending + .lock() + .unwrap_or_else(PoisonError::into_inner); + pending.resync + } } /// Resync every sync on this node whenever its Redis link may have lost a diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index c2483d6c69..76ffc21f7f 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -119,6 +119,33 @@ pub async fn filter_fanout_by_access( matches: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>, threaded: Option<&crate::state::ThreadedChannelVisibility>, ) -> Vec<(crate::subscription::ConnId, crate::subscription::SubId)> { + check_fanout_access(state, community_id, stored_event, matches, threaded) + .await + .allowed +} + +/// The result of the fan-out access gate. +struct FanoutAccess { + /// Recipients that may receive the event. + allowed: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>, + /// Recipients dropped because a visibility or membership lookup failed, + /// not because they lack access. + unverified: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>, +} + +/// `filter_fanout_by_access`, also reporting the recipients it dropped +/// only because a lookup failed. +async fn check_fanout_access( + state: &AppState, + community_id: CommunityId, + stored_event: &StoredEvent, + matches: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>, + threaded: Option<&crate::state::ThreadedChannelVisibility>, +) -> FanoutAccess { + let verified = |allowed| FanoutAccess { + allowed, + unverified: Vec::new(), + }; // First enforce the receiver-side tenant label. Subscription indexes are // community-scoped, but stale/injected matches and future fan-out helpers // must still fail closed at the send chokepoint: a connection bound to @@ -200,7 +227,7 @@ pub async fn filter_fanout_by_access( }; let Some(channel_id) = stored_event.channel_id else { - return matches; + return verified(matches); }; // Fence 3 (§4.8 phase-2): the threaded value is used only when it was // resolved under exactly this (community_id, channel_id); anything else @@ -217,17 +244,21 @@ pub async fn filter_fanout_by_access( } }; match visibility { - Ok(v) if v != "private" => return matches, + Ok(v) if v != "private" => return verified(matches), Ok(_) => {} Err(e) => { // Fail closed: if we cannot determine visibility, do not leak a // possibly-private channel's events. warn!(%channel_id, "fan-out access filter: visibility lookup failed: {e}"); - return Vec::new(); + return FanoutAccess { + allowed: Vec::new(), + unverified: matches, + }; } } let mut allowed = Vec::with_capacity(matches.len()); + let mut unverified = Vec::new(); for (conn_id, sub_id) in matches { let Some(pubkey) = state.conn_manager.pubkey_for_conn(conn_id) else { continue; @@ -240,10 +271,35 @@ pub async fn filter_fanout_by_access( Ok(false) => {} Err(e) => { warn!(%channel_id, "fan-out access filter: membership lookup failed: {e}"); + unverified.push((conn_id, sub_id)); } } } - allowed + FanoutAccess { + allowed, + unverified, + } +} + +/// Pass fan-out matches through the access gate, and return the ones that +/// receive the event itself. Syncs among the matches get its change signal +/// instead (NIP-AR rule 5), and each fan-out path calls this, so none of them +/// can skip a sync. +pub(crate) async fn route_fanout_matches( + state: &AppState, + community_id: CommunityId, + stored: &StoredEvent, + matches: Vec<(crate::subscription::ConnId, crate::subscription::SubId)>, + threaded: Option<&crate::state::ThreadedChannelVisibility>, +) -> Vec<(crate::subscription::ConnId, crate::subscription::SubId)> { + state.artifact_syncs.signal_access_change(stored, &matches); + let access = check_fanout_access(state, community_id, stored, matches, threaded).await; + // A sync that the gate could not check loses this change, so its copy is + // no longer exact. It must not stay open as if it were. + state + .artifact_syncs + .resync_matches(&access.unverified, "resync: access check failed"); + state.artifact_syncs.divert(stored, access.allowed) } /// Deliver one event to this relay's local subscribers through the access gate. @@ -267,9 +323,7 @@ pub(crate) async fn fan_out_event_to_local_subscribers( stored: &StoredEvent, ) { let matches = state.sub_registry.fan_out_scoped(community_id, stored); - state.artifact_syncs.signal_access_change(stored, &matches); - let matches = filter_fanout_by_access(state, community_id, stored, matches, None).await; - let matches = state.artifact_syncs.divert(stored, matches); + let matches = route_fanout_matches(state, community_id, stored, matches, None).await; metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64); if matches.is_empty() { return; @@ -329,9 +383,7 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub } let matches = state.sub_registry.fan_out_scoped(community_id, &stored); - state.artifact_syncs.signal_access_change(&stored, &matches); - let matches = filter_fanout_by_access(state, community_id, &stored, matches, None).await; - let matches = state.artifact_syncs.divert(&stored, matches); + let matches = route_fanout_matches(state, community_id, &stored, matches, None).await; metrics::counter!("buzz_multinode_fanout_total").increment(1); if matches.is_empty() { return; @@ -446,10 +498,7 @@ async fn dispatch_persistent_event_inner( let matches = state .sub_registry .fan_out_scoped(tenant.community(), stored_event); - state - .artifact_syncs - .signal_access_change(stored_event, &matches); - let matches = filter_fanout_by_access( + let matches = route_fanout_matches( state, tenant.community(), stored_event, @@ -457,7 +506,6 @@ async fn dispatch_persistent_event_inner( threaded_visibility.as_ref(), ) .await; - let matches = state.artifact_syncs.divert(stored_event, matches); metrics::histogram!("buzz_fanout_recipients").record(matches.len() as f64); debug!( event_id = %event_id_hex, @@ -3347,6 +3395,92 @@ mod tests { .await; assert_eq!(out, matches); } + /// A sync among the fan-out matches must resync when the access gate + /// cannot check it: the gate drops the change, so the sync's copy is + /// no longer exact. Runs through both production fan-out entries. + /// Mutation: drop the `resync_matches` call in `route_fanout_matches` + /// → the failed lookups leave every sync open with no resync → RED. + #[tokio::test] + async fn failed_access_lookup_resyncs_the_syncs_it_drops() { + let state = test_state().await; + // Every database lookup now fails at once. + state.db.pool().close().await; + let community_id = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()); + let pk = vec![1u8; 32]; + let artifact = |channel_id| { + let event = EventBuilder::new(Kind::Custom(45010), "{}") + .tags([nostr::Tag::parse(["d", &Uuid::new_v4().to_string()]).expect("d")]) + .sign_with_keys(&Keys::generate()) + .expect("sign"); + StoredEvent::new(event, Some(channel_id)) + }; + let sync_on = |channel_id| { + let conn = register_conn(&state, Some(pk.clone())); + state.sub_registry.register_channels_scoped( + community_id, + conn, + "s".to_string(), + vec![nostr::Filter::new()], + vec![channel_id], + ); + state.artifact_syncs.insert_idle(conn, "s"); + conn + }; + let private = |channel_id| { + state + .channel_visibility_cache + .insert((community_id, channel_id), "private".to_string()); + }; + + // Control: a member whose access is known gets the change signal. + let known = Uuid::new_v4(); + private(known); + state + .membership_cache + .insert((community_id, known, pk.clone()), true); + let conn = sync_on(known); + super::super::fan_out_event_to_local_subscribers( + &state, + community_id, + &artifact(known), + ) + .await; + assert_eq!(state.artifact_syncs.pending_resync(conn, "s"), None); + + // The visibility lookup fails (local fan-out). + let channel_id = Uuid::new_v4(); + let conn = sync_on(channel_id); + super::super::fan_out_event_to_local_subscribers( + &state, + community_id, + &artifact(channel_id), + ) + .await; + assert_eq!( + state.artifact_syncs.pending_resync(conn, "s"), + Some("resync: access check failed"), + "visibility lookup" + ); + + // The membership lookup fails (Redis fan-out). + let channel_id = Uuid::new_v4(); + private(channel_id); + let conn = sync_on(channel_id); + super::super::fan_out_pubsub_event( + &state, + buzz_pubsub::ChannelEvent { + community_id, + topic: buzz_pubsub::EventTopic::Channel(channel_id), + event: artifact(channel_id).event, + }, + ) + .await; + assert_eq!( + state.artifact_syncs.pending_resync(conn, "s"), + Some("resync: access check failed"), + "membership lookup" + ); + } } // --------------------------------------------------------------------- diff --git a/docs/nips/NIP-AR.md b/docs/nips/NIP-AR.md index 5b204b0da5..d5ed45a477 100644 --- a/docs/nips/NIP-AR.md +++ b/docs/nips/NIP-AR.md @@ -152,7 +152,7 @@ Rules: 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. + - 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; it could not check the reader's access to a change for that sync (for example, a failed membership lookup); 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. - If the node cannot queue the `CLOSED "resync: ..."` itself (for example, because the same send queue is full), it MUST close the connection. A client starts a new sync on any disconnect (rule 7), so a resync is never lost. - 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).