fix(ecash): reduce Minibits relay churn/privacy leak and page past a 200-DM claim backlog
fetch_relay_dms connected to all three CLAIM_RELAY_URLS (the Minibits relay plus the two public fallbacks, relay.damus.io and nos.lol) on every 8s poll, even though the module's own docs already described RELAY_URL as the primary with the public relays meant only as a fallback. In practice this meant 3 fresh WebSocket connections every poll and broadcasting the wallet's derived Nostr pubkey's DM activity to two public relays it didn't need to touch. - Query RELAY_URL alone first; only add and query the public fallbacks when it's unreachable (via try_connect_relay). Happy path is now one connection per poll instead of three, and the public relays only see this pubkey's traffic when the primary is actually down. - Page through the DM filter instead of a single limit(200) fetch: a relay returns the newest `limit` events for a filter, so a backlog of more than 200 DMs since the last poll (e.g. a long-offline node) silently skipped the older ones forever, since `since` never advanced past them. Capped at 5 pages so a relay that never stops returning full pages can't hang the poll. - Moved the ensure_mint_accepted doc comment back above its own function — it had been glued onto fetch_relay_dms by an earlier edit. - Timestamp::as_u64() -> as_secs() to clear the deprecation warning. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EZnFgeUBKY5UAfyJFsYccS
This commit is contained in:
@@ -625,53 +625,106 @@ const NO_CLAIMS: ClaimOutcome = ClaimOutcome {
|
|||||||
/// three real payments that `/claim` never surfaced. Best-effort: a relay
|
/// three real payments that `/claim` never surfaced. Best-effort: a relay
|
||||||
/// error here must not abort the poll, since `pending_claims` may still hold
|
/// error here must not abort the poll, since `pending_claims` may still hold
|
||||||
/// earlier fetches worth retrying.
|
/// 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(
|
async fn fetch_relay_dms(
|
||||||
our_pubkey: nostr_sdk::PublicKey,
|
our_pubkey: nostr_sdk::PublicKey,
|
||||||
since: u64,
|
since: u64,
|
||||||
) -> Vec<(String, u64, String)> {
|
) -> Vec<(String, u64, String)> {
|
||||||
let client = Client::default();
|
let client = Client::default();
|
||||||
for url in CLAIM_RELAY_URLS {
|
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 {
|
if let Err(e) = client.add_relay(*url).await {
|
||||||
warn!("Minibits: could not add relay {url}: {e}");
|
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
|
// Give relays a moment to finish the WebSocket handshake before the
|
||||||
// fetch's own timeout starts consuming that time.
|
// fetch's own timeout starts consuming that time.
|
||||||
tokio::time::sleep(std::time::Duration::from_millis(800)).await;
|
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
|
||||||
|
// 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()
|
let filter = Filter::new()
|
||||||
.pubkey(our_pubkey)
|
.pubkey(our_pubkey)
|
||||||
.kind(Kind::from(4u16))
|
.kind(Kind::from(4u16))
|
||||||
.since(Timestamp::from(since.saturating_add(1)))
|
.since(Timestamp::from(watermark.saturating_add(1)))
|
||||||
.limit(200);
|
.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
|
|
||||||
.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
|
|
||||||
}
|
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
warn!("Minibits: relay fetch for claim DMs failed: {e}");
|
warn!("Minibits: relay fetch for claim DMs failed: {e}");
|
||||||
Vec::new()
|
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);
|
||||||
|
}
|
||||||
|
out.extend(page_events);
|
||||||
|
|
||||||
client.shutdown().await;
|
if got < PAGE_LIMIT {
|
||||||
result
|
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;
|
||||||
|
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<()> {
|
async fn ensure_mint_accepted(data_dir: &Path, mint_url: &str) -> Result<()> {
|
||||||
let mut accepted = ecash::load_accepted_mints(data_dir).await?;
|
let mut accepted = ecash::load_accepted_mints(data_dir).await?;
|
||||||
if !accepted.mints.iter().any(|m| m == mint_url) {
|
if !accepted.mints.iter().any(|m| m == mint_url) {
|
||||||
|
|||||||
Reference in New Issue
Block a user