core/archipelago/src/wallet/minibits.rs

@@ -625,53 +625,106 @@ const NO_CLAIMS: ClaimOutcome = ClaimOutcome {
 /// three real payments that `/claim` never surfaced. Best-effort: a relay
 /// error here must not abort the poll, since `pending_claims` may still hold
 /// earlier fetches worth retrying.
+///
+/// Queries `RELAY_URL` (the service's own relay) alone first — the happy
+/// path for a 8s poll is one WebSocket connection, not three, and the
+/// wallet's derived Nostr pubkey isn't broadcast to the public fallback
+/// relays unless it's actually needed. Only when that relay is unreachable
+/// does it fall back to all of `CLAIM_RELAY_URLS`.
 async fn fetch_relay_dms(
     our_pubkey: nostr_sdk::PublicKey,
     since: u64,
 ) -> Vec<(String, u64, String)> {
     let client = Client::default();
-    for url in CLAIM_RELAY_URLS {
+    if let Err(e) = client.add_relay(RELAY_URL).await {
-        if let Err(e) = client.add_relay(*url).await {
+        warn!("Minibits: could not add relay {RELAY_URL}: {e}");
-            warn!("Minibits: could not add relay {url}: {e}");
+    }
+    let primary_reachable = client
+        .try_connect_relay(RELAY_URL, std::time::Duration::from_secs(3))
+        .await
+        .is_ok();
+    if !primary_reachable {
+        warn!("Minibits: primary relay {RELAY_URL} unreachable, falling back to public relays too");
+        for url in &CLAIM_RELAY_URLS[1..] {
+            if let Err(e) = client.add_relay(*url).await {
+                warn!("Minibits: could not add relay {url}: {e}");
+            }
         }
+        client.connect().await;
     }
-    client.connect().await;
     // Give relays a moment to finish the WebSocket handshake before the
     // fetch's own timeout starts consuming that time.
     tokio::time::sleep(std::time::Duration::from_millis(800)).await;
 
-    // `since` is inclusive in NIP-01, and `since` here is the `created_at` of
+    // Page through results instead of a single `limit(200)` fetch: relays
-    // the newest event we've already queued — so filter strictly after it,
+    // return the *newest* `limit` events for a filter, so a backlog of more
-    // or the same event gets re-fetched (and its already-spent token
+    // than 200 DMs since the last poll (a node offline a long time) would
-    // re-attempted) every poll forever.
+    // otherwise silently skip the older ones forever — `since` never moves
-    let filter = Filter::new()
+    // past them because they're never fetched. Capped at `MAX_PAGES` so a
-        .pubkey(our_pubkey)
+    // relay that never stops returning full pages can't hang the poll.
-        .kind(Kind::from(4u16))
+    const PAGE_LIMIT: usize = 200;
-        .since(Timestamp::from(since.saturating_add(1)))
+    const MAX_PAGES: usize = 5;
-        .limit(200);
+    let mut watermark = since;
+    let mut out: Vec<(String, u64, String)> = Vec::new();
+    for page in 0..MAX_PAGES {
+        // `since` is inclusive in NIP-01, and `watermark` is the `created_at`
+        // of the newest event we've already queued — so filter strictly
+        // after it, or the same event gets re-fetched (and its already-spent
+        // token re-attempted) every poll forever.
+        let filter = Filter::new()
+            .pubkey(our_pubkey)
+            .kind(Kind::from(4u16))
+            .since(Timestamp::from(watermark.saturating_add(1)))
+            .limit(PAGE_LIMIT);
 
-    let result = match client
+        let events = match client
-        .fetch_events(filter, std::time::Duration::from_secs(10))
+            .fetch_events(filter, std::time::Duration::from_secs(10))
-        .await
+            .await
-    {
+        {
-        Ok(events) => {
+            Ok(events) => events,
-            let mut out: Vec<(String, u64, String)> = events
+            Err(e) => {
-                .into_iter()
+                warn!("Minibits: relay fetch for claim DMs failed: {e}");
-                .map(|e| (e.content, e.created_at.as_u64(), e.pubkey.to_hex()))
+                break;
-                .collect();
+            }
-            out.sort_by_key(|(_, created_at, _)| *created_at);
+        };
-            out
+        let got = events.len();
+        let mut page_events: Vec<(String, u64, String)> = events
+            .into_iter()
+            .map(|e| (e.content, e.created_at.as_secs(), e.pubkey.to_hex()))
+            .collect();
+        page_events.sort_by_key(|(_, created_at, _)| *created_at);
+        if let Some((_, newest, _)) = page_events.last() {
+            watermark = watermark.max(*newest);
         }
-        Err(e) => {
+        out.extend(page_events);
-            warn!("Minibits: relay fetch for claim DMs failed: {e}");
+
-            Vec::new()
+        if got < PAGE_LIMIT {
+            break;
         }
-    };
+        if page == MAX_PAGES - 1 {
+            warn!(
+                "Minibits: hit the {MAX_PAGES}-page claim DM pagination cap; \
+                 some older DMs may remain unfetched until the next poll"
+            );
+        }
+    }
 
     client.shutdown().await;
-    result
+    out
 }
 
+/// Make sure the Minibits mint is on the accepted-mints allow-list.
+///
+/// `ecash::receive_token` checks the raw accepted-mints file directly (not
+/// the more lenient `ecash::is_mint_trusted`, which always trusts the default
+/// mint) — so an operator who edited their accepted-mints list (e.g. via the
+/// `streaming.configure-mints` RPC) and dropped the default mint would
+/// otherwise cause every Minibits claim to fail *after* the claim was already
+/// consumed server-side, permanently losing those coins with nothing but a
+/// log line to show for it. The Minibits Lightning address is inherently
+/// backed by this one mint — registering it already implies trusting the
+/// mint — so self-heal the allow-list here rather than let that combination
+/// silently strand funds.
 async fn ensure_mint_accepted(data_dir: &Path, mint_url: &str) -> Result<()> {
     let mut accepted = ecash::load_accepted_mints(data_dir).await?;
     if !accepted.mints.iter().any(|m| m == mint_url) {