From 715d9ac104ab9e2f7eb891c64a1ceae85dfa8dab Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 6 Nov 2026 12:23:47 -0500 Subject: [PATCH 1/5] fix(relay): fence pending subscriptions on close or replacement A REQ handler is spawned and can pause in its access checks, but it took its subscription ID only when it registered. A CLOSE during that pause was acknowledged and then the REQ registered anyway; an older same-ID REQ that finished late could replace the newer one or close it with its refusal. The read loop now admits each REQ and sync REQ in arrival order before it spawns the handler, recording its owner token in the connection's subscription map. Claims and in-handler refusals (CLOSED and its NOTICE) act only while the request still owns its token, so a CLOSE or a newer same-ID request fences a pending one. A pubsub retain no longer waits on a full subscriber command queue while the caller holds the lifecycle lock: it leaves a coalesced wake-up, and the subscriber subscribes every retained topic it is missing. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- crates/buzz-pubsub/src/lib.rs | 74 ++++- crates/buzz-pubsub/src/subscriber.rs | 30 +- crates/buzz-relay/src/connection.rs | 127 +++++++- .../buzz-relay/src/handlers/artifact_sync.rs | 87 +++--- .../handlers/artifact_sync_postgres_tests.rs | 14 +- crates/buzz-relay/src/handlers/close.rs | 97 +++++- crates/buzz-relay/src/handlers/req.rs | 279 ++++++++++++++---- 7 files changed, 581 insertions(+), 127 deletions(-) diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 5eb1897ceab..3d682d3ee7a 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/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, @@ -122,6 +122,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, @@ -133,6 +137,8 @@ pub struct PubSubManager { desired_topics: subscriber::DesiredTopics, subscription_tx: mpsc::Sender, subscription_rx: Mutex>>, + /// Coalesced wake-up: reconcile desired topics with the live Redis link. + subscription_changed: Arc, broadcast_tx: broadcast::Sender, cache_invalidation_tx: broadcast::Sender, conn_control_tx: broadcast::Sender, @@ -155,7 +161,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, @@ -164,6 +170,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, @@ -187,6 +194,7 @@ impl PubSubManager { self.broadcast_tx.clone(), self.desired_topics.clone(), subscription_rx, + self.subscription_changed.clone(), self.link.clone(), ) .await; @@ -238,7 +246,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 = { @@ -250,10 +259,22 @@ impl PubSubManager { }; if should_subscribe { - let _ = self + // The wake-up is lossless and coalesced: if the bounded command + // queue is full, the subscriber still reconciles the whole + // desired-topic map without waiting for a Redis reconnect. + self.subscription_changed.notify_one(); + match self .subscription_tx - .send(subscriber::SubscriptionCommand::Subscribe(topic_key)) - .await; + .try_send(subscriber::SubscriptionCommand::Subscribe(topic_key)) + { + Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + tracing::warn!( + ?topic_key, + "pubsub subscription command queue full; reconciliation scheduled" + ); + } + } } } @@ -706,6 +727,47 @@ 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. Mutation: `send().await` instead of `try_send` + /// hangs at the timeout below; no `notify_one` fails the 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; + } + // Drain the wake-ups the fillers left (coalesced into one permit). + manager.subscription_changed.notified().await; + + 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))); + } + #[tokio::test] async fn retain_release_refcounts_and_debounces_last_release() { let pool = make_test_pool(); diff --git a/crates/buzz-pubsub/src/subscriber.rs b/crates/buzz-pubsub/src/subscriber.rs index b43ccea4615..a51acc73bb5 100644 --- a/crates/buzz-pubsub/src/subscriber.rs +++ b/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, desired_topics: DesiredTopics, mut subscription_rx: mpsc::Receiver, + subscription_changed: Arc, link: watch::Sender, ) { 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, desired_topics: DesiredTopics, subscription_rx: &mut mpsc::Receiver, + subscription_changed: &Notify, link: &watch::Sender, ) -> 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) => { @@ -202,6 +215,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, +) -> Vec { + 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() diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index ace9792e1a4..6fbace1e961 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -1489,10 +1489,31 @@ async fn handle_text_message(text: String, conn: Arc, 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, 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), ); @@ -2839,6 +2878,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) -> Vec { + 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. diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index f753c2134fa..2b91dea08ae 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/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, conn: Arc, state: Arc, @@ -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, - access: &mut ReadLog, conn: &ConnectionState, state: &AppState, ) -> Option { - let mut subs = conn.subscriptions.lock().await; - if conn.cancel.is_cancelled() { - 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"); + let subs = conn.subscriptions.lock().await; + if conn.cancel.is_cancelled() || subs.get(sub_id) != Some(&owner) { 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)); diff --git a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs index 6696877c4e3..9c8e3c5936c 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs +++ b/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) -> 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), diff --git a/crates/buzz-relay/src/handlers/close.rs b/crates/buzz-relay/src/handlers/close.rs index 62e60b39536..4c7b726cce3 100644 --- a/crates/buzz-relay/src/handlers/close.rs +++ b/crates/buzz-relay/src/handlers/close.rs @@ -25,12 +25,14 @@ pub async fn handle_close(sub_id: String, conn: Arc, 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; @@ -55,18 +57,57 @@ pub(crate) async fn send_if_owner( subs.get(sub_id) == Some(&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) } +/// 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 { + 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(), owner); + 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 token of the newest admitted request, 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`). @@ -77,6 +118,13 @@ pub(crate) async fn retire_locked( 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 +137,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) { 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; diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index b5e9347941b..551844ef7df 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/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, before_ids: Vec>>, conn: Arc, state: Arc, +) { + 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, + before_ids: Vec>>, + conn: Arc, + state: Arc, ) { 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,8 @@ 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 +1744,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 +/// 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 { + 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 the owner token, or `None` once the connection is -/// closing — the caller then exits silently. -pub(crate) async fn claim_live_subscription( +/// 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 { - let mut subs = conn.subscriptions.lock().await; - if conn.cancel.is_cancelled() { + let subs = conn.subscriptions.lock().await; + if conn.cancel.is_cancelled() || subs.get(sub_id) != Some(&owner) { 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 +1804,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 { - let mut subs = conn.subscriptions.lock().await; - if conn.cancel.is_cancelled() { + 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 { + let subs = conn.subscriptions.lock().await; + if conn.cancel.is_cancelled() || subs.get(sub_id) != Some(&owner) { 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) } @@ -2182,8 +2286,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), @@ -2648,6 +2754,53 @@ mod tests { 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!(conn.subscriptions.lock().await.get("x"), 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] From f823e878a0274a9498bfd0a4ef50ba31ca9308cd Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 6 Nov 2026 12:52:51 -0500 Subject: [PATCH 2/5] fix(relay): spare a pending replacement when its old holder is evicted The lifecycle map now records whether the newest admitted request for an ID has claimed it. A channel-access eviction that removes an old holder while a newer same-ID request is still pending retires only the holder: the client already replaced it, so the relay keeps the pending request's token and sends no CLOSED for the ID. Tests: a revoke of a replaced holder keeps the pending replacement; the pubsub subscriber recovers a dropped subscribe hint on a live Redis link; the bridge search test waits for the claim itself, not for admission. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- crates/buzz-pubsub/src/lib.rs | 110 ++++++++++++++++++ crates/buzz-relay/src/api/bridge.rs | 29 +++-- crates/buzz-relay/src/connection.rs | 25 +++- .../buzz-relay/src/handlers/artifact_sync.rs | 21 ++-- crates/buzz-relay/src/handlers/close.rs | 56 ++++++++- crates/buzz-relay/src/handlers/req.rs | 106 ++++++++++++++--- .../buzz-relay/src/handlers/side_effects.rs | 12 +- 7 files changed, 314 insertions(+), 45 deletions(-) diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 3d682d3ee7a..aa09d351c7c 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -768,6 +768,116 @@ mod tests { 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, + 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; + tokio::time::sleep(Duration::from_millis(20)).await; + // Queued after the debounced unsubscribe: once it answers, the + // subscriber has processed that unsubscribe. + manager + .confirm_topic(&filler, EventTopic::Channel(Uuid::from_u128(0))) + .await + .expect("confirm"); + 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(); diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index f5d7f18cddb..cb7c6ab6ba7 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -10827,7 +10827,6 @@ 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"); let task = tokio::spawn(handle_req( "x".into(), vec![search], @@ -10835,18 +10834,26 @@ 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; - } - assert!( - fx.state + // Search takes the ID's token when it is admitted, before its access + // checks, so the token alone does not show it has claimed. Its claim + // retires the live fan-out under the lifecycle lock, and then its + // hydrate blocks on the table lock: the live entry must be gone while + // search is still running, not only at final cleanup. + tokio::time::timeout(std::time::Duration::from_secs(10), async { + while fx + .state .sub_registry .get_filters(conn.conn_id, "x") - .is_none(), - "accepting search must retire live fan-out" - ); + .is_some() + { + tokio::task::yield_now().await; + } + }) + .await + .expect("accepting search must retire live fan-out"); + // The claim's critical section, topic release included, is over once + // the lifecycle lock is free again. + drop(conn.subscriptions.lock().await); assert_eq!(fx.state.pubsub.topic_refcount(&tenant, topic).await, 0); task.await.expect("search task"); locker.rollback().await.expect("unlock"); diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 6fbace1e961..6cd277017b8 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/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>>; +/// the newest admitted request for it (see `close::SubOwner`). +pub(crate) type ConnectionSubscriptions = Arc>>; /// 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 @@ -1662,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>>; + pub(crate) type Hold = tokio::sync::oneshot::Sender< + tokio::sync::OwnedMutexGuard>, + >; tokio::task_local! { /// Hands the test this connection's subscription lock, already held, @@ -3065,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 @@ -5881,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; diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index 2b91dea08ae..fa0cadf57e2 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync.rs @@ -540,8 +540,8 @@ async fn claim( conn: &ConnectionState, state: &AppState, ) -> Option { - let subs = conn.subscriptions.lock().await; - if conn.cancel.is_cancelled() || subs.get(sub_id) != Some(&owner) { + let mut subs = conn.subscriptions.lock().await; + if !super::close::claim_locked(&mut subs, sub_id, owner, conn) { return None; } super::close::retire_holder_locked(sub_id, conn, state).await; @@ -689,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| { @@ -714,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 @@ -977,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] @@ -996,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()); } @@ -1017,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; diff --git a/crates/buzz-relay/src/handlers/close.rs b/crates/buzz-relay/src/handlers/close.rs index 4c7b726cce3..395aacfe4da 100644 --- a/crates/buzz-relay/src/handlers/close.rs +++ b/crates/buzz-relay/src/handlers/close.rs @@ -54,7 +54,7 @@ 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 takes its sub ID @@ -66,6 +66,46 @@ 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, 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, + 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 { @@ -98,7 +138,13 @@ pub(crate) async fn admit(sub_id: &str, conn: &ConnectionState) -> Result Result, + subs: &mut HashMap, sub_id: &str, conn: &ConnectionState, state: &AppState, @@ -172,7 +218,7 @@ async fn close_owned( 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 { diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 551844ef7df..48fb87d5055 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -1772,8 +1772,8 @@ pub(crate) async fn claim_admitted_live( conn: &ConnectionState, state: &AppState, ) -> Option { - let subs = conn.subscriptions.lock().await; - if conn.cancel.is_cancelled() || subs.get(sub_id) != Some(&owner) { + let mut subs = conn.subscriptions.lock().await; + if !super::close::claim_locked(&mut subs, sub_id, owner, conn) { return None; } // A sync that held the ID stops; the registry entry and topics below are @@ -1826,8 +1826,8 @@ async fn claim_admitted_search( conn: &ConnectionState, state: &AppState, ) -> Option { - let subs = conn.subscriptions.lock().await; - if conn.cancel.is_cancelled() || subs.get(sub_id) != Some(&owner) { + let mut subs = conn.subscriptions.lock().await; + if !super::close::claim_locked(&mut subs, sub_id, owner, conn) { return None; } super::close::retire_holder_locked(sub_id, conn, state).await; @@ -2210,6 +2210,15 @@ mod tests { .collect() } + /// The owner token the lifecycle map holds for `sub_id`. + async fn token(conn: &ConnectionState, sub_id: &str) -> Option { + conn.subscriptions + .lock() + .await + .get(sub_id) + .map(|held| held.owner) + } + fn registered(state: &AppState, conn: &ConnectionState, sub_id: &str) -> bool { state .sub_registry @@ -2374,7 +2383,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" @@ -2426,7 +2435,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()); @@ -2546,10 +2555,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; @@ -2604,7 +2616,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); @@ -2677,7 +2695,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 @@ -2739,7 +2757,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 @@ -2754,6 +2772,66 @@ 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. @@ -2779,7 +2857,7 @@ mod tests { let stale = claim_admitted_live("x", older, scoped(&text_filters(), Some(&[a])), &conn, &state).await; assert_eq!(stale, None); - assert_eq!(conn.subscriptions.lock().await.get("x"), Some(&newer)); + 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)); diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 0bcc267a3cb..53bbbf10574 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/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" From ad3c2266b264102a5cfc062c2d50c62f5ae990d5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 6 Nov 2026 13:04:10 -0500 Subject: [PATCH 3/5] test(relay): pin the search claim and the unsubscribe in their tests The bridge search test now checks the retired live entry at the before_search_query hook, after the claim and before the timeout CLOSED that would also retire it. The pubsub subscriber test waits for Redis to report no subscribers on the released topic before it retains it again. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- crates/buzz-pubsub/src/lib.rs | 29 +++++++++++++++++++------ crates/buzz-relay/src/api/bridge.rs | 33 +++++++++++++---------------- 2 files changed, 38 insertions(+), 24 deletions(-) diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index aa09d351c7c..daeedd77862 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -862,13 +862,30 @@ mod tests { ); manager.release_topic(&target, topic).await; - tokio::time::sleep(Duration::from_millis(20)).await; - // Queued after the debounced unsubscribe: once it answers, the - // subscriber has processed that unsubscribe. - manager - .confirm_topic(&filler, EventTopic::Channel(Uuid::from_u128(0))) + // 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("confirm"); + .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, diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index cb7c6ab6ba7..ff7481e8103 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -10827,6 +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"); + // `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], @@ -10834,27 +10838,20 @@ pub(crate) mod postgres_tests { conn.clone(), fx.state.clone(), )); - // Search takes the ID's token when it is admitted, before its access - // checks, so the token alone does not show it has claimed. Its claim - // retires the live fan-out under the lifecycle lock, and then its - // hydrate blocks on the table lock: the live entry must be gone while - // search is still running, not only at final cleanup. - tokio::time::timeout(std::time::Duration::from_secs(10), async { - while fx - .state + tokio::time::timeout(std::time::Duration::from_secs(10), arrived) + .await + .expect("search reaches its query") + .expect("hook armed"); + assert!( + fx.state .sub_registry .get_filters(conn.conn_id, "x") - .is_some() - { - tokio::task::yield_now().await; - } - }) - .await - .expect("accepting search must retire live fan-out"); - // The claim's critical section, topic release included, is over once - // the lifecycle lock is free again. - drop(conn.subscriptions.lock().await); + .is_none(), + "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"); From 8d84ac94866c0bae726e1f3375419c6075c22a30 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 6 Nov 2026 13:23:24 -0500 Subject: [PATCH 4/5] fix(pubsub): wake reconciliation only for a full queue; run the new tests in CI A queued Subscribe is processed in order, and a closed queue means the subscriber restarts and re-subscribes the whole desired map. Only a subscribe hint dropped by a full queue needs the reconciliation wake-up, so first retains no longer each make the subscriber rescan the map. The new relay lifecycle tests and the full-queue pubsub test join the infra-free unit lanes (Justfile and scripts/run-tests.sh). The Redis subscriber test runs in a Redis-backed backend-integration step, so buzz-pubsub joins that archive. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- .github/workflows/_ci-relay.yml | 8 ++++++++ Justfile | 9 +++++---- crates/buzz-pubsub/src/lib.rs | 30 ++++++++++++++++++++++-------- scripts/run-tests.sh | 10 ++++++++++ 4 files changed, 45 insertions(+), 12 deletions(-) diff --git a/.github/workflows/_ci-relay.yml b/.github/workflows/_ci-relay.yml index 3cd039baa0c..3d949448778 100644 --- a/.github/workflows/_ci-relay.yml +++ b/.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 diff --git a/Justfile b/Justfile index c9e005c71c3..599381e51ff 100644 --- a/Justfile +++ b/Justfile @@ -407,9 +407,10 @@ test-unit: cargo nextest run -p buzz-auth --test mesh_jwks_proxy cargo nextest run -p buzz-audit --lib # Cross-pod disconnect payload tests (NIP-FI and community archive - # fence) and the shared Redis connection's timeout bounds (loopback - # fake peers); all infra-free. Keep the Cargo fallback in step. - cargo nextest run -p buzz-pubsub --lib -E 'test(/^conn_control::tests::/) + test(/^connection::tests::/)' + # fence), the shared Redis connection's timeout bounds (loopback + # fake peers), and the full-queue retain (no Redis dialed); all + # infra-free. Keep the Cargo fallback in step. + cargo nextest run -p buzz-pubsub --lib -E 'test(/^conn_control::tests::/) + test(/^connection::tests::/) + test(=tests::queue_full_retain_does_not_wait_and_schedules_reconciliation)' # buzz-auth NIP-FI verifier doctests. The sealed-authority # `compile_fail` doctests prove the default-feature public API alone # cannot forge the issuer→JWKS authority; nextest does not run @@ -562,7 +563,7 @@ test-unit: # cross-node fan-out loss handling; its stalled-delivery case fakes a # database that never answers on loopback, and none use DATABASE_URL. cargo nextest run -p buzz-relay --lib --bin buzz-relay \ - -E 'test(/^api::media_access::tests::/) + test(/^handlers::media_access::tests::/) + test(/^subscription::tests::/) + test(/^handlers::multinode_fanout::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::missed_pongs_close_with_their_reason_and_count) + test(=connection::tests::client_reply_to_a_jittered_restart_closes_as_shutdown) + test(=connection::tests::client_close_racing_a_policy_disconnect_keeps_the_policy_reason) + test(=connection::tests::server_close_reason_ranks_shutdown_policy_then_cancellation) + test(=connection::tests::auth_timeout_after_a_rejected_auth_closes_as_auth_failed) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::count_fallback_fetches_one_extra_candidate) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' + -E 'test(/^api::media_access::tests::/) + test(/^handlers::media_access::tests::/) + test(/^subscription::tests::/) + test(/^handlers::multinode_fanout::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=handlers::req::tests::unqueued_eose_logs_aborted_not_served) + test(=api::bridge::postgres_tests::reject_code_keeps_only_a_known_prefix) + test(=api::bridge::postgres_tests::nip98_replay_check_fails_closed_when_guard_errors) + test(=api::bridge::postgres_tests::verify_bridge_auth_rejects_nip98_event_signed_for_wrong_communitys_host) + test(=api::bridge::postgres_tests::verify_bridge_auth_can_require_payload_tag_for_json_body_endpoints) + test(=api::bridge::postgres_tests::verify_bridge_auth_accepts_nip98_event_signed_for_matching_host) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_verifies_with_matching_query) + test(=api::bridge::postgres_tests::moderation_read_query_bearing_nip98_event_rejected_against_bare_path) + test(=api::bridge::postgres_tests::moderation_read_audit_query_bearing_nip98_event_verifies) + test(=api::bridge::postgres_tests::moderation_read_restricted_no_query_still_verifies) + test(=connection::tests::aborting_lifecycle_owner_is_drop_safe) + test(=connection::tests::shutdown_terminalizes_auth_stalled_on_database_before_late_success) + test(=connection::terminal_close_delivery_tests::upgrade_would_deny_line_without_auth_logs_absent_at_close) + test(=handlers::event::tests::observer_owner_cache_is_scoped_to_community) + test(=state::tests::shared_counter_between_direct_and_fanout) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::missed_pongs_close_with_their_reason_and_count) + test(=connection::tests::client_reply_to_a_jittered_restart_closes_as_shutdown) + test(=connection::tests::client_close_racing_a_policy_disconnect_keeps_the_policy_reason) + test(=connection::tests::server_close_reason_ranks_shutdown_policy_then_cancellation) + test(=connection::tests::auth_timeout_after_a_rejected_auth_closes_as_auth_failed) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::saturated_handler_write_line_records_whether_its_ok_was_queued) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(=api::bridge::moderation_refusal_tests::moderation_lookup_failure_is_unavailable) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_policy::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::db_read_error_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^api::git::store::tests::/) + test(/^api::git::store::fault_probe::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::revoke_of_a_replaced_holder_spares_the_pending_replacement) + test(=handlers::req::tests::superseded_or_closed_requests_cannot_claim) + test(=connection::tests::close_fences_a_req_whose_handler_has_not_run) + test(=connection::tests::only_the_newest_same_id_req_answers) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::count_fallback_fetches_one_extra_candidate) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))' # Note on audio::join::tests scope: the full suite is infra-free (no DB, # no Redis). The infra-free audio/FI regression witnesses — bootstrap # ordering barrier, CommitConfirmed arm, pending-close invisibility, diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index daeedd77862..044888826f5 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/crates/buzz-pubsub/src/lib.rs @@ -259,16 +259,19 @@ impl PubSubManager { }; if should_subscribe { - // The wake-up is lossless and coalesced: if the bounded command - // queue is full, the subscriber still reconciles the whole - // desired-topic map without waiting for a Redis reconnect. - self.subscription_changed.notify_one(); match self .subscription_tx .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" @@ -730,8 +733,11 @@ mod tests { /// 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. Mutation: `send().await` instead of `try_send` - /// hangs at the timeout below; no `notify_one` fails the wake-up check. + /// 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; @@ -741,8 +747,16 @@ mod tests { .retain_topic(&filler, EventTopic::Channel(Uuid::from_u128(id as u128))) .await; } - // Drain the wake-ups the fillers left (coalesced into one permit). - manager.subscription_changed.notified().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)); diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 17b6848455a..3f7b55396f4 100755 --- a/scripts/run-tests.sh +++ b/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 \ From 011b88e75d5471a94129020f02415a512d7a0232 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Fri, 6 Nov 2026 14:59:21 -0500 Subject: [PATCH 5/5] style(relay): format the claim calls rebased onto per-filter scopes Rebasing onto main's per-filter live scopes (#8519) lengthened the claim_admitted_live calls; rustfmt now wraps them. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- crates/buzz-relay/src/handlers/req.rs | 51 +++++++++++++++++++-------- 1 file changed, 37 insertions(+), 14 deletions(-) diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 48fb87d5055..39fb6c0fb8d 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -525,8 +525,7 @@ pub async fn handle_admitted_req( ) }) .collect(); - let Some(owner) = claim_admitted_live(&sub_id, owner, live_filters, &conn, &state).await - else { + let Some(owner) = claim_admitted_live(&sub_id, owner, live_filters, &conn, &state).await else { return; }; @@ -1762,9 +1761,9 @@ pub(crate) async fn claim_live_subscription( /// 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. +/// 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, @@ -2824,8 +2823,14 @@ mod tests { 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; + 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); @@ -2851,11 +2856,23 @@ mod tests { 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; + 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)); @@ -2863,8 +2880,14 @@ mod tests { 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 = 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"));