diff --git a/core/archipelago/src/wallet/ecash.rs b/core/archipelago/src/wallet/ecash.rs index 2015bc19..e77647a8 100644 --- a/core/archipelago/src/wallet/ecash.rs +++ b/core/archipelago/src/wallet/ecash.rs @@ -1205,6 +1205,7 @@ pub async fn receive_token(data_dir: &Path, token_str: &str) -> Result { // for the log. Remember the last one so a total failure can tell the user // *why* instead of just "nothing was received". let mut last_reason: Option = None; + let mut all_already_redeemed = true; // Swap proofs at each mint for entry in &token.token { @@ -1217,6 +1218,7 @@ pub async fn receive_token(data_dir: &Path, token_str: &str) -> Result { } Err(e) => { warn!("Failed to swap proofs from mint {}: {:#}", entry.mint, e); + all_already_redeemed &= e.is::(); last_reason = Some(e.to_string()); // Continue with other mints if any } @@ -1224,10 +1226,7 @@ pub async fn receive_token(data_dir: &Path, token_str: &str) -> Result { } if received_total == 0 { - match last_reason { - Some(reason) => anyhow::bail!("Could not receive this ecash: {}", reason), - None => anyhow::bail!("Failed to receive any proofs from token"), - } + return Err(receive_failure(last_reason, all_already_redeemed)); } wallet.record_tx( @@ -1243,6 +1242,17 @@ pub async fn receive_token(data_dir: &Path, token_str: &str) -> Result { Ok(received_total) } +fn receive_failure(last_reason: Option, all_already_redeemed: bool) -> anyhow::Error { + match last_reason { + Some(reason) if all_already_redeemed => { + anyhow::Error::new(super::mint_client::AlreadyRedeemed) + .context(format!("Could not receive this ecash: {reason}")) + } + Some(reason) => anyhow::anyhow!("Could not receive this ecash: {reason}"), + None => anyhow::anyhow!("Failed to receive any proofs from token"), + } +} + /// Receive a legacy format token (cashuSend_{amount}_{uuid}_{timestamp}). /// For backwards compatibility during migration period. async fn receive_legacy_token(data_dir: &Path, token_str: &str) -> Result { @@ -1632,6 +1642,18 @@ fn default_mint_url() -> String { #[cfg(test)] mod tests { + #[test] + fn mixed_mint_failures_do_not_discard_a_retryable_claim() { + let reason = super::super::mint_client::ALREADY_REDEEMED_MSG.to_string(); + assert!(super::receive_failure(Some(reason.clone()), true) + .is::()); + assert!(!super::receive_failure(Some(reason), false) + .is::()); + assert!( + !super::receive_failure(None, true).is::() + ); + } + use super::*; use tempfile::TempDir; diff --git a/core/archipelago/src/wallet/minibits.rs b/core/archipelago/src/wallet/minibits.rs index 65c59a5f..087178b9 100644 --- a/core/archipelago/src/wallet/minibits.rs +++ b/core/archipelago/src/wallet/minibits.rs @@ -47,7 +47,8 @@ //! key, a crash mid-loop) must not silently lose the coins, so every fetched //! token is persisted to `MinibitsState::pending_claims` *before* decrypt/ //! redeem is attempted, and stays there — retried on every later poll — until -//! it succeeds. `ClaimOutcome::failed_count` reports how many are still +//! it succeeds or every mint reports that it was already spent. +//! `ClaimOutcome::failed_count` reports how many are still //! stuck so the caller can surface it instead of it being a log-only event. //! Separately, `ensure_mint_accepted` keeps the Minibits mint on the node's //! accepted-mints allow-list: the address is inherently backed by that one @@ -167,6 +168,9 @@ pub struct MinibitsState { /// already-spent token) but wasteful and noisy. #[serde(default)] pub last_dm_seen_at: u64, + /// Resume a bounded backward scan before advancing to newer relay events. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub relay_scan: Option, /// Event ids already queued from the relay. `created_at` has only /// one-second resolution, so a strict `since = last + 1` watermark can /// permanently miss a second payment published later in the same second. @@ -627,6 +631,7 @@ async fn register_new_state( created_at: chrono::Utc::now().to_rfc3339(), pending_claims: Vec::new(), last_dm_seen_at: 0, + relay_scan: None, seen_dm_ids: Vec::new(), last_receipt_id: 0, last_receipt_sats: 0, @@ -654,13 +659,12 @@ pub struct ClaimOutcome { /// True when `ecash::receive_token` failed because the token was already /// redeemed (mint error 11001, see `mint_client::describe_mint_error_code`) — -/// a terminal condition, not a reason to retry. Seen on archy-x250-pa3, +/// a terminal condition, not a reason to retry. Seen on a deployed node, /// 2026-09-15: a claim that had already been swept kept failing this way on /// every poll forever, since nothing distinguished it from a transient /// failure worth retrying. fn is_already_redeemed(err: &anyhow::Error) -> bool { - err.to_string() - .contains(super::mint_client::ALREADY_REDEEMED_MSG) + err.is::() } const NO_CLAIMS: ClaimOutcome = ClaimOutcome { @@ -714,15 +718,15 @@ fn outcome_with_latest_receipt( /// 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`. Results are paged (capped at -/// `MAX_PAGES`) since a relay returns only the newest `limit` events for a -/// filter — a backlog bigger than one page would otherwise silently strand -/// older DMs forever, as `since` never advances past events that were never -/// fetched. +/// `CLAIM_MAX_PAGES`) since a relay returns only the newest `limit` events for +/// a filter. A durable backward cursor keeps older pages reachable even after +/// newly queued claims advance the normal forward watermark. async fn fetch_relay_dms( our_pubkey: nostr_sdk::PublicKey, server_pubkey: nostr_sdk::PublicKey, since: u64, -) -> Vec<(String, u64, String, String)> { + resume: Option, +) -> RelayBatch { let client = Client::default(); if let Err(e) = client.add_relay(RELAY_URL).await { warn!("Minibits: could not add relay {RELAY_URL}: {e}"); @@ -744,66 +748,108 @@ async fn fetch_relay_dms( // fetch's own timeout starts consuming that time. tokio::time::sleep(std::time::Duration::from_millis(400)).await; - const PAGE_LIMIT: usize = 200; - const MAX_PAGES: usize = 5; - let mut watermark = since; - let mut out: Vec<(String, u64, String, String)> = Vec::new(); - for page in 0..MAX_PAGES { - // Nostr timestamps have one-second resolution. Query the boundary - // second inclusively: a later-published payment may legitimately - // share that timestamp. `seen_dm_ids` performs the exact - // deduplication locally. - let filter = Filter::new() - .author(server_pubkey) - .pubkey(our_pubkey) - .kind(Kind::from(4u16)) - .since(Timestamp::from(watermark)) - .limit(PAGE_LIMIT); + let batch = collect_relay_pages(since, resume, |scan| { + let client = &client; + async move { + let mut filter = Filter::new() + .author(server_pubkey) + .pubkey(our_pubkey) + .kind(Kind::from(4u16)) + .since(Timestamp::from(scan.since)) + .limit(scan.limit); + if let Some(until) = scan.until { + filter = filter.until(Timestamp::from(until)); + } + let events = client + .fetch_events(filter, std::time::Duration::from_secs(5)) + .await?; + Ok(events + .into_iter() + .map(|e| { + ( + e.content, + e.created_at.as_secs(), + e.pubkey.to_hex(), + e.id.to_hex(), + ) + }) + .collect()) + } + }) + .await; + client.shutdown().await; + batch +} - let events = match client - .fetch_events(filter, std::time::Duration::from_secs(5)) - .await - { +const CLAIM_PAGE_LIMIT: usize = 200; +const CLAIM_MAX_PAGES: usize = 5; +type RelayDm = (String, u64, String, String); + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub struct RelayScan { + since: u64, + until: Option, + limit: usize, +} + +struct RelayBatch { + dms: Vec, + resume: Option, +} + +/// NIP-01 returns newest events first. Walk backward with an inclusive `until` +/// boundary, deduplicating event ids. A full boundary second needs a larger +/// limit, not `until - 1`, which would skip payments sharing that timestamp. +/// Persist the cursor at the page cap or on failure so older claims cannot be +/// hidden by the newest timestamp already queued in `last_dm_seen_at`. +async fn collect_relay_pages( + since: u64, + resume: Option, + mut fetch: F, +) -> RelayBatch +where + F: FnMut(RelayScan) -> Fut, + Fut: std::future::Future>>, +{ + let mut scan = resume.unwrap_or(RelayScan { + since, + until: None, + limit: CLAIM_PAGE_LIMIT, + }); + let mut out = Vec::new(); + let mut ids = std::collections::HashSet::new(); + let mut resume = Some(scan); + for _ in 0..CLAIM_MAX_PAGES { + let events = match fetch(scan).await { Ok(events) => events, Err(e) => { - warn!("Minibits: relay fetch for claim DMs failed: {e}"); + warn!("Minibits: relay fetch failed; preserving scan cursor: {e}"); break; } }; - let got = events.len(); - let mut page_events: Vec<(String, u64, String, String)> = events - .into_iter() - .map(|e| { - ( - e.content, - e.created_at.as_secs(), - e.pubkey.to_hex(), - e.id.to_hex(), - ) - }) - .collect(); - page_events.sort_by_key(|(_, created_at, _, _)| *created_at); - if let Some((_, newest, _, _)) = page_events.last() { - // Advance strictly past the newest event seen so a full page - // doesn't refetch its own boundary forever; `since` is inclusive - // in NIP-01. - watermark = watermark.max(newest.saturating_add(1)); + let count = events.len(); + let oldest = events.iter().map(|e| e.1).min(); + for event in events { + if ids.insert(event.3.clone()) { + out.push(event); + } } - out.extend(page_events); - - if got < PAGE_LIMIT { + if count < scan.limit { + resume = None; 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" - ); + if let Some(oldest) = oldest { + if scan.until == Some(oldest) { + scan.limit = scan.limit.saturating_add(CLAIM_PAGE_LIMIT); + } else { + scan.until = Some(oldest); + scan.limit = CLAIM_PAGE_LIMIT; + } } + resume = Some(scan); } - - client.shutdown().await; - out + out.sort_by(|a, b| (a.1, &a.3).cmp(&(b.1, &b.3))); + RelayBatch { dms: out, resume } } fn queue_relay_dm( @@ -961,8 +1007,15 @@ pub async fn claim_and_redeem(data_dir: &Path) -> Result { // NIP-04 DM on relays, not via `/claim` above. `since` is our own // watermark (Nostr events never expire off a relay, so without it we'd // re-fetch and re-attempt every claim ever sent on every poll). - let dms = fetch_relay_dms(identity.keys.public_key(), server_pk, state.last_dm_seen_at).await; - for (content, created_at, author, event_id) in dms { + let batch = fetch_relay_dms( + identity.keys.public_key(), + server_pk, + state.last_dm_seen_at, + state.relay_scan, + ) + .await; + state.relay_scan = batch.resume; + for (content, created_at, author, event_id) in batch.dms { if author != state.server_nostr_pubkey { warn!("Minibits: ignoring claim DM from unexpected pubkey {author}"); continue; @@ -1024,7 +1077,7 @@ pub async fn claim_and_redeem(data_dir: &Path) -> Result { // crash lost track of it). Retrying can never succeed, so // drop it instead of leaving `failed_count` stuck non-zero // forever — see archy-x250-pa3, 2026-09-15. - info!("Minibits claim was already redeemed; dropping (not a loss, value already received)"); + info!("Minibits mint reports this claim was already redeemed; removing it from the retry queue"); } Err(e) => { warn!("Minibits claim decrypted but failed to redeem ({e}); will retry next poll"); @@ -1056,6 +1109,123 @@ pub async fn claim_and_redeem(data_dir: &Path) -> Result { #[cfg(test)] mod tests { + fn simulated_relay_page(events: &[RelayDm], scan: RelayScan) -> Vec { + let mut page: Vec<_> = events + .iter() + .filter(|e| e.1 >= scan.since && scan.until.is_none_or(|until| e.1 <= until)) + .cloned() + .collect(); + page.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.3.cmp(&b.3))); + page.truncate(scan.limit); + page + } + + fn relay_fixture(count: usize, same_second: bool) -> Vec { + (1..=count) + .map(|n| { + ( + format!("claim-{n}"), + if same_second { 100 } else { n as u64 }, + "service".into(), + format!("id-{n:06}"), + ) + }) + .collect() + } + + #[tokio::test] + async fn relay_paging_fetches_older_claims_in_newest_first_backlog() { + let events = relay_fixture(450, false); + let batch = collect_relay_pages(0, None, |scan| { + std::future::ready(Ok(simulated_relay_page(&events, scan))) + }) + .await; + assert_eq!(batch.dms.len(), 450); + assert!(batch.resume.is_none()); + assert_eq!(batch.dms.first().unwrap().1, 1); + assert_eq!(batch.dms.last().unwrap().1, 450); + } + + #[tokio::test] + async fn relay_paging_preserves_payments_at_the_same_timestamp() { + let events = relay_fixture(250, true); + let batch = collect_relay_pages(100, None, |scan| { + std::future::ready(Ok(simulated_relay_page(&events, scan))) + }) + .await; + assert_eq!(batch.dms.len(), 250); + assert!(batch.resume.is_none()); + } + + #[tokio::test] + async fn relay_page_cap_resumes_older_claims_after_watermark_advances() { + let events = relay_fixture(1300, false); + let mut state = MinibitsState::default(); + let first = collect_relay_pages(0, None, |scan| { + std::future::ready(Ok(simulated_relay_page(&events, scan))) + }) + .await; + assert!(first.resume.is_some()); + state.relay_scan = first.resume; + let mut ids = std::collections::HashSet::new(); + for (content, time, author, id) in first.dms { + ids.insert(id.clone()); + queue_relay_dm(&mut state, content, time, id, author); + } + assert_eq!(state.last_dm_seen_at, 1300); + let state: MinibitsState = + serde_json::from_str(&serde_json::to_string(&state).unwrap()).unwrap(); + let second = collect_relay_pages(state.last_dm_seen_at, state.relay_scan, |scan| { + std::future::ready(Ok(simulated_relay_page(&events, scan))) + }) + .await; + assert!(second.resume.is_none()); + ids.extend(second.dms.into_iter().map(|e| e.3)); + assert_eq!(ids.len(), 1300); + } + + #[tokio::test] + async fn relay_fetch_failure_keeps_the_unfinished_page_cursor() { + let events = relay_fixture(450, false); + let mut requests = 0; + let first = collect_relay_pages(0, None, |scan| { + requests += 1; + std::future::ready(if requests == 1 { + Ok(simulated_relay_page(&events, scan)) + } else { + Err(anyhow!("relay timeout")) + }) + }) + .await; + assert_eq!(first.dms.len(), 200); + assert_eq!(first.resume.unwrap().until, Some(251)); + let second = collect_relay_pages(450, first.resume, |scan| { + std::future::ready(Ok(simulated_relay_page(&events, scan))) + }) + .await; + let ids: std::collections::HashSet<_> = first + .dms + .into_iter() + .chain(second.dms) + .map(|e| e.3) + .collect(); + assert_eq!(ids.len(), 450); + } + + #[test] + fn only_typed_spent_claims_are_terminal_even_with_wrapped_errors() { + let spent = anyhow::Error::new(super::super::mint_client::AlreadyRedeemed) + .context("receive token") + .context("claim failed"); + assert!(is_already_redeemed(&spent)); + assert!(!is_already_redeemed(&anyhow!( + super::super::mint_client::ALREADY_REDEEMED_MSG + ))); + assert!(!is_already_redeemed(&anyhow!( + "mint temporarily unreachable" + ))); + } + use super::*; #[test] diff --git a/core/archipelago/src/wallet/mint_client.rs b/core/archipelago/src/wallet/mint_client.rs index d9e31880..83950898 100644 --- a/core/archipelago/src/wallet/mint_client.rs +++ b/core/archipelago/src/wallet/mint_client.rs @@ -79,6 +79,16 @@ pub struct MintResult { pub const ALREADY_REDEEMED_MSG: &str = "This ecash has already been redeemed — it can't be claimed twice."; +/// Typed terminal condition: never infer spent proofs from a mint's free text. +#[derive(Debug)] +pub(super) struct AlreadyRedeemed; +impl std::fmt::Display for AlreadyRedeemed { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(ALREADY_REDEEMED_MSG) + } +} +impl std::error::Error for AlreadyRedeemed {} + fn describe_mint_error_code(code: i64) -> Option<&'static str> { Some(match code { 10001 => "The mint rejected these coins as invalid.", @@ -132,8 +142,15 @@ fn describe_mint_error_body(status: reqwest::StatusCode, body: &str) -> String { /// translation layered on top via `.context()` so `{}` — what reaches the /// wallet user — shows something actionable instead of raw mint JSON. fn mint_error(op: &str, status: reqwest::StatusCode, body: &str) -> anyhow::Error { - let friendly = describe_mint_error_body(status, body); - anyhow::anyhow!("{} failed ({}): {}", op, status, body).context(friendly) + let cause = anyhow::anyhow!("{} failed ({}): {}", op, status, body); + if serde_json::from_str::(body) + .ok() + .and_then(|v| v.get("code").and_then(|c| c.as_i64())) + == Some(11001) + { + return cause.context(AlreadyRedeemed); + } + cause.context(describe_mint_error_body(status, body)) } /// HTTP client for a single Cashu mint. @@ -811,6 +828,28 @@ impl MintClient { #[cfg(test)] mod tests { + #[test] + fn spent_condition_comes_from_code_not_remote_text_and_survives_context() { + let spent = super::mint_error( + "Swap", + reqwest::StatusCode::BAD_REQUEST, + r#"{"code":11001,"detail":"Token Already Spent"}"#, + ) + .context("Receive failed"); + assert!(spent.is::()); + let body = + serde_json::json!({"code":11002,"detail":super::ALREADY_REDEEMED_MSG}).to_string(); + assert!( + !super::mint_error("Swap", reqwest::StatusCode::BAD_REQUEST, &body) + .is::() + ); + let body = serde_json::json!({"detail":super::ALREADY_REDEEMED_MSG}).to_string(); + assert!( + !super::mint_error("Swap", reqwest::StatusCode::BAD_GATEWAY, &body) + .is::() + ); + } + use super::*; #[test] diff --git a/docs/incident-2026-09-15-minibits-already-redeemed.md b/docs/incident-2026-09-15-minibits-already-redeemed.md index 5a4f4689..f367023e 100644 --- a/docs/incident-2026-09-15-minibits-already-redeemed.md +++ b/docs/incident-2026-09-15-minibits-already-redeemed.md @@ -66,11 +66,13 @@ Two parts: - **`core/archipelago/src/wallet/mint_client.rs`**: exposed the existing NUT error-code-11001 translation as a public constant, - `ALREADY_REDEEMED_MSG`, so other modules can recognize it without - duplicating the string. + `ALREADY_REDEEMED_MSG`, and a typed `AlreadyRedeemed` condition identified + only by the structured mint error code. Remote text cannot impersonate it. - **`core/archipelago/src/wallet/minibits.rs`**: - Added `is_already_redeemed(&anyhow::Error) -> bool`, checking the error - chain for `ALREADY_REDEEMED_MSG`. + chain for the typed `AlreadyRedeemed` condition. The ecash receive path + preserves it only when all failed mint entries report already-spent proofs; + mixed terminal/transient failures remain retryable. - In the claim redeem loop, a redeem failure matching `is_already_redeemed` is now dropped (logged at `info!`, not retried) instead of being pushed back onto `pending_claims`. Every other failure @@ -79,8 +81,10 @@ Two parts: alone first via `try_connect_relay`, and only adds the two public fallback relays (`relay.damus.io`, `nos.lol`) if that primary relay is unreachable. Also paginates the DM fetch (200/page, capped at 5 pages) - so a backlog larger than one page can't silently strand older DMs - behind an un-advanced watermark. + backward with an inclusive `until` boundary. The cursor persists across + polls when capped or interrupted, independently of the forward watermark. + A full same-second boundary is fetched with a larger limit rather than + skipped, so multiple payments sharing a timestamp remain reachable. Deliberately **not** ported from the unmerged branch: its `STATE_LOCK` skip-if-busy guard and per-claim attempt-count backstop. `main`'s existing @@ -104,3 +108,12 @@ actually deployed. Before trusting a memory or changelog claim that something "shipped," check which branch the running/released build was built from (`git log ..main` / `main..`) rather than assuming a pushed branch was merged. + +## Pre-merge review regressions + +- A 450-event newest-first backlog is completely fetched. +- 250 distinct payments sharing one timestamp are preserved. +- A 1,300-event backlog resumes after the five-page cap and a state reload. +- An interrupted relay fetch retains its unfinished cursor. +- Only structured error 11001 is terminal, including when errors are wrapped; + remote free text and mixed mint failures cannot discard a retryable claim.