diff --git a/nostr-sdk/CHANGELOG.md b/nostr-sdk/CHANGELOG.md index 9ab7cc71d..a8ca8bb56 100644 --- a/nostr-sdk/CHANGELOG.md +++ b/nostr-sdk/CHANGELOG.md @@ -31,12 +31,14 @@ ### Added +- Add `SyncEvents::with_outcomes` to retain partial reconciliation progress with its terminal error. - 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 ### Fixed +- Retain progress from failed relays in aggregate reconciliation summaries. - 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. - Clean up NIP-77 request-owned subscriptions when reconciliation fails or is canceled. diff --git a/nostr-sdk/src/client/api/sync.rs b/nostr-sdk/src/client/api/sync.rs index 22d553e05..a9e9b3bb5 100644 --- a/nostr-sdk/src/client/api/sync.rs +++ b/nostr-sdk/src/client/api/sync.rs @@ -13,7 +13,10 @@ use crate::relay::{RelayCapabilities, SyncOptions, SyncSummary as RelaySyncSumma /// Client negentropy reconciliation summary /// -/// This includes the summary for all relays involved in the reconciliation process. +/// This includes observed progress from all relays involved in reconciliation, +/// including relays listed as failed in the operation's `Output::failed` map. +/// A failed relay's progress does not establish completion for the selected +/// filter/window or downstream durable admission. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct SyncSummary { /// Events that were stored locally (missing on relay) @@ -200,10 +203,113 @@ where #[cfg(test)] mod tests { - use nostr::event::Kind; + use std::future::Future; + use std::net::SocketAddr; + use std::pin::Pin; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + use nostr::event::{EventBuilder, FinalizeEvent, Kind}; + use nostr::key::Keys; + use nostr::message::MachineReadablePrefix; use super::*; use crate::error::ErrorKind; + use crate::local_relay::{LocalRelay, QueryPolicy, QueryPolicyResult}; + + #[derive(Debug, Default)] + struct RejectSecondDownload { + batches: AtomicUsize, + } + + impl QueryPolicy for RejectSecondDownload { + fn admit_query<'a>( + &'a self, + query: &'a mut Filter, + _addr: &'a SocketAddr, + ) -> Pin + Send + 'a>> { + let reject: bool = + query.ids.is_some() && self.batches.fetch_add(1, Ordering::SeqCst) > 0; + Box::pin(async move { + if reject { + QueryPolicyResult::reject( + MachineReadablePrefix::Blocked, + "second batch rejected", + ) + } else { + QueryPolicyResult::Accept + } + }) + } + } + + #[tokio::test] + async fn aggregate_sync_retains_failed_relay_progress() { + let healthy = LocalRelay::new(); + let failing = LocalRelay::builder() + .query_policy(RejectSecondDownload::default()) + .build(); + healthy.run().await.unwrap(); + failing.run().await.unwrap(); + let keys: Keys = Keys::generate(); + for index in 0..101 { + let event = EventBuilder::new(Kind::TextNote, format!("remote {index}")) + .finalize(&keys) + .unwrap(); + healthy.add_event(event.clone()).await.unwrap(); + failing.add_event(event).await.unwrap(); + } + let healthy_url = healthy.url().await; + let failing_url = failing.url().await; + let client: Client = Client::new(); + client.add_relay(&healthy_url).and_connect().await.unwrap(); + client.add_relay(&failing_url).and_connect().await.unwrap(); + let output = client + .sync(Filter::new().kind(Kind::TextNote)) + .opts( + SyncOptions::new() + .initial_timeout(Duration::from_secs(2)) + .idle_timeout(Duration::from_secs(2)), + ) + .await + .unwrap(); + assert!(output.success.contains_key(&healthy_url)); + assert!( + output + .failed + .get(&failing_url) + .unwrap() + .contains("second batch rejected") + ); + assert_eq!(output.received.len(), 101); + assert_eq!( + output + .received + .values() + .filter(|urls| urls.contains(&healthy_url)) + .count(), + 101 + ); + assert_eq!( + output + .received + .values() + .filter(|urls| urls.contains(&failing_url)) + .count(), + 100 + ); + assert_eq!( + output + .remote + .values() + .filter(|urls| urls.contains(&failing_url)) + .count(), + 101 + ); + client.shutdown().await; + healthy.shutdown(); + failing.shutdown(); + } #[tokio::test] async fn test_sync_with_empty_list_of_relays() { diff --git a/nostr-sdk/src/pool/mod.rs b/nostr-sdk/src/pool/mod.rs index df5952e1d..2469da0e6 100644 --- a/nostr-sdk/src/pool/mod.rs +++ b/nostr-sdk/src/pool/mod.rs @@ -706,7 +706,7 @@ impl RelayPool { .sync(filter) .items(items) .opts(opts.clone()) - .into_future(), + .with_outcomes(), ); } @@ -717,9 +717,12 @@ impl RelayPool { for (url, result) in urls.into_iter().zip(list) { match result { Ok(reconciliation) => { - // Success, insert relay url in 'success' set result - output.success.insert(url.clone(), ()); - output.merge_relay_summary(url, reconciliation); + if let Some(error) = reconciliation.error { + output.failed.insert(url.clone(), error.to_string()); + } else { + output.success.insert(url.clone(), ()); + } + output.merge_relay_summary(url, reconciliation.summary); } Err(e) => { output.failed.insert(url, e.to_string()); diff --git a/nostr-sdk/src/relay/api/sync.rs b/nostr-sdk/src/relay/api/sync.rs index 43e401a9b..3fcdf674c 100644 --- a/nostr-sdk/src/relay/api/sync.rs +++ b/nostr-sdk/src/relay/api/sync.rs @@ -83,6 +83,20 @@ pub struct SyncSummary { // pub receive: HashMap>, } +/// Reconciliation progress and its terminal result for one relay. +/// +/// Progress may be useful after failure. A missing terminal error means the +/// reconciliation loop completed; inspect `summary.send_failures` for individual +/// publication failures. Received events are not evidence of downstream durable +/// admission. +#[derive(Debug)] +pub struct RelaySyncOutcome { + /// Progress observed before completion or failure. + pub summary: SyncSummary, + /// Failure that interrupted reconciliation, if any. + pub error: Option, +} + /// Sync events with relay /// /// @@ -123,6 +137,30 @@ impl<'relay> SyncEvents<'relay> { self.opts = opts; self } + + /// Reconcile while retaining partial progress if the operation fails. + /// + /// Preflight errors still return `Err` before reconciliation begins. + pub async fn with_outcomes(self) -> Result { + self.relay.inner.ensure_operational()?; + if !self.relay.inner.capabilities.can_read() { + return Err(Error::read_disabled()); + } + + let items: Vec<(EventId, Timestamp)> = match self.items { + Some(items) => items, + None => { + let database = self.relay.inner.state.database(); + database.negentropy_items(self.filter.clone()).await? + } + }; + + let mut summary: SyncSummary = SyncSummary::default(); + let error: Option = sync(self.relay, &self.filter, items, &self.opts, &mut summary) + .await + .err(); + Ok(RelaySyncOutcome { summary, error }) + } } #[inline] @@ -668,35 +706,11 @@ impl<'relay> IntoFuture for SyncEvents<'relay> { fn into_future(self) -> Self::IntoFuture { Box::pin(async move { - // Check if relay is operational - self.relay.inner.ensure_operational()?; - - // Check if relay can read - if !self.relay.inner.capabilities.can_read() { - return Err(Error::read_disabled()); + let outcome = self.with_outcomes().await?; + match outcome.error { + Some(error) => Err(error), + None => Ok(outcome.summary), } - - let items: Vec<(EventId, Timestamp)> = match self.items { - Some(items) => items, - None => { - // Get negentropy items - let database = self.relay.inner.state.database(); - database.negentropy_items(self.filter.clone()).await? - } - }; - - let mut output: SyncSummary = SyncSummary::default(); - - sync( - self.relay, - &self.filter, - items.clone(), - &self.opts, - &mut output, - ) - .await?; - - Ok(output) }) } } @@ -721,7 +735,7 @@ mod tests { use super::*; use crate::error::{Error, ErrorKind}; use crate::local_relay::{LocalRelay, MockRelay, QueryPolicy, QueryPolicyResult}; - use crate::relay::{SyncDirection, SyncOptions}; + use crate::relay::{RelayCapabilities, SyncDirection, SyncOptions}; use crate::transport::websocket::{ DefaultWebsocketTransport, WebSocketSink, WebSocketStream, WebSocketTransport, }; @@ -1143,7 +1157,7 @@ mod tests { relay.shutdown(); } - async fn assert_rejected_download_after(allowed_batches: usize) { + async fn assert_rejected_download_after(allowed_batches: usize, report_outcomes: bool) { let event_count = allowed_batches * NEGENTROPY_BATCH_SIZE_DOWN + 1; let local = LocalRelay::builder() @@ -1181,7 +1195,18 @@ mod tests { let opts: SyncOptions = SyncOptions::new() .initial_timeout(Duration::from_secs(2)) .idle_timeout(Duration::from_secs(2)); - let err = relay.sync(filter).opts(opts).await.unwrap_err(); + let err = if report_outcomes { + let outcome: RelaySyncOutcome = + relay.sync(filter).opts(opts).with_outcomes().await.unwrap(); + assert_eq!(outcome.summary.remote.len(), event_count); + assert_eq!( + outcome.summary.received.len(), + allowed_batches * NEGENTROPY_BATCH_SIZE_DOWN + ); + outcome.error.unwrap() + } else { + relay.sync(filter).opts(opts).await.unwrap_err() + }; assert_eq!(err, Error::relay_msg(String::from("blocked: reads denied"))); assert_eq!( @@ -1204,12 +1229,45 @@ mod tests { #[tokio::test] async fn rejected_first_download_batch_fails_sync() { - assert_rejected_download_after(0).await; + assert_rejected_download_after(0, false).await; } #[tokio::test] async fn rejected_later_download_batch_fails_sync() { - assert_rejected_download_after(1).await; + assert_rejected_download_after(1, false).await; + } + + #[tokio::test] + async fn sync_outcomes_retain_progress_before_rejection() { + assert_rejected_download_after(0, true).await; + assert_rejected_download_after(1, true).await; + } + + #[tokio::test] + async fn sync_outcomes_report_success_without_an_error() { + let local = LocalRelay::new(); + local.run().await.unwrap(); + let relay = Relay::new(local.url().await); + relay + .try_connect() + .timeout(Duration::from_secs(2)) + .await + .unwrap(); + let outcome: RelaySyncOutcome = relay.sync(Filter::new()).with_outcomes().await.unwrap(); + assert!(outcome.error.is_none()); + assert_eq!(outcome.summary, SyncSummary::default()); + relay.capabilities().remove(RelayCapabilities::READ); + let error = relay.sync(Filter::new()).with_outcomes().await.unwrap_err(); + assert_eq!(error, Error::read_disabled()); + relay.shutdown(); + local.shutdown(); + } + + #[tokio::test] + async fn sync_outcomes_reject_preflight_failure() { + let relay = Relay::new(RelayUrl::parse("wss://relay.example.com").unwrap()); + let error = relay.sync(Filter::new()).with_outcomes().await.unwrap_err(); + assert_eq!(error, Error::not_ready()); } #[tokio::test]