From 64a1fa3bcfe9b0cb1e76b386417af34c3368eb46 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 30 Sep 2026 09:18:48 +0200 Subject: [PATCH 1/3] sdk: clean up interrupted reconciliation and fail rejected downloads --- nostr-sdk/CHANGELOG.md | 6 + nostr-sdk/src/relay/api/sync.rs | 299 +++++++++++++++++++++++++++++--- nostr-sdk/src/relay/inner.rs | 5 + 3 files changed, 290 insertions(+), 20 deletions(-) diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index 871813789..b21c9d9ce 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -35,6 +35,12 @@ - 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 +### Fixed + +- Clean up NIP-77 request-owned subscriptions when reconciliation fails or is cancelled. +- Fail NIP-77 reconciliation when a download batch is rejected or closes before + its requested events arrive, without closing unrelated subscriptions. + ### Deprecated - Deprecate `LocalRelayBuilder::max_query_results` and `LocalRelayBuilder::default_filter_limit` (https://github.com/nostrdevkit/nostr/pull/1461) diff --git a/nostr-sdk/src/relay/api/sync.rs b/nostr-sdk/src/relay/api/sync.rs index 968b087f3..b1f158209 100644 --- a/nostr-sdk/src/relay/api/sync.rs +++ b/nostr-sdk/src/relay/api/sync.rs @@ -3,7 +3,7 @@ use std::cmp; use std::collections::{HashMap, HashSet}; use std::future::IntoFuture; -use async_utility::time; +use async_utility::{task, time}; use negentropy::{Id, Negentropy, NegentropyStorageVector}; use nostr::event::EventId; use nostr::filter::Filter; @@ -103,6 +103,49 @@ fn neg_id_to_event_id(id: Id) -> EventId { EventId::from_byte_array(id.to_bytes()) } +// A dropped sync future must not keep its request-owned subscription registered. +// Network CLOSE is best effort because the relay may already be disconnected. +struct SyncCleanup { + relay: Relay, + neg_id: SubscriptionId, + down_id: SubscriptionId, + armed: bool, +} + +impl SyncCleanup { + fn new(relay: &Relay, neg_id: SubscriptionId, down_id: SubscriptionId) -> Self { + Self { + relay: relay.clone(), + neg_id, + down_id, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for SyncCleanup { + fn drop(&mut self) { + if !self.armed { + return; + } + + let relay = self.relay.clone(); + let neg_id = self.neg_id.clone(); + let down_id = self.down_id.clone(); + task::spawn(async move { + relay.inner.remove_subscription(&down_id).await; + let _ = relay + .send_msg(ClientMessage::Close(Cow::Borrowed(&down_id))) + .await; + let _ = send_neg_close(&relay, &neg_id).await; + }); + } +} + #[inline(never)] async fn handle_neg_msg( relay: &Relay, @@ -218,7 +261,8 @@ async fn req_neg_events( relay: &Relay, need_ids: &mut Vec, in_flight_down: &mut bool, - down_sub_id: &SubscriptionId, + pending_down_ids: &mut HashSet, + cleanup: &mut SyncCleanup, opts: &SyncOptions, ) -> Result<(), Error> { // Check if it should skip the download @@ -226,6 +270,11 @@ async fn req_neg_events( return Ok(()); } + // Each batch needs its own ID: a relay may send CLOSED after EOSE, and a + // late CLOSED from the previous batch must not terminate the next one. + cleanup.down_id = SubscriptionId::generate(); + let down_sub_id = &cleanup.down_id; + let capacity: usize = cmp::min(need_ids.len(), NEGENTROPY_BATCH_SIZE_DOWN); let mut ids: Vec = Vec::with_capacity(capacity); @@ -249,6 +298,7 @@ async fn req_neg_events( }); } + let requested_ids: HashSet = ids.iter().copied().collect(); let filter = Filter::new().ids(ids); let msg: ClientMessage = ClientMessage::Req { subscription_id: Cow::Borrowed(down_sub_id), @@ -270,6 +320,7 @@ async fn req_neg_events( return Err(e); } + *pending_down_ids = requested_ids; *in_flight_down = true; Ok(()) @@ -334,16 +385,17 @@ pub(super) async fn sync( initial_message: Cow::Owned(faster_hex::hex_string(&initial_message)), }; relay.send_msg(open_msg).await?; + let mut cleanup = SyncCleanup::new(relay, sub_id.clone(), SubscriptionId::generate()); // Check if negentropy is supported check_negentropy_support(&sub_id, opts, &mut temp_notifications).await?; let mut in_flight_up: HashSet = HashSet::new(); let mut in_flight_down: bool = false; + let mut pending_down_ids: HashSet = HashSet::new(); let mut sync_done: bool = false; let mut have_ids: Vec = Vec::new(); let mut need_ids: Vec = Vec::new(); - let down_sub_id: SubscriptionId = SubscriptionId::generate(); let mut last_relevant_msg: Instant = Instant::now(); // Start reconciliation @@ -430,8 +482,9 @@ pub(super) async fn sync( event, } => { #[allow(clippy::collapsible_match)] - if subscription_id.as_ref() == &down_sub_id { + if subscription_id.as_ref() == &cleanup.down_id { output.received.insert(event.id); + pending_down_ids.remove(&event.id); // Relevant to this sync true @@ -442,15 +495,16 @@ pub(super) async fn sync( } RelayMessage::EndOfStoredEvents(subscription_id) => { #[allow(clippy::collapsible_match)] - if subscription_id.as_ref() == &down_sub_id { + if subscription_id.as_ref() == &cleanup.down_id { in_flight_down = false; + pending_down_ids.clear(); // Remove subscription - relay.inner.remove_subscription(&down_sub_id).await; + relay.inner.remove_subscription(&cleanup.down_id).await; // Close subscription relay - .send_msg(ClientMessage::Close(Cow::Borrowed(&down_sub_id))) + .send_msg(ClientMessage::Close(Cow::Borrowed(&cleanup.down_id))) .await?; // Relevant to this sync @@ -461,19 +515,27 @@ pub(super) async fn sync( } } RelayMessage::Closed { - subscription_id, .. + subscription_id, + message, } => { #[allow(clippy::collapsible_match)] - if subscription_id.as_ref() == &down_sub_id { - in_flight_down = false; - - // NOTE: the subscription is removed in the `InnerRelay::handle_relay_message` method, - // so there is no need to try to remove it also here. - - // Relevant to this sync - true + if subscription_id.as_ref() == &cleanup.down_id && in_flight_down { + // An empty CLOSED can end a finite explicit-ID + // batch only after every requested event arrived. + if message.is_empty() && pending_down_ids.is_empty() { + in_flight_down = false; + true + } else { + return Err(if message.is_empty() { + Error::relay_msg( + "download subscription closed before all requested events arrived" + .to_owned(), + ) + } else { + Error::relay_msg(message.into_owned()) + }); + } } else { - // Not relevant to this sync false } } @@ -488,7 +550,8 @@ pub(super) async fn sync( relay, &mut need_ids, &mut in_flight_down, - &down_sub_id, + &mut pending_down_ids, + &mut cleanup, opts, ) .await?; @@ -516,6 +579,8 @@ pub(super) async fn sync( tracing::info!(url = %relay.url(), "Negentropy reconciliation terminated."); + cleanup.disarm(); + Ok(()) } @@ -631,17 +696,144 @@ impl<'relay> IntoFuture for SyncEvents<'relay> { #[cfg(test)] mod tests { use std::collections::{HashMap, HashSet}; + use std::future::Future; + use std::net::SocketAddr; + use std::pin::Pin; use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; + use futures::StreamExt; + use nostr::message::MachineReadablePrefix; use nostr_memory::prelude::*; - use tokio::sync::broadcast; + use tokio::sync::{Notify, broadcast}; use super::*; + use crate::client::Client; use crate::error::ErrorKind; - use crate::local_relay::*; + use crate::local_relay::{LocalRelay, MockRelay, QueryPolicy, QueryPolicyResult}; use crate::relay::{SyncDirection, SyncOptions}; + #[derive(Debug)] + struct RejectDownloadAfter { + allowed_batches: usize, + seen_batches: AtomicUsize, + } + + impl QueryPolicy for RejectDownloadAfter { + fn admit_query<'a>( + &'a self, + query: &'a mut Filter, + _addr: &'a SocketAddr, + ) -> Pin + Send + 'a>> { + let reject = query.ids.is_some() + && self.seen_batches.fetch_add(1, Ordering::SeqCst) >= self.allowed_batches; + Box::pin(async move { + if reject { + QueryPolicyResult::reject(MachineReadablePrefix::Blocked, "reads denied") + } else { + QueryPolicyResult::Accept + } + }) + } + } + + #[derive(Debug)] + struct HoldDownload { + entered: Arc, + release: Arc, + } + + impl QueryPolicy for HoldDownload { + fn admit_query<'a>( + &'a self, + query: &'a mut Filter, + _addr: &'a SocketAddr, + ) -> Pin + Send + 'a>> { + Box::pin(async move { + if query.ids.is_some() { + self.entered.notify_one(); + self.release.notified().await; + } + QueryPolicyResult::Accept + }) + } + } + + #[tokio::test] + async fn cancelled_sync_cleans_only_its_subscription() { + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let local = LocalRelay::builder() + .query_policy(HoldDownload { + entered: entered.clone(), + release: release.clone(), + }) + .build(); + local.run().await.unwrap(); + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::TextNote, "download") + .finalize(&keys) + .unwrap(); + local.add_event(event).await.unwrap(); + let client = Client::new(); + let url = local.url().await; + client.add_relay(&url).and_connect().await.unwrap(); + let relay = client.relay(&url).await.unwrap().unwrap(); + let unrelated_id = relay + .subscribe(Filter::new().kind(Kind::Metadata)) + .await + .unwrap(); + let mut notifications = relay.notifications(); + let mut reconciliation = Box::pin( + relay + .sync(Filter::new().kind(Kind::TextNote)) + .opts(SyncOptions::new().idle_timeout(Duration::from_secs(5))) + .into_future(), + ); + tokio::time::timeout(Duration::from_secs(2), async { + tokio::select! { + _ = entered.notified() => (), + result = &mut reconciliation => panic!("sync ended before cancellation: {result:?}"), + } + }) + .await + .unwrap(); + assert_eq!(relay.inner.active_subscription_count().await, 2); + + drop(reconciliation); + tokio::time::timeout(Duration::from_secs(2), async { + while relay.inner.active_subscription_count().await != 1 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(relay.inner.has_subscription(&unrelated_id).await); + release.notify_one(); + + let live = EventBuilder::new(Kind::Metadata, "{}") + .finalize(&keys) + .unwrap(); + local.add_event(live.clone()).await.unwrap(); + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if let Some(RelayNotification::Event { + subscription_id, + event, + }) = notifications.next().await + { + if subscription_id == unrelated_id && event.id == live.id { + break; + } + } + } + }) + .await + .unwrap(); + client.shutdown().await; + } + #[tokio::test] async fn test_check_negentropy_support_times_out() { let (_tx, mut rx) = broadcast::channel(1); @@ -670,6 +862,73 @@ mod tests { assert_eq!(error.kind(), ErrorKind::Other); } + #[tokio::test] + async fn rejected_download_fails_sync_and_preserves_unrelated_subscription() { + for (allowed_batches, event_count) in [(0, 1), (1, NEGENTROPY_BATCH_SIZE_DOWN + 1)] { + let local = LocalRelay::builder() + .query_policy(RejectDownloadAfter { + allowed_batches, + seen_batches: AtomicUsize::new(0), + }) + .build(); + local.run().await.unwrap(); + let keys = Keys::generate(); + for index in 0..event_count { + let event = EventBuilder::new(Kind::TextNote, format!("remote {index}")) + .finalize(&keys) + .unwrap(); + local.add_event(event).await.unwrap(); + } + + let url = local.url().await; + let client = Client::builder() + .database(MemoryDatabase::unbounded()) + .build(); + client.add_relay(&url).and_connect().await.unwrap(); + let relay = client.relay(&url).await.unwrap().unwrap(); + let unrelated_id = SubscriptionId::generate(); + relay + .subscribe(Filter::new().kind(Kind::Metadata)) + .with_id(unrelated_id.clone()) + .await + .unwrap(); + + let output = client + .sync(Filter::new().kind(Kind::TextNote)) + .with([url.clone()]) + .opts( + SyncOptions::new() + .initial_timeout(Duration::from_secs(2)) + .idle_timeout(Duration::from_secs(2)), + ) + .await + .unwrap(); + assert!(output.success.is_empty()); + assert!( + output.failed[&url].contains("blocked: reads denied"), + "allowed_batches={allowed_batches}, failed={:?}", + output.failed, + ); + assert_eq!( + client + .database() + .count(Filter::new().kind(Kind::TextNote)) + .await + .unwrap(), + allowed_batches * NEGENTROPY_BATCH_SIZE_DOWN + ); + tokio::time::timeout(Duration::from_secs(2), async { + while relay.inner.active_subscription_count().await != 1 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(relay.inner.has_subscription(&unrelated_id).await); + client.shutdown().await; + } + } + #[tokio::test] async fn test_negentropy_sync() { // Mock relay diff --git a/nostr-sdk/src/relay/inner.rs b/nostr-sdk/src/relay/inner.rs index 7a716b0bc..f2bdcba80 100644 --- a/nostr-sdk/src/relay/inner.rs +++ b/nostr-sdk/src/relay/inner.rs @@ -310,6 +310,11 @@ impl InnerRelay { subscription.contains_key(id) } + #[cfg(test)] + pub(crate) async fn active_subscription_count(&self) -> usize { + self.atomic.subscriptions.read().await.len() + } + /// Returns all long-lived (non-auto-closing) subscriptions pub async fn subscriptions(&self) -> HashMap> { let subscription = self.atomic.subscriptions.read().await; From fe23bcff6d083ef73fa5d88f98745bdca123562f Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 30 Sep 2026 15:59:00 +0200 Subject: [PATCH 2/3] test: verify empty CLOSED download completion requirements --- nostr-sdk/src/relay/api/sync.rs | 90 +++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/nostr-sdk/src/relay/api/sync.rs b/nostr-sdk/src/relay/api/sync.rs index b1f158209..411277bb1 100644 --- a/nostr-sdk/src/relay/api/sync.rs +++ b/nostr-sdk/src/relay/api/sync.rs @@ -713,6 +713,96 @@ mod tests { use crate::error::ErrorKind; use crate::local_relay::{LocalRelay, MockRelay, QueryPolicy, QueryPolicyResult}; use crate::relay::{SyncDirection, SyncOptions}; + use crate::transport::websocket::{ + DefaultWebsocketTransport, WebSocketSink, WebSocketStream, WebSocketTransport, + }; + + #[derive(Debug)] + struct CloseInsteadOfEose { + omit_events: bool, + } + + impl WebSocketTransport for CloseInsteadOfEose { + fn support_ping(&self) -> bool { + true + } + + fn connect<'a>( + &'a self, + url: &'a Url, + proxy: Option, + ) -> BoxedFuture<'a, Result<(WebSocketSink, WebSocketStream), Error>> { + Box::pin(async move { + let (sink, stream) = DefaultWebsocketTransport.connect(url, proxy).await?; + let omit_events = self.omit_events; + let stream = stream.filter_map(move |message| async move { + match message { + Ok(async_wsocket::Message::Text(text)) => { + match RelayMessage::from_json(&text) { + Ok(RelayMessage::EndOfStoredEvents(id)) => { + Some(Ok(async_wsocket::Message::Text( + RelayMessage::closed(id.into_owned(), "").as_json(), + ))) + } + Ok(RelayMessage::Event { .. }) if omit_events => None, + _ => Some(Ok(async_wsocket::Message::Text(text))), + } + } + other => Some(other), + } + }); + Ok((sink, Box::pin(stream) as WebSocketStream)) + }) + } + } + + #[tokio::test] + async fn empty_closed_requires_all_requested_download_events() { + for omit_events in [false, true] { + let local = LocalRelay::new(); + local.run().await.unwrap(); + let event = EventBuilder::new(Kind::TextNote, "explicit download") + .finalize(&Keys::generate()) + .unwrap(); + local.add_event(event.clone()).await.unwrap(); + let relay = Relay::builder(local.url().await) + .websocket_transport(CloseInsteadOfEose { omit_events }) + .build(); + relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap(); + + let result = relay + .sync(Filter::new().kind(Kind::TextNote)) + .opts( + SyncOptions::new() + .initial_timeout(Duration::from_secs(2)) + .idle_timeout(Duration::from_secs(2)), + ) + .await; + if omit_events { + assert!( + result + .unwrap_err() + .to_string() + .contains("closed before all requested events arrived") + ); + } else { + assert_eq!(result.unwrap().received, HashSet::from([event.id])); + } + tokio::time::timeout(Duration::from_secs(2), async { + while relay.inner.active_subscription_count().await != 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + relay.shutdown(); + local.shutdown(); + } + } #[derive(Debug)] struct RejectDownloadAfter { From f9b8c305975cbbae5d1868ab0a460a87947b158a Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 30 Sep 2026 16:59:43 +0200 Subject: [PATCH 3/3] test: disambiguate sync transport error on stable Rust --- nostr-sdk/src/relay/api/sync.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nostr-sdk/src/relay/api/sync.rs b/nostr-sdk/src/relay/api/sync.rs index 411277bb1..ddafd6d80 100644 --- a/nostr-sdk/src/relay/api/sync.rs +++ b/nostr-sdk/src/relay/api/sync.rs @@ -710,7 +710,7 @@ mod tests { use super::*; use crate::client::Client; - use crate::error::ErrorKind; + use crate::error::{Error, ErrorKind}; use crate::local_relay::{LocalRelay, MockRelay, QueryPolicy, QueryPolicyResult}; use crate::relay::{SyncDirection, SyncOptions}; use crate::transport::websocket::{