fix(relay): fence pending subscriptions on close or replacement (#2382)

## Summary

A REQ handler runs as a spawned task and can pause in its access checks.
Before this change it took its subscription ID only when it registered.
So:

- `REQ x`, then `CLOSE x` while the REQ was still in its access checks:
the relay sent `CLOSED x`, and then the REQ registered `x` anyway and
sent events and `EOSE` under an ID the client had closed.
- `REQ x` (A), then a newer `REQ x` (B): if A finished its checks after
B registered, A replaced B, or A's refusal retired B and sent a `CLOSED`
the client could not tell from B failing.

Now the read loop admits each REQ and sync REQ, in arrival order, before
it spawns the handler: the newest request owns the ID from the moment it
arrives. A `CLOSE` or a newer same-ID request fences an older one that
is still pending. A fenced request does not register, and does not send
`NOTICE`, `CLOSED`, `EVENT` or `EOSE` for the ID.

This builds on the owner token and lifecycle lock that #8522 added. That
PR fenced output after registration; this one moves the token to
admission, so the same fence also covers the time before registration.

The pubsub retain no longer waits on a full subscriber command queue.
Callers hold a connection's lifecycle lock while they retain. A retain
past a full queue now leaves a coalesced wake-up, and the subscriber
subscribes every retained topic it is missing. A retain that fits in the
queue leaves no wake-up, so first retains do not each make the
subscriber rescan the map.

## Behavior

- `close::admit` runs in the read loop before spawn. It records the
request's token in the connection's subscription map, with a flag that
says whether the request has claimed yet. It also applies the
subscription limit, which moves there from the handlers.
- Claims (`claim_admitted_live`, `claim_admitted_search`, the sync
claim) register only while the request still owns its token.
- Refusals after admission go through `close_if_owner`. The two REQ
refusals that also send a `NOTICE` send it in the same critical section.
- Refusals before admission (rate limit, admission quota) still retire
the old holder, as before.
- The old holder stays registered until the newer request claims or is
refused, as before.
- A channel-access eviction that removes an old holder while a newer
same-ID request is still pending retires only the old holder. It sends
no `CLOSED` for the ID, because the client already replaced that holder,
and the newer request still claims.

Client-visible changes to the subscription limit:

- The limit is checked on arrival, before scope, auth and filter checks.
On a full connection, a REQ for a new ID that would also fail those
checks now gets `CLOSED "error: too many subscriptions"` instead of the
other refusal.
- Requests still in their access checks count toward the limit.

## Scope

Live fan-out does not check the token. A same-ID request that has
arrived but not yet claimed does not stop live events to the old holder.
That path is unchanged here.

If a handler panics before it claims, its token stays in the map until a
`CLOSE`, a same-ID REQ, or the disconnect removes it. Until then it
counts toward the subscription limit.

## Testing

- `close_fences_a_req_whose_handler_has_not_run` (REQ and sync REQ):
`REQ x` then `CLOSE x` before the handler runs gives only `CLOSED x` and
no registration.
- `only_the_newest_same_id_req_answers`: of two same-ID REQs, only the
newer one answers.
- `superseded_or_closed_requests_cannot_claim`: an older request cannot
claim after a newer one, and a closed request cannot claim or close.
- `revoke_of_a_replaced_holder_spares_the_pending_replacement`: a revoke
of the old holder sends no `CLOSED`, keeps the pending request's token,
and the pending request then claims its own channel.
- `queue_full_retain_does_not_wait_and_schedules_reconciliation`
(pubsub).
- `live_subscriber_recovers_a_dropped_subscribe_hint` (pubsub, requires
Redis): it runs the real subscriber on a live link and drops a topic's
subscribe hint with a full queue. The topic is still delivered without a
reconnect and without `Confirm`. The same holds after the topic is
released, unsubscribed and retained again.
- Mutations: with no owner check in `close_owned`, both connection tests
fail. With no owner check in `claim_admitted_live`, the claim test
fails. With no `replaced` check in the eviction, the revoke test fails.
With no reconcile arm in the subscriber, or no `notify_one` in
`retain_topic`, the Redis subscriber test fails.
- CI: the four new relay tests and the full-queue pubsub test are added
to `just test-unit` and `scripts/run-tests.sh`. The Redis subscriber
test runs in a new Redis-backed backend-integration step.
- `cargo test -p buzz-relay --lib` (with `--skip config::tests`, which
hangs locally when run in one process), `cargo test -p buzz-pubsub --
--include-ignored` against a local Redis, `cargo clippy -p buzz-relay -p
buzz-pubsub --all-targets -- -D warnings`, `cargo fmt --all -- --check`.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

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

.github/workflows/_ci-relay.yml

@@ -121,6 +121,7 @@ jobs:
           cargo nextest archive \
             --cargo-profile ci \
             -p buzz-db \
+            -p buzz-pubsub \
             -p buzz-relay \
             -p buzz-relay-mesh \
             -p buzz-test-client \
@@ -583,6 +584,13 @@ jobs:
             --archive-file target/ci/backend-integration-tests.tar.zst \
             -E 'package(buzz-relay-mesh) and test(/^registry::tests::external_infra_redis::/)' \
             --run-ignored ignored-only
+      - name: Pubsub subscriber reconciliation regression
+        # Runs the live Redis subscriber; the test dials redis://127.0.0.1:6379.
+        run: |
+          cargo nextest run \
+            --archive-file target/ci/backend-integration-tests.tar.zst \
+            -E 'package(buzz-pubsub) and test(=tests::live_subscriber_recovers_a_dropped_subscribe_hint)' \
+            --run-ignored ignored-only
       - name: Workflow message provenance unit tests
         # The relay's workflow_sink suite is not selected by the infra-free
         # unit job. Its ignored database cases run in the isolated PostgreSQL

crates/buzz-pubsub/src/lib.rs

@@ -56,7 +56,7 @@ use std::time::Duration;
 
 use buzz_core::TenantContext;
 use nostr::PublicKey;
-use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex};
+use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex, Notify};
 
 use crate::cache_invalidation::{
     cache_invalidation_channel, CacheInvalidation, ScopedCacheInvalidation,
@@ -119,6 +119,10 @@ impl PubSubConfig {
     }
 }
 
+/// Capacity of the subscriber command queue. A full queue loses only a
+/// subscribe hint: the desired-topic map is reconciled separately.
+const SUBSCRIPTION_COMMAND_CAPACITY: usize = 4096;
+
 /// Central pub/sub manager for a Buzz relay instance.
 pub struct PubSubManager {
     pool: crate::RedisConnection,
@@ -130,6 +134,8 @@ pub struct PubSubManager {
     desired_topics: subscriber::DesiredTopics,
     subscription_tx: mpsc::Sender<subscriber::SubscriptionCommand>,
     subscription_rx: Mutex<Option<mpsc::Receiver<subscriber::SubscriptionCommand>>>,
+    /// Coalesced wake-up: reconcile desired topics with the live Redis link.
+    subscription_changed: Arc<Notify>,
     broadcast_tx: broadcast::Sender<ChannelEvent>,
     cache_invalidation_tx: broadcast::Sender<ScopedCacheInvalidation>,
     conn_control_tx: broadcast::Sender<ScopedConnControl>,
@@ -152,7 +158,7 @@ impl PubSubManager {
         let (cache_invalidation_tx, _) = broadcast::channel(4096);
         let (conn_control_tx, _) = broadcast::channel(4096);
         let (nip_fi_disconnect_tx, _) = broadcast::channel(4096);
-        let (subscription_tx, subscription_rx) = mpsc::channel(4096);
+        let (subscription_tx, subscription_rx) = mpsc::channel(SUBSCRIPTION_COMMAND_CAPACITY);
 
         Ok(Self {
             pool,
@@ -161,6 +167,7 @@ impl PubSubManager {
             desired_topics: Arc::new(Mutex::new(HashMap::new())),
             subscription_tx,
             subscription_rx: Mutex::new(Some(subscription_rx)),
+            subscription_changed: Arc::new(Notify::new()),
             broadcast_tx,
             cache_invalidation_tx,
             conn_control_tx,
@@ -184,6 +191,7 @@ impl PubSubManager {
             self.broadcast_tx.clone(),
             self.desired_topics.clone(),
             subscription_rx,
+            self.subscription_changed.clone(),
             self.link.clone(),
         )
         .await;
@@ -235,7 +243,8 @@ impl PubSubManager {
     /// Retain local interest in a scoped Redis event topic.
     ///
     /// The first retain for a topic asks the subscriber task to `SUBSCRIBE`.
-    /// Additional retains only increment the local desired refcount.
+    /// Additional retains only increment the local desired refcount. It never
+    /// waits on the subscriber: callers hold a connection's lifecycle lock.
     pub async fn retain_topic(&self, ctx: &TenantContext, topic: EventTopic) {
         let topic_key = EventTopicKey::from_context(ctx, topic);
         let should_subscribe = {
@@ -247,10 +256,25 @@ impl PubSubManager {
         };
 
         if should_subscribe {
-            let _ = self
+            match self
                 .subscription_tx
-                .send(subscriber::SubscriptionCommand::Subscribe(topic_key))
-                .await;
+                .try_send(subscriber::SubscriptionCommand::Subscribe(topic_key))
+            {
+                // A queued command subscribes in order; a closed queue means
+                // the subscriber is gone, and a restart re-subscribes the
+                // whole desired-topic map.
+                Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {}
+                Err(mpsc::error::TrySendError::Full(_)) => {
+                    // The wake-up is lossless and coalesced: the subscriber
+                    // reconciles the desired-topic map on the live link
+                    // instead of waiting for a Redis reconnect.
+                    self.subscription_changed.notify_one();
+                    tracing::warn!(
+                        ?topic_key,
+                        "pubsub subscription command queue full; reconciliation scheduled"
+                    );
+                }
+            }
         }
     }
 
@@ -703,6 +727,185 @@ mod tests {
         assert_eq!(manager.topic_refcount(&ctx_b, topic).await, 0);
     }
 
+    /// A retain past a full command queue neither waits for the subscriber
+    /// (callers hold a lifecycle lock) nor loses the topic: it leaves a
+    /// reconciliation wake-up, and the topic is in the desired map the
+    /// subscriber re-reads. Retains that fit in the queue leave no wake-up,
+    /// so first retains do not each make the subscriber rescan the map.
+    /// Mutation: `send().await` instead of `try_send` hangs at the timeout
+    /// below; no `notify_one` fails the wake-up check; `notify_one` on every
+    /// first retain fails the no-wake-up check.
+    #[tokio::test]
+    async fn queue_full_retain_does_not_wait_and_schedules_reconciliation() {
+        let manager = make_manager().await;
+        let filler = ctx(0xaaaa, "a.example");
+        for id in 0..SUBSCRIPTION_COMMAND_CAPACITY {
+            manager
+                .retain_topic(&filler, EventTopic::Channel(Uuid::from_u128(id as u128)))
+                .await;
+        }
+        // Retains that fit in the queue leave no reconciliation wake-up.
+        assert!(
+            tokio::time::timeout(
+                Duration::from_millis(10),
+                manager.subscription_changed.notified()
+            )
+            .await
+            .is_err(),
+            "a queued subscribe needs no reconciliation"
+        );
+
+        let dropped_ctx = ctx(0xbbbb, "b.example");
+        let dropped = EventTopic::Channel(Uuid::from_u128(0xcccc));
+        tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.retain_topic(&dropped_ctx, dropped),
+        )
+        .await
+        .expect("a retain must not wait on a full command queue");
+        assert_eq!(manager.topic_refcount(&dropped_ctx, dropped).await, 1);
+
+        tokio::time::timeout(
+            Duration::from_millis(10),
+            manager.subscription_changed.notified(),
+        )
+        .await
+        .expect("a queue-full retain must leave a reconciliation wake-up");
+        let missing = subscriber::missing_desired_topics(
+            &manager.desired_topics,
+            &std::collections::HashSet::new(),
+        )
+        .await;
+        assert!(missing.contains(&EventTopicKey::from_context(&dropped_ctx, dropped)));
+    }
+
+    /// Fill the command queue with first retains of fresh topics, then retain
+    /// `topic`, without yielding: on this single-threaded runtime the
+    /// subscriber cannot drain the queue in between, so `topic`'s subscribe
+    /// hint is dropped.
+    async fn retain_with_dropped_hint(
+        manager: &PubSubManager,
+        filler: &TenantContext,
+        first_filler: u128,
+        target: &TenantContext,
+        topic: EventTopic,
+    ) {
+        tokio::task::unconstrained(async {
+            let mut id = first_filler;
+            while manager.subscription_tx.capacity() > 0 {
+                manager
+                    .retain_topic(filler, EventTopic::Channel(Uuid::from_u128(id)))
+                    .await;
+                id += 1;
+            }
+            manager.retain_topic(target, topic).await;
+        })
+        .await;
+    }
+
+    /// Publish to `topic` until the live link delivers it.
+    async fn delivered(
+        manager: &PubSubManager,
+        rx: &mut broadcast::Receiver<ChannelEvent>,
+        ctx: &TenantContext,
+        topic: EventTopic,
+    ) -> bool {
+        let event = EventBuilder::new(Kind::TextNote, "reconciled")
+            .sign_with_keys(&Keys::generate())
+            .expect("sign");
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                manager
+                    .publish_event(ctx, topic, &event)
+                    .await
+                    .expect("publish");
+                let wait = tokio::time::sleep(Duration::from_millis(100));
+                tokio::pin!(wait);
+                loop {
+                    tokio::select! {
+                        () = &mut wait => break,
+                        received = rx.recv() => match received {
+                            Ok(received) if received.event.id == event.id => return,
+                            Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => {}
+                            Err(broadcast::error::RecvError::Closed) => panic!("closed"),
+                        },
+                    }
+                }
+            }
+        })
+        .await
+        .is_ok()
+    }
+
+    /// A subscribe hint lost to a full command queue is recovered by the
+    /// running subscriber on the same Redis link: first for a new topic, then
+    /// for one released, unsubscribed, and retained again. The link never
+    /// reconnects and nothing sends `Confirm` for the topic, so only the
+    /// reconciliation wake-up can subscribe it. Mutation: drop the
+    /// subscriber's `subscription_changed` arm, or the `notify_one` in
+    /// `retain_topic` → the topic is never delivered → RED.
+    #[tokio::test]
+    #[ignore = "requires Redis"]
+    async fn live_subscriber_recovers_a_dropped_subscribe_hint() {
+        let manager = Arc::new(
+            PubSubManager::with_config(
+                PubSubConfig::new("redis://127.0.0.1:6379")
+                    .with_unsubscribe_debounce(Duration::from_millis(1)),
+                make_test_pool(),
+            )
+            .await
+            .unwrap(),
+        );
+        let mut rx = manager.subscribe_local();
+        let mut link = manager.link_state();
+        let subscriber = tokio::spawn(Arc::clone(&manager).run_subscriber());
+        link.wait_for(|state| state.up).await.expect("link up");
+        let epoch = link.borrow().epoch;
+
+        let filler = ctx(Uuid::new_v4().as_u128(), "filler.example");
+        let target = ctx(Uuid::new_v4().as_u128(), "target.example");
+        let topic = EventTopic::Channel(Uuid::new_v4());
+
+        retain_with_dropped_hint(&manager, &filler, 0, &target, topic).await;
+        assert!(
+            delivered(&manager, &mut rx, &target, topic).await,
+            "a dropped hint for a new topic must be reconciled"
+        );
+
+        manager.release_topic(&target, topic).await;
+        // Wait until Redis shows the debounced unsubscribe took effect, so
+        // the re-retain below starts from an unsubscribed topic.
+        let channel = EventTopicKey::from_context(&target, topic).redis_channel();
+        let mut redis = redis::Client::open("redis://127.0.0.1:6379")
+            .unwrap()
+            .get_multiplexed_async_connection()
+            .await
+            .expect("redis");
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                let (_, subscribers): (String, i64) = redis::cmd("PUBSUB")
+                    .arg("NUMSUB")
+                    .arg(&channel)
+                    .query_async(&mut redis)
+                    .await
+                    .expect("numsub");
+                if subscribers == 0 {
+                    break;
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("the released topic must be unsubscribed");
+        retain_with_dropped_hint(&manager, &filler, 1 << 32, &target, topic).await;
+        assert!(
+            delivered(&manager, &mut rx, &target, topic).await,
+            "a dropped hint after an unsubscribe must be reconciled"
+        );
+        assert_eq!(link.borrow().epoch, epoch, "the link must not reconnect");
+        subscriber.abort();
+    }
+
     #[tokio::test]
     async fn retain_release_refcounts_and_debounces_last_release() {
         let pool = make_test_pool();

crates/buzz-pubsub/src/subscriber.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 
 use futures_util::StreamExt;
 use nostr::JsonUtil;
-use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex};
+use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex, Notify};
 
 use crate::topic::EventTopicKey;
 use crate::{ChannelEvent, LinkState};
@@ -51,6 +51,7 @@ pub(crate) async fn run_subscriber(
     broadcast_tx: broadcast::Sender<ChannelEvent>,
     desired_topics: DesiredTopics,
     mut subscription_rx: mpsc::Receiver<SubscriptionCommand>,
+    subscription_changed: Arc<Notify>,
     link: watch::Sender<LinkState>,
 ) {
     let mut backoff_secs = BACKOFF_INITIAL_SECS;
@@ -61,6 +62,7 @@ pub(crate) async fn run_subscriber(
             &broadcast_tx,
             desired_topics.clone(),
             &mut subscription_rx,
+            &subscription_changed,
             &link,
         )
         .await;
@@ -95,6 +97,7 @@ async fn connect_and_subscribe(
     broadcast_tx: &broadcast::Sender<ChannelEvent>,
     desired_topics: DesiredTopics,
     subscription_rx: &mut mpsc::Receiver<SubscriptionCommand>,
+    subscription_changed: &Notify,
     link: &watch::Sender<LinkState>,
 ) -> Result<(), redis::RedisError> {
     let client = redis::Client::open(redis_url)?;
@@ -124,6 +127,16 @@ async fn connect_and_subscribe(
 
     loop {
         tokio::select! {
+            () = subscription_changed.notified() => {
+                // Retains signal this besides their best-effort queue hint.
+                // Re-read the source of truth so a hint lost to a full queue
+                // cannot leave the live connection missing a topic.
+                for topic in missing_desired_topics(&desired_topics, &active_topics).await {
+                    let channel = topic.redis_channel();
+                    sink.subscribe(&channel).await?;
+                    active_topics.insert(channel);
+                }
+            }
             Some(command) = subscription_rx.recv() => {
                 match command {
                     SubscriptionCommand::Subscribe(topic) => {
@@ -201,6 +214,21 @@ async fn connect_and_subscribe(
     }
 }
 
+/// Retained topics the live connection has not subscribed yet.
+pub(crate) async fn missing_desired_topics(
+    desired_topics: &DesiredTopics,
+    active_topics: &HashSet<String>,
+) -> Vec<EventTopicKey> {
+    desired_topics
+        .lock()
+        .await
+        .iter()
+        .filter_map(|(topic, count)| {
+            (*count > 0 && !active_topics.contains(&topic.redis_channel())).then_some(*topic)
+        })
+        .collect()
+}
+
 async fn desired_refcount(desired_topics: &DesiredTopics, topic: EventTopicKey) -> usize {
     desired_topics
         .lock()

crates/buzz-relay/src/api/bridge.rs

@@ -10827,7 +10827,10 @@ pub(crate) mod postgres_tests {
             .expect("lock replica events");
         let search: nostr::Filter =
             serde_json::from_value(fx.search_filter()).expect("search filter");
-        let live_owner = *conn.subscriptions.lock().await.get("x").expect("x owned");
+        // `before_search_query` runs after the claim and before the query, so
+        // the live entry must already be gone there: the timeout CLOSED,
+        // which also retires it, has not happened yet.
+        let (arrived, release) = crate::nip_fi_test_hooks::search_query_hook::arm(fx.community);
         let task = tokio::spawn(handle_req(
             "x".into(),
             vec![search],
@@ -10835,11 +10838,10 @@ pub(crate) mod postgres_tests {
             conn.clone(),
             fx.state.clone(),
         ));
-        // Once search has claimed `x` its hydrate blocks on the lock. The old
-        // live fan-out must already be retired then, not only at final cleanup.
-        while conn.subscriptions.lock().await.get("x") == Some(&live_owner) {
-            tokio::task::yield_now().await;
-        }
+        tokio::time::timeout(std::time::Duration::from_secs(10), arrived)
+            .await
+            .expect("search reaches its query")
+            .expect("hook armed");
         assert!(
             fx.state
                 .sub_registry
@@ -10848,6 +10850,8 @@ pub(crate) mod postgres_tests {
             "accepting search must retire live fan-out"
         );
         assert_eq!(fx.state.pubsub.topic_refcount(&tenant, topic).await, 0);
+        assert!(rx.is_empty(), "nothing sent before the query");
+        release.notify_one();
         task.await.expect("search task");
         locker.rollback().await.expect("unlock");
 

crates/buzz-relay/src/connection.rs

@@ -37,8 +37,8 @@ const AUTH_TIMEOUT: Duration = Duration::from_secs(5);
 pub(crate) const WS_TERMINAL_FLUSH_TIMEOUT: Duration = Duration::from_secs(1);
 
 /// Shared mutable subscription map for a single WebSocket connection: sub ID →
-/// owner token of the request that currently holds it (see `close::next_owner`).
-pub(crate) type ConnectionSubscriptions = Arc<Mutex<HashMap<String, u64>>>;
+/// the newest admitted request for it (see `close::SubOwner`).
+pub(crate) type ConnectionSubscriptions = Arc<Mutex<HashMap<String, handlers::close::SubOwner>>>;
 
 /// Pre-built NIP-FI session components created before the `is_community_active`
 /// bootstrap await. Passed from the HTTP-layer wrapper into the active handler so
@@ -1489,10 +1489,31 @@ async fn handle_text_message(text: String, conn: Arc<ConnectionState>, state: Ar
                     return;
                 }
             };
+            // Admit before spawning: arrival order decides which request owns
+            // the ID, so a later same-ID REQ or a CLOSE fences this one even
+            // while its handler is still in access checks.
+            let owner = match handlers::close::admit(&sub_id, &conn).await {
+                Ok(owner) => owner,
+                Err(not_admitted) => {
+                    if not_admitted == handlers::close::NotAdmitted::TooManySubscriptions {
+                        crate::access::ReadLog::refused_read(
+                            "req",
+                            &conn,
+                            &sub_id,
+                            &filters,
+                            "too_many_subscriptions",
+                        );
+                    }
+                    return;
+                }
+            };
             let span = tracing::info_span!("ws.req", conn_id = %conn.conn_id, sub_id = %sub_id);
             tokio::spawn(
                 async move {
-                    handlers::req::handle_req(sub_id, filters, before_ids, conn, state).await;
+                    handlers::req::handle_admitted_req(
+                        sub_id, owner, filters, before_ids, conn, state,
+                    )
+                    .await;
                     drop(permit);
                 }
                 .instrument(span),
@@ -1521,11 +1542,29 @@ async fn handle_text_message(text: String, conn: Arc<ConnectionState>, state: Ar
                     return;
                 }
             };
+            // Admitted before spawning, like a REQ.
+            let owner = match handlers::close::admit(&sub_id, &conn).await {
+                Ok(owner) => owner,
+                Err(not_admitted) => {
+                    if not_admitted == handlers::close::NotAdmitted::TooManySubscriptions {
+                        crate::access::ReadLog::refused_sync(
+                            &conn,
+                            &sub_id,
+                            filter.as_ref().ok(),
+                            "too_many_subscriptions",
+                        );
+                    }
+                    return;
+                }
+            };
             let span =
                 tracing::info_span!("ws.artifact_sync", conn_id = %conn.conn_id, sub_id = %sub_id);
             tokio::spawn(
                 async move {
-                    handlers::artifact_sync::handle_sync(sub_id, filter, conn, state, permit).await;
+                    handlers::artifact_sync::handle_sync(
+                        sub_id, owner, filter, conn, state, permit,
+                    )
+                    .await;
                 }
                 .instrument(span),
             );
@@ -1623,8 +1662,9 @@ fn log_aborted_auth(conn: &ConnectionState, claimed: nostr::PublicKey) {
 pub(crate) mod test_seam {
     use super::*;
 
-    pub(crate) type Hold =
-        tokio::sync::oneshot::Sender<tokio::sync::OwnedMutexGuard<HashMap<String, u64>>>;
+    pub(crate) type Hold = tokio::sync::oneshot::Sender<
+        tokio::sync::OwnedMutexGuard<HashMap<String, handlers::close::SubOwner>>,
+    >;
 
     tokio::task_local! {
         /// Hands the test this connection's subscription lock, already held,
@@ -2839,6 +2879,90 @@ pub(crate) mod tests {
         }
     }
 
+    /// Let every spawned REQ handler finish: each holds a handler permit
+    /// until it returns.
+    async fn wait_for_handlers(state: &AppState, permits: usize) {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while state.handler_semaphore.available_permits() < permits {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("spawned handlers finish");
+    }
+
+    fn frames(rx: &mut mpsc::Receiver<WsMessage>) -> Vec<serde_json::Value> {
+        std::iter::from_fn(|| rx.try_recv().ok())
+            .map(|message| match message {
+                WsMessage::Text(text) => serde_json::from_str(&text).expect("valid JSON frame"),
+                other => panic!("unexpected websocket message: {other:?}"),
+            })
+            .collect()
+    }
+
+    /// `REQ x` whose handler has not run yet, then `CLOSE x`: the request is
+    /// retired with the ID. When its handler runs, it sends nothing after the
+    /// acknowledgement and registers nothing. The read loop admits before it
+    /// spawns, so this holds however the handler is scheduled.
+    /// Mutation: admit inside the spawned handler instead → the handler's
+    /// refusal follows `CLOSED` → RED.
+    #[tokio::test]
+    async fn close_fences_a_req_whose_handler_has_not_run() {
+        for sync in [false, true] {
+            let state = crate::state::tests::test_state().await;
+            let permits = state.handler_semaphore.available_permits();
+            // Unauthenticated: admission quotas do not apply, and the handler
+            // refuses without a database.
+            let (conn, mut rx) = test_conn_with_auth(AuthState::Failed);
+            let filter = if sync {
+                serde_json::json!({"artifact": "current", "#h": [uuid::Uuid::new_v4().to_string()]})
+            } else {
+                serde_json::json!({"kinds": [1]})
+            };
+            let req = serde_json::json!(["REQ", "x", filter]).to_string();
+            handle_text_message(req, Arc::clone(&conn), Arc::clone(&state)).await;
+            let close = serde_json::json!(["CLOSE", "x"]).to_string();
+            handle_text_message(close, Arc::clone(&conn), Arc::clone(&state)).await;
+            wait_for_handlers(&state, permits).await;
+
+            assert_eq!(
+                frames(&mut rx),
+                vec![serde_json::json!(["CLOSED", "x", ""])],
+                "sync: {sync}"
+            );
+            assert!(conn.subscriptions.lock().await.is_empty(), "sync: {sync}");
+            assert!(
+                state.sub_registry.get_filters(conn.conn_id, "x").is_none(),
+                "sync: {sync}"
+            );
+        }
+    }
+
+    /// Two same-ID REQs whose handlers have not run: only the newer one may
+    /// answer. The older one's refusal must not close the ID the newer one
+    /// owns. Mutation: refuse in-handler with an unconditional retire → two
+    /// NOTICE/CLOSED pairs → RED.
+    #[tokio::test]
+    async fn only_the_newest_same_id_req_answers() {
+        let state = crate::state::tests::test_state().await;
+        let permits = state.handler_semaphore.available_permits();
+        let (conn, mut rx) = test_conn_with_auth(AuthState::Failed);
+        for kind in [1, 7] {
+            let req = serde_json::json!(["REQ", "x", {"kinds": [kind]}]).to_string();
+            handle_text_message(req, Arc::clone(&conn), Arc::clone(&state)).await;
+        }
+        wait_for_handlers(&state, permits).await;
+
+        assert_eq!(
+            frames(&mut rx),
+            vec![
+                serde_json::json!(["NOTICE", "auth-required: authenticate before subscribing"]),
+                serde_json::json!(["CLOSED", "x", "auth-required: not authenticated"]),
+            ]
+        );
+        assert!(conn.subscriptions.lock().await.is_empty());
+    }
+
     /// An ordinary REQ that `handle_req` itself refuses, before it claims its
     /// ID, still replaces the old holder (NIP-01): a live subscription or an
     /// artifact sync must not keep using an ID the client was told is closed.
@@ -2942,7 +3066,13 @@ pub(crate) mod tests {
         {
             let mut subs = conn.subscriptions.lock().await;
             for index in 0..handlers::req::MAX_SUBSCRIPTIONS {
-                subs.insert(format!("held-{index}"), 0);
+                subs.insert(
+                    format!("held-{index}"),
+                    handlers::close::SubOwner {
+                        owner: 0,
+                        claimed: true,
+                    },
+                );
             }
         }
         while conn
@@ -5758,7 +5888,13 @@ pub(crate) mod terminal_close_delivery_tests {
         });
         drop(effect_rx.await.expect("effect permit"));
         let mut subscriptions = subs_rx.await.expect("subscription lock");
-        subscriptions.insert("s".to_owned(), 1);
+        subscriptions.insert(
+            "s".to_owned(),
+            handlers::close::SubOwner {
+                owner: 1,
+                claimed: true,
+            },
+        );
         let mut socket = connected.await.expect("raw client");
 
         let elapsed = time_to_eof(&mut socket, std::time::Instant::now()).await;

crates/buzz-relay/src/handlers/artifact_sync.rs

@@ -20,7 +20,6 @@ use tokio_util::sync::CancellationToken;
 use tracing::warn;
 use uuid::Uuid;
 
-use super::close::refuse;
 use crate::access::ReadLog;
 use crate::connection::{AuthState, ConnectionState};
 use crate::protocol::RelayMessage;
@@ -353,10 +352,14 @@ pub(crate) fn live_filter(channels: &[Uuid]) -> Filter {
         )
 }
 
-/// Serve one sync `REQ`. `permit` is the handler slot; it is held only for
-/// the initial read.
+/// Serve one sync `REQ` that `close::admit` admitted with `owner`. `permit`
+/// is the handler slot; it is held only for the initial read. Every exit
+/// before the claim refuses through `close_if_owner`, so a sync that a newer
+/// same-ID request or a `CLOSE` superseded sends nothing and registers
+/// nothing.
 pub async fn handle_sync(
     sub_id: String,
+    owner: u64,
     filter: Result<ArtifactSyncFilter, &'static str>,
     conn: Arc<ConnectionState>,
     state: Arc<AppState>,
@@ -365,7 +368,14 @@ pub async fn handle_sync(
     let filter = match filter {
         Ok(filter) => filter,
         Err(reason) => {
-            refuse(&sub_id, &format!("invalid: {reason}"), &conn, &state).await;
+            super::close::close_if_owner(
+                &sub_id,
+                owner,
+                Some(&format!("invalid: {reason}")),
+                &conn,
+                &state,
+            )
+            .await;
             ReadLog::refused_sync(&conn, &sub_id, None, "invalid_filter");
             return;
         }
@@ -375,7 +385,14 @@ pub async fn handle_sync(
     let (reader, bound) = match conn.auth_state_snapshot() {
         AuthState::Authenticated(ctx) => {
             if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) {
-                refuse(&sub_id, "restricted: insufficient scope", &conn, &state).await;
+                super::close::close_if_owner(
+                    &sub_id,
+                    owner,
+                    Some("restricted: insufficient scope"),
+                    &conn,
+                    &state,
+                )
+                .await;
                 access.deny("insufficient_scope");
                 return;
             }
@@ -383,7 +400,8 @@ pub async fn handle_sync(
             // 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)) {
-                    refuse(&sub_id, RESTRICTED, &conn, &state).await;
+                    super::close::close_if_owner(&sub_id, owner, Some(RESTRICTED), &conn, &state)
+                        .await;
                     access.deny("not_channel_member");
                     return;
                 }
@@ -391,7 +409,14 @@ pub async fn handle_sync(
             (ctx.pubkey.to_bytes().to_vec(), ctx.channel_ids)
         }
         _ => {
-            refuse(&sub_id, "auth-required: not authenticated", &conn, &state).await;
+            super::close::close_if_owner(
+                &sub_id,
+                owner,
+                Some("auth-required: not authenticated"),
+                &conn,
+                &state,
+            )
+            .await;
             access.deny("auth_required");
             return;
         }
@@ -399,7 +424,14 @@ pub async fn handle_sync(
     let _effect = match conn.nip_fi_gate.acquire_effect().await {
         Ok(permit) => permit,
         Err(crate::nip_fi_gate::SessionExpired) => {
-            refuse(&sub_id, "restricted: authorization denied", &conn, &state).await;
+            super::close::close_if_owner(
+                &sub_id,
+                owner,
+                Some("restricted: authorization denied"),
+                &conn,
+                &state,
+            )
+            .await;
             access.deny("session_expired");
             return;
         }
@@ -408,16 +440,7 @@ pub async fn handle_sync(
     // Rule 1: live interest first, confirmed by Redis on an unbroken link.
     let start = *state.pubsub.link_state().borrow();
     let handle = Arc::new(SyncHandle::new());
-    let Some(owner) = claim(
-        &sub_id,
-        &filter.channels,
-        &handle,
-        &mut access,
-        &conn,
-        &state,
-    )
-    .await
-    else {
+    let Some(owner) = claim(&sub_id, owner, &filter.channels, &handle, &conn, &state).await else {
         return;
     };
     if !start.up || !confirm(&filter.channels, start.epoch, &conn, &state).await {
@@ -505,35 +528,23 @@ enum Stop {
     Retired,
 }
 
-/// Claim `sub_id` under the lifecycle lock. The sync's inbox is installed
-/// before fan-out registration, so no event frame is ever sent to it.
+/// Claim `sub_id` for the sync admitted with `owner`, under the lifecycle
+/// lock, retiring the previous holder. The sync's inbox is installed before
+/// fan-out registration, so no event frame is ever sent to it. `None` once the
+/// connection is closing or this sync was superseded.
 async fn claim(
     sub_id: &str,
+    owner: u64,
     channels: &[Uuid],
     handle: &Arc<SyncHandle>,
-    access: &mut ReadLog,
     conn: &ConnectionState,
     state: &AppState,
 ) -> Option<u64> {
     let mut subs = conn.subscriptions.lock().await;
-    if conn.cancel.is_cancelled() {
+    if !super::close::claim_locked(&mut subs, sub_id, owner, conn) {
         return None;
     }
-    if !subs.contains_key(sub_id) && subs.len() >= super::req::MAX_SUBSCRIPTIONS {
-        // A new ID has no old holder to retire, but a lost CLOSED would leave
-        // the client waiting on a refused sync.
-        if !conn.send(RelayMessage::closed(
-            sub_id,
-            "error: too many subscriptions",
-        )) {
-            conn.cancel.cancel();
-        }
-        access.deny("too_many_subscriptions");
-        return None;
-    }
-    super::close::retire_locked(&mut subs, sub_id, conn, state).await;
-    let owner = super::close::next_owner();
-    subs.insert(sub_id.to_owned(), owner);
+    super::close::retire_holder_locked(sub_id, conn, state).await;
     state
         .artifact_syncs
         .insert(conn.conn_id, sub_id, Arc::clone(handle));
@@ -678,7 +689,7 @@ async fn send_reserved(
         .await
         .map_err(|_| Stop::Close("error: connection closed"))?;
     let subs = conn.subscriptions.lock().await;
-    if subs.get(sub_id) != Some(&owner) {
+    if !super::close::owns(&subs, sub_id, owner) {
         return Err(Stop::Retired);
     }
     let pending = unless_resync.map(|handle| {
@@ -703,7 +714,7 @@ async fn try_send_owned(
     conn: &ConnectionState,
 ) -> Result<(), Stop> {
     let subs = conn.subscriptions.lock().await;
-    if subs.get(sub_id) != Some(&owner) {
+    if !super::close::owns(&subs, sub_id, owner) {
         return Err(Stop::Retired);
     }
     conn.send_tx
@@ -966,10 +977,17 @@ mod tests {
         conn.subscriptions
             .try_lock()
             .unwrap()
-            .insert("s".into(), owner);
+            .insert("s".into(), held(owner));
         (conn, rx)
     }
 
+    fn held(owner: u64) -> super::super::close::SubOwner {
+        super::super::close::SubOwner {
+            owner,
+            claimed: true,
+        }
+    }
+
     /// A live frame goes out only while its request still owns the ID.
     /// Mutation: drop the owner check in `try_send_owned` → RED.
     #[tokio::test]
@@ -985,7 +1003,7 @@ mod tests {
             Err(Stop::Retired)
         ));
         assert!(rx.try_recv().is_err());
-        conn.subscriptions.lock().await.insert("s".into(), 1);
+        conn.subscriptions.lock().await.insert("s".into(), held(1));
         assert!(try_send_owned("s", 1, "x".into(), &conn).await.is_ok());
         assert!(rx.try_recv().is_ok());
     }
@@ -1006,7 +1024,7 @@ mod tests {
         // Polled once, the frame is now waiting in `reserve`.
         assert!(futures_util::poll!(&mut waiting).is_pending());
         // A replacing REQ takes the ID while the frame waits.
-        conn.subscriptions.lock().await.insert("s".into(), 2);
+        conn.subscriptions.lock().await.insert("s".into(), held(2));
         rx.recv().await.unwrap();
         assert!(matches!(waiting.await, Err(Stop::Retired)));
         let mut rest = 0;

crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs

@@ -153,13 +153,11 @@ fn start_sync(
         .clone()
         .try_acquire_owned()
         .expect("permit");
-    tokio::spawn(super::artifact_sync::handle_sync(
-        "s".into(),
-        filter,
-        Arc::clone(conn),
-        Arc::clone(&f.state),
-        permit,
-    ))
+    let (conn, state) = (Arc::clone(conn), Arc::clone(&f.state));
+    tokio::spawn(async move {
+        let owner = super::close::admit("s", &conn).await.expect("admit");
+        super::artifact_sync::handle_sync("s".into(), owner, filter, conn, state, permit).await;
+    })
 }
 
 async fn frame(rx: &mut mpsc::Receiver<WsMessage>) -> Value {
@@ -496,10 +494,12 @@ async fn close_during_a_live_head_read_sends_nothing_more() {
         .try_acquire_owned()
         .expect("permit");
     let pause = Arc::new(super::close::test_seam::Pause::default());
+    let token = super::close::admit("s", &conn).await.expect("admit");
     let task = tokio::spawn(super::artifact_sync::test_seam::LIVE_SEND_PAUSE.scope(
         Arc::clone(&pause),
         super::artifact_sync::handle_sync(
             "s".into(),
+            token,
             filter,
             Arc::clone(&conn),
             Arc::clone(&f.state),

crates/buzz-relay/src/handlers/close.rs

@@ -25,12 +25,14 @@ pub async fn handle_close(sub_id: String, conn: Arc<ConnectionState>, state: Arc
     debug!(conn_id = %conn.conn_id, sub_id = %sub_id, "Subscription closed");
 }
 
-/// Refuse a `REQ` or 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. Every refusal before a request claims its ID goes through here. If
-/// the frame cannot be queued, the connection is cancelled: the client would
-/// otherwise never learn that the old holder is gone.
+/// Refuse a `REQ` or sync `REQ` as it arrives, before it is admitted, 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 and any
+/// request still pending for the ID are retired before the frame, under the
+/// lifecycle lock. Only the connection's read loop calls this, in arrival
+/// order; an admitted request refuses with `close_if_owner`. If the frame
+/// cannot be queued, the connection is cancelled: the client would otherwise
+/// never learn that the old holder is gone.
 pub(crate) async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, state: &AppState) {
     let mut subs = conn.subscriptions.lock().await;
     retire_locked(&mut subs, sub_id, conn, state).await;
@@ -52,31 +54,123 @@ pub(crate) async fn send_if_owner(
     #[cfg(test)]
     test_seam::pause_at(&test_seam::SEND_PAUSE).await;
     let subs = conn.subscriptions.lock().await;
-    subs.get(sub_id) == Some(&owner) && conn.send(frame)
+    owns(&subs, sub_id, owner) && conn.send(frame)
 }
 
-/// Monotonic source of subscription owner tokens. Each REQ claims its sub ID
-/// with a fresh token, so a superseded request can tell it no longer owns it.
+/// Monotonic source of subscription owner tokens. Each REQ takes its sub ID
+/// with a fresh token when it arrives, so a request that a later same-ID REQ
+/// or a `CLOSE` superseded can tell it no longer owns the ID.
 static NEXT_OWNER: AtomicU64 = AtomicU64::new(1);
 
 pub(crate) fn next_owner() -> u64 {
     NEXT_OWNER.fetch_add(1, Ordering::Relaxed)
 }
 
+/// What the lifecycle map holds for a subscription ID: the token of the
+/// newest admitted request, and whether that request has claimed the ID.
+/// Until it claims, whatever is registered for the ID belongs to an older
+/// request it is replacing.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct SubOwner {
+    /// The request's token from `next_owner`.
+    pub owner: u64,
+    /// The request has claimed the ID: what is registered for it is its own.
+    pub claimed: bool,
+}
+
+/// Whether the request holding `owner` still owns `sub_id`. The caller holds
+/// the lifecycle lock.
+pub(crate) fn owns(subs: &HashMap<String, SubOwner>, sub_id: &str, owner: u64) -> bool {
+    subs.get(sub_id).is_some_and(|held| held.owner == owner)
+}
+
+/// Claim `sub_id` for the request holding `owner`: `false` once the
+/// connection is closing or a `CLOSE` or newer same-ID REQ superseded it.
+/// The caller holds the lifecycle lock and, on `true`, replaces whatever is
+/// registered for the ID in the same critical section.
+pub(crate) fn claim_locked(
+    subs: &mut HashMap<String, SubOwner>,
+    sub_id: &str,
+    owner: u64,
+    conn: &ConnectionState,
+) -> bool {
+    if conn.cancel.is_cancelled() {
+        return false;
+    }
+    match subs.get_mut(sub_id) {
+        Some(held) if held.owner == owner => {
+            held.claimed = true;
+            true
+        }
+        _ => false,
+    }
+}
+
+/// Why `admit` did not admit a request.
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum NotAdmitted {
+    /// The connection is closing; the request exits silently.
+    Closing,
+    /// A new ID past `MAX_SUBSCRIPTIONS`; `CLOSED` was queued.
+    TooManySubscriptions,
+}
+
+/// Admit a `REQ` or sync `REQ` for `sub_id` as it arrives, before its handler
+/// is spawned, and return its owner token. Admission order is arrival order:
+/// the newest request owns the ID from this point, so an older request still
+/// in its access checks can no longer claim, refuse, or send for the ID, and a
+/// `CLOSE` retires a request that has not yet claimed. The previous holder
+/// stays registered until this request claims or is refused.
+pub(crate) async fn admit(sub_id: &str, conn: &ConnectionState) -> Result<u64, NotAdmitted> {
+    let mut subs = conn.subscriptions.lock().await;
+    if conn.cancel.is_cancelled() {
+        return Err(NotAdmitted::Closing);
+    }
+    if !subs.contains_key(sub_id) && subs.len() >= super::req::MAX_SUBSCRIPTIONS {
+        // A new ID has no old holder to retire, but a lost CLOSED would leave
+        // the client waiting on a refused request.
+        if !conn.send(RelayMessage::closed(
+            sub_id,
+            "error: too many subscriptions",
+        )) {
+            conn.cancel.cancel();
+        }
+        return Err(NotAdmitted::TooManySubscriptions);
+    }
+    let owner = next_owner();
+    subs.insert(
+        sub_id.to_owned(),
+        SubOwner {
+            owner,
+            claimed: false,
+        },
+    );
+    Ok(owner)
+}
+
 // Lifecycle lock: `conn.subscriptions` serializes every change to a
 // connection's subscription state — map token, registry entry, topic
 // retention, and the terminal CLOSED frame. Neither the registry nor pubsub
-// takes it, so holding it across their awaits cannot invert lock order.
+// takes it, so holding it across their awaits cannot invert lock order. The
+// map holds, per ID, the newest admitted request's `SubOwner`, whether it has
+// claimed (registered) yet or is still pending.
 
 /// Drop `sub_id` from the map, the fan-out index, and its topic retention.
 /// The caller holds the lifecycle lock (`subs`).
 pub(crate) async fn retire_locked(
-    subs: &mut HashMap<String, u64>,
+    subs: &mut HashMap<String, SubOwner>,
     sub_id: &str,
     conn: &ConnectionState,
     state: &AppState,
 ) {
     subs.remove(sub_id);
+    retire_holder_locked(sub_id, conn, state).await;
+}
+
+/// Drop the registered holder of `sub_id` from the fan-out index and its topic
+/// retention, but keep the map token: the newer request that owns the ID is
+/// replacing it. The caller holds the lifecycle lock.
+pub(crate) async fn retire_holder_locked(sub_id: &str, conn: &ConnectionState, state: &AppState) {
     state.artifact_syncs.remove(conn.conn_id, sub_id);
     if let Some(removed) = state.sub_registry.remove_subscription(conn.conn_id, sub_id) {
         #[cfg(test)]
@@ -89,18 +183,47 @@ pub(crate) async fn retire_locked(
 /// send `closed` (if any) before releasing the lifecycle lock, so a replacing
 /// REQ can never claim the ID between the teardown and its terminal frame.
 /// Returns whether this request still owned the ID. A superseded request
-/// touches nothing and says nothing.
+/// touches nothing and says nothing. An admitted request refuses through here:
+/// owning the ID, it retires the previous holder (NIP-01 replacement) before
+/// its `CLOSED`.
 pub(crate) async fn close_if_owner(
     sub_id: &str,
     owner: u64,
     closed: Option<&str>,
     conn: &ConnectionState,
     state: &AppState,
+) -> bool {
+    close_owned(sub_id, owner, None, closed, conn, state).await
+}
+
+/// `close_if_owner` with `CLOSED reason`, preceded by `NOTICE notice` in the
+/// same critical section, so a superseded request sends neither.
+pub(crate) async fn refuse_if_owner_with_notice(
+    sub_id: &str,
+    owner: u64,
+    notice: &str,
+    reason: &str,
+    conn: &ConnectionState,
+    state: &AppState,
+) -> bool {
+    close_owned(sub_id, owner, Some(notice), Some(reason), conn, state).await
+}
+
+async fn close_owned(
+    sub_id: &str,
+    owner: u64,
+    notice: Option<&str>,
+    closed: Option<&str>,
+    conn: &ConnectionState,
+    state: &AppState,
 ) -> bool {
     let mut subs = conn.subscriptions.lock().await;
-    if subs.get(sub_id) != Some(&owner) {
+    if !owns(&subs, sub_id, owner) {
         return false;
     }
+    if let Some(notice) = notice {
+        conn.send(RelayMessage::notice(notice));
+    }
     retire_locked(&mut subs, sub_id, conn, state).await;
     #[cfg(test)]
     test_seam::pause_at(&test_seam::PAUSE).await;

crates/buzz-relay/src/handlers/req.rs

@@ -53,50 +53,69 @@ pub(crate) const FILTER_QUERY_CONCURRENCY: usize = 4;
 // the range fails the build.
 const _: () = assert!(FILTER_QUERY_CONCURRENCY >= 2 && FILTER_QUERY_CONCURRENCY <= 8);
 
-/// Handle a REQ message: register the subscription, deliver historical events, then send EOSE.
+/// Admit a `REQ` and handle it inline. The read loop admits before it spawns
+/// the handler, so arrival order decides ownership; this is the same path for
+/// callers that run the handler in place.
+#[cfg(test)]
 pub async fn handle_req(
     sub_id: String,
     filters: Vec<Filter>,
     before_ids: Vec<Option<Vec<u8>>>,
     conn: Arc<ConnectionState>,
     state: Arc<AppState>,
+) {
+    let Ok(owner) = super::close::admit(&sub_id, &conn).await else {
+        return;
+    };
+    handle_admitted_req(sub_id, owner, filters, before_ids, conn, state).await;
+}
+
+/// Handle a REQ message: register the subscription, deliver historical events, then send EOSE.
+///
+/// The REQ was admitted by `close::admit` with `owner`. Every exit before
+/// the claim refuses through `close_if_owner`, so a request that a newer
+/// same-ID `REQ` or a `CLOSE` superseded during its access checks sends
+/// nothing and registers nothing.
+pub async fn handle_admitted_req(
+    sub_id: String,
+    owner: u64,
+    filters: Vec<Filter>,
+    before_ids: Vec<Option<Vec<u8>>>,
+    conn: Arc<ConnectionState>,
+    state: Arc<AppState>,
 ) {
     let mut access = crate::access::ReadLog::req(&conn, &sub_id, &filters);
     let (conn_id, pubkey_bytes, token_channel_ids) = {
         match conn.auth_state_snapshot() {
             AuthState::Authenticated(ctx) => {
                 if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) {
-                    conn.send(RelayMessage::notice("restricted: insufficient scope"));
-                    super::close::refuse(&sub_id, "restricted: insufficient scope", &conn, &state)
-                        .await;
+                    super::close::refuse_if_owner_with_notice(
+                        &sub_id,
+                        owner,
+                        "restricted: insufficient scope",
+                        "restricted: insufficient scope",
+                        &conn,
+                        &state,
+                    )
+                    .await;
                     access.deny("insufficient_scope");
                     return;
                 }
 
                 let pk_bytes = ctx.pubkey.to_bytes().to_vec();
 
-                let subs = conn.subscriptions.lock().await;
-                if !subs.contains_key(&sub_id) && subs.len() >= MAX_SUBSCRIPTIONS {
-                    // A new ID has no old holder to retire, but a lost CLOSED
-                    // would leave the client waiting on a refused request.
-                    if !conn.send(RelayMessage::closed(
-                        &sub_id,
-                        "error: too many subscriptions",
-                    )) {
-                        conn.cancel.cancel();
-                    }
-                    access.deny("too_many_subscriptions");
-                    return;
-                }
-
                 (conn.conn_id, pk_bytes, ctx.channel_ids.clone())
             }
             _ => {
-                conn.send(RelayMessage::notice(
+                super::close::refuse_if_owner_with_notice(
+                    &sub_id,
+                    owner,
                     "auth-required: authenticate before subscribing",
-                ));
-                super::close::refuse(&sub_id, "auth-required: not authenticated", &conn, &state)
-                    .await;
+                    "auth-required: not authenticated",
+                    &conn,
+                    &state,
+                )
+                .await;
                 access.deny("auth_required");
                 return;
             }
@@ -107,9 +126,10 @@ pub async fn handle_req(
     let requested_channel_ids = match extract_channel_ids_from_filters_limited(&filters) {
         Ok(ids) => ids,
         Err(()) => {
-            super::close::refuse(
+            super::close::close_if_owner(
                 &sub_id,
-                "restricted: too many explicit channels",
+                owner,
+                Some("restricted: too many explicit channels"),
                 &conn,
                 &state,
             )
@@ -138,7 +158,14 @@ pub async fn handle_req(
             Ok(ids) => ids,
             Err(e) => {
                 warn!(conn_id = %conn_id, error_kind = %e.kind(), "Failed to get accessible channels");
-                super::close::refuse(&sub_id, DATABASE_ERROR_CLOSED, &conn, &state).await;
+                super::close::close_if_owner(
+                    &sub_id,
+                    owner,
+                    Some(DATABASE_ERROR_CLOSED),
+                    &conn,
+                    &state,
+                )
+                .await;
                 access.error("database_error");
                 return;
             }
@@ -163,7 +190,14 @@ pub async fn handle_req(
         Ok(ids) => ids,
         Err(e) => {
             warn!(conn_id = %conn_id, error_kind = %e.kind(), "Failed to get readable named-only channels");
-            super::close::refuse(&sub_id, DATABASE_ERROR_CLOSED, &conn, &state).await;
+            super::close::close_if_owner(
+                &sub_id,
+                owner,
+                Some(DATABASE_ERROR_CLOSED),
+                &conn,
+                &state,
+            )
+            .await;
             access.error("database_error");
             return;
         }
@@ -212,7 +246,14 @@ pub async fn handle_req(
                     error_kind = %e.kind(),
                     "Channel membership confirmation failed"
                 );
-                super::close::refuse(&sub_id, DATABASE_ERROR_CLOSED, &conn, &state).await;
+                super::close::close_if_owner(
+                    &sub_id,
+                    owner,
+                    Some(DATABASE_ERROR_CLOSED),
+                    &conn,
+                    &state,
+                )
+                .await;
                 access.error("database_error");
                 return;
             }
@@ -232,7 +273,14 @@ pub async fn handle_req(
         .await
         {
             warn!(conn_id = %conn_id, error_kind = %e.kind(), "Failed to classify repaired channels");
-            super::close::refuse(&sub_id, DATABASE_ERROR_CLOSED, &conn, &state).await;
+            super::close::close_if_owner(
+                &sub_id,
+                owner,
+                Some(DATABASE_ERROR_CLOSED),
+                &conn,
+                &state,
+            )
+            .await;
             access.error("database_error");
             return;
         }
@@ -258,7 +306,14 @@ pub async fn handle_req(
         .as_ref()
         .is_some_and(|authorized| authorized.is_empty())
     {
-        super::close::refuse(&sub_id, "restricted: not a channel member", &conn, &state).await;
+        super::close::close_if_owner(
+            &sub_id,
+            owner,
+            Some("restricted: not a channel member"),
+            &conn,
+            &state,
+        )
+        .await;
         access.deny("not_channel_member");
         return;
     }
@@ -278,14 +333,20 @@ pub async fn handle_req(
             Err(crate::nip_fi_gate::SessionExpired) => {
                 // Fix 4: [FI-TRACE-DENIAL-ORACLE] gate is off_mode when no assertion
                 // exists, so SessionExpired here always implies an active FI session.
-                super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state)
-                    .await;
+                super::close::close_if_owner(
+                    &sub_id,
+                    owner,
+                    Some("restricted: authorization denied"),
+                    &conn,
+                    &state,
+                )
+                .await;
                 access.deny("session_expired");
                 return;
             }
         };
         // One-shot like search: accepting it replaces the ID's old holder.
-        let Some(owner) = claim_search_subscription(&sub_id, &conn, &state).await else {
+        let Some(owner) = claim_admitted_search(&sub_id, owner, &conn, &state).await else {
             return;
         };
         unless_cancelled(
@@ -316,9 +377,10 @@ pub async fn handle_req(
     if channel_id.is_none() {
         let authed_pubkey_hex = hex::encode(&pubkey_bytes);
         if !p_gated_filters_authorized(&filters, &authed_pubkey_hex) {
-            super::close::refuse(
+            super::close::close_if_owner(
                 &sub_id,
-                "restricted: p-gated events require #p matching your pubkey",
+                owner,
+                Some("restricted: p-gated events require #p matching your pubkey"),
                 &conn,
                 &state,
             )
@@ -327,9 +389,10 @@ pub async fn handle_req(
             return;
         }
         if !engram_filters_authorized(&filters, &authed_pubkey_hex) {
-            super::close::refuse(
+            super::close::close_if_owner(
                 &sub_id,
-                "restricted: agent-engram reads require authors=[self] or #p=[self]",
+                owner,
+                Some("restricted: agent-engram reads require authors=[self] or #p=[self]"),
                 &conn,
                 &state,
             )
@@ -338,9 +401,10 @@ pub async fn handle_req(
             return;
         }
         if !author_only_filters_authorized(&filters, &authed_pubkey_hex) {
-            super::close::refuse(
+            super::close::close_if_owner(
                 &sub_id,
-                "restricted: author-only kinds require authors=[self]",
+                owner,
+                Some("restricted: author-only kinds require authors=[self]"),
                 &conn,
                 &state,
             )
@@ -357,9 +421,10 @@ pub async fn handle_req(
     let has_search = filters.iter().any(|f| f.search.is_some());
     if has_search {
         if filters.iter().any(|f| f.search.is_none()) {
-            super::close::refuse(
+            super::close::close_if_owner(
                 &sub_id,
-                "error: mixed search and non-search filters not supported",
+                owner,
+                Some("error: mixed search and non-search filters not supported"),
                 &conn,
                 &state,
             )
@@ -376,13 +441,19 @@ pub async fn handle_req(
             Ok(permit) => permit,
             Err(crate::nip_fi_gate::SessionExpired) => {
                 // Fix 4: [FI-TRACE-DENIAL-ORACLE]
-                super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state)
-                    .await;
+                super::close::close_if_owner(
+                    &sub_id,
+                    owner,
+                    Some("restricted: authorization denied"),
+                    &conn,
+                    &state,
+                )
+                .await;
                 access.deny("session_expired");
                 return;
             }
         };
-        let Some(owner) = claim_search_subscription(&sub_id, &conn, &state).await else {
+        let Some(owner) = claim_admitted_search(&sub_id, owner, &conn, &state).await else {
             return;
         };
         let search = async {
@@ -425,7 +496,14 @@ pub async fn handle_req(
         Ok(permit) => permit,
         Err(crate::nip_fi_gate::SessionExpired) => {
             // Fix 4: [FI-TRACE-DENIAL-ORACLE]
-            super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state).await;
+            super::close::close_if_owner(
+                &sub_id,
+                owner,
+                Some("restricted: authorization denied"),
+                &conn,
+                &state,
+            )
+            .await;
             access.deny("session_expired");
             return;
         }
@@ -447,7 +525,7 @@ pub async fn handle_req(
             )
         })
         .collect();
-    let Some(owner) = claim_live_subscription(&sub_id, live_filters, &conn, &state).await else {
+    let Some(owner) = claim_admitted_live(&sub_id, owner, live_filters, &conn, &state).await else {
         return;
     };
 
@@ -1665,24 +1743,38 @@ async fn handle_huddle_liveness_req(
     }
 }
 
-/// Claim `sub_id` for a live REQ under the lifecycle lock: record a fresh
-/// owner token, register for fan-out (replacing any same-ID entry), retain its
-/// topics, and release the replaced scope's. Each filter carries its own
-/// server-resolved live scope; the subscription retains each topic in their
-/// union once. Returns the owner token, or `None` once the connection is
-/// closing — the caller then exits silently.
+/// Admit and claim `sub_id` for a live REQ in one step, as a handler run in
+/// place would. `None` if the connection is closing or a newer request took
+/// the ID between the two steps.
+#[cfg(test)]
 pub(crate) async fn claim_live_subscription(
     sub_id: &str,
     filters: Vec<(Filter, SubscriptionScope)>,
     conn: &ConnectionState,
     state: &AppState,
+) -> Option<u64> {
+    let owner = super::close::admit(sub_id, conn).await.ok()?;
+    claim_admitted_live(sub_id, owner, filters, conn, state).await
+}
+
+/// Claim `sub_id` for the live REQ admitted with `owner`, under the lifecycle
+/// lock: register for fan-out (replacing any same-ID entry), retain its
+/// topics, and release the replaced scope's. Each filter carries its own
+/// server-resolved live scope; the subscription retains each topic in their
+/// union once. Returns `owner`, or `None` once the connection is closing or a
+/// newer same-ID REQ or a `CLOSE` superseded this request — the caller then
+/// exits silently.
+pub(crate) async fn claim_admitted_live(
+    sub_id: &str,
+    owner: u64,
+    filters: Vec<(Filter, SubscriptionScope)>,
+    conn: &ConnectionState,
+    state: &AppState,
 ) -> Option<u64> {
     let mut subs = conn.subscriptions.lock().await;
-    if conn.cancel.is_cancelled() {
+    if !super::close::claim_locked(&mut subs, sub_id, owner, conn) {
         return None;
     }
-    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);
@@ -1711,22 +1803,33 @@ pub(crate) async fn claim_live_subscription(
     Some(owner)
 }
 
-/// Claim `sub_id` for a one-shot search or huddle-liveness REQ under the
-/// lifecycle lock. Accepting it retires whatever holds the ID (NIP-01
-/// replacement), even though neither registers for fan-out. `None` once the
-/// connection is closing.
+/// Admit and claim `sub_id` for a one-shot REQ in one step, as a handler run
+/// in place would.
+#[cfg(test)]
 async fn claim_search_subscription(
     sub_id: &str,
     conn: &ConnectionState,
     state: &AppState,
+) -> Option<u64> {
+    let owner = super::close::admit(sub_id, conn).await.ok()?;
+    claim_admitted_search(sub_id, owner, conn, state).await
+}
+
+/// Claim `sub_id` for the one-shot search or huddle-liveness REQ admitted with
+/// `owner`, under the lifecycle lock. Accepting it retires whatever holds the
+/// ID (NIP-01 replacement), even though neither registers for fan-out. `None`
+/// once the connection is closing or this request was superseded.
+async fn claim_admitted_search(
+    sub_id: &str,
+    owner: u64,
+    conn: &ConnectionState,
+    state: &AppState,
 ) -> Option<u64> {
     let mut subs = conn.subscriptions.lock().await;
-    if conn.cancel.is_cancelled() {
+    if !super::close::claim_locked(&mut subs, sub_id, owner, conn) {
         return None;
     }
-    super::close::retire_locked(&mut subs, sub_id, conn, state).await;
-    let owner = super::close::next_owner();
-    subs.insert(sub_id.to_string(), owner);
+    super::close::retire_holder_locked(sub_id, conn, state).await;
     Some(owner)
 }
 
@@ -2106,6 +2209,15 @@ mod tests {
             .collect()
     }
 
+    /// The owner token the lifecycle map holds for `sub_id`.
+    async fn token(conn: &ConnectionState, sub_id: &str) -> Option<u64> {
+        conn.subscriptions
+            .lock()
+            .await
+            .get(sub_id)
+            .map(|held| held.owner)
+    }
+
     fn registered(state: &AppState, conn: &ConnectionState, sub_id: &str) -> bool {
         state
             .sub_registry
@@ -2182,8 +2294,10 @@ mod tests {
                 .acquire_owned()
                 .await
                 .expect("permit");
+            let owner = super::super::close::admit("s", &conn).await.expect("admit");
             super::super::artifact_sync::handle_sync(
                 "s".into(),
+                owner,
                 filter,
                 Arc::clone(&conn),
                 Arc::clone(&state),
@@ -2268,7 +2382,7 @@ mod tests {
 
         close_timed_out_subscription("x", old, &conn, &state).await;
 
-        assert_eq!(conn.subscriptions.lock().await.get("x"), Some(&new));
+        assert_eq!(token(&conn, "x").await, Some(new));
         assert!(
             registered(&state, &conn, "x"),
             "replacement must stay registered"
@@ -2320,7 +2434,7 @@ mod tests {
         .expect("claim");
         close_timed_out_subscription("x", search, &conn, &state).await;
 
-        assert_eq!(conn.subscriptions.lock().await.get("x"), Some(&live));
+        assert_eq!(token(&conn, "x").await, Some(live));
         assert!(registered(&state, &conn, "x"));
         assert_eq!(state.pubsub.topic_refcount(&conn.tenant, topic).await, 1);
         assert!(closed_frames(&mut send_rx).is_empty());
@@ -2440,10 +2554,13 @@ mod tests {
                     _ = pause.reached.notified() => {}
                 }
                 if case == "replaced" {
-                    conn.subscriptions
-                        .lock()
-                        .await
-                        .insert("s".to_string(), owner + 1_000_000);
+                    conn.subscriptions.lock().await.insert(
+                        "s".to_string(),
+                        crate::handlers::close::SubOwner {
+                            owner: owner + 1_000_000,
+                            claimed: false,
+                        },
+                    );
                 }
                 pause.resume.notify_one();
                 search.await;
@@ -2498,7 +2615,13 @@ mod tests {
             task.await.expect("task");
         }
 
-        let owner = *conn.subscriptions.lock().await.get("x").expect("x owned");
+        let owner = conn
+            .subscriptions
+            .lock()
+            .await
+            .get("x")
+            .expect("x owned")
+            .owner;
         assert_ne!(owner, stale);
         assert!(registered(&state, &conn, "x"));
         assert_eq!(state.pubsub.topic_refcount(&conn.tenant, topic).await, 1);
@@ -2571,7 +2694,7 @@ mod tests {
             closed_frames(&mut send_rx),
             vec![format!(r#"["CLOSED","x","{QUERY_TIMED_OUT_CLOSED}"]"#)]
         );
-        assert_eq!(conn.subscriptions.lock().await.get("x"), Some(&new));
+        assert_eq!(token(&conn, "x").await, Some(new));
         assert!(subscribed(&state, &conn, b) && !subscribed(&state, &conn, a));
         let refcount = |c| {
             state
@@ -2633,7 +2756,7 @@ mod tests {
         revoke.await;
         let new = replacement.await.expect("claim");
 
-        assert_eq!(conn.subscriptions.lock().await.get("x"), Some(&new));
+        assert_eq!(token(&conn, "x").await, Some(new));
         assert!(subscribed(&state, &conn, b) && !subscribed(&state, &conn, a));
         let refcount = |c| {
             state
@@ -2648,6 +2771,137 @@ mod tests {
         state.conn_manager.deregister(conn.conn_id);
     }
 
+    /// A revoke that evicts a holder a newer same-ID REQ is already replacing
+    /// retires the holder only: the client replaced it, so no `CLOSED` names
+    /// the ID, and the pending replacement keeps its token and still claims.
+    /// Mutation: drop the `replaced` check from the eviction → the
+    /// replacement's token is removed, its claim fails, and a `CLOSED` is
+    /// sent → RED.
+    #[tokio::test]
+    async fn revoke_of_a_replaced_holder_spares_the_pending_replacement() {
+        let state = crate::state::tests::test_state().await;
+        let (conn, mut send_rx) = lifecycle_conn();
+        let conn = Arc::new(conn);
+        state.conn_manager.register(
+            conn.conn_id,
+            conn.send_tx.clone(),
+            conn.ctrl_tx.clone(),
+            tokio::sync::mpsc::channel(1).0,
+            None,
+            conn.cancel.clone(),
+            conn.tenant.community(),
+            Arc::clone(&conn.backpressure_count),
+            Arc::clone(&conn.subscriptions),
+            3,
+            crate::state::CommunityConnectionControl::new(conn.cancel.clone()),
+        );
+        let (a, b) = (uuid::Uuid::new_v4(), uuid::Uuid::new_v4());
+        claim_live_subscription("x", scoped(&text_filters(), Some(&[a])), &conn, &state)
+            .await
+            .expect("claim");
+        // The replacement is admitted and still in its access checks.
+        let pending = super::super::close::admit("x", &conn).await.expect("admit");
+
+        crate::handlers::side_effects::evict_conn_channel_subscriptions(
+            &conn.tenant,
+            &state,
+            a,
+            conn.conn_id,
+        )
+        .await;
+
+        let refcount = |c| {
+            state
+                .pubsub
+                .topic_refcount(&conn.tenant, EventTopic::Channel(c))
+        };
+        assert!(
+            !registered(&state, &conn, "x"),
+            "the revoked holder retires"
+        );
+        assert_eq!(refcount(a).await, 0);
+        assert_eq!(token(&conn, "x").await, Some(pending));
+        assert!(closed_frames(&mut send_rx).is_empty());
+
+        let claimed = claim_admitted_live(
+            "x",
+            pending,
+            scoped(&text_filters(), Some(&[b])),
+            &conn,
+            &state,
+        )
+        .await;
+        assert_eq!(claimed, Some(pending));
+        assert!(subscribed(&state, &conn, b));
+        assert_eq!(refcount(b).await, 1);
+        state.conn_manager.deregister(conn.conn_id);
+    }
+
+    /// A request admitted before a newer same-ID request, or before a
+    /// `CLOSE`, can no longer claim: an older REQ finishing its access checks
+    /// late must not replace the newer one, and a closed ID must stay closed.
+    /// Mutation: drop the owner check from `claim_admitted_live` → the stale
+    /// claim registers → RED.
+    #[tokio::test]
+    async fn superseded_or_closed_requests_cannot_claim() {
+        let state = crate::state::tests::test_state().await;
+        let (conn, mut send_rx) = lifecycle_conn();
+        let conn = Arc::new(conn);
+        let (a, b) = (uuid::Uuid::new_v4(), uuid::Uuid::new_v4());
+        let refcount = |c| {
+            state
+                .pubsub
+                .topic_refcount(&conn.tenant, EventTopic::Channel(c))
+        };
+
+        let older = super::super::close::admit("x", &conn).await.expect("admit");
+        let newer = super::super::close::admit("x", &conn).await.expect("admit");
+        claim_admitted_live(
+            "x",
+            newer,
+            scoped(&text_filters(), Some(&[b])),
+            &conn,
+            &state,
+        )
+        .await
+        .expect("the newest request claims");
+        let stale = claim_admitted_live(
+            "x",
+            older,
+            scoped(&text_filters(), Some(&[a])),
+            &conn,
+            &state,
+        )
+        .await;
+        assert_eq!(stale, None);
+        assert_eq!(token(&conn, "x").await, Some(newer));
+        assert!(subscribed(&state, &conn, b) && !subscribed(&state, &conn, a));
+        assert_eq!((refcount(a).await, refcount(b).await), (0, 1));
+
+        let pending = super::super::close::admit("y", &conn).await.expect("admit");
+        super::super::close::handle_close("y".into(), Arc::clone(&conn), Arc::clone(&state)).await;
+        let late = claim_admitted_live(
+            "y",
+            pending,
+            scoped(&text_filters(), Some(&[a])),
+            &conn,
+            &state,
+        )
+        .await;
+        let late_search = claim_admitted_search("y", pending, &conn, &state).await;
+        assert_eq!((late, late_search), (None, None));
+        assert!(!registered(&state, &conn, "y"));
+        assert_eq!(refcount(a).await, 0);
+        assert!(
+            !super::super::close::close_if_owner("y", pending, Some("late"), &conn, &state).await
+        );
+        assert_eq!(
+            closed_frames(&mut send_rx),
+            vec![r#"["CLOSED","y",""]"#.to_string()],
+            "only the CLOSE acknowledgement"
+        );
+    }
+
     /// Connection cleanup fences later claims: a detached REQ task resuming
     /// after it gets `None` and leaves nothing behind.
     #[tokio::test]

crates/buzz-relay/src/handlers/side_effects.rs

@@ -268,7 +268,8 @@ async fn disable_departed_member_workflows(
 }
 
 /// Close every live channel-scoped subscription on `conn_id`, removing them from
-/// the connection's local map and sending `CLOSED restricted` for each. Runs
+/// the connection's local map and sending `CLOSED restricted` for each; a
+/// holder that a pending same-ID REQ is replacing is only retired. Runs
 /// under the connection's lifecycle lock so a concurrent same-ID claim is
 /// either fully before (and revoked) or fully after (and untouched).
 pub(crate) async fn evict_conn_channel_subscriptions(
@@ -289,6 +290,13 @@ pub(crate) async fn evict_conn_channel_subscriptions(
         channel_id,
     );
     for update in removed {
+        // A newer same-ID REQ that has not claimed yet owns the ID and is
+        // replacing this holder, so the client already treats the holder as
+        // gone (NIP-01): retire it without touching the newer request's token
+        // or sending a frame for the ID.
+        let replaced = conn_subscriptions
+            .get(&update.sub_id)
+            .is_some_and(|held| !held.claimed);
         // A multi-channel sub keeps its other channels (and its map token).
         state
             .pubsub
@@ -305,7 +313,7 @@ pub(crate) async fn evict_conn_channel_subscriptions(
                 super::close::release_scope_topics(state, tenant, &rest.scope).await;
             }
         }
-        if update.removed || sync {
+        if (update.removed || sync) && !replaced {
             conn_subscriptions.remove(&update.sub_id);
             let reason = if sync {
                 "resync: channel access changed"

scripts/run-tests.sh

@@ -103,6 +103,12 @@ run_unit_tests() {
   run_test_step "buzz-pubsub connection tests" \
     cargo test -p buzz-pubsub --lib connection::tests:: -- --nocapture
 
+  # A retain past a full subscriber command queue (no Redis dialed;
+  # infra-free). Mirrors `just test-unit`.
+  run_test_step "buzz-pubsub full-queue retain test" \
+    cargo test -p buzz-pubsub --lib -- --exact --nocapture \
+      tests::queue_full_retain_does_not_wait_and_schedules_reconciliation
+
   run_test_step "buzz-voice tests" \
     cargo test -p buzz-voice --lib -- --nocapture
 
@@ -304,6 +310,10 @@ run_unit_tests() {
       handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner \
       handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim \
       handlers::req::tests::revoke_then_replacement_keeps_replacement_whole \
+      handlers::req::tests::revoke_of_a_replaced_holder_spares_the_pending_replacement \
+      handlers::req::tests::superseded_or_closed_requests_cannot_claim \
+      connection::tests::close_fences_a_req_whose_handler_has_not_run \
+      connection::tests::only_the_newest_same_id_req_answers \
       handlers::req::tests::claims_after_connection_cleanup_are_refused \
       handlers::req::tests::dropped_terminal_frame_cancels_connection \
       handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection \