diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index 13cc6633..e3563a36 100644 --- a/src/app/bond/payout.rs +++ b/src/app/bond/payout.rs @@ -79,15 +79,6 @@ use super::db::{find_bond_by_id, find_bonds_by_state}; use super::model::Bond; use super::types::{BondSlashReason, BondState}; -/// Per-message ceiling for the `send_payment` status stream. LND -/// streams periodic InFlight updates while a payment is routing; if no -/// update lands inside this window the channel is treated as dead and -/// the attempt is routed through `on_send_payment_failure`. Picked to -/// be longer than the typical InFlight cadence (a few seconds) but -/// short enough to keep a single bond from blocking a scheduler task -/// indefinitely. -const PAYMENT_STATUS_RECV_TIMEOUT: Duration = Duration::from_secs(120); - /// One full pass over every bond in [`BondState::PendingPayout`]. /// /// Mirror of `dev_fee::run_dev_fee_cycle`: each tick walks the work @@ -664,82 +655,139 @@ async fn pay_counterparty( // send_payment. The helper caps the fee via `routing_fee_cap_sats`, // the same value persisted above as `payout_routing_fee_sats`. + // `send_payment` consumes LND's payment stream until a terminal state, + // which a locked-in but unresolved HTLC (hold invoice as the winner's + // payout invoice, or an HTLC stuck in route) can delay indefinitely, so + // it is bounded with the same timeout as the buyer payout; LND stops + // launching route attempts at 60s, so past 75s only an unresolved HTLC + // keeps the stream open. + // + // The bounded send and the status drain run CONCURRENTLY (`join!`): + // `send_payment` forwards every LND update through `tx` and blocks when + // the channel fills, so draining only after it returned could deadlock a + // chatty stream (>100 updates) until the timeout. Same watcher-before-send + // pattern as the buyer payout in `do_payment`, kept on this task (no + // spawn) because the scheduler job needs a single combined outcome. The + // drain always terminates: when the send future ends — normal return, RPC + // error, or dropped by the timeout — `tx` drops and `rx.recv()` yields + // `None`. let (tx, mut rx) = channel(100); - let send_outcome = ln_client - .send_payment(invoice, counterparty_share, tx) - .await; - if let Err(e) = send_outcome { - // The RPC call itself errored. We cannot be sure the payment did - // not partially enter LND, so treat it as indeterminate: keep - // the invoice + hash for reconciliation rather than risk a - // double payout by re-prompting. - return on_send_payment_failure( - pool, - bond, - max_retries, - claim_window_seconds, - PaymentFailureKind::Indeterminate, - &format!("{e}"), - ) - .await; - } + + let send_fut = timeout( + crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT, + ln_client.send_payment(invoice, counterparty_share, tx), + ); // Collect the first terminal status from the stream. Mirrors - // dev_fee::send_dev_fee_payment, but each recv is bounded by - // `PAYMENT_STATUS_RECV_TIMEOUT` so a wedged LND stream (no terminal - // update, no EOF, no InFlight churn) does not pin the scheduler - // task forever. We track *why* the stream ended: only an explicit - // `PaymentStatus::Failed` is terminal. A timeout or clean EOF leaves - // the payment outcome unknown (it may still be in flight), so it is - // routed as `Indeterminate` — `on_send_payment_failure` then keeps - // the invoice + hash for reconciliation instead of re-prompting. - let mut succeeded = false; - let mut failure: Option<(PaymentFailureKind, String)> = None; - loop { - match timeout(PAYMENT_STATUS_RECV_TIMEOUT, rx.recv()).await { - Err(_) => { - failure = Some(( - PaymentFailureKind::Indeterminate, - format!( - "payment status stream timed out after {}s without a terminal update", - PAYMENT_STATUS_RECV_TIMEOUT.as_secs() - ), - )); - break; - } - Ok(None) => break, - Ok(Some(msg)) => { - if let Ok(status) = PaymentStatus::try_from(msg.payment.status) { - match status { - PaymentStatus::Succeeded => { - succeeded = true; - break; - } - PaymentStatus::Failed => { - failure = Some(( - PaymentFailureKind::Terminal, - format!("payment failed: reason {}", msg.payment.failure_reason), - )); - break; - } - _ => {} + // dev_fee::send_dev_fee_payment. The drain is bounded transitively by the + // send-side timeout above: when the send future ends — normal return, RPC + // error, or dropped at the 75s bound — `tx` drops and `recv()` yields + // `None`. We track *why* the stream ended: only an explicit + // `PaymentStatus::Failed` is terminal. A clean EOF leaves the payment + // outcome unknown (it may still be in flight), so it is routed as + // `Indeterminate` — `on_send_payment_failure` then keeps the invoice + + // hash for reconciliation instead of re-prompting. + let drain_fut = async move { + let mut succeeded = false; + let mut failure: Option<(PaymentFailureKind, String)> = None; + while let Some(msg) = rx.recv().await { + if let Ok(status) = PaymentStatus::try_from(msg.payment.status) { + match status { + PaymentStatus::Succeeded => { + succeeded = true; + break; + } + PaymentStatus::Failed => { + failure = Some(( + PaymentFailureKind::Terminal, + format!("payment failed: reason {}", msg.payment.failure_reason), + )); + break; } + _ => {} } } } + // Unblock a send that is still pushing updates into a channel we are + // done reading: dropping `rx` fails its next `listener.send`, so the + // send future returns immediately instead of riding out the 75s + // bound for a payment whose verdict we already hold. + drop(rx); + (succeeded, failure) + }; + + let (send_outcome, (succeeded, stream_failure)) = tokio::join!(send_fut, drain_fut); + + match classify_send_verdict(send_outcome, succeeded, stream_failure) { + SendVerdict::Settled => slash_after_success(pool, bond, counterparty_share).await, + SendVerdict::Failure(kind, msg) => { + on_send_payment_failure(pool, bond, max_retries, claim_window_seconds, kind, &msg).await + } } +} + +/// Combined verdict of a bounded `send_payment` and its concurrent status +/// drain (see `pay_counterparty`). +#[derive(Debug, PartialEq)] +enum SendVerdict { + /// The stream reported `Succeeded`: the payment settled — finalize the + /// slash. + Settled, + /// No settlement: route through `on_send_payment_failure` with this kind + /// and cause. + Failure(PaymentFailureKind, String), +} +/// Classify the joint outcome of the bounded send future and the status +/// drain into a single verdict. +/// +/// A terminal verdict from the stream is the payment's actual outcome and +/// takes priority over however the send future ended: a `Succeeded` +/// delivered just before the timeout finalizes the slash immediately instead +/// of deferring to reconciliation, and an explicit `Failed` is safe to act +/// on regardless of the send-side result. +/// +/// With no terminal verdict, classify by the most specific cause. Every such +/// branch is indeterminate — the payment may still settle, so the caller +/// keeps the invoice + hash for reconciliation and never re-prompts the +/// winner against a payment that may still succeed. (A locked-in HTLC cannot +/// be cancelled by the sender; dropping the send future on timeout closes +/// our side of the gRPC stream only.) +fn classify_send_verdict( + send_outcome: Result, tokio::time::error::Elapsed>, + succeeded: bool, + stream_failure: Option<(PaymentFailureKind, String)>, +) -> SendVerdict { if succeeded { - return slash_after_success(pool, bond, counterparty_share).await; + return SendVerdict::Settled; } - - // EOF with no terminal status (the `Ok(None)` break above) is also - // indeterminate: the stream closed without telling us the outcome. - let (kind, msg) = failure.unwrap_or(( - PaymentFailureKind::Indeterminate, - "payment stream ended without terminal status".to_string(), - )); - on_send_payment_failure(pool, bond, max_retries, claim_window_seconds, kind, &msg).await + let stream_failure = match stream_failure { + Some((PaymentFailureKind::Terminal, msg)) => { + return SendVerdict::Failure(PaymentFailureKind::Terminal, msg); + } + other => other, + }; + let (kind, msg) = match send_outcome { + Err(_) => ( + PaymentFailureKind::Indeterminate, + format!( + "send_payment reached no terminal state after {}s", + crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + ), + ), + Ok(Err(e)) => { + // The RPC call itself errored. We cannot be sure the payment did + // not partially enter LND. + (PaymentFailureKind::Indeterminate, format!("{e}")) + } + // EOF with no terminal status: the stream closed without telling us + // the outcome. + Ok(Ok(())) => stream_failure.unwrap_or(( + PaymentFailureKind::Indeterminate, + "payment stream ended without terminal status".to_string(), + )), + }; + SendVerdict::Failure(kind, msg) } /// Flip a `PendingPayout` row to `Slashed` after a confirmed payment. @@ -2373,6 +2421,97 @@ mod tests { assert_eq!(after.payout_payment_hash.as_deref(), Some("cafebabe")); } + /// Produce a real `tokio::time::error::Elapsed` (it has no public + /// constructor): a zero-duration timeout over a pending future. + async fn elapsed() -> tokio::time::error::Elapsed { + timeout(std::time::Duration::ZERO, std::future::pending::<()>()) + .await + .unwrap_err() + } + + #[tokio::test] + async fn classify_stream_succeeded_wins_over_send_timeout() { + // A Succeeded delivered just before the 75s cutoff is the payment's + // real outcome: finalize the slash immediately instead of deferring + // to reconciliation, no matter how the send future ended. + let verdict = classify_send_verdict(Err(elapsed().await), true, None); + assert_eq!(verdict, SendVerdict::Settled); + } + + #[tokio::test] + async fn classify_stream_terminal_failed_maps_to_terminal() { + let verdict = classify_send_verdict( + Ok(Ok(())), + false, + Some(( + PaymentFailureKind::Terminal, + "payment failed: reason 1".to_string(), + )), + ); + assert_eq!( + verdict, + SendVerdict::Failure( + PaymentFailureKind::Terminal, + "payment failed: reason 1".to_string() + ) + ); + } + + #[tokio::test] + async fn classify_send_timeout_is_indeterminate() { + // The Elapsed branch: dropping the send future does not cancel a + // locked-in HTLC, so the verdict must be Indeterminate (keep the + // invoice + hash for reconciliation), never Terminal. + let verdict = classify_send_verdict(Err(elapsed().await), false, None); + assert_eq!( + verdict, + SendVerdict::Failure( + PaymentFailureKind::Indeterminate, + format!( + "send_payment reached no terminal state after {}s", + crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + ) + ) + ); + } + + #[tokio::test] + async fn classify_send_rpc_error_is_indeterminate() { + let rpc_err = MostroInternalErr(ServiceError::LnPaymentError("boom".to_string())); + let verdict = classify_send_verdict(Ok(Err(rpc_err)), false, None); + match verdict { + SendVerdict::Failure(PaymentFailureKind::Indeterminate, msg) => { + assert!( + msg.contains("boom"), + "cause must carry the RPC error: {msg}" + ); + } + other => panic!("expected indeterminate failure, got {other:?}"), + } + } + + #[tokio::test] + async fn classify_stream_succeeded_wins_over_send_error() { + // Once the drain sees Succeeded it drops `rx`, which fails the send's + // next `listener.send` — the resulting Ok(Err(..)) from the send + // future must not shadow the settled verdict. + let send_err = MostroInternalErr(ServiceError::LnNodeError("channel closed".to_string())); + let verdict = classify_send_verdict(Ok(Err(send_err)), true, None); + assert_eq!(verdict, SendVerdict::Settled); + } + + #[tokio::test] + async fn classify_stream_eof_is_indeterminate() { + let verdict = classify_send_verdict(Ok(Ok(())), false, None); + assert_eq!( + verdict, + SendVerdict::Failure( + PaymentFailureKind::Indeterminate, + "payment stream ended without terminal status".to_string() + ) + ); + } + #[tokio::test] async fn finalize_node_only_transitions_to_slashed() { // `slash_node_share_pct = 1.0` style row: counterparty share is diff --git a/src/app/release.rs b/src/app/release.rs index 3b017d8e..272ebd60 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -3,7 +3,7 @@ use crate::app::context::AppContext; use crate::app::dispute::close_dispute_after_user_resolution; use crate::escrow::EscrowBackend; use crate::lightning::invoice::{decode_invoice, validate_payout_invoice}; -use crate::lightning::LndConnector; +use crate::lightning::{LndConnector, PaymentMessage, PAYOUT_SEND_PAYMENT_TIMEOUT}; use crate::lnurl::resolv_ln_address; use crate::nip33::{new_order_event_with_created_at, order_to_tags}; use crate::util::{ @@ -23,8 +23,24 @@ use sqlx::{Pool, Sqlite}; use std::cmp::Ordering; use std::str::FromStr; use tokio::sync::mpsc::channel; +use tokio::sync::Semaphore; +use tokio::time::timeout; use tracing::{info, warn}; +/// Cap on concurrently running payout send tasks. Since `do_payment` returns +/// right after claiming, the scheduler retry loop can fan a backlog of N +/// failed payouts into N background tasks, each holding its own LND gRPC +/// connection and payment stream for up to [`PAYOUT_SEND_PAYMENT_TIMEOUT`]; +/// this semaphore makes a backlog queue instead of fanning out. The queue puts +/// an unbounded wait between the claim and the send, during which +/// reconciliation may re-arm the claim — and the buyer may then supply a fresh +/// invoice with a *different* hash, a case neither the pre-send duplicate +/// guard nor LND's duplicate rejection can catch; that window is closed by +/// `touch_order_payout_claim` right after the permit: a task whose claim was +/// re-armed or replaced while it queued drops its send. Fixed for now; could +/// become a settings knob later. +static PAYOUT_DISPATCH_SEMAPHORE: Semaphore = Semaphore::const_new(8); + /// Run [`check_failure_retries`] and surface bookkeeping failures instead of /// silently dropping them. On success, preserves the existing retry-count log. async fn check_failure_retries_or_log(ctx: &AppContext, order: &Order, request_id: Option) { @@ -524,20 +540,29 @@ async fn handle_child_order( Ok(()) } -/// Pay the buyer invoice for a settled-hold-invoice order. +/// Dispatch the buyer payout for a settled-hold-invoice order. +/// +/// `Ok(())` means the payout was *dispatched* (or a claim already exists), +/// not that it settled: everything up to and including the idempotency claim +/// runs inline — bounded work only — and the `send_payment` call itself runs +/// in a background task so a payment that never reaches a terminal state +/// (hold invoice, HTLC stuck in route) cannot freeze the event loop. The +/// task is bounded by [`PAYOUT_SEND_PAYMENT_TIMEOUT`]; on timeout the claim +/// marker is kept and `reconcile_inflight_payout` owns the outcome. /// /// Lightning Addresses **and** bech32 LNURLs are resolved via /// [`resolv_ln_address`] under the LNURL host policy. A non-empty `pr` must /// decode as BOLT11 and pass [`validate_payout_invoice`] (chain match, final -/// CLTV bound, not already expired) before LND submission; resolve, decode and -/// validation failures plus `send_payment` RPC errors go through -/// [`check_failure_retries_or_log`] and return `Err` (no empty status-watcher -/// spawn). Streamed `PaymentStatus::Failed` updates also bump retry -/// bookkeeping. Callers such as `release_action` typically ignore the error -/// after hold settlement — retries are driven by the failed-payment job. +/// CLTV bound, not already expired) before LND submission; resolve, decode +/// and validation failures — all pre-claim — go through +/// [`check_failure_retries_or_log`] and return `Err`. `send_payment` RPC +/// errors and streamed `PaymentStatus::Failed` updates bump the same retry +/// bookkeeping from the background task. Callers such as `release_action` +/// ignore the result after hold settlement — retries are driven by the +/// failed-payment job. pub async fn do_payment( ctx: &AppContext, - mut order: Order, + order: Order, request_id: Option, ) -> Result<(), MostroError> { let payment_request = match order.buyer_invoice.as_ref() { @@ -609,6 +634,11 @@ pub async fn do_payment( None => payment_request, }; + // Resolve the buyer pubkey *before* claiming: a malformed order fails + // here without a claim, so no marker is ever left set for a payout that + // was never dispatched. + let buyer_pubkey = order.get_buyer_pubkey().map_err(MostroInternalErr)?; + // Connect to LND *before* claiming: if the connection fails, `?` returns // here without a claim, so a transient connect blip never leaves a marker // set with no payment behind it. @@ -622,8 +652,10 @@ pub async fn do_payment( // racing) — only the winner pays. Cleared on a confirmed-terminal outcome or // by reconciliation; the timestamp keeps reconciliation from acting on this // payout until LND has surely registered it (closing the reconcile-vs-send - // race). Placed right before `send_payment` so the window between claim and - // LND registering the payment is only the send call itself. + // race). The dispatch task re-validates and refreshes this claim + // (`touch_order_payout_claim`) after its semaphore wait, so the window + // between the (refreshed) claim and LND registering the payment is only + // the send call itself even when the task queued behind a backlog. let payout_hash = decode_invoice(&payment_request) .map(|inv| bytes_to_string(inv.payment_hash().as_ref())) .map_err(|_| MostroInternalErr(ServiceError::InvoiceInvalidError))?; @@ -637,115 +669,211 @@ pub async fn do_payment( return Ok(()); }; - let (tx, mut rx) = channel(100); - - let payment_task = ln_client_payment.send_payment(&payment_request, amount as i64, tx); - if let Err(payment_result) = payment_task.await { - warn!("Error during ln payment : {}", payment_result); - // `send_payment` returned before spawning the status watcher, so the - // claim we just set would otherwise stay locked (blocking retry and - // AddInvoice) until the grace-delayed reconciliation job runs. Ask LND - // what actually happened to this hash and resolve the claim inline: - // - in flight / succeeded / lookup error -> KEEP the marker; the - // payment may still settle, so reconciliation owns the outcome and - // no second payout is ever dispatched. - // - not registered / failed -> re-arm retry now (and - // notify the buyer) instead of waiting. - let keep_marker = match Vec::::from_hex(&payout_hash) { - Ok(bytes) if bytes.len() == 32 => matches!( - ln_client_payment.lookup_payment_status(&bytes).await, - Ok(Some(PaymentStatus::InFlight)) | Ok(Some(PaymentStatus::Succeeded)) | Err(_) - ), - // Should not happen (we just built this hash), but if it is - // unusable we cannot confirm an in-flight payment — re-arm. - _ => false, - }; - if !keep_marker - && crate::db::fail_order_payout( - ctx.pool(), - order.id, - &payout_hash, - Some(payout_claimed_at), - ) - .await - .unwrap_or(false) - { - check_failure_retries_or_log(ctx, &order, request_id).await; - } - // Do not spawn the status watcher or report Ok. - return Err(payment_result); - } - // Get Mostro keys from context let my_keys = ctx.keys().clone(); - // Get buyer and seller pubkeys - let buyer_pubkey = order.get_buyer_pubkey().map_err(MostroInternalErr)?; - - // Clone ctx for the async closure + // Clone ctx for the background task let ctx = ctx.clone(); - let payment = { - async move { - // We redeclare vars to use inside this block - // Receiving msgs from send_payment() - while let Some(msg) = rx.recv().await { - if let Ok(status) = PaymentStatus::try_from(msg.payment.status) { - match status { - PaymentStatus::Succeeded => { - info!( - "Order Id {}: Invoice with hash: {} paid!", - order.id, msg.payment.payment_hash - ); - // Release our claim only if the order actually - // reached Success. If finalization fails, keep the - // marker so reconciliation retries it — clearing it - // here would strand a paid order with no recovery. - if payment_success(&ctx, &mut order, buyer_pubkey, &my_keys, request_id) + // From here on the payout runs OFF the event loop: `send_payment` waits + // for LND's payment stream to reach a terminal state, which a locked-in + // but unresolved HTLC (hold invoice, HTLC stuck in route) can delay + // indefinitely — awaiting it inline froze the whole daemon. The claim + // persisted above is what makes backgrounding safe: whatever happens to + // this task (RPC error, timeout, process restart), reconciliation can + // always finish or fail the payout by hash, so it is never lost or paid + // twice. + tokio::spawn(async move { + // Bound concurrent sends (see PAYOUT_DISPATCH_SEMAPHORE). The + // semaphore is static and never closed, so acquire() only errs if it + // were closed; proceeding unpermitted in that impossible case beats + // silently dropping a claimed payout. The permit is held for the + // whole task (send, watcher drain, RPC-error reconcile) via RAII. + let _permit = PAYOUT_DISPATCH_SEMAPHORE.acquire().await; + + // Re-validate the claim now that the queue wait is over, refreshing + // its timestamp in the same CAS. If reconciliation re-armed the claim + // while this task queued (the buyer may already have supplied a fresh + // invoice under a different hash), a newer payout owns the order and + // this invoice must NOT be sent. On a DB error, dropping the send is + // also the safe direction: the kept marker is recoverable by + // reconciliation, a blind send is not. The refreshed timestamp is the + // claim token from here on — it restarts the reconcile grace clock + // and invalidates any pre-touch snapshot a reconciler already holds. + let payout_claimed_at = match crate::db::touch_order_payout_claim( + ctx.pool(), + order.id, + &payout_hash, + Some(payout_claimed_at), + ) + .await + { + Ok(Some(refreshed_at)) => refreshed_at, + Ok(None) => { + warn!( + "Order {}: payout claim was re-armed or replaced while queued; dropping stale dispatch of hash {}", + order.id, payout_hash + ); + return; + } + Err(e) => { + warn!( + "Order {}: could not re-validate payout claim after queue ({e}); dropping dispatch of hash {} — reconciliation will resolve the kept marker", + order.id, payout_hash + ); + return; + } + }; + + let (tx, mut rx) = channel::(100); + + // Start the status watcher BEFORE `send_payment` and run them + // concurrently: `send_payment` forwards every LND update through `tx` + // and blocks when the channel fills, so a watcher started only after + // it returns could deadlock the payment on a chatty stream. The + // watcher ends on its own when `tx` drops (send_payment returned or + // its future was dropped by the timeout). + let watcher = { + let ctx = ctx.clone(); + let payout_hash = payout_hash.clone(); + let mut order = order.clone(); + async move { + // Receiving msgs from send_payment() + while let Some(msg) = rx.recv().await { + if let Ok(status) = PaymentStatus::try_from(msg.payment.status) { + match status { + PaymentStatus::Succeeded => { + info!( + "Order Id {}: Invoice with hash: {} paid!", + order.id, msg.payment.payment_hash + ); + // Release our claim only if the order actually + // reached Success. If finalization fails, keep the + // marker so reconciliation retries it — clearing it + // here would strand a paid order with no recovery. + if payment_success( + &ctx, + &mut order, + buyer_pubkey, + &my_keys, + request_id, + ) .await .unwrap_or(false) - { - let _ = crate::db::clear_order_payout( + { + let _ = crate::db::clear_order_payout( + ctx.pool(), + order.id, + &payout_hash, + Some(payout_claimed_at), + ) + .await; + } + } + PaymentStatus::Failed => { + warn!( + "Order Id {}: Invoice with hash: {} has failed!", + order.id, msg.payment.payment_hash + ); + + // Release our own claim (scoped to this hash and + // the per-claim timestamp) and re-arm retry. Only + // do the failure bookkeeping and buyer + // notification if we still owned the claim, so a + // stale watcher never pollutes a newer payout's + // retry state or notifies against it — even when + // the retry reused the same invoice/hash. + if crate::db::fail_order_payout( ctx.pool(), order.id, &payout_hash, Some(payout_claimed_at), ) - .await; - } - } - PaymentStatus::Failed => { - warn!( - "Order Id {}: Invoice with hash: {} has failed!", - order.id, msg.payment.payment_hash - ); - - // Release our own claim (scoped to this hash and the - // per-claim timestamp) and re-arm retry. Only do the - // failure bookkeeping and buyer notification if we - // still owned the claim, so a stale watcher never - // pollutes a newer payout's retry state or notifies - // against it — even when the retry reused the same - // invoice/hash. - if crate::db::fail_order_payout( - ctx.pool(), - order.id, - &payout_hash, - Some(payout_claimed_at), - ) - .await - .unwrap_or(false) - { - check_failure_retries_or_log(&ctx, &order, request_id).await; + .await + .unwrap_or(false) + { + check_failure_retries_or_log(&ctx, &order, request_id).await; + } } + _ => {} } - _ => {} } } } + }; + tokio::spawn(watcher); + + match timeout( + PAYOUT_SEND_PAYMENT_TIMEOUT, + ln_client_payment.send_payment(&payment_request, amount as i64, tx), + ) + .await + { + // The send stream ended. Usually a terminal update was delivered + // and the watcher finishes the bookkeeping — but send_payment's + // `while let Ok(Some(..))` swallows a mid-stream gRPC error, so + // this branch also covers a stream that died or EOF'd with no + // terminal update. In that case the watcher exits without acting, + // the claim marker stays set, and reconciliation resolves the + // real outcome by payment hash. + Ok(Ok(())) => {} + Ok(Err(payment_result)) => { + warn!("Error during ln payment : {}", payment_result); + // `send_payment` failed at the RPC level, so the claim set + // above would otherwise stay locked (blocking retry and + // AddInvoice) until the grace-delayed reconciliation job runs. + // Ask LND what actually happened to this hash and resolve the + // claim now: + // - in flight / succeeded / lookup error -> KEEP the marker; + // the payment may still settle, so reconciliation owns the + // outcome and no second payout is ever dispatched. + // - not registered / failed -> re-arm retry now + // (and notify the buyer) instead of waiting. + let keep_marker = match Vec::::from_hex(&payout_hash) { + Ok(bytes) if bytes.len() == 32 => matches!( + ln_client_payment.lookup_payment_status(&bytes).await, + Ok(Some(PaymentStatus::InFlight)) + | Ok(Some(PaymentStatus::Succeeded)) + | Err(_) + ), + // Should not happen (we just built this hash), but if it is + // unusable we cannot confirm an in-flight payment — re-arm. + _ => false, + }; + if !keep_marker + && crate::db::fail_order_payout( + ctx.pool(), + order.id, + &payout_hash, + Some(payout_claimed_at), + ) + .await + .unwrap_or(false) + { + check_failure_retries_or_log(&ctx, &order, request_id).await; + } + } + Err(_) => { + // Timed out without a terminal state. Dropping the + // `send_payment` future closes our side of the gRPC stream but + // does NOT cancel the payment: a locked-in HTLC cannot be + // cancelled by the sender and may still settle later (up to + // its CLTV). Do NOT call `fail_order_payout` here — re-arming + // a retry against a payment that may still succeed risks a + // double payout. Keep the marker: reconciliation looks the + // hash up in LND and finalizes or fails the order with the + // real outcome, so a slow-but-successful payment is delayed by + // at most the reconciler cadence, never lost. + warn!( + "Order Id {}: payout with hash {} got no terminal state after {}s; keeping claim marker for reconciliation", + order.id, + payout_hash, + PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + ); + } } - }; - tokio::spawn(payment); + }); + Ok(()) } diff --git a/src/db.rs b/src/db.rs index f46e0723..6f3a166e 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1229,6 +1229,49 @@ pub async fn claim_order_payout( Ok((result.rows_affected() > 0).then_some(claimed_at)) } +/// Re-validate ownership of a payout claim and refresh its timestamp, as one +/// CAS. +/// +/// Used by the dispatch task after waiting in the send-semaphore queue: a task +/// can queue past the reconcile grace window, and reconciliation may then +/// re-arm the claim (buyer prompted for a fresh invoice → different payment +/// hash), in which case a newer payout owns the order and the queued invoice +/// must not be sent. Refreshing `payout_claimed_at` — rather than merely +/// checking it — also restarts the reconcile grace clock, restoring the +/// invariant that a claim is never older than its send by more than the send +/// itself; a mere ownership check would leave reconciliation free to re-arm in +/// the instant between the check and LND registering the payment. The refresh +/// also invalidates any pre-touch snapshot a reconciler already holds: its +/// release CAS is scoped to the timestamp it observed, which no longer +/// matches. +/// +/// Scoped to hash + token + `settled-hold-invoice` status: a marker lingering +/// on an order that has since gone terminal must never turn into a send. +/// Returns `Some(refreshed_at)` — the new per-claim token every later release +/// must be scoped to — when the caller still owned the claim, or `None` when +/// the claim was re-armed or replaced while queued (drop the send). +pub async fn touch_order_payout_claim( + pool: &SqlitePool, + order_id: Uuid, + payment_hash: &str, + claimed_at: Option, +) -> Result, MostroError> { + let refreshed_at = chrono::Utc::now().timestamp(); + let result = sqlx::query( + "UPDATE orders SET payout_claimed_at = ?1 \ + WHERE id = ?2 AND payout_payment_hash = ?3 AND payout_claimed_at IS ?4 \ + AND status = 'settled-hold-invoice'", + ) + .bind(refreshed_at) + .bind(order_id) + .bind(payment_hash) + .bind(claimed_at) + .execute(pool) + .await + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + Ok((result.rows_affected() > 0).then_some(refreshed_at)) +} + /// Clear the in-flight payout marker for `order_id` after a successful terminal /// outcome (or as tidy-up once the order has moved to `Success`). /// @@ -3178,6 +3221,140 @@ mod tests { ); } + async fn claimed_at_of(pool: &SqlitePool, id: uuid::Uuid) -> Option { + sqlx::query_scalar::<_, Option>("SELECT payout_claimed_at FROM orders WHERE id = ?") + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn test_touch_payout_claim_refreshes_token() { + // An owned claim (backdated, as if the dispatch task queued for a + // while) is revalidated: the touch returns a fresh token and the row + // reflects it, restarting the reconcile grace clock. + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let hash = "a".repeat(64); + insert_inflight_order(&pool, id, &hash, Some(1000)).await; + + let refreshed = super::touch_order_payout_claim(&pool, id, &hash, Some(1000)) + .await + .unwrap() + .expect("owner must keep its claim"); + assert!(refreshed > 1000, "token must be refreshed forward"); + assert_eq!(claimed_at_of(&pool, id).await, Some(refreshed)); + assert_eq!( + payout_hash_of(&pool, id).await.as_deref(), + Some(hash.as_str()) + ); + } + + #[tokio::test] + async fn test_touch_payout_claim_loses_after_rearm() { + // Reconciliation re-armed (cleared) the claim while the task queued: + // the touch must lose and leave the row alone. + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let hash = "b".repeat(64); + insert_inflight_order(&pool, id, &hash, Some(1000)).await; + assert!(super::fail_order_payout(&pool, id, &hash, Some(1000)) + .await + .unwrap()); + + let touched = super::touch_order_payout_claim(&pool, id, &hash, Some(1000)) + .await + .unwrap(); + assert!(touched.is_none(), "a re-armed claim must not be touchable"); + assert!(payout_hash_of(&pool, id).await.is_none()); + assert_eq!(claimed_at_of(&pool, id).await, None); + } + + #[tokio::test] + async fn test_touch_payout_claim_loses_to_replacement_claim() { + // The double-payout scenario: while the task queued, its claim was + // re-armed and a NEW payout (fresh invoice, different hash) claimed + // the order. The stale task's touch must lose and must not disturb + // the replacement claim. + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let old_hash = "c".repeat(64); + let new_hash = "d".repeat(64); + insert_inflight_order(&pool, id, &old_hash, Some(1000)).await; + assert!(super::fail_order_payout(&pool, id, &old_hash, Some(1000)) + .await + .unwrap()); + let new_token = super::claim_order_payout(&pool, id, &new_hash) + .await + .unwrap() + .unwrap(); + + let touched = super::touch_order_payout_claim(&pool, id, &old_hash, Some(1000)) + .await + .unwrap(); + assert!( + touched.is_none(), + "the stale dispatch must drop its send once a newer payout owns the order" + ); + assert_eq!( + payout_hash_of(&pool, id).await.as_deref(), + Some(new_hash.as_str()), + "the replacement claim must be untouched" + ); + assert_eq!(claimed_at_of(&pool, id).await, Some(new_token)); + } + + #[tokio::test] + async fn test_touch_payout_claim_refuses_terminal_status() { + // A marker lingering on an order that already went terminal must + // never be refreshed into a send. + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let hash = "e".repeat(64); + insert_inflight_order(&pool, id, &hash, Some(1000)).await; + sqlx::query("UPDATE orders SET status = 'success' WHERE id = ?") + .bind(id) + .execute(&pool) + .await + .unwrap(); + + let touched = super::touch_order_payout_claim(&pool, id, &hash, Some(1000)) + .await + .unwrap(); + assert!(touched.is_none()); + assert_eq!(claimed_at_of(&pool, id).await, Some(1000), "row untouched"); + } + + #[tokio::test] + async fn test_touch_payout_claim_invalidates_pre_touch_snapshot() { + // A reconciler that read the claim BEFORE the touch holds a stale + // token: its scoped release must lose after the refresh, so it cannot + // re-arm a payout whose send is imminent. + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let hash = "f".repeat(64); + insert_inflight_order(&pool, id, &hash, Some(1000)).await; + + let refreshed = super::touch_order_payout_claim(&pool, id, &hash, Some(1000)) + .await + .unwrap() + .unwrap(); + // Reconciler acts on its pre-touch snapshot (token 1000): loses. + assert!(!super::fail_order_payout(&pool, id, &hash, Some(1000)) + .await + .unwrap()); + assert_eq!( + payout_hash_of(&pool, id).await.as_deref(), + Some(hash.as_str()) + ); + // The dispatch task's own release with the refreshed token still wins. + assert!(super::clear_order_payout(&pool, id, &hash, Some(refreshed)) + .await + .unwrap()); + assert!(payout_hash_of(&pool, id).await.is_none()); + } + async fn insert_inflight_order( pool: &SqlitePool, id: uuid::Uuid, diff --git a/src/lightning/mod.rs b/src/lightning/mod.rs index 60ac5b28..abe2490d 100644 --- a/src/lightning/mod.rs +++ b/src/lightning/mod.rs @@ -10,16 +10,43 @@ use fedimint_tonic_lnd::invoicesrpc::{ SettleInvoiceMsg, SettleInvoiceResp, }; use fedimint_tonic_lnd::lnrpc::{ - invoice::InvoiceState, GetInfoRequest, GetInfoResponse, InvoiceHtlcState, Payment, PaymentHash, + invoice::InvoiceState, payment, GetInfoRequest, GetInfoResponse, InvoiceHtlcState, Payment, + PaymentHash, }; use fedimint_tonic_lnd::routerrpc::{SendPaymentRequest, TrackPaymentRequest}; use fedimint_tonic_lnd::Client; use mostro_core::prelude::*; use rand::{self, RngCore}; use std::cmp::Ordering; +use std::time::Duration; use tokio::sync::mpsc::Sender; +use tokio::time::timeout; use tracing::info; +/// Seconds LND keeps launching route attempts for a payment +/// (`SendPaymentRequest.timeout_seconds`). Past this window an open payment +/// stream is only kept alive by an HTLC that is locked-in but unresolved, +/// which the sender cannot cancel. +pub(crate) const LND_PAYMENT_ROUTE_TIMEOUT_SECS: i32 = 60; + +/// Upper bound on how long a payout waits for `send_payment` to reach a +/// terminal state: LND's own route-attempt window plus margin, DERIVED so +/// that raising [`LND_PAYMENT_ROUTE_TIMEOUT_SECS`] can never silently +/// undercut LND's retries. Hitting this bound does NOT fail the payout — +/// the payment may still settle, so callers keep their claim/hash and let +/// reconciliation resolve the real outcome. Fixed for now; could become a +/// settings knob later. +pub(crate) const PAYOUT_SEND_PAYMENT_TIMEOUT: Duration = + Duration::from_secs(LND_PAYMENT_ROUTE_TIMEOUT_SECS as u64 + 15); + +/// Bound on the duplicate-guard lookup inside `send_payment`. The guard is +/// advisory — on timeout or transport error the send proceeds, because LND +/// itself rejects a genuine duplicate for an in-flight/settled hash — so it +/// must never eat a caller's whole budget: `dev_fee::send_dev_fee_payment` +/// wraps `send_payment` in a 5s total timeout, and 2s leaves the majority of +/// that for the send itself. +const DUPLICATE_GUARD_LOOKUP_TIMEOUT: Duration = Duration::from_secs(2); + #[derive(Clone)] pub struct LndConnector { pub client: Client, @@ -266,7 +293,11 @@ impl LndConnector { listener: Sender, ) -> Result<(), MostroError> { let invoice = decode_invoice(payment_request)?; - let payment_hash = invoice.signable_hash(); + // The BOLT11 payment hash — the key LND indexes payments by. NOT + // `signable_hash()`, which is the invoice's signature digest and is + // never known to LND, so a guard keyed on it can never fire. + let payment_hash_ref: &[u8] = invoice.payment_hash().as_ref(); + let payment_hash = payment_hash_ref.to_vec(); let hash = bytes_to_string(&payment_hash); // We need to set a max fee amount. `routing_fee_cap_sats` is the @@ -274,29 +305,37 @@ impl LndConnector { // debugging always matches what LND actually enforces. let max_fee = routing_fee_cap_sats(amount); - let track_payment_req = TrackPaymentRequest { - payment_hash: payment_hash.to_vec(), - no_inflight_updates: true, - }; - - let track = self - .client - .router() - .track_payment_v2(track_payment_req) - .await - .map_err(|e| MostroInternalErr(ServiceError::LnPaymentError(e.to_string()))); - - // We only send the payment if it wasn't attempted before - if track.is_ok() { - info!("Aborting paying invoice with hash {} to buyer", hash); - return Err(MostroInternalErr(ServiceError::LnPaymentError( - "Track error".to_string(), - ))); + // Duplicate-dispatch guard: refuse to send only when LND reports this + // hash as already in flight or settled. A Failed/Unknown/absent record + // must NOT abort — the retry flow legitimately re-sends the same + // invoice after a failure. A lookup transport error or timeout also + // proceeds: LND itself rejects a duplicate SendPaymentV2 for an + // in-flight or settled hash (the hard backstop behind this check), + // and if LND is truly unreachable the send below fails anyway. The + // lookup is bounded so this advisory check can never eat a caller's + // budget (see DUPLICATE_GUARD_LOOKUP_TIMEOUT). + match timeout( + DUPLICATE_GUARD_LOOKUP_TIMEOUT, + self.lookup_payment_status(&payment_hash), + ) + .await + { + Ok(Ok(Some(payment::PaymentStatus::InFlight))) + | Ok(Ok(Some(payment::PaymentStatus::Succeeded))) => { + info!( + "Aborting payment for hash {}: already in flight or settled", + hash + ); + return Err(MostroInternalErr(ServiceError::LnPaymentError( + "payment already dispatched for this hash".to_string(), + ))); + } + _ => {} } let mut request = SendPaymentRequest { payment_request: payment_request.to_string(), - timeout_seconds: 60, + timeout_seconds: LND_PAYMENT_ROUTE_TIMEOUT_SECS, fee_limit_sat: max_fee, ..Default::default() };