diff --git a/core/archipelago/src/wallet/minibits.rs b/core/archipelago/src/wallet/minibits.rs index 8f6f2030..a3956c7b 100644 --- a/core/archipelago/src/wallet/minibits.rs +++ b/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(*url).await { - warn!("Minibits: could not add relay {url}: {e}"); + if let Err(e) = client.add_relay(RELAY_URL).await { + warn!("Minibits: could not add relay {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 - // 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(since.saturating_add(1))) - .limit(200); + // Page through results instead of a single `limit(200)` fetch: relays + // return the *newest* `limit` events for a filter, so a backlog of more + // than 200 DMs since the last poll (a node offline a long time) would + // otherwise silently skip the older ones forever — `since` never moves + // past them because they're never fetched. Capped at `MAX_PAGES` so a + // relay that never stops returning full pages can't hang the poll. + const PAGE_LIMIT: usize = 200; + const MAX_PAGES: usize = 5; + 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 - .fetch_events(filter, std::time::Duration::from_secs(10)) - .await - { - Ok(events) => { - let mut out: Vec<(String, u64, String)> = events - .into_iter() - .map(|e| (e.content, e.created_at.as_u64(), e.pubkey.to_hex())) - .collect(); - out.sort_by_key(|(_, created_at, _)| *created_at); - out + let events = match client + .fetch_events(filter, std::time::Duration::from_secs(10)) + .await + { + Ok(events) => events, + Err(e) => { + warn!("Minibits: relay fetch for claim DMs failed: {e}"); + break; + } + }; + 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) => { - warn!("Minibits: relay fetch for claim DMs failed: {e}"); - Vec::new() + out.extend(page_events); + + 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) {