diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index f502064c6..8667f5962 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -37,6 +37,10 @@ ### Fixed +- Preserve reconnect requests made while a terminated connection task is still + releasing ownership, instead of leaving the relay pending with no task. +- Restore long-lived subscriptions on immediate reconnect even when both + connections occur within one second. - Require a correlated COUNT response before `Relay::count_events` returns a count. - Preserve the broadcast lag or closure cause in event `OK` and authentication waiters instead of returning a generic premature-exit error. diff --git a/nostr-sdk/src/client/api/send_event.rs b/nostr-sdk/src/client/api/send_event.rs index 02556c8e3..a3a86522b 100644 --- a/nostr-sdk/src/client/api/send_event.rs +++ b/nostr-sdk/src/client/api/send_event.rs @@ -446,6 +446,8 @@ where #[cfg(test)] mod tests { + use std::time::Duration; + use nostr::nips::nip17::InboxRelayList; use nostr::nips::nip65::RelayList; use nostr::prelude::*; @@ -454,9 +456,84 @@ mod tests { use nostr_gossip_memory::store::NostrGossipMemory; use super::*; + use crate::authenticator::SignerAuthenticator; use crate::client::{GossipConfig, GossipRelayLimits}; use crate::error::ErrorKind; use crate::local_relay::*; + use crate::test_utils::{setup_client, setup_client_with_authenticator}; + + #[tokio::test] + async fn auth_gated_clients_keep_independent_signers_across_removal_and_reconnect() { + let local = LocalRelay::builder() + .nip42(LocalRelayBuilderNip42::write()) + .build(); + local.run().await.unwrap(); + let url = local.url().await; + + let first_keys = Keys::generate(); + let second_keys = Keys::generate(); + let first = setup_client_with_authenticator( + url.clone(), + SignerAuthenticator::new(first_keys.clone()), + ) + .await; + let second = setup_client_with_authenticator( + url.clone(), + SignerAuthenticator::new(second_keys.clone()), + ) + .await; + let unauthenticated = setup_client(url.clone()).await; + + for (client, keys, content) in [ + (&first, &first_keys, "first account"), + (&second, &second_keys, "second account"), + ] { + let event = EventBuilder::new(Kind::TextNote, content) + .tag(Tag::protected()) + .finalize(keys) + .unwrap(); + let output = client + .send_event(&event) + .ok_timeout(Duration::from_secs(2)) + .authentication_timeout(Duration::from_secs(2)) + .await + .unwrap(); + assert!(output.failed.is_empty()); + assert!(output.success.get(&url).unwrap().is_ack()); + } + + let unauthenticated_event = EventBuilder::new(Kind::TextNote, "no signer") + .tag(Tag::protected()) + .finalize(&Keys::generate()) + .unwrap(); + let output = unauthenticated + .send_event(&unauthenticated_event) + .ok_timeout(Duration::from_secs(2)) + .await + .unwrap(); + assert!(output.success.is_empty()); + assert!(output.failed.contains_key(&url)); + + first.remove_relay(&url).force().await.unwrap(); + second.disconnect().await; + second.connect().and_wait(Duration::from_secs(3)).await; + assert_eq!( + second.relay(&url).await.unwrap().unwrap().status(), + crate::relay::RelayStatus::Connected + ); + let event = EventBuilder::new(Kind::TextNote, "second account after reconnect") + .tag(Tag::protected()) + .finalize(&second_keys) + .unwrap(); + let output = second + .send_event(&event) + .ok_timeout(Duration::from_secs(2)) + .authentication_timeout(Duration::from_secs(2)) + .await + .unwrap(); + assert!(output.failed.is_empty(), "{:?}", output.failed); + assert!(output.success.get(&url).unwrap().is_ack()); + } #[tokio::test] async fn unverified_event_is_rejected_before_sending() { diff --git a/nostr-sdk/src/client/mod.rs b/nostr-sdk/src/client/mod.rs index 7910123e1..f2575fce7 100644 --- a/nostr-sdk/src/client/mod.rs +++ b/nostr-sdk/src/client/mod.rs @@ -421,7 +421,12 @@ impl Client { let _ = self.remove_all_relays().force().await; } - /// Connect to a previously added relay + /// Connect to a previously added relay. + /// + /// This queues a connection request and returns before the relay is + /// connected. After [`Client::disconnect_relay`], it also works while the + /// previous connection task is still closing. Observe a fresh connected + /// status or live subscription traffic before treating the relay as ready. #[inline] pub async fn connect_relay<'a, U>(&self, url: U) -> Result<(), Error> where @@ -432,7 +437,13 @@ impl Client { self.pool().connect_relay(&url).await } - /// Try to connect to a previously added relay + /// Try to connect to a previously added relay. + /// + /// This is a one-shot connection attempt, not a task-exit barrier. If the + /// previous task still owns the relay after [`Client::disconnect_relay`], + /// it returns a state error and queues a fresh connection. Use + /// [`Client::connect_relay`] for immediate reactivation without retiring + /// and replacing the relay object. #[inline] pub async fn try_connect_relay<'a, U>(&self, url: U, timeout: Duration) -> Result<(), Error> where @@ -443,7 +454,10 @@ impl Client { self.pool().try_connect_relay(&url, timeout).await } - /// Disconnect relay + /// Disconnect relay. + /// + /// This requests termination and publishes `Terminated` status, but does + /// not wait for the connection task or WebSocket close to finish. #[inline] pub async fn disconnect_relay<'a, U>(&self, url: U) -> Result<(), Error> where diff --git a/nostr-sdk/src/relay/api/try_connect.rs b/nostr-sdk/src/relay/api/try_connect.rs index 99f26bf28..9256b4442 100644 --- a/nostr-sdk/src/relay/api/try_connect.rs +++ b/nostr-sdk/src/relay/api/try_connect.rs @@ -14,6 +14,19 @@ pub struct TryConnect<'relay> { timeout: Duration, } +struct ReservedConnection<'relay> { + relay: &'relay Relay, + transferred: bool, +} + +impl Drop for ReservedConnection<'_> { + fn drop(&mut self) { + if !self.transferred { + self.relay.inner.finish_reserved_connection_task(); + } + } +} + impl<'relay> TryConnect<'relay> { #[inline] pub(crate) fn new(relay: &'relay Relay) -> Self { @@ -63,6 +76,16 @@ impl<'relay> IntoFuture for TryConnect<'relay> { return Err(Error::connection_rejected(reason)); } + // Reserve the relay before dialing. Otherwise a retiring task can + // still own it after the handshake publishes Connected. + if !self.relay.inner.reserve_try_connect()? { + return Ok(()); + } + let mut reservation = ReservedConnection { + relay: self.relay, + transferred: false, + }; + // Try to connect // This will set the status to "terminated" if the connection fails let stream: (WebSocketSink, WebSocketStream) = self @@ -71,8 +94,8 @@ impl<'relay> IntoFuture for TryConnect<'relay> { ._try_connect(self.timeout, RelayStatus::Terminated) .await?; - // Spawn connection task - self.relay.inner.spawn_connection_task(Some(stream)); + self.relay.inner.spawn_reserved_connection_task(stream); + reservation.transferred = true; Ok(()) }) @@ -81,12 +104,142 @@ impl<'relay> IntoFuture for TryConnect<'relay> { #[cfg(test)] mod tests { + use std::net::SocketAddr; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use async_utility::time; - use nostr::types::RelayUrl; + use nostr::types::{RelayUrl, Url}; + use tokio::sync::Notify; use super::*; use crate::error::ErrorKind; use crate::local_relay::*; + use crate::policy::AdmitPolicy; + use crate::transport::websocket::WebSocketTransport; + + #[derive(Debug, Clone, Default)] + struct FirstConnectionGate { + blocked: Arc, + entered: Arc, + release: Arc, + } + + impl AdmitPolicy for FirstConnectionGate { + fn admit_connection<'a>( + &'a self, + _url: &'a RelayUrl, + ) -> BoxedFuture<'a, Result> { + Box::pin(async move { + if !self.blocked.swap(true, Ordering::SeqCst) { + self.entered.notify_one(); + self.release.notified().await; + } + Ok(AdmitStatus::Success) + }) + } + } + + #[tokio::test] + async fn try_connect_rechecks_state_after_connection_policy() { + for status in [ + RelayStatus::Shutdown, + RelayStatus::Banned, + RelayStatus::Connected, + ] { + let mock = MockRelay::run().await.unwrap(); + let gate = FirstConnectionGate::default(); + let relay = Relay::builder(mock.url().await) + .admit_policy(gate.clone()) + .build(); + let mut attempt = relay.try_connect().into_future(); + tokio::time::timeout(Duration::from_secs(2), async { + tokio::select! { + _ = gate.entered.notified() => (), + result = &mut attempt => panic!("attempt ended before policy release: {result:?}"), + } + }).await.unwrap(); + + match status { + RelayStatus::Shutdown => relay.shutdown(), + RelayStatus::Banned => relay.ban(), + RelayStatus::Connected => relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap(), + _ => unreachable!(), + } + gate.release.notify_one(); + let result = attempt.await; + assert_eq!(relay.status(), status); + if status.is_connected() { + result.unwrap(); + assert_eq!(relay.stats().success(), 1); + assert!(relay.inner.is_running()); + } else { + assert!(result.is_err()); + assert_eq!(relay.stats().attempts(), 0); + assert!(!relay.inner.is_running()); + } + relay.shutdown(); + mock.shutdown(); + } + } + + #[derive(Debug, Clone, Default)] + struct PendingTransport { + entered: Arc, + } + + impl WebSocketTransport for PendingTransport { + fn support_ping(&self) -> bool { + false + } + + fn connect<'a>( + &'a self, + _url: &'a Url, + _proxy: Option, + ) -> BoxedFuture<'a, Result<(WebSocketSink, WebSocketStream), Error>> { + Box::pin(async move { + self.entered.notify_one(); + std::future::pending().await + }) + } + } + + #[tokio::test] + async fn cancelled_and_timed_out_dials_release_connection_ownership() { + let transport = PendingTransport::default(); + let relay = Relay::builder("wss://relay.example.com".parse().unwrap()) + .websocket_transport(transport.clone()) + .build(); + let mut attempt = relay.try_connect().into_future(); + tokio::time::timeout(Duration::from_secs(2), async { + tokio::select! { + _ = transport.entered.notified() => (), + result = &mut attempt => panic!("dial ended before cancellation: {result:?}"), + } + }) + .await + .unwrap(); + assert_eq!(relay.status(), RelayStatus::Connecting); + assert!(relay.inner.is_running()); + drop(attempt); + assert_eq!(relay.status(), RelayStatus::Terminated); + assert!(!relay.inner.is_running()); + + let error = relay + .try_connect() + .timeout(Duration::from_millis(20)) + .await + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::Timeout); + assert_eq!(relay.stats().attempts(), 2); + assert_eq!(relay.status(), RelayStatus::Terminated); + assert!(!relay.inner.is_running()); + } #[tokio::test] async fn test_try_connect() { diff --git a/nostr-sdk/src/relay/inner.rs b/nostr-sdk/src/relay/inner.rs index 254c823ac..e1d8c6358 100644 --- a/nostr-sdk/src/relay/inner.rs +++ b/nostr-sdk/src/relay/inner.rs @@ -2,8 +2,8 @@ use std::borrow::Cow; use std::cmp; use std::collections::HashMap; use std::net::SocketAddr; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use async_utility::{task, time}; @@ -105,12 +105,19 @@ impl RelayChannels { pub fn terminate(&self) { self.terminate.notify_one() } + + fn clear_termination_permit(&self) { + let mut notified = std::pin::pin!(self.terminate.notified()); + // `enable` consumes a stored permit, if any. The old task no longer + // waits for it when its replacement is reserved. + let _ = notified.as_mut().enable(); + } } #[derive(Debug)] struct SubscriptionData { pub filters: Vec, - pub subscribed_at: Timestamp, + pub subscribed_connection: usize, pub is_auto_closing: bool, /// Received EOSE msg pub received_eose: bool, @@ -122,10 +129,10 @@ struct SubscriptionData { impl SubscriptionData { #[inline] - fn long_lived(filters: Vec) -> Self { + fn long_lived(filters: Vec, subscribed_connection: usize) -> Self { Self { filters, - subscribed_at: Timestamp::now(), + subscribed_connection, is_auto_closing: false, received_eose: false, received_events: AtomicUsize::new(0), @@ -137,7 +144,7 @@ impl SubscriptionData { fn auto_closing(filters: Vec) -> Self { Self { filters, - subscribed_at: Timestamp::zero(), + subscribed_connection: 0, is_auto_closing: true, received_eose: false, received_events: AtomicUsize::new(0), @@ -153,7 +160,24 @@ pub(super) struct AtomicPrivateData { status: AtomicRelayStatus, channels: RelayChannels, subscriptions: RwLock>, - running: AtomicBool, + connection_task: StdMutex, + #[cfg(test)] + teardown_hook: StdMutex>>, +} + +#[derive(Debug, Default)] +struct ConnectionTaskOwnership { + running: bool, + restart_requested: bool, +} + +#[cfg(test)] +#[derive(Debug, Default)] +struct TeardownHook { + started: Notify, + entered: Notify, + release: Notify, + finished: Notify, } #[derive(Debug, Clone)] @@ -184,7 +208,9 @@ impl InnerRelay { status: AtomicRelayStatus::default(), channels: RelayChannels::new(), subscriptions: RwLock::new(HashMap::new()), - running: AtomicBool::new(false), + connection_task: StdMutex::new(ConnectionTaskOwnership::default()), + #[cfg(test)] + teardown_hook: StdMutex::new(None), }), capabilities: Arc::new(AtomicRelayCapabilities::new(capabilities)), opts, @@ -209,8 +235,9 @@ impl InnerRelay { /// Check if the connection task is running #[inline] + #[cfg(test)] pub(super) fn is_running(&self) -> bool { - self.atomic.running.load(Ordering::SeqCst) + self.atomic.connection_task.lock().unwrap().running } #[inline] @@ -341,7 +368,10 @@ impl InnerRelay { return Err(Error::invalid_msg("subscription ID already exists")); } - subscriptions.insert(id, SubscriptionData::long_lived(filters)); + subscriptions.insert( + id, + SubscriptionData::long_lived(filters, self.stats.success()), + ); Ok(()) } @@ -367,7 +397,7 @@ impl InnerRelay { &self, id: &SubscriptionId, filters: Vec, - update_subscribed_at: bool, + mark_subscribed: bool, ) -> Result<(), Error> { let mut subscriptions = self.atomic.subscriptions.write().await; let data = subscriptions @@ -375,8 +405,8 @@ impl InnerRelay { .ok_or_else(|| Error::not_found("subscription not found"))?; data.filters = filters; - if update_subscribed_at { - data.subscribed_at = Timestamp::now(); + if mark_subscribed { + data.subscribed_connection = self.stats.success(); data.closed = false; } @@ -419,20 +449,21 @@ impl InnerRelay { let subscriptions = self.atomic.subscriptions.read().await; match subscriptions.get(id) { Some(SubscriptionData { - subscribed_at, + subscribed_connection, closed, is_auto_closing: false, .. }) => { // Never subscribed -> SHOULD subscribe // Subscription closed by relay -> SHOULD subscribe - if subscribed_at.is_zero() || *closed { + if *closed { return true; } - // First connection and subscribed_at != 0 -> SHOULD NOT re-subscribe - // Many connections and subscription NOT done in current websocket session -> SHOULD re-subscribe - self.stats.connected_at() > *subscribed_at && self.stats.success() > 1 + // A request queued before the first connection is already in the + // outbound channel. Later connections must restore it once. + let current_connection = self.stats.success(); + current_connection > 1 && current_connection > *subscribed_connection } // NOT subscribe if auto-closing subscription or subscription not found Some(SubscriptionData { @@ -548,48 +579,82 @@ impl InnerRelay { tracing::debug!(url = %self.url, "Waking up sleeping relay."); - // Update status to pending - // TODO: is this needed here? - self.set_status(RelayStatus::Pending, false); - - // Spawn a new connection task - self.spawn_connection_task(None); + self.request_connect(); // Update the wake-up timestamp self.stats.just_woke_up(); } - pub(super) fn spawn_connection_task(&self, stream: Option<(WebSocketSink, WebSocketStream)>) { - // Check if the connection task is already running - // This is checked also later, but it's checked also here to avoid a full-clone if we know that is already running. - if self.is_running() { - tracing::warn!(url = %self.url, "Connection task is already running."); + pub(super) fn reserve_try_connect(&self) -> Result { + let mut ownership = self.atomic.connection_task.lock().unwrap(); + let status = self.status(); + if status.is_shutdown() { + return Err(Error::shutdown()); + } + if status.is_banned() { + return Err(Error::banned()); + } + if !status.can_connect() { + return Ok(false); + } + if ownership.running { + ownership.restart_requested = true; + self.set_status(RelayStatus::Pending, false); + return Err(Error::state_msg( + "previous connection task is stopping; reconnect queued", + )); + } + self.atomic.channels.clear_termination_permit(); + ownership.running = true; + self.set_status(RelayStatus::Pending, false); + Ok(true) + } + + pub(super) fn spawn_reserved_connection_task(&self, stream: (WebSocketSink, WebSocketStream)) { + task::spawn(self.clone().connection_task(Some(stream))); + } + + pub(super) fn finish_reserved_connection_task(&self) { + if self.status() == RelayStatus::Connecting { + self.set_status(RelayStatus::Terminated, false); + } + self.finish_connection_task(); + } + + pub(super) fn request_connect(&self) { + let mut ownership = self.atomic.connection_task.lock().unwrap(); + if !self.status().can_connect() { return; } - // Full-clone - let relay: InnerRelay = self.clone(); + self.set_status(RelayStatus::Pending, false); + if ownership.running { + ownership.restart_requested = true; + return; + } - // Spawn task - task::spawn(relay.connection_task(stream)); + self.atomic.channels.clear_termination_permit(); + ownership.running = true; + drop(ownership); + task::spawn(self.clone().connection_task(None)); } - /// This **MUST** be called only by the [`InnerRelay::spawn_connection_task`] method! + /// This **MUST** be called only after connection-task ownership is reserved. async fn connection_task(self, mut stream: Option<(WebSocketSink, WebSocketStream)>) { - // Set the connection task as running and get the previous value. - let is_running: bool = self.atomic.running.swap(true, Ordering::SeqCst); - - // Re-check if the connection task is already running. - // This is required because may happen that two tasks are spawned at the exact same moment. - // Not use the "assert" macro since will cause the task to panic. - if is_running { - tracing::warn!(url = %self.url, "Connection task is already running."); + let status = self.status(); + if status.is_terminated() || status.is_banned() || status.is_shutdown() { + self.finish_connection_task(); return; } // Lock receiver let mut rx_nostr = self.atomic.channels.rx_nostr().await; + #[cfg(test)] + if let Some(hook) = self.atomic.teardown_hook.lock().unwrap().as_ref() { + hook.started.notify_one(); + } + // Last websocket error // Store it to avoid printing every time the same connection error let mut last_ws_error = None; @@ -621,6 +686,18 @@ impl InnerRelay { // Get status let status: RelayStatus = self.status(); + // A connect requested during teardown belongs to a replacement + // task. Do not turn it into an ordinary delayed retry here. + if self + .atomic + .connection_task + .lock() + .unwrap() + .restart_requested + { + break; + } + // If the relay is terminated, banned or sleeping, break the loop. if status.is_terminated() || status.is_banned() @@ -664,12 +741,48 @@ impl InnerRelay { } } - // Mark the connection task as stopped. - self.atomic.running.store(false, Ordering::SeqCst); + drop(rx_nostr); + + #[cfg(test)] + let hook = self.atomic.teardown_hook.lock().unwrap().take(); + #[cfg(test)] + if let Some(hook) = hook.as_ref() { + hook.entered.notify_one(); + hook.release.notified().await; + } + + self.finish_connection_task(); + + #[cfg(test)] + if let Some(hook) = hook { + hook.finished.notify_one(); + } tracing::debug!(url = %self.url, "Auto connect loop terminated."); } + fn finish_connection_task(&self) { + let mut ownership = self.atomic.connection_task.lock().unwrap(); + let status = self.status(); + let restart = ownership.restart_requested + && !status.is_banned() + && !status.is_shutdown() + && !status.is_sleeping(); + + ownership.restart_requested = false; + if restart { + if status.is_terminated() { + self.set_status(RelayStatus::Pending, false); + } + self.atomic.channels.clear_termination_permit(); + // Keep ownership reserved until the replacement task starts. + drop(ownership); + task::spawn(self.clone().connection_task(None)); + } else { + ownership.running = false; + } + } + /// Depending on attempts and success, use default or incremental retry interval fn calculate_retry_interval(&self) -> Duration { // Check if the incremental interval is enabled @@ -720,8 +833,18 @@ impl InnerRelay { timeout: Duration, status_on_failure: RelayStatus, ) -> Result<(WebSocketSink, WebSocketStream), Error> { - // Update status - self.set_status(RelayStatus::Connecting, true); + { + let ownership = self.atomic.connection_task.lock().unwrap(); + let status = self.status(); + if ownership.restart_requested + || status.is_terminated() + || status.is_banned() + || status.is_shutdown() + { + return Err(Error::state_msg("connection attempt superseded")); + } + self.set_status(RelayStatus::Connecting, true); + } // Increase the attempts self.stats.new_attempt(); @@ -740,7 +863,12 @@ impl InnerRelay { // Connect res = fut => match res { Some(Ok((ws_tx, ws_rx))) => { - // Update status + // A disconnect or replacement request may have superseded + // this dial while the transport was connecting. + let _ownership = self.atomic.connection_task.lock().unwrap(); + if self.status() != RelayStatus::Connecting { + return Err(Error::state_msg("connection attempt superseded")); + } self.set_status(RelayStatus::Connected, true); // Increment success stats @@ -749,15 +877,19 @@ impl InnerRelay { Ok((ws_tx, ws_rx)) } Some(Err(e)) => { - // Update status - self.set_status(status_on_failure, false); + let _ownership = self.atomic.connection_task.lock().unwrap(); + if self.status() == RelayStatus::Connecting { + self.set_status(status_on_failure, false); + } // Return error Err(Error::transport(e)) } None => { - // Update status - self.set_status(status_on_failure, false); + let _ownership = self.atomic.connection_task.lock().unwrap(); + if self.status() == RelayStatus::Connecting { + self.set_status(status_on_failure, false); + } // Return error Err(Error::timeout()) @@ -1331,6 +1463,7 @@ impl InnerRelay { } pub fn disconnect(&self) { + let mut ownership = self.atomic.connection_task.lock().unwrap(); let status = self.status(); // Check if it's already terminated, banned or shutdown @@ -1343,9 +1476,11 @@ impl InnerRelay { // Update status self.set_status(RelayStatus::Terminated, true); + ownership.restart_requested = false; } pub fn ban(&self) { + let mut ownership = self.atomic.connection_task.lock().unwrap(); let status = self.status(); // Check if it's already terminated, banned or shutdown @@ -1358,9 +1493,11 @@ impl InnerRelay { // Update status self.set_status(RelayStatus::Banned, true); + ownership.restart_requested = false; } pub(super) fn shutdown(&self) { + let mut ownership = self.atomic.connection_task.lock().unwrap(); let status = self.status(); // Check if it's already terminated, banned or shutdown @@ -1373,6 +1510,7 @@ impl InnerRelay { // Update status self.set_status(RelayStatus::Shutdown, true); + ownership.restart_requested = false; } #[inline] @@ -1889,9 +2027,242 @@ mod tests { use super::*; use crate::authenticator::SignerAuthenticator; use crate::error::ErrorKind; + use crate::local_relay::MockRelay; use crate::policy::{AdmitPolicy, AdmitStatus}; use crate::relay::{Relay, RelayOptions}; + #[tokio::test] + async fn reconnect_requested_during_task_teardown_is_not_lost() { + reconnect_during_teardown(true, false).await; + } + + #[tokio::test] + async fn explicit_reconnect_during_teardown_works_without_automatic_reconnect() { + reconnect_during_teardown(false, false).await; + } + + #[tokio::test] + async fn queued_reconnect_ignores_stale_termination_permit() { + reconnect_during_teardown(false, true).await; + } + + #[tokio::test] + async fn connect_after_idle_disconnect_ignores_stored_termination() { + let local = MockRelay::run().await.unwrap(); + let relay = Relay::builder(local.url().await) + .opts(RelayOptions::default().reconnect(false)) + .build(); + relay.disconnect(); + assert_eq!(relay.status(), RelayStatus::Terminated); + + relay.connect(); + relay.wait_for_connection(Duration::from_secs(2)).await; + assert_eq!(relay.status(), RelayStatus::Connected); + assert_eq!(relay.stats().success(), 1); + } + + async fn reconnect_during_teardown(automatic_reconnect: bool, stale_termination: bool) { + let local = MockRelay::run().await.unwrap(); + let relay = Relay::builder(local.url().await) + .opts(RelayOptions::default().reconnect(automatic_reconnect)) + .build(); + let hook = Arc::new(TeardownHook::default()); + *relay.inner.atomic.teardown_hook.lock().unwrap() = Some(hook.clone()); + relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap(); + + tokio::time::timeout(Duration::from_secs(2), hook.started.notified()) + .await + .unwrap(); + let mut notifications = relay.notifications(); + let keys = Keys::generate(); + let subscription_id = relay + .subscribe(Filter::new().author(keys.public_key())) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let notification = notifications + .next() + .await + .expect("notification stream closed before EOSE"); + if matches!(notification, RelayNotification::Message { message } if matches!(*message, RelayMessage::EndOfStoredEvents(ref id) if id.as_ref() == &subscription_id)) { + break; + } + } + }) + .await + .unwrap(); + + relay.disconnect(); + tokio::time::timeout(Duration::from_secs(2), hook.entered.notified()) + .await + .unwrap(); + assert_eq!(relay.status(), RelayStatus::Terminated); + assert!(relay.inner.is_running()); + + if stale_termination { + // The old task no longer polls the termination Notify at this gate. + relay.inner.atomic.channels.terminate(); + } + + let ((), ()) = tokio::join!(async { relay.connect() }, async { relay.connect() }); + assert_eq!(relay.status(), RelayStatus::Pending); + hook.release.notify_one(); + + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let notification = notifications + .next() + .await + .expect("notification stream closed before reconnect"); + if matches!( + notification, + RelayNotification::RelayStatus { + status: RelayStatus::Connected + } + ) { + break; + } + } + }) + .await + .unwrap(); + assert_eq!(relay.status(), RelayStatus::Connected); + assert!(relay.inner.is_running()); + + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let notification = notifications + .next() + .await + .expect("notification stream closed before restored EOSE"); + if matches!(notification, RelayNotification::Message { message } if matches!(*message, RelayMessage::EndOfStoredEvents(ref id) if id.as_ref() == &subscription_id)) { + break; + } + } + }) + .await + .unwrap(); + + let event = EventBuilder::new(Kind::TextNote, "after reconnect") + .finalize(&keys) + .unwrap(); + local.add_event(event.clone()).await.unwrap(); + let received = tokio::time::timeout(Duration::from_secs(2), async { + loop { + let notification = notifications + .next() + .await + .expect("notification stream closed"); + if let RelayNotification::Event { + subscription_id: received_id, + event, + } = notification + { + if received_id == subscription_id { + break event.id; + } + } + } + }) + .await + .unwrap(); + assert_eq!(received, event.id); + } + + #[tokio::test] + async fn shutdown_cancels_a_queued_reconnect() { + let local = MockRelay::run().await.unwrap(); + let relay = Relay::new(local.url().await); + let hook = Arc::new(TeardownHook::default()); + *relay.inner.atomic.teardown_hook.lock().unwrap() = Some(hook.clone()); + relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(2), hook.started.notified()) + .await + .unwrap(); + + relay.disconnect(); + tokio::time::timeout(Duration::from_secs(2), hook.entered.notified()) + .await + .unwrap(); + relay.connect(); + relay.shutdown(); + hook.release.notify_one(); + tokio::time::timeout(Duration::from_secs(2), hook.finished.notified()) + .await + .unwrap(); + + assert_eq!(relay.status(), RelayStatus::Shutdown); + assert!(!relay.inner.is_running()); + } + + #[tokio::test] + async fn try_connect_reports_when_old_task_still_owns_connection() { + let local = MockRelay::run().await.unwrap(); + let relay = Relay::new(local.url().await); + let hook = Arc::new(TeardownHook::default()); + *relay.inner.atomic.teardown_hook.lock().unwrap() = Some(hook.clone()); + relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(2), hook.started.notified()) + .await + .unwrap(); + + relay.disconnect(); + tokio::time::timeout(Duration::from_secs(2), hook.entered.notified()) + .await + .unwrap(); + + let success_before = relay.stats().success(); + let mut status_updates = relay.inner.internal_notification_sender.subscribe(); + let error = relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::State); + assert_eq!(relay.status(), RelayStatus::Pending); + assert_eq!(relay.stats().success(), success_before); + loop { + match status_updates.try_recv() { + Ok(RelayNotification::RelayStatus { + status: RelayStatus::Connected, + }) => panic!("discarded try_connect socket reported Connected"), + Ok(_) => {} + Err(broadcast::error::TryRecvError::Empty) => break, + Err(error) => panic!("unexpected status receiver error: {error}"), + } + } + let mut notifications = relay.notifications(); + hook.release.notify_one(); + + tokio::time::timeout(Duration::from_secs(5), async { + while let Some(notification) = notifications.next().await { + if matches!( + notification, + RelayNotification::RelayStatus { + status: RelayStatus::Connected + } + ) { + break; + } + } + }) + .await + .unwrap(); + } + #[derive(Debug)] struct CountingAdmitPolicy(Arc); diff --git a/nostr-sdk/src/relay/mod.rs b/nostr-sdk/src/relay/mod.rs index 8ed6f3854..ddd108e7a 100644 --- a/nostr-sdk/src/relay/mod.rs +++ b/nostr-sdk/src/relay/mod.rs @@ -254,23 +254,15 @@ impl Relay { /// Otherwise, the connection task will be spawned, which will attempt to connect to relay. /// /// This method returns immediately and doesn't provide any information on if the connection was successful or not. + /// A request made while the previous task is stopping is retained and starts + /// after that task releases ownership. /// /// # Automatic reconnection /// /// By default, in case of disconnection, the connection task will automatically attempt to reconnect. /// This behavior can be disabled by changing [`RelayOptions::reconnect`] option. pub fn connect(&self) { - // Immediately return if can't connect - if !self.status().can_connect() { - return; - } - - // Update status - // Change it to pending to avoid issues with the health check (initialized check) - self.inner.set_status(RelayStatus::Pending, false); - - // Spawn connection task - self.inner.spawn_connection_task(None); + self.inner.request_connect(); } /// Waits for relay connection @@ -328,6 +320,8 @@ impl Relay { /// regardless of whether the initial connection succeeds. /// /// Returns an error if the connection fails or if the relay has been banned. + /// If an older connection task still owns the relay, this returns a state + /// error and queues a fresh reconnect rather than claiming the new socket. /// /// # Automatic reconnection /// @@ -339,6 +333,10 @@ impl Relay { } /// Disconnect from relay and set status to [`RelayStatus::Terminated`]. + /// + /// This requests termination; the old connection task may still be + /// releasing its resources when this method returns. A subsequent + /// [`Relay::connect`] request is retained during that interval. #[inline] pub fn disconnect(&self) { self.inner.disconnect() @@ -825,7 +823,6 @@ mod tests { relay.ban(); assert_eq!(relay.status(), RelayStatus::Banned); - assert!(!relay.inner.is_running()); // Retry to connect let res = relay.try_connect().timeout(Duration::from_secs(2)).await; diff --git a/nostr-sdk/src/relay/status.rs b/nostr-sdk/src/relay/status.rs index bf17e15bc..23b34931d 100644 --- a/nostr-sdk/src/relay/status.rs +++ b/nostr-sdk/src/relay/status.rs @@ -55,7 +55,11 @@ pub enum RelayStatus { Connected = 3, /// The connection failed, but another attempt will occur soon. Disconnected = 4, - /// The connection has been terminated and no retry will occur. + /// Termination was requested and no automatic retry will occur. + /// + /// This status does not mean that the previous connection task has exited + /// or that its WebSocket has finished closing. Use [`Relay::connect`](crate::relay::Relay::connect) + /// to queue a new connection while that task releases ownership. Terminated = 5, /// The relay has been banned. Banned = 6,