From f7ebc0a88ea27ac4e3d74b941e3a44ddb10b306c Mon Sep 17 00:00:00 2001 From: agent-p1p <292420120+agent-p1p@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:11:02 +0000 Subject: [PATCH] sdk: enhance subscription handling with restoration fixes and atomic reissue - Return `SubscriptionData` from client subscription APIs for better visibility of subscription state (e.g., EOSE and closure status). - Add coverage for restoration edge cases (auto-closing, unregistered, and failed restorations). - Ensure atomic subscription restoration by queuing tracked REQs under a single lock to avoid duplicate REQs for open IDs. - Introduce `Relay::reissue_subscription` to atomically queue CLOSE and REQ for a subscription, ensuring proper replay behavior. - Reset tracking for REQs resent on the same connection, ensuring accurate state and stored-events query handling. Pull-Request: https://github.com/nostrdevkit/nostr/pull/1495 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Yuki Kishimoto --- nostr-sdk/CHANGELOG.md | 4 + nostr-sdk/src/client/api/subscribe.rs | 43 +- nostr-sdk/src/client/mod.rs | 13 +- nostr-sdk/src/pool/mod.rs | 25 +- nostr-sdk/src/relay/api/mod.rs | 2 + .../src/relay/api/reissue_subscription.rs | 33 + nostr-sdk/src/relay/api/subscribe.rs | 10 +- nostr-sdk/src/relay/inner.rs | 733 ++++++++++++++++-- nostr-sdk/src/relay/mod.rs | 29 +- 9 files changed, 803 insertions(+), 89 deletions(-) create mode 100644 nostr-sdk/src/relay/api/reissue_subscription.rs diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index a900d64bd..dd6647386 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -35,6 +35,7 @@ - Redesign `Client::try_connect_relay` API - Return `NotificationStream` from client and relay `notifications()` methods instead of a boxed stream. - Remove `RelayStatus::Pending` and rename `RelayStatus::Terminated` to `RelayStatus::Idle` (https://github.com/nostrdevkit/nostr/pull/1491). +- Return `SubscriptionData` from `Relay::subscriptions`, `Relay::subscription`, `Client::subscriptions` and `Client::subscription` (one per relay) instead of the filters. It also tells whether the current `REQ` was answered with `EOSE` and whether the relay closed the subscription. ### Changed @@ -47,6 +48,7 @@ - Add `LocalRelayBuilderNip42::relay_url` to make it possible to configure a custom `relay_url` for the nip42 challenge validation (https://github.com/nostrdevkit/nostr/pull/1476) - Add `LocalRelayBuilder::new_event_channel_size` for customizing the size of the channel used to notify new received events - Add opt-in receiver gap reporting with `NotificationStream::with_gaps()`. +- Add `Relay::reissue_subscription` to queue `CLOSE` and `REQ` for a long-lived subscription as a single unit, so the relay sends its stored events again. ### Fixed @@ -60,6 +62,8 @@ - Restore subscriptions after reconnections within the same second (https://github.com/nostrdevkit/nostr/pull/1491). - Reset subscription EOSE and event-limit tracking when a subscription is restored, so a relay isn't flagged for resending stored events. - Avoid duplicate subscription `REQ`s after reconnecting. +- Reset subscription EOSE and event-limit tracking when a `REQ` is sent again on the same connection. +- Avoid a duplicate subscription `REQ` when another `REQ` for it is queued while it's being restored after a reconnection. ### Deprecated diff --git a/nostr-sdk/src/client/api/subscribe.rs b/nostr-sdk/src/client/api/subscribe.rs index 8bfb4205d..aaf6e82c8 100644 --- a/nostr-sdk/src/client/api/subscribe.rs +++ b/nostr-sdk/src/client/api/subscribe.rs @@ -70,20 +70,61 @@ where #[cfg(test)] mod tests { + use std::collections::HashMap; use std::time::Duration; + use std::{future, slice}; use futures::StreamExt; use nostr::event::{EventBuilder, FinalizeEvent, Kind}; use nostr::filter::Filter; use nostr::key::Keys; - use nostr::message::SubscriptionId; + use nostr::message::{RelayMessage, SubscriptionId}; + use nostr::types::RelayUrl; use crate::authenticator::SignerAuthenticator; use crate::client::ClientNotification; + use crate::local_relay::MockRelay; + use crate::relay::SubscriptionData; use crate::test_utils::{ setup_client, setup_client_with_authenticator, setup_nip42_read_local_relay, }; + #[tokio::test] + async fn test_client_subscription_returns_data_registered_on_each_relay() { + let mock = MockRelay::run().await.unwrap(); + let url: RelayUrl = mock.url().await; + let client = setup_client(url.clone()).await; + let filter: Filter = Filter::new().kind(Kind::TextNote); + let id: SubscriptionId = SubscriptionId::new("client-subscription-data"); + + let notifications = client.notifications(); + + client + .subscribe(filter.clone()) + .with_id(id.clone()) + .await + .unwrap(); + + let eose = notifications.any(|notification| { + future::ready(matches!(notification, ClientNotification::Message { ref message, .. } + if matches!(message.as_ref(), RelayMessage::EndOfStoredEvents(received) if received.as_ref() == &id))) + }); + assert!( + tokio::time::timeout(Duration::from_secs(5), eose) + .await + .unwrap() + ); + + let subscription: HashMap = client.subscription(&id).await; + let data: &SubscriptionData = subscription.get(&url).unwrap(); + assert_eq!(data.filters(), slice::from_ref(&filter)); + assert!(data.received_eose()); + assert!(!data.is_closed()); + + let subscriptions = client.subscriptions().await; + assert_eq!(subscriptions[&id][&url].filters(), [filter]); + } + #[tokio::test] async fn test_client_subscribe_dont_resubscribes_after_auth_required_closed_without_authenticator() { diff --git a/nostr-sdk/src/client/mod.rs b/nostr-sdk/src/client/mod.rs index f75eeb359..717d40276 100644 --- a/nostr-sdk/src/client/mod.rs +++ b/nostr-sdk/src/client/mod.rs @@ -29,7 +29,8 @@ use crate::pool::{RelayPool, RelayPoolBuilder}; #[cfg(not(target_arch = "wasm32"))] use crate::proxy::Proxy; use crate::relay::{ - Relay, RelayCapabilities, RelayLimits, RelayOptions, SleepWhenIdle, SyncOptions, + Relay, RelayCapabilities, RelayLimits, RelayOptions, SleepWhenIdle, SubscriptionData, + SyncOptions, }; use crate::stream::NotificationStream; @@ -591,15 +592,17 @@ impl Client { self.pool().disconnect().await } - /// Get subscriptions + /// Get long-lived subscriptions, with the data registered on each relay #[inline] - pub async fn subscriptions(&self) -> HashMap>> { + pub async fn subscriptions( + &self, + ) -> HashMap> { self.pool().subscriptions().await } - /// Get subscription + /// Get subscription by [`SubscriptionId`], with the data registered on each relay #[inline] - pub async fn subscription(&self, id: &SubscriptionId) -> HashMap> { + pub async fn subscription(&self, id: &SubscriptionId) -> HashMap { self.pool().subscription(id).await } diff --git a/nostr-sdk/src/pool/mod.rs b/nostr-sdk/src/pool/mod.rs index 22b2b1232..cb5dd676c 100644 --- a/nostr-sdk/src/pool/mod.rs +++ b/nostr-sdk/src/pool/mod.rs @@ -25,7 +25,7 @@ use crate::monitor::Monitor; use crate::policy::AdmitStatus; use crate::relay::{ AtomicRelayCapabilities, Relay, RelayCapabilities, RelayOptions, RelayStreamEvent, - ReqExitPolicy, SubscribeAutoCloseOptions, SyncOptions, + ReqExitPolicy, SubscribeAutoCloseOptions, SubscriptionData, SyncOptions, }; use crate::shared::SharedState; use crate::stream::ReceiverStream; @@ -388,11 +388,11 @@ impl RelayPool { #[inline] pub(crate) async fn subscriptions( &self, - ) -> HashMap>> { + ) -> HashMap> { // Lock with read shared access let relays = self.relays.read().await; - let mut subscriptions: HashMap>> = + let mut subscriptions: HashMap> = HashMap::new(); for (url, relay) in relays.iter() { @@ -400,11 +400,11 @@ impl RelayPool { let relay_subscriptions = relay.subscriptions().await; // Iterate relay subscriptions and populate the general subscriptions map - for (id, list) in relay_subscriptions.into_iter() { + for (id, data) in relay_subscriptions.into_iter() { subscriptions .entry(id) .or_default() - .insert(url.clone(), list); + .insert(url.clone(), data); } } @@ -412,21 +412,24 @@ impl RelayPool { } #[inline] - pub(crate) async fn subscription(&self, id: &SubscriptionId) -> HashMap> { + pub(crate) async fn subscription( + &self, + id: &SubscriptionId, + ) -> HashMap { // Lock with read shared access let relays = self.relays.read().await; - let mut filters: HashMap> = HashMap::new(); + let mut subscription: HashMap = HashMap::new(); - // Iterate relays and populate filters + // Iterate relays and collect the subscription registered on each for (url, relay) in relays.iter() { // try to get subscription by ID from the relay - if let Some(list) = relay.subscription(id).await { - filters.insert(url.clone(), list); + if let Some(data) = relay.subscription(id).await { + subscription.insert(url.clone(), data); } } - filters + subscription } pub(crate) async fn send_msg( diff --git a/nostr-sdk/src/relay/api/mod.rs b/nostr-sdk/src/relay/api/mod.rs index 6e49bc5f5..8c1fef8cd 100644 --- a/nostr-sdk/src/relay/api/mod.rs +++ b/nostr-sdk/src/relay/api/mod.rs @@ -1,5 +1,6 @@ mod count_events; mod fetch_events; +mod reissue_subscription; mod send_event; mod send_msg; mod stream_events; @@ -11,6 +12,7 @@ mod unsubscribe_all; pub use self::count_events::*; pub use self::fetch_events::*; +pub use self::reissue_subscription::*; pub use self::send_event::*; pub use self::send_msg::*; pub use self::stream_events::*; diff --git a/nostr-sdk/src/relay/api/reissue_subscription.rs b/nostr-sdk/src/relay/api/reissue_subscription.rs new file mode 100644 index 000000000..0c5c9003c --- /dev/null +++ b/nostr-sdk/src/relay/api/reissue_subscription.rs @@ -0,0 +1,33 @@ +use std::future::IntoFuture; + +use nostr::message::SubscriptionId; + +use crate::error::Error; +use crate::future::BoxedFuture; +use crate::relay::Relay; + +/// Reissue a long-lived subscription +#[must_use = "Does nothing unless you await!"] +pub struct ReissueSubscription<'relay, 'id> { + relay: &'relay Relay, + id: &'id SubscriptionId, +} + +impl<'relay, 'id> ReissueSubscription<'relay, 'id> { + #[inline] + pub(crate) fn new(relay: &'relay Relay, id: &'id SubscriptionId) -> Self { + Self { relay, id } + } +} + +impl<'relay, 'id> IntoFuture for ReissueSubscription<'relay, 'id> +where + 'id: 'relay, +{ + type Output = Result<(), Error>; + type IntoFuture = BoxedFuture<'relay, Self::Output>; + + fn into_future(self) -> Self::IntoFuture { + Box::pin(async move { self.relay.inner.reissue_subscription(self.id).await }) + } +} diff --git a/nostr-sdk/src/relay/api/subscribe.rs b/nostr-sdk/src/relay/api/subscribe.rs index 1c4542b51..757144216 100644 --- a/nostr-sdk/src/relay/api/subscribe.rs +++ b/nostr-sdk/src/relay/api/subscribe.rs @@ -313,7 +313,10 @@ mod tests { assert_eq!(err.kind(), ErrorKind::Invalid); assert_eq!(err.to_string(), "subscription ID already exists"); - assert_eq!(relay.subscription(&id).await.unwrap(), vec![first_filter]); + assert_eq!( + relay.subscription(&id).await.unwrap().filters(), + [first_filter] + ); } #[tokio::test] @@ -348,7 +351,10 @@ mod tests { assert_eq!(err.kind(), ErrorKind::Invalid); assert_eq!(err.to_string(), "subscription ID already exists"); - assert_eq!(relay.subscription(&id).await.unwrap(), vec![first_filter]); + assert_eq!( + relay.subscription(&id).await.unwrap().filters(), + [first_filter] + ); } #[tokio::test] diff --git a/nostr-sdk/src/relay/inner.rs b/nostr-sdk/src/relay/inner.rs index a00e4f769..a50fa532c 100644 --- a/nostr-sdk/src/relay/inner.rs +++ b/nostr-sdk/src/relay/inner.rs @@ -21,7 +21,7 @@ use rand::RngExt; use rand::rand_core::UnwrapErr; use rand::rngs::SysRng; use tokio::sync::futures::Notified; -use tokio::sync::mpsc::{self, Receiver, Sender}; +use tokio::sync::mpsc::{self, PermitIterator, Receiver, Sender}; use tokio::sync::{Mutex, MutexGuard, Notify, RwLock, RwLockWriteGuard, broadcast, oneshot, watch}; use universal_time::Instant; @@ -232,6 +232,21 @@ impl RelayChannels { .map_err(|_| Error::state_msg("can't send message to the transport dispatcher")) } + /// Queue the messages in order, or none of them if the channel can't fit them all. + fn send_client_msgs(&self, msgs: [JsonMessageItem; N]) -> Result<(), Error> { + let permits: PermitIterator<'_, JsonMessageItem> = self + .nostr + .0 + .try_reserve_many(N) + .map_err(|_| Error::state_msg("can't send message to the transport dispatcher"))?; + + for (permit, msg) in permits.zip(msgs) { + permit.send(msg); + } + + Ok(()) + } + #[inline] pub async fn rx_nostr(&self) -> MutexGuard<'_, Receiver> { self.nostr.1.lock().await @@ -243,17 +258,19 @@ impl RelayChannels { } } +/// Subscription registered on a relay. #[derive(Debug)] -struct SubscriptionData { - pub filters: Vec, - pub subscribed_connection: usize, - pub is_auto_closing: bool, +pub struct SubscriptionData { + filters: Vec, + subscribed_connection: usize, + is_auto_closing: bool, /// Received EOSE msg - pub received_eose: bool, + received_eose: bool, /// Number of received events - pub received_events: AtomicUsize, + // TODO: replace this AtomicUsize with a usize + received_events: AtomicUsize, /// Subscription closed by relay - pub closed: bool, + closed: bool, /// Counter used for long-lived subscriptions, for tracking the number of the pending ones pending_req_counter: Option, } @@ -284,6 +301,76 @@ impl SubscriptionData { pending_req_counter: None, } } + + /// Get the filters of the subscription. + #[inline] + pub fn filters(&self) -> &[Filter] { + &self.filters + } + + /// Check if the relay answered the current `REQ` of the subscription with `EOSE`. + /// + /// `EOSE` is recorded even when notification receivers fall behind, + /// so this tells a receiver that missed notifications whether the relay + /// finished sending the stored events of the subscription. + /// + /// It's `false` from when a new `REQ` for the subscription is sent + /// (e.g., when it is restored after a reconnection or reissued with + /// [`Relay::reissue_subscription`](super::Relay::reissue_subscription)) + /// until the relay answers it with `EOSE`. + #[inline] + pub fn received_eose(&self) -> bool { + self.received_eose + } + + /// Check if the relay closed the subscription without the SDK removing it + /// (e.g., with an `auth-required:` or `rate-limited:` reason). + /// + /// Its `REQ` is sent again the next time subscriptions are restored, + /// after a reconnection or an authentication. + #[inline] + pub fn is_closed(&self) -> bool { + self.closed + } + + /// Check if the subscription must be restored on the current WebSocket session. + fn needs_restore(&self, current_connection: usize) -> bool { + if self.is_auto_closing { + return false; + } + + if self + .pending_req_counter + .as_ref() + .is_some_and(PendingReqCounter::has_pending) + { + return false; + } + + if self.closed { + return true; + } + + // Restore subscriptions dispatched on an earlier WebSocket session, + // even when reconnecting within one second. + current_connection > self.subscribed_connection + } +} + +// TODO: derive the Clone when the AtomicUsize will be replaced by a usize +// `AtomicUsize` isn't `Clone`: copy its current value. +impl Clone for SubscriptionData { + fn clone(&self) -> Self { + Self { + filters: self.filters.clone(), + subscribed_connection: self.subscribed_connection, + is_auto_closing: self.is_auto_closing, + received_eose: self.received_eose, + received_events: AtomicUsize::new(self.received_events.load(Ordering::SeqCst)), + closed: self.closed, + pending_req_counter: self.pending_req_counter.clone(), + } + } } // Instead of wrap every field in an `Arc`, which increases the number of atomic operations, @@ -475,17 +562,17 @@ impl InnerRelay { } /// Returns all long-lived (non-auto-closing) subscriptions - pub async fn subscriptions(&self) -> HashMap> { + pub async fn subscriptions(&self) -> HashMap { let subscription = self.atomic.subscriptions.read().await; subscription .iter() - .filter_map(|(k, v)| (!v.is_auto_closing).then_some((k.clone(), v.filters.clone()))) + .filter_map(|(k, v)| (!v.is_auto_closing).then_some((k.clone(), v.clone()))) .collect() } - pub async fn subscription(&self, id: &SubscriptionId) -> Option> { + pub async fn subscription(&self, id: &SubscriptionId) -> Option { let subscription = self.atomic.subscriptions.read().await; - subscription.get(id).map(|d| d.filters.clone()) + subscription.get(id).cloned() } pub(super) async fn remove_subscription(&self, id: &SubscriptionId) { @@ -530,6 +617,7 @@ impl InnerRelay { Ok(()) } + #[cfg(test)] pub(crate) async fn update_subscription( &self, id: &SubscriptionId, @@ -586,38 +674,12 @@ impl InnerRelay { } /// Check if it should subscribe for current websocket session + #[cfg(test)] pub(crate) async fn should_resubscribe(&self, id: &SubscriptionId) -> bool { let subscriptions = self.atomic.subscriptions.read().await; - match subscriptions.get(id) { - Some(SubscriptionData { - subscribed_connection, - closed, - is_auto_closing: false, - pending_req_counter, - .. - }) => { - if let Some(counter) = pending_req_counter { - if counter.has_pending() { - return false; - } - } - - if *closed { - return true; - } - - // Restore subscriptions dispatched on an earlier WebSocket session, - // even when reconnecting within one second. - let current_connection: usize = self.stats.success(); - current_connection > *subscribed_connection - } - // NOT subscribe if auto-closing subscription or subscription not found - Some(SubscriptionData { - is_auto_closing: true, - .. - }) - | None => false, - } + subscriptions + .get(id) + .is_some_and(|data| data.needs_restore(self.stats.success())) } async fn subscription_req_started(&self, request: &PendingReq) { @@ -635,15 +697,18 @@ impl InnerRelay { return; } - // A queued REQ may have survived an earlier session without being dispatched. - let connection: usize = self.stats.success(); - - if connection > data.subscribed_connection { - data.received_eose = false; - *data.received_events.get_mut() = 0; - } - - data.subscribed_connection = connection; + // Every REQ starts a new stored-events query, also when sent again on the same + // connection. Frames the relay sent for the previous REQ before handling the + // CLOSE may still arrive after this reset: + // - its EOSE sets the flag early, which is acceptable because the filters are + // the same, so that EOSE covers the same stored events; + // - its events count toward the limit of the new REQ. + data.received_eose = false; + *data.received_events.get_mut() = 0; + + // Record the session that sends the REQ, since a queued REQ may have survived + // an earlier session without being dispatched. + data.subscribed_connection = self.stats.success(); data.closed = false; } @@ -1861,20 +1926,8 @@ impl InnerRelay { pub async fn resubscribe(&self) -> Result<(), Error> { // TODO: avoid subscriptions clone let subscriptions = self.subscriptions().await; - for (id, filters) in subscriptions.into_iter() { - if !filters.is_empty() && self.should_resubscribe(&id).await { - self.update_subscription(&id, filters.clone(), true).await?; - - let msg = ClientMessage::Req { - subscription_id: Cow::Borrowed(&id), - filters: filters.into_iter().map(Cow::Owned).collect(), - }; - - if let Err(e) = self.send_msg(msg, None).await { - self.subscription_closed(&id).await; - return Err(e); - } - } else { + for id in subscriptions.into_keys() { + if !self.queue_resubscription(&id).await? { tracing::debug!("Skip re-subscription of '{id}'"); } } @@ -1882,6 +1935,109 @@ impl InnerRelay { Ok(()) } + /// Queue the `REQ` that restores a subscription, if it still needs one. + /// + /// Returns `false` if the subscription doesn't need to be restored. + async fn queue_resubscription(&self, id: &SubscriptionId) -> Result { + // Decide and enqueue under the lock every tracked REQ is queued under, + // so a REQ queued meanwhile (e.g., by `reissue_subscription`) can't be + // followed by a duplicate, which relays may reject by closing the ID. + let mut subscriptions = self.atomic.subscriptions.write().await; + + let current_connection: usize = self.stats.success(); + + let Some(data) = subscriptions.get_mut(id) else { + return Ok(false); + }; + + if data.filters.is_empty() || !data.needs_restore(current_connection) { + return Ok(false); + } + + let msg: ClientMessage<'_> = ClientMessage::Req { + subscription_id: Cow::Borrowed(id), + filters: data.filters.iter().map(Cow::Borrowed).collect(), + }; + let item: JsonMessageItem = JsonMessageItem { + json: msg.as_json(), + confirmation: None, + pending_req: data + .pending_req_counter + .as_ref() + .map(|counter| counter.track(id.clone())), + }; + + // The relay answers the new REQ with a new stored-events query. + data.subscribed_connection = current_connection; + data.closed = false; + data.received_eose = false; + *data.received_events.get_mut() = 0; + + let queued: Result<(), Error> = self.ensure_operational().and_then(|()| { + if !self.capabilities.can_read() { + return Err(Error::read_disabled()); + } + + self.atomic.channels.send_client_msg(item) + }); + + if let Err(e) = queued { + // Keep it restorable on the next attempt. + data.closed = true; + return Err(e); + } + + Ok(true) + } + + pub(super) async fn reissue_subscription(&self, id: &SubscriptionId) -> Result<(), Error> { + self.ensure_operational()?; + + if !self.capabilities.can_read() { + return Err(Error::read_disabled()); + } + + // Like `queue_msg`, hold the lock until the REQ is queued and tracked, + // so the restoration on reconnect can't send another REQ in between. + let subscriptions = self.atomic.subscriptions.write().await; + + let data: &SubscriptionData = subscriptions + .get(id) + .ok_or_else(|| Error::not_found("subscription not found"))?; + + if data.is_auto_closing { + return Err(Error::invalid_msg( + "auto-closing subscriptions can't be reissued", + )); + } + + let close: ClientMessage<'_> = ClientMessage::Close(Cow::Borrowed(id)); + let req: ClientMessage<'_> = ClientMessage::Req { + subscription_id: Cow::Borrowed(id), + filters: data.filters.iter().map(Cow::Borrowed).collect(), + }; + + let pending_req: Option = data + .pending_req_counter + .as_ref() + .map(|counter| counter.track(id.clone())); + + // A lone CLOSE would leave the relay without the subscription, and a lone REQ + // would repeat an open ID, which some relays reject as a duplicate. + self.atomic.channels.send_client_msgs([ + JsonMessageItem { + json: close.as_json(), + confirmation: None, + pending_req: None, + }, + JsonMessageItem { + json: req.as_json(), + confirmation: None, + pending_req, + }, + ]) + } + pub(super) fn spawn_auto_closing_handler( &self, id: SubscriptionId, @@ -2716,7 +2872,8 @@ mod tests { .handle_relay_message(r#"["CLOSED","test","auth-required: you must auth"]"#, &tx) .await; - assert!(relay.inner.subscription(&subscription_id).await.is_some()); + let data: SubscriptionData = relay.inner.subscription(&subscription_id).await.unwrap(); + assert!(data.is_closed()); assert!(relay.inner.should_resubscribe(&subscription_id).await); relay @@ -2725,6 +2882,8 @@ mod tests { .await .unwrap(); + let data: SubscriptionData = relay.inner.subscription(&subscription_id).await.unwrap(); + assert!(!data.is_closed()); assert!(!relay.inner.should_resubscribe(&subscription_id).await); } @@ -3868,6 +4027,54 @@ mod tests { assert_eq!(data.received_events.load(Ordering::SeqCst), 0); } + #[tokio::test] + async fn req_resent_on_same_connection_resets_limit_tracking_when_dispatched() { + let ControlledRelay { + relay, + mut transport, + } = ControlledRelay::with_opts(RelayOptions::default().verify_subscriptions(true)); + let id: SubscriptionId = SubscriptionId::new("live"); + let filter: Filter = Filter::new().kind(Kind::TextNote).limit(1); + + relay.connect(); + + let (connection, recorded) = recording_connection(); + assert!(transport.next_dial().await.send(Ok(connection)).is_ok()); + wait_for_status(&relay, RelayStatus::Connected).await; + + relay + .subscribe(filter.clone()) + .with_id(id.clone()) + .await + .unwrap(); + flush_outbound(&relay).await; + + let mut notifications: NotificationStream = relay.notifications(); + let event: Event = EventBuilder::new(Kind::TextNote, "stored event") + .finalize(&Keys::generate()) + .unwrap(); + recorded.reply(RelayMessage::event(id.clone(), event)); + recorded.reply(RelayMessage::eose(id.clone())); + wait_for_eose(&mut notifications, &id).await; + + relay + .send_msg(ClientMessage::close(id.clone())) + .await + .unwrap(); + relay + .send_msg(ClientMessage::req(id.clone(), filter)) + .await + .unwrap(); + flush_outbound(&relay).await; + + // The relay answers the new REQ with a new stored-events query. + let subscriptions = relay.inner.atomic.subscriptions.read().await; + let data: &SubscriptionData = subscriptions.get(&id).unwrap(); + assert_eq!(data.subscribed_connection, 1); + assert!(!data.received_eose); + assert_eq!(data.received_events.load(Ordering::SeqCst), 0); + } + #[test] fn pending_req_counter_does_not_underflow() { let counter = PendingReqCounter::default(); @@ -3982,6 +4189,400 @@ mod tests { .count(); assert_eq!(reqs, 1, "frames on first session: {frames:?}"); } + + /// Returns the EOSE flag of the subscription, or `None` if it isn't registered. + async fn subscription_eose(relay: &Relay, id: &SubscriptionId) -> Option { + relay + .subscription(id) + .await + .map(|data| data.received_eose()) + } + + #[tokio::test] + async fn subscription_data_tracks_eose_of_current_req() { + let ControlledRelay { + relay, + mut transport, + } = ControlledRelay::new(); + let id: SubscriptionId = SubscriptionId::new("live"); + let filter: Filter = Filter::new().kind(Kind::TextNote); + + assert!(relay.subscription(&id).await.is_none()); + + relay.connect(); + + let (connection, recorded) = recording_connection(); + assert!(transport.next_dial().await.send(Ok(connection)).is_ok()); + wait_for_status(&relay, RelayStatus::Connected).await; + + relay + .subscribe(filter.clone()) + .with_id(id.clone()) + .await + .unwrap(); + flush_outbound(&relay).await; + + let data: SubscriptionData = relay.subscription(&id).await.unwrap(); + assert_eq!(data.filters(), [filter]); + assert!(!data.received_eose()); + assert!(!data.is_closed()); + + let mut notifications: NotificationStream = relay.notifications(); + recorded.reply(RelayMessage::eose(id.clone())); + wait_for_eose(&mut notifications, &id).await; + + assert_eq!(subscription_eose(&relay, &id).await, Some(true)); + } + + /// Returns the `CLOSE` and `REQ` frames that reissue the subscription. + fn reissue_frames(id: &SubscriptionId, filter: &Filter) -> [String; 2] { + [ + ClientMessage::close(id.clone()).as_json(), + ClientMessage::req(id.clone(), filter.clone()).as_json(), + ] + } + + #[tokio::test] + async fn reissued_subscription_restarts_eose_and_limit_tracking() { + let ControlledRelay { + relay, + mut transport, + } = ControlledRelay::with_opts(RelayOptions::default().verify_subscriptions(true)); + let id: SubscriptionId = SubscriptionId::new("live"); + let filter: Filter = Filter::new().kind(Kind::TextNote).limit(1); + + relay.connect(); + + let (connection, mut recorded) = recording_connection(); + assert!(transport.next_dial().await.send(Ok(connection)).is_ok()); + wait_for_status(&relay, RelayStatus::Connected).await; + + relay + .subscribe(filter.clone()) + .with_id(id.clone()) + .await + .unwrap(); + flush_outbound(&relay).await; + recorded.sent_text(); + + let mut notifications: NotificationStream = relay.notifications(); + let stored: Event = EventBuilder::new(Kind::TextNote, "stored event") + .finalize(&Keys::generate()) + .unwrap(); + recorded.reply(RelayMessage::event(id.clone(), stored.clone())); + recorded.reply(RelayMessage::eose(id.clone())); + wait_for_eose(&mut notifications, &id).await; + + assert_eq!(subscription_eose(&relay, &id).await, Some(true)); + + relay.reissue_subscription(&id).await.unwrap(); + let barrier: SubscriptionId = flush_outbound(&relay).await; + + let [close, req] = reissue_frames(&id, &filter); + assert_eq!( + recorded.sent_text(), + vec![close, req, ClientMessage::close(barrier).as_json()] + ); + + // Dispatching the REQ on the same connection starts a new stored-events query. + { + let subscriptions = relay.inner.atomic.subscriptions.read().await; + let data: &SubscriptionData = subscriptions.get(&id).unwrap(); + assert!(!data.received_eose); + assert_eq!(data.received_events.load(Ordering::SeqCst), 0); + } + + // The relay sends the stored event again, within the limit of the new query. + recorded.reply(RelayMessage::event(id.clone(), stored)); + recorded.reply(RelayMessage::eose(id.clone())); + wait_for_eose(&mut notifications, &id).await; + + assert_eq!(subscription_eose(&relay, &id).await, Some(true)); + + let subscriptions = relay.inner.atomic.subscriptions.read().await; + let data: &SubscriptionData = subscriptions.get(&id).unwrap(); + assert_eq!(data.received_events.load(Ordering::SeqCst), 1); + } + + async fn reissue_queues_nothing_without_room_for_both_frames(free_slots: usize) { + let controlled = ControlledRelay::new(); + let relay = &controlled.relay; + // Holding the queue keeps the connection task from draining it. + let mut queue = relay.inner.atomic.channels.rx_nostr().await; + let sender: &Sender = &relay.inner.atomic.channels.nostr.0; + let filler: ClientMessage = ClientMessage::close(SubscriptionId::new("filler")); + + relay.connect(); + let id: SubscriptionId = relay.subscribe(Filter::new()).await.unwrap(); + drop(queue.try_recv().unwrap()); + + while sender.capacity() > free_slots { + relay.send_msg(filler.clone()).await.unwrap(); + } + + let err: Error = relay.reissue_subscription(&id).await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::State); + assert_eq!(sender.capacity(), free_slots); + + while let Ok(item) = queue.try_recv() { + assert_eq!(item.json, filler.as_json()); + } + + let subscriptions = relay.inner.atomic.subscriptions.read().await; + let data: &SubscriptionData = subscriptions.get(&id).unwrap(); + assert!(!data.pending_req_counter.as_ref().unwrap().has_pending()); + } + + #[tokio::test] + async fn reissue_into_full_queue_queues_nothing() { + reissue_queues_nothing_without_room_for_both_frames(0).await; + } + + #[tokio::test] + async fn reissue_into_queue_with_one_free_slot_queues_nothing() { + reissue_queues_nothing_without_room_for_both_frames(1).await; + } + + #[tokio::test] + async fn subscription_reissued_while_disconnected_is_sent_once_after_reconnect() { + let opts: RelayOptions = RelayOptions::default() + .adjust_retry_interval(false) + .retry_interval(Duration::from_secs(1)); + + let ControlledRelay { + relay, + mut transport, + } = ControlledRelay::with_opts(opts); + let id: SubscriptionId = SubscriptionId::new("live"); + let filter: Filter = Filter::new().kind(Kind::TextNote); + + relay.connect(); + + let (connection, first) = recording_connection(); + assert!(transport.next_dial().await.send(Ok(connection)).is_ok()); + wait_for_status(&relay, RelayStatus::Connected).await; + + relay + .subscribe(filter.clone()) + .with_id(id.clone()) + .await + .unwrap(); + flush_outbound(&relay).await; + + let mut notifications: NotificationStream = relay.notifications(); + first.reply(RelayMessage::eose(id.clone())); + wait_for_eose(&mut notifications, &id).await; + + first.close(); + wait_for_status(&relay, RelayStatus::Disconnected).await; + + relay.reissue_subscription(&id).await.unwrap(); + + // Queueing alone leaves the tracking of the previous REQ intact. + assert_eq!(subscription_eose(&relay, &id).await, Some(true)); + + let (connection, mut second) = recording_connection(); + assert!(transport.next_dial().await.send(Ok(connection)).is_ok()); + wait_for_status(&relay, RelayStatus::Connected).await; + let barrier: SubscriptionId = flush_outbound(&relay).await; + + // The queued frames replace the automatic restoration of the subscription. + let [close, req] = reissue_frames(&id, &filter); + let frames: Vec = second.sent_text(); + assert_eq!( + frames, + vec![close, req, ClientMessage::close(barrier.clone()).as_json()] + ); + + assert_eq!(subscription_eose(&relay, &id).await, Some(false)); + + reply_rejecting_duplicate_reqs(&second, &frames); + second.reply(RelayMessage::eose(barrier.clone())); + wait_for_eose(&mut notifications, &barrier).await; + + assert!(relay.inner.has_subscription(&id).await); + assert_eq!(subscription_eose(&relay, &id).await, Some(true)); + } + + #[tokio::test] + async fn reissue_requires_registered_long_lived_subscription() { + let controlled = ControlledRelay::new(); + let relay = &controlled.relay; + let queue = relay.inner.atomic.channels.rx_nostr().await; + + relay.connect(); + + let err: Error = relay + .reissue_subscription(&SubscriptionId::new("unknown")) + .await + .unwrap_err(); + assert_eq!(err.kind(), ErrorKind::NotFound); + assert_eq!(err.to_string(), "subscription not found"); + + let id: SubscriptionId = SubscriptionId::new("auto-closing"); + relay + .inner + .add_auto_closing_subscription(id.clone(), vec![Filter::new()]) + .await + .unwrap(); + + let err: Error = relay.reissue_subscription(&id).await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::Invalid); + assert_eq!( + err.to_string(), + "auto-closing subscriptions can't be reissued" + ); + + assert!(queue.is_empty()); + } + + #[tokio::test] + async fn reissue_requires_read_capability() { + let controlled = ControlledRelay::new(); + let relay = &controlled.relay; + let mut queue = relay.inner.atomic.channels.rx_nostr().await; + + relay.connect(); + let id: SubscriptionId = relay.subscribe(Filter::new()).await.unwrap(); + drop(queue.try_recv().unwrap()); + + relay.capabilities().remove(RelayCapabilities::READ); + + let err: Error = relay.reissue_subscription(&id).await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::Unsupported); + assert_eq!(err.to_string(), "read actions are disabled"); + + assert!(queue.is_empty()); + } + + #[tokio::test] + async fn reissue_requires_operational_relay() { + let relay: Relay = Relay::new(RelayUrl::parse("wss://relay.example.com").unwrap()); + let id: SubscriptionId = SubscriptionId::new("live"); + + relay + .inner + .add_long_lived_subscription(id.clone(), vec![Filter::new()]) + .await + .unwrap(); + + let err: Error = relay.reissue_subscription(&id).await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::State); + assert_eq!(err.to_string(), "relay is initialized but not ready"); + } + + #[tokio::test] + async fn restoration_does_not_repeat_a_req_queued_during_its_check() { + let controlled = ControlledRelay::new(); + let relay = &controlled.relay; + // Holding the queue keeps the connection task from draining it. + let mut queue = relay.inner.atomic.channels.rx_nostr().await; + let id: SubscriptionId = SubscriptionId::new("live"); + + relay.connect(); + relay + .subscribe(Filter::new().kind(Kind::TextNote)) + .with_id(id.clone()) + .await + .unwrap(); + drop(queue.try_recv().unwrap()); + + // The REQ belongs to an earlier session, so restoration wants to send it again. + relay.inner.stats.new_success(); + relay.inner.stats.new_success(); + assert!(relay.inner.should_resubscribe(&id).await); + + // Park restoration on the subscriptions lock, and queue a reissue behind it. + let subscriptions = relay.inner.atomic.subscriptions.read().await; + + let mut restore = Box::pin(relay.inner.resubscribe()); + assert!(futures::poll!(restore.as_mut()).is_pending()); + + let mut reissue = Box::pin(relay.reissue_subscription(&id).into_future()); + assert!(futures::poll!(reissue.as_mut()).is_pending()); + + drop(subscriptions); + + let (restored, reissued) = tokio::join!(restore, reissue); + restored.unwrap(); + reissued.unwrap(); + + // The relay must never receive a REQ for an ID it still has open. + let mut frames: Vec = Vec::new(); + let mut open: bool = false; + while let Ok(item) = queue.try_recv() { + match ClientMessage::from_json(&item.json).unwrap() { + ClientMessage::Req { .. } => { + assert!(!open, "duplicate REQ in {frames:?} + {}", item.json); + open = true; + } + ClientMessage::Close(_) => open = false, + _ => {} + } + frames.push(item.json); + } + assert!(open, "no REQ queued: {frames:?}"); + } + + #[tokio::test] + async fn restoration_skips_auto_closing_and_unregistered_subscriptions() { + let relay: Relay = Relay::new(RelayUrl::parse("wss://relay.example.com").unwrap()); + let auto_closing: SubscriptionId = SubscriptionId::new("auto-closing"); + + relay + .inner + .add_auto_closing_subscription(auto_closing.clone(), vec![Filter::new()]) + .await + .unwrap(); + + relay.inner.stats.new_success(); + relay.inner.stats.new_success(); + + assert!(!relay.inner.should_resubscribe(&auto_closing).await); + assert!( + !relay + .inner + .queue_resubscription(&auto_closing) + .await + .unwrap() + ); + + let unregistered: SubscriptionId = SubscriptionId::new("unregistered"); + assert!(!relay.inner.should_resubscribe(&unregistered).await); + assert!( + !relay + .inner + .queue_resubscription(&unregistered) + .await + .unwrap() + ); + } + + #[tokio::test] + async fn failed_restoration_keeps_subscription_restorable() { + let controlled = ControlledRelay::new(); + let relay = &controlled.relay; + // Holding the queue keeps the connection task from draining it. + let mut queue = relay.inner.atomic.channels.rx_nostr().await; + + relay.connect(); + let id: SubscriptionId = relay.subscribe(Filter::new()).await.unwrap(); + drop(queue.try_recv().unwrap()); + + relay.inner.stats.new_success(); + relay.inner.stats.new_success(); + relay.capabilities().remove(RelayCapabilities::READ); + + let err: Error = relay.inner.resubscribe().await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::Unsupported); + assert!(queue.is_empty()); + + // Nothing was queued, so the next restoration tries again. + let data: SubscriptionData = relay.subscription(&id).await.unwrap(); + assert!(data.is_closed()); + assert!(!data.pending_req_counter.as_ref().unwrap().has_pending()); + assert!(relay.inner.should_resubscribe(&id).await); + } } #[cfg(bench)] diff --git a/nostr-sdk/src/relay/mod.rs b/nostr-sdk/src/relay/mod.rs index eb6b47813..26037a437 100644 --- a/nostr-sdk/src/relay/mod.rs +++ b/nostr-sdk/src/relay/mod.rs @@ -29,6 +29,7 @@ pub use self::builder::*; pub use self::capabilities::*; pub(crate) use self::constants::DEFAULT_FETCH_EVENTS_LIMIT; use self::inner::InnerRelay; +pub use self::inner::SubscriptionData; pub use self::limits::*; pub use self::notification::*; pub use self::options::*; @@ -173,15 +174,15 @@ impl Relay { &self.inner.capabilities } - /// Get subscriptions + /// Get long-lived subscriptions #[inline] - pub async fn subscriptions(&self) -> HashMap> { + pub async fn subscriptions(&self) -> HashMap { self.inner.subscriptions().await } - /// Get filters by [SubscriptionId] + /// Get subscription by [SubscriptionId] #[inline] - pub async fn subscription(&self, id: &SubscriptionId) -> Option> { + pub async fn subscription(&self, id: &SubscriptionId) -> Option { self.inner.subscription(id).await } @@ -371,6 +372,26 @@ impl Relay { UnsubscribeAll::new(self) } + /// Reissue a long-lived subscription, so the relay sends its stored events again. + /// + /// Queues `CLOSE` followed by a `REQ` with the stored filters, both or neither. + /// The relay is never left without the subscription, + /// and never receives a `REQ` for an ID it still has open. + /// + /// If the relay isn't connected, both messages wait in the queue + /// and are sent on the next connection instead of the automatic restoration of the subscription, + /// so the relay still receives a single `REQ`. + /// + /// Returns an error if the subscription isn't registered or is auto-closing, + /// if the relay isn't operational or can't read, or if the queue can't fit both messages. + #[inline] + pub fn reissue_subscription<'id>( + &self, + id: &'id SubscriptionId, + ) -> ReissueSubscription<'_, 'id> { + ReissueSubscription::new(self, id) + } + /// Stream events from relay #[inline] pub fn stream_events(&self, filters: F) -> StreamEvents<'_>