sdk: retain partial reconciliation progress with relay outcomes

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.

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<Box<dyn Future<Output = QueryPolicyResult> + 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() {

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());

nostr-sdk/src/relay/api/sync.rs

@@ -83,6 +83,20 @@ pub struct SyncSummary {
     // pub receive: HashMap<EventId, Vec<String>>,
 }
 
+/// 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<Error>,
+}
+
 /// Sync events with relay
 ///
 /// <https://github.com/nostr-protocol/nips/blob/master/77.md>
@@ -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<RelaySyncOutcome, Error> {
+        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<Error> = 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]