From 71ce09f5dbfc11c2664fa9c20fc40f1006cc7aec Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Sun, 16 Aug 2026 23:37:13 -0600 Subject: [PATCH 01/10] fix: resolve buyer pubkey before claiming the payout get_buyer_pubkey() ran after send_payment, so a malformed order (no buyer pubkey) failed only after the payment was already dispatched and the payout claim was set. Compute it before LndConnector::new() and the claim CAS so a bad order fails fast without leaving a payout marker for a payment that was never sent, preserving the no marker without a payment, behind it invariant. Preparatory for dispatching the buyer payout off the event loop: the background task will capture this value instead of computing it after send_payment returns. --- src/app/release.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/app/release.rs b/src/app/release.rs index 3b017d8e..da5ef436 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -609,6 +609,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. @@ -679,9 +684,6 @@ pub async fn do_payment( // 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 let ctx = ctx.clone(); From efa3d06ecf9d4919402ec497cc6b64b25acefa4f Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Sun, 16 Aug 2026 23:46:27 -0600 Subject: [PATCH 02/10] fix: dispatch buyer payout off the event loop with a bounded send_payment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The event loop awaited do_payment's send_payment inline, which consumes LND's payment stream until a terminal state. A payout HTLC that never resolves (buyer-supplied hold invoice, or an HTLC stuck at a routing node) kept that await pending forever and froze all message processing for the whole daemon, while spawned tasks kept logging normally. Seen twice in production (orders d4b04bcd and 33423dc5) and reproduced on regtest. Run everything past the idempotency claim in a background task instead. The inline path now only does bounded work (resolve invoice/LNURL, connect to LND, persist the payout claim) and returns dispatched; the persisted claim is what makes this safe — whatever happens to the task, reconcile_inflight_payout can finish or fail the payout by hash, so it is never lost or paid twice. Inside the task: - start the status watcher before send_payment and run them concurrently, so a chatty stream can no longer deadlock on the full status channel with the watcher not yet started - bound send_payment with a 75s timeout (LND stops routing attempts at 60s); on timeout, keep the claim marker and let reconciliation resolve the real outcome — a locked-in HTLC cannot be cancelled by the sender and may still settle, so failing the payout here could double-pay - on an RPC-level send_payment error, keep the existing inline reconciliation (lookup by hash, keep marker or re-arm retry), now running in the task All three do_payment callers (release_action, admin_settle and the scheduler retry job) already ignore or merely log the result, so the new dispatched semantics changes no caller behavior, and all of them stop being able to freeze on a payout. --- src/app/release.rs | 269 +++++++++++++++++++++++++++++---------------- 1 file changed, 172 insertions(+), 97 deletions(-) diff --git a/src/app/release.rs b/src/app/release.rs index da5ef436..906e9a9e 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}; use crate::lnurl::resolv_ln_address; use crate::nip33::{new_order_event_with_created_at, order_to_tags}; use crate::util::{ @@ -22,9 +22,22 @@ use nostr_sdk::prelude::*; use sqlx::{Pool, Sqlite}; use std::cmp::Ordering; use std::str::FromStr; +use std::time::Duration; use tokio::sync::mpsc::channel; +use tokio::time::timeout; use tracing::{info, warn}; +/// Upper bound on how long the background payout task waits for +/// `send_payment` to reach a terminal state. LND itself stops launching new +/// route attempts after `timeout_seconds: 60`, so 75s leaves margin for LND to +/// exhaust its own attempts; past that, the stream is only kept open by an +/// HTLC that is locked-in but unresolved (hold invoice or stuck route), which +/// the sender cannot cancel anyway. Hitting this timeout does NOT fail the +/// payout: the payment may still settle in LND, so the claim marker is kept +/// and `reconcile_inflight_payout` resolves the outcome by payment hash. +/// Fixed for now; could become a settings knob later. +pub(crate) const PAYOUT_SEND_PAYMENT_TIMEOUT: Duration = Duration::from_secs(75); + /// 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 +537,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() { @@ -642,112 +664,165 @@ 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(); - // 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 { + 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 stream reached a terminal state; the watcher drains the + // remaining updates and finishes the bookkeeping. + 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(()) } From a95ede4df56f1113414308295eea85c13db3f3b8 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Sun, 16 Aug 2026 23:50:27 -0600 Subject: [PATCH 03/10] fix: bound bond payout send_payment with the payout timeout The bond payout job awaited send_payment unbounded, which consumes LND's payment stream until a terminal state. A payout HTLC that never resolves (hold invoice as the dispute winner's payout invoice, or an HTLC stuck in route) would pin the scheduler task forever, and since the status channel is drained only after send_payment returns, a stream with >100 updates could also deadlock on the full channel. The existing PAYMENT_STATUS_RECV_TIMEOUT only bounds the drain loop, not the send itself. Wrap the call in the same 75s PAYOUT_SEND_PAYMENT_TIMEOUT used by the buyer payout and route the elapsed case through the existing PaymentFailureKind::Indeterminate handling: the invoice + hash are kept for the reconciliation branch to resolve the real outcome on the next tick, so the winner is never re-prompted against a payment that may still settle and no double payout is possible. --- src/app/bond/payout.rs | 40 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 37 insertions(+), 3 deletions(-) diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index 13cc6633..b2e8498a 100644 --- a/src/app/bond/payout.rs +++ b/src/app/bond/payout.rs @@ -664,10 +664,44 @@ 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 — + // unbounded, that pins this scheduler task forever and, because the + // status channel is drained only after `send_payment` returns, a chatty + // stream could also deadlock on the full channel. Bound it 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. let (tx, mut rx) = channel(100); - let send_outcome = ln_client - .send_payment(invoice, counterparty_share, tx) - .await; + let send_outcome = timeout( + crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT, + ln_client.send_payment(invoice, counterparty_share, tx), + ) + .await; + let send_outcome = match send_outcome { + Ok(outcome) => outcome, + Err(_) => { + // Timed out without a terminal state. Dropping the 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. Indeterminate keeps the invoice + hash so the + // reconciliation branch above resolves the real outcome on the + // next tick — never re-prompt the winner against a payment that + // may still succeed. + return on_send_payment_failure( + pool, + bond, + max_retries, + claim_window_seconds, + PaymentFailureKind::Indeterminate, + &format!( + "send_payment reached no terminal state after {}s", + crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + ), + ) + .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 From d74c1113c8331c1fa3c5645632b9ca7116d87963 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:19:25 -0600 Subject: [PATCH 04/10] fix(lightning): key the duplicate-payment guard on the real payment hash MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pre-send guard queried track_payment_v2 with invoice.signable_hash() — the invoice's signature digest, which LND never indexes — so it could never find a prior payment and has never blocked a duplicate dispatch. Key it on invoice.payment_hash() and decide by status instead of existence: only an InFlight or Succeeded payment aborts the send. A Failed/Unknown/absent record proceeds, preserving the retry flow that legitimately re-sends the same invoice after a failure; a lookup transport error also proceeds, since LND itself rejects a duplicate SendPaymentV2 for an in-flight or settled hash and the subsequent send fails anyway if LND is unreachable. Reuses lookup_payment_status, the same primitive the payout and bond reconciliation paths already use. --- src/lightning/mod.rs | 46 +++++++++++++++++++++++++------------------- 1 file changed, 26 insertions(+), 20 deletions(-) diff --git a/src/lightning/mod.rs b/src/lightning/mod.rs index 60ac5b28..9c6ab116 100644 --- a/src/lightning/mod.rs +++ b/src/lightning/mod.rs @@ -10,7 +10,8 @@ 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; @@ -266,7 +267,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,24 +279,25 @@ 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 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. + match self.lookup_payment_status(&payment_hash).await { + Ok(Some(payment::PaymentStatus::InFlight)) + | 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 { From 9883aad8947d279c23ca4c76c2e533d6225e77b4 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:25:26 -0600 Subject: [PATCH 05/10] fix(bond): drain the payment status stream concurrently with send_payment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bond payout drained the status channel only after send_payment returned, so a stream with more than 100 updates filled the channel, blocked the sender, and rode the 75s timeout into Indeterminate for a payment that could have completed. Run the bounded send and the drain concurrently with tokio::join! — the same watcher-before-send pattern do_payment uses, kept on this task because the scheduler job needs a single combined outcome. A terminal verdict from the stream now takes priority over how the send future ended: a Succeeded delivered just before the timeout finalizes the slash immediately instead of deferring to reconciliation. The per-recv PAYMENT_STATUS_RECV_TIMEOUT stays as a second line of defense. --- src/app/bond/payout.rs | 197 ++++++++++++++++++++++------------------- 1 file changed, 105 insertions(+), 92 deletions(-) diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index b2e8498a..b1f0544e 100644 --- a/src/app/bond/payout.rs +++ b/src/app/bond/payout.rs @@ -666,113 +666,126 @@ async fn pay_counterparty( // 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 — - // unbounded, that pins this scheduler task forever and, because the - // status channel is drained only after `send_payment` returns, a chatty - // stream could also deadlock on the full channel. Bound it 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. + // 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 = timeout( + + let send_fut = timeout( crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT, ln_client.send_payment(invoice, counterparty_share, tx), - ) - .await; - let send_outcome = match send_outcome { - Ok(outcome) => outcome, - Err(_) => { - // Timed out without a terminal state. Dropping the 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. Indeterminate keeps the invoice + hash so the - // reconciliation branch above resolves the real outcome on the - // next tick — never re-prompt the winner against a payment that - // may still succeed. - return on_send_payment_failure( - pool, - bond, - max_retries, - claim_window_seconds, - PaymentFailureKind::Indeterminate, - &format!( - "send_payment reached no terminal state after {}s", - crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() - ), - ) - .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; - } + ); // 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; each recv stays bounded by + // `PAYMENT_STATUS_RECV_TIMEOUT` as a second line of defense, although the + // send-side timeout above already bounds the whole exchange. 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 drain_fut = async { + 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; + } + _ => {} } - _ => {} } } } } - } + (succeeded, failure) + }; + + let (send_outcome, (succeeded, stream_failure)) = tokio::join!(send_fut, drain_fut); + // 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. if succeeded { return slash_after_success(pool, bond, counterparty_share).await; } + if let Some((PaymentFailureKind::Terminal, msg)) = &stream_failure { + return on_send_payment_failure( + pool, + bond, + max_retries, + claim_window_seconds, + PaymentFailureKind::Terminal, + msg, + ) + .await; + } - // 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(), - )); + // No terminal verdict from the stream: classify why, most specific cause + // first. Every branch is indeterminate — the payment may still settle, so + // keep the invoice + hash for the reconciliation branch above and never + // re-prompt 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.) + let (kind, msg) = match send_outcome { + Err(_) => ( + PaymentFailureKind::Indeterminate, + format!( + "send_payment reached no terminal state after {}s", + crate::app::release::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 (or recv timeout) 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(), + )), + }; on_send_payment_failure(pool, bond, max_retries, claim_window_seconds, kind, &msg).await } From bc1e1f956044f663b1522b23b577c101b95628ba Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:32:42 -0600 Subject: [PATCH 06/10] fix: bound concurrent buyer-payout dispatch with a semaphore Since do_payment returns right after claiming, the scheduler retry loop can fan a backlog of N failed payouts into N concurrent background tasks, each holding its own LND gRPC connection and payment stream for up to PAYOUT_SEND_PAYMENT_TIMEOUT. Gate the send phase behind a static 8-permit semaphore acquired at the top of the spawned task, so a backlog queues instead of fanning out. A task queued past the reconcile grace window can lose its claim to re-arm-and-redispatch; that is safe: the pre-send duplicate guard (now keyed on the real payment hash) and LND's own duplicate rejection stop the late sender from double-paying --- src/app/release.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/src/app/release.rs b/src/app/release.rs index 906e9a9e..9ace5841 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -24,6 +24,7 @@ use std::cmp::Ordering; use std::str::FromStr; use std::time::Duration; use tokio::sync::mpsc::channel; +use tokio::sync::Semaphore; use tokio::time::timeout; use tracing::{info, warn}; @@ -38,6 +39,17 @@ use tracing::{info, warn}; /// Fixed for now; could become a settings knob later. pub(crate) const PAYOUT_SEND_PAYMENT_TIMEOUT: Duration = Duration::from_secs(75); +/// 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. A task queued +/// past the reconcile grace window can lose its claim to re-arm-and-redispatch; +/// that is safe: the pre-send duplicate guard in `send_payment` (keyed on the +/// real payment hash) and LND's own duplicate rejection stop the late sender +/// from double-paying. 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) { @@ -679,6 +691,13 @@ pub async fn do_payment( // 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; + let (tx, mut rx) = channel::(100); // Start the status watcher BEFORE `send_payment` and run them From 05a1665d7e4e8e119c1d9eaab8e5c2f8d93fb903 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:47:27 -0600 Subject: [PATCH 07/10] test(bond): cover the payout send-outcome classification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract the post-join! verdict logic of pay_counterparty into a pure classify_send_verdict and cover its full matrix: a stream-delivered Succeeded wins over a send timeout, an explicit Failed maps to Terminal, and the three no-verdict cases (send timeout, RPC error, stream EOF/recv-timeout) all classify as Indeterminate with their specific message. The timeout branch was previously unreachable in tests — a dead endpoint fails fast, it never hangs — and the Elapsed value is produced with a zero-duration timeout, no mock needed. Behavior of pay_counterparty is unchanged. --- src/app/bond/payout.rs | 175 +++++++++++++++++++++++++++++++++++------ 1 file changed, 150 insertions(+), 25 deletions(-) diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index b1f0544e..64e3f32f 100644 --- a/src/app/bond/payout.rs +++ b/src/app/bond/payout.rs @@ -740,32 +740,55 @@ async fn pay_counterparty( let (send_outcome, (succeeded, stream_failure)) = tokio::join!(send_fut, drain_fut); - // 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. - if succeeded { - return slash_after_success(pool, bond, counterparty_share).await; - } - if let Some((PaymentFailureKind::Terminal, msg)) = &stream_failure { - return on_send_payment_failure( - pool, - bond, - max_retries, - claim_window_seconds, - PaymentFailureKind::Terminal, - msg, - ) - .await; + 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 + } } +} - // No terminal verdict from the stream: classify why, most specific cause - // first. Every branch is indeterminate — the payment may still settle, so - // keep the invoice + hash for the reconciliation branch above and never - // re-prompt 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.) +/// 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 SendVerdict::Settled; + } + 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, @@ -786,7 +809,7 @@ async fn pay_counterparty( "payment stream ended without terminal status".to_string(), )), }; - on_send_payment_failure(pool, bond, max_retries, claim_window_seconds, kind, &msg).await + SendVerdict::Failure(kind, msg) } /// Flip a `PendingPayout` row to `Slashed` after a confirmed payment. @@ -2420,6 +2443,108 @@ 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::app::release::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_recv_timeout_is_indeterminate() { + // The drain's own recv-timeout (second line of defense) surfaces its + // message when the send side ended cleanly. + let verdict = classify_send_verdict( + Ok(Ok(())), + false, + Some(( + PaymentFailureKind::Indeterminate, + "payment status stream timed out after 120s without a terminal update".to_string(), + )), + ); + assert_eq!( + verdict, + SendVerdict::Failure( + PaymentFailureKind::Indeterminate, + "payment status stream timed out after 120s without a terminal update".to_string() + ) + ); + } + + #[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 From 47ea73cfd1261626bb894487d0a4fcbd7bb38c4f Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 13:53:39 -0600 Subject: [PATCH 08/10] refactor(lightning): derive the payout timeout from LND's route timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move PAYOUT_SEND_PAYMENT_TIMEOUT next to send_payment in the lightning layer — the 60s window it must exceed lives there — and derive it as the named LND route-attempt timeout plus a 15s margin instead of a free-standing 75. Raising LND's timeout_seconds now automatically raises the payout bound, removing the silent coupling between the two values, and the bond module stops reaching into app::release for a lightning-layer constant. No behavior change. --- src/app/bond/payout.rs | 6 +++--- src/app/release.rs | 14 +------------- src/lightning/mod.rs | 19 ++++++++++++++++++- 3 files changed, 22 insertions(+), 17 deletions(-) diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index 64e3f32f..28ee0a70 100644 --- a/src/app/bond/payout.rs +++ b/src/app/bond/payout.rs @@ -683,7 +683,7 @@ async fn pay_counterparty( let (tx, mut rx) = channel(100); let send_fut = timeout( - crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT, + crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT, ln_client.send_payment(invoice, counterparty_share, tx), ); @@ -794,7 +794,7 @@ fn classify_send_verdict( PaymentFailureKind::Indeterminate, format!( "send_payment reached no terminal state after {}s", - crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() ), ), Ok(Err(e)) => { @@ -2491,7 +2491,7 @@ mod tests { PaymentFailureKind::Indeterminate, format!( "send_payment reached no terminal state after {}s", - crate::app::release::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() ) ) ); diff --git a/src/app/release.rs b/src/app/release.rs index 9ace5841..33837f2e 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, PaymentMessage}; +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::{ @@ -22,23 +22,11 @@ use nostr_sdk::prelude::*; use sqlx::{Pool, Sqlite}; use std::cmp::Ordering; use std::str::FromStr; -use std::time::Duration; use tokio::sync::mpsc::channel; use tokio::sync::Semaphore; use tokio::time::timeout; use tracing::{info, warn}; -/// Upper bound on how long the background payout task waits for -/// `send_payment` to reach a terminal state. LND itself stops launching new -/// route attempts after `timeout_seconds: 60`, so 75s leaves margin for LND to -/// exhaust its own attempts; past that, the stream is only kept open by an -/// HTLC that is locked-in but unresolved (hold invoice or stuck route), which -/// the sender cannot cancel anyway. Hitting this timeout does NOT fail the -/// payout: the payment may still settle in LND, so the claim marker is kept -/// and `reconcile_inflight_payout` resolves the outcome by payment hash. -/// Fixed for now; could become a settings knob later. -pub(crate) const PAYOUT_SEND_PAYMENT_TIMEOUT: Duration = Duration::from_secs(75); - /// 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 diff --git a/src/lightning/mod.rs b/src/lightning/mod.rs index 9c6ab116..c030c409 100644 --- a/src/lightning/mod.rs +++ b/src/lightning/mod.rs @@ -18,9 +18,26 @@ 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 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); + #[derive(Clone)] pub struct LndConnector { pub client: Client, @@ -302,7 +319,7 @@ impl LndConnector { 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() }; From 5b8b4a97d6f13f84904848bb8da9fdfa25c91a37 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 15:52:11 -0600 Subject: [PATCH 09/10] fix: revalidate the payout claim after the dispatch queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dispatch semaphore put an unbounded wait between the idempotency claim and the send, so a task queued past the reconcile grace window could lose its claim to re-arm — and the re-armed payout may use a fresh invoice, whose different payment hash escapes both the pre-send duplicate guard and LND's duplicate rejection: two invoices could settle against one escrow. Close it with touch_order_payout_claim, a single CAS that re-validates claim ownership after the permit AND refreshes payout_claimed_at. A re-armed or replaced claim makes the queued task drop the send; a still-owned claim restarts the reconcile grace clock, so the claim is once again never older than its send by more than the send itself — and a reconciler holding a pre-touch snapshot loses its scoped release CAS. On a DB error the task also drops the send: the kept marker is recoverable by reconciliation, a blind send is not. The watcher and all later releases scope to the refreshed token. Covered by five DB-level tests, including the replacement-claim scenario and the pre-touch reconciler snapshot. --- src/app/release.rs | 53 ++++++++++++-- src/db.rs | 177 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 223 insertions(+), 7 deletions(-) diff --git a/src/app/release.rs b/src/app/release.rs index 33837f2e..87abc391 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -31,11 +31,14 @@ use tracing::{info, warn}; /// 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. A task queued -/// past the reconcile grace window can lose its claim to re-arm-and-redispatch; -/// that is safe: the pre-send duplicate guard in `send_payment` (keyed on the -/// real payment hash) and LND's own duplicate rejection stop the late sender -/// from double-paying. Fixed for now; could become a settings knob later. +/// 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 @@ -649,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))?; @@ -686,6 +691,40 @@ pub async fn do_payment( // 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 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, From 49190c499d0f0cb2e2117f310770bb6ec438e3d6 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:04:15 -0600 Subject: [PATCH 10/10] fix: address the should-fix review notes on the payout paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - bond: the drain now owns rx and drops it once it has a terminal verdict, so a send still pushing updates fails its next listener.send and returns immediately instead of riding out the 75s bound for a payment whose outcome is already known. PAYMENT_STATUS_RECV_TIMEOUT is removed outright — behind the 75s send bound it could never fire — and the test that exercised that unreachable branch is replaced by one covering the new shape (a stream-delivered Succeeded wins over the send error caused by dropping rx). - lightning: the duplicate-guard lookup is bounded at 2s at its call site, so this advisory check can never eat a caller's budget (dev_fee wraps send_payment in 5s total). On timeout the guard degrades to proceed — the direction already documented as safe, since LND itself rejects a genuine duplicate for an in-flight or settled hash. - release: the Ok(Ok(())) comment no longer claims the stream reached a terminal state — send_payment's while let Ok(Some(..)) swallows a mid-stream gRPC error, so that branch also covers a dead/EOF'd stream; the behavior was already safe (marker kept, reconciliation resolves by hash), only the comment was wrong. --- src/app/bond/payout.rs | 107 ++++++++++++++--------------------------- src/app/release.rs | 9 +++- src/lightning/mod.rs | 30 +++++++++--- 3 files changed, 67 insertions(+), 79 deletions(-) diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index 28ee0a70..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 @@ -688,53 +679,40 @@ async fn pay_counterparty( ); // Collect the first terminal status from the stream. Mirrors - // dev_fee::send_dev_fee_payment; each recv stays bounded by - // `PAYMENT_STATUS_RECV_TIMEOUT` as a second line of defense, although the - // send-side timeout above already bounds the whole exchange. 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 drain_fut = async { + // 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; - 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; - } - _ => {} - } + 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) }; @@ -802,8 +780,8 @@ fn classify_send_verdict( // not partially enter LND. (PaymentFailureKind::Indeterminate, format!("{e}")) } - // EOF (or recv timeout) with no terminal status: the stream closed - // without telling us the outcome. + // 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(), @@ -2513,24 +2491,13 @@ mod tests { } #[tokio::test] - async fn classify_stream_recv_timeout_is_indeterminate() { - // The drain's own recv-timeout (second line of defense) surfaces its - // message when the send side ended cleanly. - let verdict = classify_send_verdict( - Ok(Ok(())), - false, - Some(( - PaymentFailureKind::Indeterminate, - "payment status stream timed out after 120s without a terminal update".to_string(), - )), - ); - assert_eq!( - verdict, - SendVerdict::Failure( - PaymentFailureKind::Indeterminate, - "payment status stream timed out after 120s without a terminal update".to_string() - ) - ); + 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] diff --git a/src/app/release.rs b/src/app/release.rs index 87abc391..272ebd60 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -809,8 +809,13 @@ pub async fn do_payment( ) .await { - // The stream reached a terminal state; the watcher drains the - // remaining updates and finishes the bookkeeping. + // 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); diff --git a/src/lightning/mod.rs b/src/lightning/mod.rs index c030c409..abe2490d 100644 --- a/src/lightning/mod.rs +++ b/src/lightning/mod.rs @@ -20,6 +20,7 @@ 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 @@ -38,6 +39,14 @@ pub(crate) const LND_PAYMENT_ROUTE_TIMEOUT_SECS: i32 = 60; 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, @@ -299,13 +308,20 @@ impl LndConnector { // 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 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. - match self.lookup_payment_status(&payment_hash).await { - Ok(Some(payment::PaymentStatus::InFlight)) - | Ok(Some(payment::PaymentStatus::Succeeded)) => { + // 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