diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index 9ab7cc71d..b9c3d357a 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -31,6 +31,8 @@ ### Added +- Add `Client::notifications_with_gaps` and `Relay::notifications_with_gaps` + to report receiver-local notification loss while continuing reception. - Add `LocalRelay::connections_left` (https://github.com/nostrdevkit/nostr/pull/1459) - 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 diff --git a/nostr-sdk/Cargo.toml b/nostr-sdk/Cargo.toml index 1a1dfaebe..8f97db5c7 100644 --- a/nostr-sdk/Cargo.toml +++ b/nostr-sdk/Cargo.toml @@ -75,6 +75,9 @@ name = "gossip" [[example]] name = "monitor" +[[example]] +name = "notification-gaps" + [[example]] name = "nostr-connect" diff --git a/nostr-sdk/README.md b/nostr-sdk/README.md index 955b49622..43674bc9b 100644 --- a/nostr-sdk/README.md +++ b/nostr-sdk/README.md @@ -15,6 +15,14 @@ The SDK can be used to build both sides of a nostr application: - clients, bots, and services that connect to existing relays; - local relays that run inside your process, including mock relays for tests. +For notification consumers that need to detect receiver buffer overflow, use +`Client::notifications_with_gaps` or `Relay::notifications_with_gaps`. A `Lagged` +item reports the number of notifications skipped by that receiver; subsequent +notifications remain available. The count is not a count of unique events, and +the stream does not establish complete relay history or downstream storage. +See the [notification gaps example](examples/notification-gaps.rs) for a minimal +consumer that continues after a gap while marking its coverage invalid. + ## Getting started ### Client diff --git a/nostr-sdk/examples/notification-gaps.rs b/nostr-sdk/examples/notification-gaps.rs new file mode 100644 index 000000000..4f3328d02 --- /dev/null +++ b/nostr-sdk/examples/notification-gaps.rs @@ -0,0 +1,32 @@ +use nostr_sdk::prelude::*; + +#[tokio::main] +async fn main() -> Result<(), Box> { + let relay_url = std::env::args().nth(1).expect("pass a relay URL"); + let client = Client::new(); + client.add_relay(relay_url).and_connect().await?; + + // Subscribe to notifications before asking for events. + let mut notifications = client.notifications_with_gaps(); + client + .subscribe(Filter::new().kind(Kind::TextNote).since(Timestamp::now())) + .await?; + + let mut coverage_invalid = false; + while let Some(update) = notifications.next().await { + match update { + NotificationUpdate::Lagged { skipped } => { + coverage_invalid = true; + eprintln!("This receiver missed {skipped} notifications; reacquisition is needed"); + } + NotificationUpdate::Notification(ClientNotification::Event { event, .. }) => { + // Events after a gap are still useful, but they do not repair it. + println!("{} (coverage invalid: {coverage_invalid})", event.id); + } + NotificationUpdate::Notification(ClientNotification::Shutdown) => break, + NotificationUpdate::Notification(_) => {} + } + } + + Ok(()) +} diff --git a/nostr-sdk/src/client/mod.rs b/nostr-sdk/src/client/mod.rs index 7910123e1..c2e777870 100644 --- a/nostr-sdk/src/client/mod.rs +++ b/nostr-sdk/src/client/mod.rs @@ -34,7 +34,7 @@ use crate::proxy::Proxy; use crate::relay::{ Relay, RelayCapabilities, RelayLimits, RelayOptions, SleepWhenIdle, SyncOptions, }; -use crate::stream::NotificationStream; +use crate::stream::{NotificationStream, NotificationUpdate, ReportingNotificationStream}; #[derive(Debug)] struct ClientConfig { @@ -194,6 +194,9 @@ impl Client { /// /// The stream terminates when the client shutdowns. /// + /// This legacy stream silently skips notifications lost when its receiver + /// falls behind. Use [`Client::notifications_with_gaps`] when coverage matters. + /// ///
When you call this method, you subscribe to the notifications channel from that precise moment. Anything received by relay/s before that moment is not included in the channel!
#[inline] pub fn notifications(&self) -> Pin + Send>> { @@ -222,6 +225,42 @@ impl Client { ) } + /// Get a notification stream that reports loss to this receiver. + /// + /// A [`NotificationUpdate::Lagged`] item means this receiver missed the + /// reported number of notifications. It does not count unique events or + /// identify affected subscriptions. Reception continues after a gap; callers + /// must explicitly reacquire any coverage they require. + /// + /// The stream starts at the moment this method is called. It emits + /// [`ClientNotification::Shutdown`] and then terminates on client shutdown. + /// A stream created after shutdown is empty. Dropping the stream ends only + /// this receiver; it does not stop the client. + #[inline] + pub fn notifications_with_gaps( + &self, + ) -> Pin> + Send>> { + if self.is_shutdown() { + return Box::pin(futures::stream::empty()); + } + + let rx = self.pool().notifications(); + let (tx, rx_done) = oneshot::channel(); + let mut tx: Option> = Some(tx); + + Box::pin( + ReportingNotificationStream::new(rx) + .inspect(move |update| { + if let NotificationUpdate::Notification(ClientNotification::Shutdown) = update { + if let Some(tx) = tx.take() { + let _ = tx.send(()); + } + } + }) + .take_until(rx_done), + ) + } + /// Get relays from the relay pool. /// /// # Configuration @@ -1190,12 +1229,192 @@ impl Client { #[cfg(test)] mod tests { + use std::num::NonZeroUsize; + + use nostr::event::{EventBuilder, FinalizeEvent, Kind}; + use nostr::key::Keys; use nostr_gossip_memory::prelude::*; use super::*; use crate::error::ErrorKind; use crate::local_relay::*; - use crate::relay::RelayStatus; + use crate::relay::{RelayNotification, RelayStatus}; + + #[tokio::test] + async fn test_notification_gap_reporting_is_receiver_local() { + let local = LocalRelay::builder().build(); + local.run().await.unwrap(); + + let url = local.url().await; + let client = Client::builder() + .notification_channel_size(NonZeroUsize::new(4).unwrap()) + .build(); + client + .add_relay(&url) + .notification_channel_size(4) + .and_connect() + .await + .unwrap(); + + let relay = client.relay(&url).await.unwrap().unwrap(); + let mut fast_client = client.notifications_with_gaps(); + let mut slow_client = client.notifications_with_gaps(); + let mut fast_relay = relay.notifications_with_gaps(); + let mut slow_relay = relay.notifications_with_gaps(); + let keys = Keys::generate(); + // Ignore another request's EOSE while waiting for the selected subscription. + client + .subscribe(Filter::new().kind(Kind::Metadata)) + .await + .unwrap(); + let id = client + .subscribe(Filter::new().author(keys.public_key())) + .await + .unwrap() + .value; + + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let update = fast_client.next().await.expect("active client stream closed"); + assert!(matches!(update, NotificationUpdate::Notification(_)), "active client receiver lagged"); + if matches!(update, NotificationUpdate::Notification(ClientNotification::Message { message, .. }) + if matches!(*message, RelayMessage::EndOfStoredEvents(ref received) if received.as_ref() == &id)) { + break; + } + } + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let update = fast_relay.next().await.expect("active relay stream closed"); + assert!(matches!(update, NotificationUpdate::Notification(_)), "active relay receiver lagged"); + if matches!(update, NotificationUpdate::Notification(RelayNotification::Message { message, .. }) + if matches!(*message, RelayMessage::EndOfStoredEvents(ref received) if received.as_ref() == &id)) { + break; + } + } + }) + .await + .unwrap(); + + let mut sent = Vec::new(); + for n in 0..64 { + let event = EventBuilder::new(Kind::TextNote, format!("notification-{n}")) + .finalize(&keys) + .unwrap(); + sent.push(event.id); + local.add_event(event).await.unwrap(); + + let received_client = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let update = fast_client + .next() + .await + .expect("active client stream closed"); + assert!( + matches!(update, NotificationUpdate::Notification(_)), + "active client receiver lagged" + ); + if let NotificationUpdate::Notification(ClientNotification::Event { + event, + .. + }) = update + { + break event.id; + } + } + }) + .await + .unwrap(); + let received_relay = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let update = fast_relay.next().await.expect("active relay stream closed"); + assert!( + matches!(update, NotificationUpdate::Notification(_)), + "active relay receiver lagged" + ); + if let NotificationUpdate::Notification(RelayNotification::Event { + event, + .. + }) = update + { + break event.id; + } + } + }) + .await + .unwrap(); + assert_eq!(received_client, sent[n]); + assert_eq!(received_relay, sent[n]); + } + + let first_client = tokio::time::timeout(Duration::from_secs(5), slow_client.next()) + .await + .unwrap() + .unwrap(); + let first_relay = tokio::time::timeout(Duration::from_secs(5), slow_relay.next()) + .await + .unwrap() + .unwrap(); + assert!(matches!(first_client, NotificationUpdate::Lagged { skipped } if skipped > 0)); + assert!(matches!(first_relay, NotificationUpdate::Lagged { skipped } if skipped > 0)); + assert!(matches!( + slow_client.next().await, + Some(NotificationUpdate::Notification(_)) + )); + assert!(matches!( + slow_relay.next().await, + Some(NotificationUpdate::Notification(_)) + )); + + client.shutdown().await; + + tokio::time::timeout(Duration::from_secs(5), async { + let mut saw_client_shutdown = false; + while let Some(update) = slow_client.next().await { + if matches!( + update, + NotificationUpdate::Notification(ClientNotification::Shutdown) + ) { + saw_client_shutdown = true; + } + } + assert!(saw_client_shutdown); + }) + .await + .unwrap(); + + tokio::time::timeout(Duration::from_secs(5), async { + let mut saw_relay_shutdown = false; + while let Some(update) = slow_relay.next().await { + if matches!(update, NotificationUpdate::Notification(RelayNotification::RelayStatus { status }) if status.is_shutdown()) { + saw_relay_shutdown = true; + } + } + assert!(saw_relay_shutdown); + }) + .await + .unwrap(); + } + + #[tokio::test] + async fn gap_streams_created_after_termination_are_empty() { + let client: Client = Client::new(); + client.shutdown().await; + assert!(client.notifications_with_gaps().next().await.is_none()); + + for banned in [false, true] { + let relay = + crate::relay::Relay::new(RelayUrl::parse("wss://relay.example.com").unwrap()); + if banned { + relay.ban(); + } else { + relay.shutdown(); + } + assert!(relay.notifications_with_gaps().next().await.is_none()); + } + } #[tokio::test] async fn test_shutdown() { diff --git a/nostr-sdk/src/lib.rs b/nostr-sdk/src/lib.rs index 0b82cc81d..bcb0fe541 100644 --- a/nostr-sdk/src/lib.rs +++ b/nostr-sdk/src/lib.rs @@ -24,3 +24,5 @@ mod stream; #[cfg(test)] mod test_utils; pub mod transport; + +pub use crate::stream::NotificationUpdate; diff --git a/nostr-sdk/src/relay/mod.rs b/nostr-sdk/src/relay/mod.rs index 8ed6f3854..8ecc860da 100644 --- a/nostr-sdk/src/relay/mod.rs +++ b/nostr-sdk/src/relay/mod.rs @@ -40,7 +40,7 @@ pub use self::status::*; use crate::client::ClientNotification; use crate::error::Error; use crate::shared::SharedState; -use crate::stream::NotificationStream; +use crate::stream::{NotificationStream, NotificationUpdate, ReportingNotificationStream}; /// Subscription auto-closed reason #[derive(Debug, Clone, PartialEq, Eq)] @@ -213,6 +213,9 @@ impl Relay { /// /// The stream terminates when the relay shutdowns or is banned. /// + /// This legacy stream silently skips notifications lost when its receiver + /// falls behind. Use [`Relay::notifications_with_gaps`] when coverage matters. + /// ///
When you call this method, you subscribe to the notifications channel from that precise moment. Anything received by relay/s before that moment is not included in the channel!
#[inline] pub fn notifications(&self) -> Pin + Send>> { @@ -245,6 +248,43 @@ impl Relay { ) } + /// Get a relay notification stream that reports loss to this receiver. + /// + /// A [`NotificationUpdate::Lagged`] item reports notifications skipped by + /// this receiver only. The stream continues after a gap. It starts at the + /// moment this method is called and terminates when the relay is shut down + /// or banned. Dropping it does not stop the relay. + #[inline] + pub fn notifications_with_gaps( + &self, + ) -> Pin> + Send>> { + let status = self.status(); + if status.is_banned() || status.is_shutdown() { + return Box::pin(futures::stream::empty()); + } + + let rx = self.inner.internal_notification_sender.subscribe(); + let (tx, rx_done) = oneshot::channel(); + let mut tx: Option> = Some(tx); + + Box::pin( + ReportingNotificationStream::new(rx) + .inspect(move |update| { + if let NotificationUpdate::Notification(RelayNotification::RelayStatus { + status, + }) = update + { + if status.is_banned() || status.is_shutdown() { + if let Some(tx) = tx.take() { + let _ = tx.send(()); + } + } + } + }) + .take_until(rx_done), + ) + } + /// Connect to the relay /// /// # Overview diff --git a/nostr-sdk/src/stream.rs b/nostr-sdk/src/stream.rs index af0ff33d7..93155871c 100644 --- a/nostr-sdk/src/stream.rs +++ b/nostr-sdk/src/stream.rs @@ -5,6 +5,23 @@ use futures::{Stream, StreamExt}; use tokio::sync::broadcast; use tokio::sync::mpsc::Receiver; use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::errors::BroadcastStreamRecvError; + +/// An item from a notification stream that reports receiver-local loss. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum NotificationUpdate { + /// A notification received by this stream. + Notification(T), + /// This stream missed `skipped` notifications because its buffer overflowed. + /// + /// The count is for this receiver only. It is not a count of unique events, + /// nor does it identify which subscriptions or relays need reacquisition. + /// Later notifications can still be received. + Lagged { + /// Number of notifications skipped by this receiver. + skipped: u64, + }, +} pub(crate) struct ReceiverStream { inner: Receiver, @@ -26,7 +43,7 @@ impl Stream for ReceiverStream { } pub(crate) struct NotificationStream { - inner: BroadcastStream, + inner: ReportingNotificationStream, } impl NotificationStream @@ -36,7 +53,7 @@ where #[inline] pub(crate) fn new(inner: broadcast::Receiver) -> Self { Self { - inner: BroadcastStream::new(inner), + inner: ReportingNotificationStream::new(inner), } } } @@ -50,12 +67,126 @@ where fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { loop { match self.inner.poll_next_unpin(cx) { - Poll::Ready(Some(Ok(notification))) => return Poll::Ready(Some(notification)), - // Skip errors for now - Poll::Ready(Some(Err(..))) => continue, + Poll::Ready(Some(NotificationUpdate::Notification(notification))) => { + return Poll::Ready(Some(notification)); + } + Poll::Ready(Some(NotificationUpdate::Lagged { .. })) => continue, Poll::Ready(None) => return Poll::Ready(None), Poll::Pending => return Poll::Pending, } } } } + +pub(crate) struct ReportingNotificationStream { + inner: BroadcastStream, +} + +impl ReportingNotificationStream +where + T: Clone + Send + 'static, +{ + #[inline] + pub(crate) fn new(inner: broadcast::Receiver) -> Self { + Self { + inner: BroadcastStream::new(inner), + } + } +} + +impl Stream for ReportingNotificationStream +where + T: Clone + Send + 'static, +{ + type Item = NotificationUpdate; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match self.inner.poll_next_unpin(cx) { + Poll::Ready(Some(Ok(notification))) => { + Poll::Ready(Some(NotificationUpdate::Notification(notification))) + } + Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(skipped)))) => { + Poll::Ready(Some(NotificationUpdate::Lagged { skipped })) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } +} + +#[cfg(test)] +mod tests { + use futures::StreamExt; + use tokio::sync::broadcast; + + use super::{NotificationStream, NotificationUpdate, ReportingNotificationStream}; + + #[tokio::test] + async fn reports_repeated_receiver_gaps_and_continues() { + let (tx, rx) = broadcast::channel(2); + let mut stream = ReportingNotificationStream::new(rx); + + for value in 0..8 { + tx.send(value).unwrap(); + } + assert_eq!( + stream.next().await, + Some(NotificationUpdate::Lagged { skipped: 6 }) + ); + assert_eq!( + stream.next().await, + Some(NotificationUpdate::Notification(6)) + ); + assert_eq!( + stream.next().await, + Some(NotificationUpdate::Notification(7)) + ); + + for value in 8..12 { + tx.send(value).unwrap(); + } + assert_eq!( + stream.next().await, + Some(NotificationUpdate::Lagged { skipped: 2 }) + ); + assert_eq!( + stream.next().await, + Some(NotificationUpdate::Notification(10)) + ); + assert_eq!( + stream.next().await, + Some(NotificationUpdate::Notification(11)) + ); + + drop(tx); + assert_eq!(stream.next().await, None); + assert_eq!(stream.next().await, None); + } + + #[tokio::test] + async fn legacy_stream_skips_gaps_and_continues() { + let (tx, rx) = broadcast::channel(2); + let mut stream = NotificationStream::new(rx); + for value in 0..8 { + tx.send(value).unwrap(); + } + assert_eq!(stream.next().await, Some(6)); + assert_eq!(stream.next().await, Some(7)); + drop(tx); + assert_eq!(stream.next().await, None); + } + + #[tokio::test] + async fn dropping_a_receiver_does_not_interrupt_another() { + let (tx, rx) = broadcast::channel(2); + let mut retained = ReportingNotificationStream::new(rx); + let dropped = ReportingNotificationStream::new(tx.subscribe()); + drop(dropped); + + tx.send(42).unwrap(); + assert_eq!( + retained.next().await, + Some(NotificationUpdate::Notification(42)) + ); + } +}