fix(relay): avoid subscription lock inversion (#3413)

## Summary
- drop the `subs` DashMap guard before mutating subscription indexes
- snapshot fan-out candidate vectors so index guards are dropped before
looking up `subs`
- add concurrent fan-out/replacement regression coverage

## Why
`fan_out_scoped` previously held an index guard while `push_match`
acquired `subs`, while CLOSE and same-ID replacement held `subs` while
removing from an index. The reverse ordering made an AB/BA deadlock
reachable and could synchronously park all Tokio workers.

## Validation
- `rustup run 1.95.0 cargo test -p buzz-relay` — 769 library tests
passed, 33 ignored; 11 binary tests passed; doc tests passed
- push hooks with pinned Rust 1.95 — branch-skew, repository Rust
suites, and desktop Tauri suite passed
- `git diff --check`

## Residual risk
Fan-out now clones bounded candidate vectors before matching. This adds
allocation/copy cost proportional to the indexed candidate set, in
exchange for eliminating nested DashMap guards. This fixes the concrete
lock cycle but does not prove every observed production wedge had this
cause.

---------

Signed-off-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@buzz.block.builderlab.xyz>
Co-authored-by: npub12gtutshhh76rx0jx697f32f9tffd4hhp3hx58fp4x6u4uemkm7sqf8f757 <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@buzz.block.builderlab.xyz>

Author: 3 people

crates/buzz-relay/src/subscription.rs

@@ -164,17 +164,30 @@ impl SubscriptionRegistry {
         conn_id: ConnId,
         sub_id: &str,
     ) -> Option<RemovedSubscription> {
-        if let Some(mut conn_subs) = self.subs.get_mut(&conn_id) {
-            if let Some((filters, community_id, channel_id)) = conn_subs.remove(sub_id) {
-                self.remove_from_index(conn_id, sub_id, &filters, community_id, channel_id);
-                metrics::gauge!("buzz_subscriptions_active").decrement(1.0);
-                return Some(RemovedSubscription {
-                    community_id,
-                    channel_id,
-                });
-            }
-        }
-        None
+        self.remove_subscription_inner(conn_id, sub_id, || {})
+    }
+
+    fn remove_subscription_inner<F>(
+        &self,
+        conn_id: ConnId,
+        sub_id: &str,
+        after_remove: F,
+    ) -> Option<RemovedSubscription>
+    where
+        F: FnOnce(),
+    {
+        let mut conn_subs = self.subs.get_mut(&conn_id)?;
+        let (filters, community_id, channel_id) = conn_subs.remove(sub_id)?;
+
+        after_remove();
+        self.remove_from_index(conn_id, sub_id, &filters, community_id, channel_id);
+        drop(conn_subs);
+
+        metrics::gauge!("buzz_subscriptions_active").decrement(1.0);
+        Some(RemovedSubscription {
+            community_id,
+            channel_id,
+        })
     }
 
     /// Remove all subscriptions for a connection and clean up index entries.
@@ -275,15 +288,37 @@ impl SubscriptionRegistry {
                 channel_id,
                 kind: event.event.kind,
             };
-            if let Some(candidates) = self.channel_kind_index.get(&(community_id, key)) {
-                for (conn_id, sub_id) in candidates.iter() {
-                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
+            if let Some(candidates) = self
+                .channel_kind_index
+                .get(&(community_id, key))
+                .map(|entry| entry.value().clone())
+            {
+                for (conn_id, sub_id) in candidates {
+                    self.push_match(
+                        conn_id,
+                        &sub_id,
+                        community_id,
+                        event,
+                        &mut results,
+                        &mut seen,
+                    );
                 }
             }
             // Also check wildcard (channel-only, kindless) index.
-            if let Some(wildcards) = self.channel_wildcard_index.get(&(community_id, channel_id)) {
-                for (conn_id, sub_id) in wildcards.iter() {
-                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
+            if let Some(wildcards) = self
+                .channel_wildcard_index
+                .get(&(community_id, channel_id))
+                .map(|entry| entry.value().clone())
+            {
+                for (conn_id, sub_id) in wildcards {
+                    self.push_match(
+                        conn_id,
+                        &sub_id,
+                        community_id,
+                        event,
+                        &mut results,
+                        &mut seen,
+                    );
                 }
             }
         } else {
@@ -296,24 +331,54 @@ impl SubscriptionRegistry {
                     kind: event.event.kind,
                     p,
                 };
-                if let Some(candidates) = self.global_p_kind_index.get(&key) {
-                    for (conn_id, sub_id) in candidates.iter() {
-                        self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
+                if let Some(candidates) = self
+                    .global_p_kind_index
+                    .get(&key)
+                    .map(|entry| entry.value().clone())
+                {
+                    for (conn_id, sub_id) in candidates {
+                        self.push_match(
+                            conn_id,
+                            &sub_id,
+                            community_id,
+                            event,
+                            &mut results,
+                            &mut seen,
+                        );
                     }
                 }
             }
             if let Some(candidates) = self
                 .global_kind_index
                 .get(&(community_id, event.event.kind))
+                .map(|entry| entry.value().clone())
             {
-                for (conn_id, sub_id) in candidates.iter() {
-                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
+                for (conn_id, sub_id) in candidates {
+                    self.push_match(
+                        conn_id,
+                        &sub_id,
+                        community_id,
+                        event,
+                        &mut results,
+                        &mut seen,
+                    );
                 }
             }
             // Also check global wildcard (kindless global subs).
-            if let Some(wildcards) = self.global_wildcard_index.get(&community_id) {
-                for (conn_id, sub_id) in wildcards.iter() {
-                    self.push_match(*conn_id, sub_id, event, &mut results, &mut seen);
+            if let Some(wildcards) = self
+                .global_wildcard_index
+                .get(&community_id)
+                .map(|entry| entry.value().clone())
+            {
+                for (conn_id, sub_id) in wildcards {
+                    self.push_match(
+                        conn_id,
+                        &sub_id,
+                        community_id,
+                        event,
+                        &mut results,
+                        &mut seen,
+                    );
                 }
             }
         }
@@ -370,13 +435,20 @@ impl SubscriptionRegistry {
         &self,
         conn_id: ConnId,
         sub_id: &str,
+        community_id: CommunityId,
         event: &StoredEvent,
         results: &mut Vec<(ConnId, SubId)>,
         seen: &mut HashSet<(ConnId, SubId)>,
     ) {
         if let Some(conn_subs) = self.subs.get(&conn_id) {
-            if let Some((filters, _, _)) = conn_subs.get(sub_id) {
-                if filters_match(filters, event) {
+            if let Some((filters, sub_community_id, sub_channel_id)) = conn_subs.get(sub_id) {
+                // Candidate snapshots can become stale while a same-ID replacement
+                // moves the subscription. Re-check its authoritative scope before
+                // matching so an old index entry cannot deliver across scopes.
+                if *sub_community_id == community_id
+                    && *sub_channel_id == event.channel_id
+                    && filters_match(filters, event)
+                {
                     let entry = (conn_id, sub_id.to_string());
                     if seen.insert(entry.clone()) {
                         results.push(entry);
@@ -576,6 +648,8 @@ mod tests {
     use buzz_core::StoredEvent;
     use chrono::Utc;
     use nostr::{EventBuilder, Keys, Kind, Tag};
+    use std::sync::Arc;
+    use std::time::{Duration, Instant};
 
     fn make_stored_event(kind: Kind, channel_id: Option<Uuid>) -> StoredEvent {
         let keys = Keys::generate();
@@ -629,6 +703,141 @@ mod tests {
         assert!(matches.is_empty());
     }
 
+    #[test]
+    fn test_subscription_removal_cannot_delete_replacement_index() {
+        let registry = Arc::new(SubscriptionRegistry::new());
+        let conn_id = Uuid::new_v4();
+        let channel_id = Uuid::new_v4();
+        let sub_id = "same-id".to_string();
+        let filters = vec![Filter::new().kind(Kind::TextNote)];
+        registry.register(conn_id, sub_id.clone(), filters.clone(), Some(channel_id));
+
+        let (removed_tx, removed_rx) = std::sync::mpsc::sync_channel(0);
+        let (resume_tx, resume_rx) = std::sync::mpsc::sync_channel(0);
+        let remove_registry = Arc::clone(&registry);
+        let remove_sub_id = sub_id.clone();
+        let remove = std::thread::spawn(move || {
+            remove_registry.remove_subscription_inner(conn_id, &remove_sub_id, || {
+                removed_tx.send(()).expect("signal authoritative removal");
+                resume_rx.recv().expect("resume index cleanup");
+            })
+        });
+
+        removed_rx.recv().expect("old subscription removed");
+
+        let (registered_tx, registered_rx) = std::sync::mpsc::sync_channel(0);
+        let register_registry = Arc::clone(&registry);
+        let register_sub_id = sub_id.clone();
+        let register = std::thread::spawn(move || {
+            register_registry.register(conn_id, register_sub_id, filters, Some(channel_id));
+            registered_tx
+                .send(())
+                .expect("signal replacement registration");
+        });
+
+        let replacement_finished_early = registered_rx
+            .recv_timeout(Duration::from_millis(100))
+            .is_ok();
+        resume_tx.send(()).expect("resume old cleanup");
+        remove.join().expect("removal thread completes");
+        if !replacement_finished_early {
+            registered_rx
+                .recv_timeout(Duration::from_secs(1))
+                .expect("replacement registration completes");
+        }
+        register.join().expect("registration thread completes");
+        assert!(
+            !replacement_finished_early,
+            "replacement must wait until old index cleanup is complete"
+        );
+
+        let event = make_stored_event(Kind::TextNote, Some(channel_id));
+        assert_eq!(
+            registry.fan_out(&event),
+            vec![(conn_id, sub_id)],
+            "replacement must remain reachable through its index"
+        );
+    }
+
+    #[test]
+    fn test_stale_candidate_snapshot_does_not_cross_subscription_scope() {
+        let registry = SubscriptionRegistry::new();
+        let conn_id = Uuid::new_v4();
+        let channel_a = Uuid::new_v4();
+        let channel_b = Uuid::new_v4();
+        let sub_id = "same-id".to_string();
+        let filters = vec![Filter::new().kind(Kind::TextNote)];
+        registry.register(conn_id, sub_id.clone(), filters.clone(), Some(channel_a));
+
+        // Reproduce fan-out's unlocked candidate snapshot, then move the same
+        // subscription ID before the authoritative subscription lookup.
+        let key = IndexKey {
+            channel_id: channel_a,
+            kind: Kind::TextNote,
+        };
+        let candidates = registry
+            .channel_kind_index
+            .get(&(test_community(), key))
+            .expect("channel A candidate exists")
+            .value()
+            .clone();
+        registry.register(conn_id, sub_id, filters, Some(channel_b));
+
+        let event = make_stored_event(Kind::TextNote, Some(channel_a));
+        let mut results = Vec::new();
+        let mut seen = HashSet::new();
+        for (candidate_conn_id, candidate_sub_id) in candidates {
+            registry.push_match(
+                candidate_conn_id,
+                &candidate_sub_id,
+                test_community(),
+                &event,
+                &mut results,
+                &mut seen,
+            );
+        }
+
+        assert!(
+            results.is_empty(),
+            "replacement on channel B received channel A event through stale snapshot"
+        );
+    }
+
+    #[test]
+    fn test_fan_out_concurrent_with_subscription_replacement_completes() {
+        let registry = Arc::new(SubscriptionRegistry::new());
+        let conn_id = Uuid::new_v4();
+        let channel_id = Uuid::new_v4();
+        let sub_id = "sub1".to_string();
+        let filters = vec![Filter::new().kind(Kind::TextNote)];
+        registry.register(conn_id, sub_id.clone(), filters.clone(), Some(channel_id));
+        let event = Arc::new(make_stored_event(Kind::TextNote, Some(channel_id)));
+        let deadline = Instant::now() + Duration::from_secs(2);
+
+        let fan_out_registry = Arc::clone(&registry);
+        let fan_out_event = Arc::clone(&event);
+        let fan_out = std::thread::spawn(move || {
+            while Instant::now() < deadline {
+                let _ = fan_out_registry.fan_out(&fan_out_event);
+            }
+        });
+
+        let replace_registry = Arc::clone(&registry);
+        let replace = std::thread::spawn(move || {
+            while Instant::now() < deadline {
+                replace_registry.register(
+                    conn_id,
+                    sub_id.clone(),
+                    filters.clone(),
+                    Some(channel_id),
+                );
+            }
+        });
+
+        fan_out.join().expect("fan-out thread completes");
+        replace.join().expect("replacement thread completes");
+    }
+
     #[test]
     fn test_subscription_registry_remove_connection() {
         let registry = SubscriptionRegistry::new();