update

crates/community/src/history.rs

@@ -1,19 +1,24 @@
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::future::Future;
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::{Duration, Instant};
 
-use anyhow::{Result, anyhow};
+use anyhow::Result;
 use concord::cord01::KIND_WRAP_EPHEMERAL;
 use concord::cord03::{self, ChatRumor, plane_keys};
+use concord::state::ChannelCursor;
 use concord::{ChannelId, Epoch};
-use futures::future::{Either, join_all, select};
+use futures::future::{Either, select};
 use nostr_sdk::prelude::*;
 
 use crate::cache::cache_rumor;
+use crate::sync::connect_relays;
 
 /// How long one relay is given to answer one page of history.
 const PAGE_TIMEOUT: Duration = Duration::from_secs(10);
+/// How far below a cursor a warm window reaches back.
+pub const CURSOR_OVERLAP_MS: u64 = 60_000;
 
 /// The region of history to read.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -43,6 +48,87 @@ impl Window {
             since_ms: Some(since_ms),
         }
     }
+
+    /// The window a channel is opened with.
+    pub fn opening(cursor: ChannelCursor) -> Self {
+        match cursor.newest_ms {
+            Some(newest_ms) => Self {
+                since_ms: Some(newest_ms.saturating_sub(CURSOR_OVERLAP_MS)),
+                until_ms: None,
+            },
+            None => Self::default(),
+        }
+    }
+}
+
+/// What a relay said about one page subscription.
+#[derive(Debug, Clone)]
+pub enum Settled {
+    Replayed,
+    Refused(String),
+}
+
+/// One relay's verdict on one page subscription.
+#[derive(Debug, Clone)]
+pub struct PageReport {
+    pub relay: RelayUrl,
+    pub outcome: Settled,
+}
+
+/// The page subscriptions in flight, by subscription id.
+pub struct PageRegistry {
+    pages: Arc<Mutex<HashMap<SubscriptionId, flume::Sender<PageReport>>>>,
+}
+
+impl Default for PageRegistry {
+    fn default() -> Self {
+        Self {
+            pages: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+}
+
+impl Clone for PageRegistry {
+    fn clone(&self) -> Self {
+        Self {
+            pages: Arc::clone(&self.pages),
+        }
+    }
+}
+
+impl PageRegistry {
+    pub fn register(&self, id: SubscriptionId, sender: flume::Sender<PageReport>) {
+        self.lock().insert(id, sender);
+    }
+
+    pub fn unregister(&self, id: &SubscriptionId) {
+        self.lock().remove(id);
+    }
+
+    /// Forget every page still in flight, because its round is gone.
+    pub fn clear(&self) {
+        self.lock().clear();
+    }
+
+    /// Hand a relay's verdict to the page that owns `id`, when it is still waiting.
+    pub fn deliver(&self, id: &SubscriptionId, relay: RelayUrl, outcome: Settled) {
+        let sender = self.lock().get(id).cloned();
+
+        let Some(sender) = sender else {
+            return;
+        };
+
+        if let Err(error) = sender.try_send(PageReport { relay, outcome }) {
+            log::debug!("community: a page report was not delivered: {error}");
+        }
+    }
+
+    fn lock(&self) -> MutexGuard<'_, HashMap<SubscriptionId, flume::Sender<PageReport>>> {
+        match self.pages.lock() {
+            Ok(pages) => pages,
+            Err(poisoned) => poisoned.into_inner(),
+        }
+    }
 }
 
 /// What one paged fetch saw.
@@ -58,8 +144,10 @@ pub struct WrapPage {
 }
 
 /// Walks a channel's history back over the community's own relays.
+#[allow(clippy::too_many_arguments)]
 pub async fn page(
     client: &Client,
+    pages: &PageRegistry,
     channel: &ChannelId,
     held: &[(Epoch, [u8; 32])],
     relays: &[RelayUrl],
@@ -77,6 +165,9 @@ pub async fn page(
         });
     }
 
+    // A REQ can only target a relay the pool already knows about.
+    connect_relays(client, relays).await;
+
     let mut walk = Walk::new(relays, window);
     let mut opened = Vec::new();
 
@@ -91,17 +182,19 @@ pub async fn page(
             .map(|index| (index, walk.url(index).clone()))
             .collect();
 
-        let answers = join_all(
-            asked
-                .iter()
-                .map(|(_, url)| ingest_page(client, url, &filter)),
-        )
-        .await;
+        let answers = ask_page(client, pages, &asked, &filter).await;
 
-        for ((index, url), answer) in asked.iter().zip(answers) {
-            if let Err(error) = answer {
-                log::warn!("community: relay {url} did not answer a history page: {error}");
-                walk.reject(*index);
+        for (index, url) in &asked {
+            match answers.get(url) {
+                Some(Settled::Replayed) => {}
+                Some(Settled::Refused(reason)) => {
+                    log::warn!("community: relay {url} refused a history page: {reason}");
+                    walk.reject(*index);
+                }
+                None => {
+                    log::warn!("community: relay {url} did not finish a history page");
+                    walk.reject(*index);
+                }
             }
         }
 
@@ -144,88 +237,99 @@ fn wrap_filter(authors: &[PublicKey], window: Window, limit: usize) -> Filter {
     filter
 }
 
-/// Ask one relay for one page and wait until it has answered.
-async fn ingest_page(client: &Client, url: &RelayUrl, filter: &Filter) -> Result<()> {
-    let relay = match client.relay(url).await? {
-        Some(relay) => relay,
-        None => {
-            client.add_relay(url).and_connect().await?;
-            client
-                .relay(url)
-                .await?
-                .ok_or_else(|| anyhow!("the relay was not added to the pool"))?
-        }
-    };
+/// One page's verdicts, by relay.
+type PageAnswers = BTreeMap<RelayUrl, Settled>;
 
-    let id = history_subscription();
-    let mut notifications = relay.notifications();
+async fn ask_page(
+    client: &Client,
+    pages: &PageRegistry,
+    asked: &[(usize, RelayUrl)],
+    filter: &Filter,
+) -> PageAnswers {
+    let mut answers = PageAnswers::new();
+    let distinct: BTreeSet<RelayUrl> = asked.iter().map(|(_, url)| url.clone()).collect();
+    let mut targets: Vec<(RelayUrl, Vec<Filter>)> = Vec::with_capacity(distinct.len());
 
-    // The SDK closes and unregisters the subscription itself once the relay says
-    // it is done, so nothing has to be unwound here.
-    relay
-        .subscribe(vec![filter.clone()])
-        .with_id(id.clone())
-        .close_on(
-            SubscribeAutoCloseOptions::default()
-                .exit_policy(ReqExitPolicy::ExitOnEOSE)
-                .timeout(Some(PAGE_TIMEOUT)),
-        )
-        .await?;
-
-    // One deadline for the whole page, so a relay that keeps talking cannot
-    // extend the wait past it.
-    let deadline = Instant::now() + PAGE_TIMEOUT;
-
-    loop {
-        let remaining = deadline.saturating_duration_since(Instant::now());
-
-        let Some(notification) = within(remaining, notifications.next()).await else {
-            return Err(anyhow!("the relay did not finish the page"));
-        };
-
-        let Some(notification) = notification else {
-            return Err(anyhow!("the relay's notification stream ended"));
-        };
-
-        let RelayNotification::Message { message } = notification else {
-            continue;
-        };
-
-        match *message {
-            RelayMessage::EndOfStoredEvents(ended) => {
-                if ended.as_ref() == &id {
-                    return Ok(());
-                }
+    for url in &distinct {
+        match client.relay(url).await {
+            Ok(Some(_)) => targets.push((url.clone(), vec![filter.clone()])),
+            Ok(None) => {
+                log::warn!("community: relay {url} is not in the pool for a history page");
+                answers.insert(
+                    url.clone(),
+                    Settled::Refused("not in the relay pool".to_owned()),
+                );
             }
-            RelayMessage::Closed {
-                subscription_id,
-                message,
-            } if subscription_id.as_ref() == &id && !auth_required(&message) => {
-                return Err(anyhow!("the relay closed the page: {message}"));
+            Err(error) => {
+                log::warn!("community: relay {url} could not be looked up: {error}");
+                answers.insert(url.clone(), Settled::Refused(error.to_string()));
             }
-            _ => {}
         }
     }
+
+    if targets.is_empty() {
+        return answers;
+    }
+
+    let id = history_subscription();
+    let (sender, receiver) = flume::bounded(distinct.len());
+    pages.register(id.clone(), sender);
+
+    let options = SubscribeAutoCloseOptions::default()
+        .exit_policy(ReqExitPolicy::ExitOnEOSE)
+        .timeout(Some(PAGE_TIMEOUT));
+
+    match client
+        .subscribe(ReqTarget::manual(targets))
+        .with_id(id.clone())
+        .close_on(options)
+        .await
+    {
+        Ok(output) => {
+            for (url, reason) in output.failed {
+                answers.insert(url, Settled::Refused(reason));
+            }
+
+            let deadline = Instant::now() + PAGE_TIMEOUT;
+
+            while answers.len() < distinct.len() {
+                let remaining = deadline.saturating_duration_since(Instant::now());
+
+                let Some(report) = within(remaining, receiver.recv_async()).await else {
+                    break;
+                };
+
+                match report {
+                    Ok(report) => {
+                        answers.insert(report.relay, report.outcome);
+                    }
+                    Err(error) => {
+                        log::warn!("community: a history page's reports were lost: {error}");
+                        break;
+                    }
+                }
+            }
+        }
+        Err(error) => {
+            log::warn!("community: a history page REQ was refused: {error}");
+        }
+    }
+
+    pages.unregister(&id);
+
+    answers
 }
 
-/// Whether a CLOSED reason is the NIP-42 `auth-required` one, the only reason the
-/// SDK recovers from on its own.
-fn auth_required(reason: &str) -> bool {
+pub(crate) fn auth_required(reason: &str) -> bool {
     matches!(
         MachineReadablePrefix::parse(reason),
         Some(MachineReadablePrefix::AuthRequired)
     )
 }
 
-/// A subscription id for one page of one relay, unique so a page's answer is
-/// never confused with the community's standing subscription or another page's.
 fn history_subscription() -> SubscriptionId {
     static NEXT: AtomicU64 = AtomicU64::new(0);
-
-    SubscriptionId::new(format!(
-        "concord-history-{}",
-        NEXT.fetch_add(1, Ordering::Relaxed)
-    ))
+    SubscriptionId::new(format!("history-{}", NEXT.fetch_add(1, Ordering::Relaxed)))
 }
 
 /// Await `future`, giving up after `limit`.
@@ -513,4 +617,46 @@ mod tests {
             .finalize(&keys)
             .expect("signs")
     }
+
+    /// A cold channel asks wide; a warm one resumes at the overlap above its
+    /// cursor, and `older_than` never includes the boundary event itself.
+    #[test]
+    fn a_cold_window_is_open_and_a_warm_one_resumes_at_the_overlap() {
+        assert_eq!(Window::opening(ChannelCursor::default()), Window::default());
+
+        let warm = Window::opening(ChannelCursor {
+            newest_ms: Some(2_000_000),
+            oldest_ms: Some(1_000),
+            exhausted: false,
+        });
+        assert_eq!(
+            warm,
+            Window {
+                since_ms: Some(2_000_000 - CURSOR_OVERLAP_MS),
+                until_ms: None,
+            }
+        );
+
+        assert_eq!(
+            Window::older_than(1_000),
+            Window {
+                since_ms: None,
+                until_ms: Some(999),
+            }
+        );
+    }
+
+    /// A page whose round has moved on is simply a report nobody reads.
+    #[test]
+    fn a_page_that_moved_on_receives_nothing() {
+        let pages = PageRegistry::default();
+        let id = SubscriptionId::new("concord-history-9");
+        let (sender, receiver) = flume::bounded(1);
+
+        pages.register(id.clone(), sender);
+        pages.unregister(&id);
+        pages.deliver(&id, relay_url("history"), Settled::Replayed);
+
+        assert!(receiver.try_recv().is_err());
+    }
 }