fix(nip-ar): resync a sync whose fan-out access check failed

The fan-out access gate fails closed: when a channel visibility or
membership lookup fails, it drops the recipient. For an ordinary
subscription that only loses one live event. For an artifact sync it
loses a change signal, so the sync stays open with a copy that is no
longer exact, and nothing tells the client.

The gate now also reports the recipients it dropped only because a
lookup failed. route_fanout_matches, which every fan-out path (local,
Redis and persistent dispatch) uses, sends CLOSED "resync: access check
failed" to each sync among them. NIP-AR rule 5 lists this loss.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

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

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<AppState>, 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"
+            );
+        }
     }
 
     // ---------------------------------------------------------------------

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).