From 6cf06b47beef42235d97e5915ec50b492285355e Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 22:46:12 -0600 Subject: [PATCH 1/5] fix: heartbeat the payout claim while queued for a send permit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit touch_order_payout_claim closed the double-payout window the dispatch queue opened, but not the spurious-failure one: a task queued past the reconcile grace window still lost its claim to re-arm, burning the buyer's retry budget — and eventually re-prompting for a fresh invoice — for a payout that was never sent. Reachable on stock settings whenever more than 8 payouts ride out their 75s bound at once, which is precisely the stuck-payout scenario this code exists to survive. Keep the claim younger than grace instead: while waiting for a permit, re-validate and re-stamp it every PAYOUT_QUEUE_HEARTBEAT (derived as 2/3 of the reconciler's MIN_GRACE_SECS, now promoted to a module constant so the relation is structural). The acquire future is pinned outside the select! loop so the task keeps its FIFO position in the semaphore queue across heartbeats. A lost claim or a DB error drops the dispatch — the same safe directions the post-permit touch already uses — and that final touch stays as the gate, since the last heartbeat can be a full cadence old. Also narrows the semaphore/permit docs: the bound covers concurrent payment streams (not LND connections, a deliberate fail-fast-before- claim trade) and the permit spans the send and RPC-error reconcile (the watcher is a sibling task). --- src/app/release.rs | 118 ++++++++++++++++++++++++++++++++++----------- src/scheduler.rs | 15 ++++-- 2 files changed, 101 insertions(+), 32 deletions(-) diff --git a/src/app/release.rs b/src/app/release.rs index 272ebd60..5876286b 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -22,6 +22,7 @@ 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; @@ -29,18 +30,36 @@ 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. +/// failed payouts into N background tasks; this semaphore makes the sends +/// queue instead of fanning out. It bounds concurrent *payment streams*, NOT +/// LND connections: `LndConnector::new()` runs before the claim (a connect +/// blip must never leave a marker set), so a backlog still opens N +/// connections that sit idle while queued — a deliberate trade for the +/// fail-fast-before-claim property. +/// +/// The queue puts an unbounded wait between the claim and the send, with two +/// hazards, both closed by `touch_order_payout_claim`: +/// - **double payout**: reconciliation re-arms the claim and the buyer +/// supplies a fresh invoice under a *different* hash — a case neither the +/// pre-send duplicate guard nor LND's duplicate rejection can catch. +/// Closed by the revalidating touch right after the permit: a task whose +/// claim was re-armed or replaced while it queued drops its send. +/// - **spurious failure**: a queued-but-never-sent payout ages past the +/// reconcile grace window and is re-armed as failed, burning the buyer's +/// retry budget and eventually re-prompting for an invoice that was never +/// needed. Closed by the heartbeat touch while waiting (see +/// [`PAYOUT_QUEUE_HEARTBEAT`]), which keeps the claim younger than grace. +/// +/// Fixed for now; could become a settings knob later. static PAYOUT_DISPATCH_SEMAPHORE: Semaphore = Semaphore::const_new(8); +/// Refresh cadence for a payout claim while its task waits for a send +/// permit. Derived from — and strictly below — the reconciler's minimum +/// grace window, so a queued claim is always re-stamped before it becomes +/// eligible for reconciliation. +const PAYOUT_QUEUE_HEARTBEAT: Duration = + Duration::from_secs(crate::scheduler::MIN_GRACE_SECS as u64 * 2 / 3); + /// 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) { @@ -684,23 +703,68 @@ 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; - - // 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( + // The claim token; refreshed by every successful touch below. + let mut payout_claimed_at = payout_claimed_at; + + // Bound concurrent sends (see PAYOUT_DISPATCH_SEMAPHORE). While + // queued, heartbeat the claim: re-validate and re-stamp it every + // PAYOUT_QUEUE_HEARTBEAT so it never ages past the reconcile grace + // window — without this, reconciliation would treat a queued-but- + // never-sent payout as failed, burning the buyer's retry budget for + // a payment that was never attempted. The acquire future is pinned + // OUTSIDE the loop so the task keeps its FIFO position in the + // semaphore queue across heartbeats. 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 then held for the send and the + // RPC-error reconcile via RAII — the watcher is a sibling task and + // finishes its bookkeeping outside the bound. + let acquire = PAYOUT_DISPATCH_SEMAPHORE.acquire(); + tokio::pin!(acquire); + let _permit = loop { + tokio::select! { + permit = &mut acquire => break permit, + _ = tokio::time::sleep(PAYOUT_QUEUE_HEARTBEAT) => { + match crate::db::touch_order_payout_claim( + ctx.pool(), + order.id, + &payout_hash, + Some(payout_claimed_at), + ) + .await + { + Ok(Some(refreshed_at)) => payout_claimed_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 heartbeat payout claim while queued ({e}); dropping dispatch of hash {} — reconciliation will resolve the kept marker", + order.id, payout_hash + ); + return; + } + } + } + } + }; + + // Final re-validation now that the queue wait is over, refreshing + // the timestamp in the same CAS (the last heartbeat may be up to a + // full PAYOUT_QUEUE_HEARTBEAT old). 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. + payout_claimed_at = match crate::db::touch_order_payout_claim( ctx.pool(), order.id, &payout_hash, diff --git a/src/scheduler.rs b/src/scheduler.rs index fcbb4d59..46df134a 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -244,6 +244,16 @@ async fn job_retry_failed_payments(ctx: AppContext) { }); } +/// Floor of the payout-reconcile grace window, in seconds. +/// +/// `LightningSettings::default()` has `payment_retries_interval = 0`, and a +/// 1s grace would be narrower than the claim→register window it guards, +/// re-opening the reconcile-vs-dispatch race the grace exists to close. +/// `release::PAYOUT_QUEUE_HEARTBEAT` is derived from this so a claim queued +/// for a send permit is always re-stamped before it becomes eligible for +/// reconciliation — keep that relation in mind before lowering it. +pub(crate) const MIN_GRACE_SECS: u32 = 30; + /// Reconcile buyer payouts left in flight (`payout_payment_hash` set) against /// LND. Complements `job_retry_failed_payments`: that job only dispatches fresh /// payouts (marker NULL), while this one resolves the durable marker so a @@ -260,11 +270,6 @@ async fn job_reconcile_inflight_payouts(ctx: AppContext) { // reconciled yet, so LND has surely registered it before we ever act on a // `None`/`Failed` lookup. Tied to the retry cadence — comfortably larger // than the sub-second claim→register window. - // - // Floor it: `LightningSettings::default()` has payment_retries_interval = 0, - // and a 1s grace would be narrower than the claim→register window it guards, - // re-opening the reconcile-vs-dispatch race the grace exists to close. - const MIN_GRACE_SECS: u32 = 30; let grace_secs = ctx .settings() .lightning From 77d4a544a35ed43073c801cd8588e61a16b29df1 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 22:49:46 -0600 Subject: [PATCH 2/5] fix(lightning): log the duplicate guard's degraded paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The guard's catch-all silently absorbed the two cases worth knowing about: a lookup timeout and a lookup transport error both fell through to proceed with no trace, so an LND whose track_payment_v2 consistently exceeds the 2s bound would leave the guard permanently disabled and nothing in the logs would say so — the abort path logged, the degraded path did not. Give both arms an info! line; the remaining catch-all now covers only Failed/Unknown/no-record, the normal go-ahead. --- src/lightning/mod.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/src/lightning/mod.rs b/src/lightning/mod.rs index abe2490d..1860b85f 100644 --- a/src/lightning/mod.rs +++ b/src/lightning/mod.rs @@ -330,6 +330,19 @@ impl LndConnector { "payment already dispatched for this hash".to_string(), ))); } + // The degraded paths must not be silent: an LND whose + // track_payment_v2 consistently exceeds the bound would leave + // this guard permanently disabled with no trace in the logs. + Err(_) => info!( + "Duplicate guard lookup for hash {} timed out after {}s; proceeding (LND rejects real duplicates)", + hash, + DUPLICATE_GUARD_LOOKUP_TIMEOUT.as_secs() + ), + Ok(Err(e)) => info!( + "Duplicate guard lookup for hash {} failed ({e}); proceeding", + hash + ), + // Failed / Unknown / no record: the normal go-ahead. _ => {} } From 7e7e351fb9d31d4f42536c14397cb842019d9d04 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:10:45 -0600 Subject: [PATCH 3/5] refactor(bond): collapse the drain result into StreamOutcome MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removing PAYMENT_STATUS_RECV_TIMEOUT left classify_send_verdict admitting states its only caller could no longer produce: the drain yields exactly None or Some((Terminal, _)), so the Indeterminate stream-failure arm and its unwrap_or were dead code, and the (succeeded, Option) pair could still encode the impossible (true, Some(Terminal)) pairing the function had to pick a winner for. Replace the pair with StreamOutcome { Succeeded, Failed(String), Ended } — precisely what the drain observes — and classify over that. The verdict logic is unchanged; the six tests port 1:1 and their matrix is now total over the real input domain instead of total over the reachable subset --- src/app/bond/payout.rs | 106 ++++++++++++++++++++++------------------- 1 file changed, 56 insertions(+), 50 deletions(-) diff --git a/src/app/bond/payout.rs b/src/app/bond/payout.rs index 4d7b130f..7f4824c3 100644 --- a/src/app/bond/payout.rs +++ b/src/app/bond/payout.rs @@ -688,19 +688,18 @@ async fn pay_counterparty( // `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; + let mut outcome = StreamOutcome::Ended; while let Some(msg) = rx.recv().await { if let Ok(status) = PaymentStatus::try_from(msg.payment.status) { match status { PaymentStatus::Succeeded => { - succeeded = true; + outcome = StreamOutcome::Succeeded; break; } PaymentStatus::Failed => { - failure = Some(( - PaymentFailureKind::Terminal, - format!("payment failed: reason {}", msg.payment.failure_reason), + outcome = StreamOutcome::Failed(format!( + "payment failed: reason {}", + msg.payment.failure_reason )); break; } @@ -713,12 +712,12 @@ async fn pay_counterparty( // send future returns immediately instead of riding out the 75s // bound for a payment whose verdict we already hold. drop(rx); - (succeeded, failure) + outcome }; - let (send_outcome, (succeeded, stream_failure)) = tokio::join!(send_fut, drain_fut); + let (send_outcome, stream) = tokio::join!(send_fut, drain_fut); - match classify_send_verdict(send_outcome, succeeded, stream_failure) { + match classify_send_verdict(send_outcome, stream) { 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 @@ -726,6 +725,21 @@ async fn pay_counterparty( } } +/// What the status drain actually observed — exactly the three states its +/// loop can produce, so `classify_send_verdict`'s input domain has no +/// unrepresentable-but-typeable values (no `(succeeded, Some(failure))` +/// pairing, no indeterminate stream failure that nothing emits). +#[derive(Debug, PartialEq)] +enum StreamOutcome { + /// A `PaymentStatus::Succeeded` update was delivered. + Succeeded, + /// A `PaymentStatus::Failed` update was delivered, with its reason. + Failed(String), + /// The channel closed (send returned, errored, or was dropped by the + /// timeout) without a terminal update. + Ended, +} + /// Combined verdict of a bounded `send_payment` and its concurrent status /// drain (see `pay_counterparty`). #[derive(Debug, PartialEq)] @@ -755,39 +769,35 @@ enum SendVerdict { /// 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)>, + stream: StreamOutcome, ) -> 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, - 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}")) + match stream { + StreamOutcome::Succeeded => SendVerdict::Settled, + StreamOutcome::Failed(msg) => SendVerdict::Failure(PaymentFailureKind::Terminal, msg), + StreamOutcome::Ended => { + 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}")) + } + // Clean EOF with no terminal status: the stream closed + // without telling us the outcome. + Ok(Ok(())) => ( + PaymentFailureKind::Indeterminate, + "payment stream ended without terminal status".to_string(), + ), + }; + SendVerdict::Failure(kind, msg) } - // 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. @@ -2354,7 +2364,7 @@ mod tests { // 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); + let verdict = classify_send_verdict(Err(elapsed().await), StreamOutcome::Succeeded); assert_eq!(verdict, SendVerdict::Settled); } @@ -2362,11 +2372,7 @@ mod tests { 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(), - )), + StreamOutcome::Failed("payment failed: reason 1".to_string()), ); assert_eq!( verdict, @@ -2382,7 +2388,7 @@ mod tests { // 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); + let verdict = classify_send_verdict(Err(elapsed().await), StreamOutcome::Ended); assert_eq!( verdict, SendVerdict::Failure( @@ -2398,7 +2404,7 @@ mod tests { #[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); + let verdict = classify_send_verdict(Ok(Err(rpc_err)), StreamOutcome::Ended); match verdict { SendVerdict::Failure(PaymentFailureKind::Indeterminate, msg) => { assert!( @@ -2416,13 +2422,13 @@ mod tests { // 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); + let verdict = classify_send_verdict(Ok(Err(send_err)), StreamOutcome::Succeeded); assert_eq!(verdict, SendVerdict::Settled); } #[tokio::test] async fn classify_stream_eof_is_indeterminate() { - let verdict = classify_send_verdict(Ok(Ok(())), false, None); + let verdict = classify_send_verdict(Ok(Ok(())), StreamOutcome::Ended); assert_eq!( verdict, SendVerdict::Failure( From da1ae64774e833fa24de25399b7bf7656719a35c Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:19:43 -0600 Subject: [PATCH 4/5] test: put the dispatch task's claim decision under test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dispatch path had no coverage above the DB layer: the do_payment tests stop at pre-claim failures, so the timeout-keeps-the-marker invariant — the central safety argument of the background dispatch — lived only in a comment. Apply the same move classify_send_verdict got on the bond side: extract the post-send claim decision into a pure classify_dispatch(send_outcome, lookup) returning StreamEnded / KeepMarker / ReArm, with the LND status lookup as a parameter. Nine tests cover the full matrix: a timed-out send always keeps the claim; an RPC-level failure keeps it when LND reports the payment in flight or settled (or the lookup itself fails), and re-arms retry when LND reports failed/unknown/no record or the hash was unusable. The caller keeps the side effects (scoped fail_order_payout + retry bookkeeping); the warn causes now travel inside the verdict, a minor log reshuffle with no behavior change. --- src/app/release.rs | 267 +++++++++++++++++++++++++++++++++++---------- 1 file changed, 209 insertions(+), 58 deletions(-) diff --git a/src/app/release.rs b/src/app/release.rs index 5876286b..49711f55 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -867,73 +867,57 @@ pub async fn do_payment( }; tokio::spawn(watcher); - match timeout( + let send_outcome = timeout( PAYOUT_SEND_PAYMENT_TIMEOUT, ln_client_payment.send_payment(&payment_request, amount as i64, tx), ) - .await - { + .await; + + // The status lookup is only meaningful after an RPC-level send + // failure: it is what decides between keeping the marker and + // re-arming. `None` alongside an Ok(Err) send means the hash was + // unusable (should not happen — it was built from the invoice). + let lookup = match &send_outcome { + Ok(Err(_)) => match Vec::::from_hex(&payout_hash) { + Ok(bytes) if bytes.len() == 32 => { + Some(ln_client_payment.lookup_payment_status(&bytes).await) + } + _ => None, + }, + _ => None, + }; + + match classify_dispatch(send_outcome, lookup) { // 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; - } + // this 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. + DispatchVerdict::StreamEnded => {} + DispatchVerdict::KeepMarker(cause) => { + warn!( + "Order Id {}: keeping payout claim for hash {}: {cause}", + order.id, payout_hash + ); } - 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. + DispatchVerdict::ReArm(cause) => { 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() + "Order Id {}: payout dispatch failed ({cause}); re-arming retry for hash {}", + order.id, payout_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; + } } } }); @@ -941,6 +925,64 @@ pub async fn do_payment( Ok(()) } +/// What the dispatch task must do with its claim once the bounded send has +/// ended (see `do_payment`). +#[derive(Debug, PartialEq)] +enum DispatchVerdict { + /// The send stream ended; the watcher owns any bookkeeping. Nothing to + /// do with the claim here. + StreamEnded, + /// KEEP the claim marker (with this cause): the payment may still + /// settle, so reconciliation owns the outcome and no second payout is + /// ever dispatched. + KeepMarker(String), + /// Release the claim (scoped to hash + token) and re-arm retry now, with + /// this cause: LND confirms nothing is or will be in flight for it. + ReArm(String), +} + +/// Classify the outcome of the bounded `send_payment` into what happens to +/// the payout claim. Pure — the LND status lookup is a parameter — so the +/// central safety invariant of the background dispatch ("a timed-out send +/// keeps the claim") is under test rather than under a comment. +/// +/// - A timed-out send KEEPS the marker: dropping the send 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). Re-arming against it risks a double payout; kept, the +/// payout is delayed by at most the reconciler cadence, never lost. +/// - An RPC-level send failure resolves the claim by what LND reports for +/// the hash: in flight / settled / lookup error → KEEP (the payment may +/// still settle); failed / unknown / no record / unusable hash → re-arm +/// retry now (and notify the buyer) instead of waiting for the +/// grace-delayed reconciliation job. +fn classify_dispatch( + send_outcome: Result, tokio::time::error::Elapsed>, + lookup: Option, MostroError>>, +) -> DispatchVerdict { + match send_outcome { + Ok(Ok(())) => DispatchVerdict::StreamEnded, + Err(_) => DispatchVerdict::KeepMarker(format!( + "no terminal state after {}s; a locked-in HTLC cannot be cancelled by the sender and may still settle — reconciliation will resolve it", + PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + )), + Ok(Err(send_err)) => match lookup { + Some(Ok(Some(PaymentStatus::InFlight))) | Some(Ok(Some(PaymentStatus::Succeeded))) => { + DispatchVerdict::KeepMarker(format!( + "send errored ({send_err}) but LND reports the payment in flight or settled" + )) + } + Some(Err(lookup_err)) => DispatchVerdict::KeepMarker(format!( + "send errored ({send_err}) and the status lookup failed ({lookup_err}); the payment may still settle" + )), + // Failed / Unknown / no record — or an unusable hash (None), + // which cannot confirm an in-flight payment: nothing to wait + // for. + Some(Ok(_)) | None => DispatchVerdict::ReArm(format!("{send_err}")), + }, + } +} + /// Finalize a paid order: transition `settled-hold-invoice` → `Success` and /// notify the buyer, but only after the status CAS actually commits. /// @@ -2325,6 +2367,115 @@ mod tests { server.abort(); } + /// 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(Duration::ZERO, std::future::pending::<()>()) + .await + .unwrap_err() + } + + fn send_err() -> MostroError { + MostroInternalErr(ServiceError::LnPaymentError("boom".to_string())) + } + + #[tokio::test] + async fn dispatch_stream_end_needs_no_claim_action() { + assert_eq!( + classify_dispatch(Ok(Ok(())), None), + DispatchVerdict::StreamEnded + ); + } + + #[tokio::test] + async fn dispatch_timeout_keeps_the_marker() { + // The central safety invariant of the background dispatch: a + // timed-out send must NEVER re-arm retry — the HTLC may still + // settle, and re-arming against it risks a double payout. + match classify_dispatch(Err(elapsed().await), None) { + DispatchVerdict::KeepMarker(cause) => assert!( + cause.contains(&format!( + "no terminal state after {}s", + PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs() + )), + "cause must name the timeout: {cause}" + ), + other => panic!("a timed-out send must keep the marker, got {other:?}"), + } + } + + #[tokio::test] + async fn dispatch_rpc_error_with_inflight_payment_keeps_the_marker() { + match classify_dispatch(Ok(Err(send_err())), Some(Ok(Some(PaymentStatus::InFlight)))) { + DispatchVerdict::KeepMarker(cause) => { + assert!(cause.contains("boom"), "cause must carry the send error") + } + other => panic!("an in-flight payment must keep the marker, got {other:?}"), + } + } + + #[tokio::test] + async fn dispatch_rpc_error_with_settled_payment_keeps_the_marker() { + assert!(matches!( + classify_dispatch( + Ok(Err(send_err())), + Some(Ok(Some(PaymentStatus::Succeeded))) + ), + DispatchVerdict::KeepMarker(_) + )); + } + + #[tokio::test] + async fn dispatch_rpc_error_with_failed_lookup_keeps_the_marker() { + // An unanswerable lookup cannot rule out an in-flight payment, so + // the conservative direction is to keep the claim. + assert!(matches!( + classify_dispatch( + Ok(Err(send_err())), + Some(Err(MostroInternalErr(ServiceError::LnPaymentError( + "lookup down".to_string() + )))) + ), + DispatchVerdict::KeepMarker(_) + )); + } + + #[tokio::test] + async fn dispatch_rpc_error_with_failed_payment_rearms() { + match classify_dispatch(Ok(Err(send_err())), Some(Ok(Some(PaymentStatus::Failed)))) { + DispatchVerdict::ReArm(cause) => { + assert!(cause.contains("boom"), "cause must carry the send error") + } + other => panic!("a failed payment must re-arm retry, got {other:?}"), + } + } + + #[tokio::test] + async fn dispatch_rpc_error_with_unknown_payment_rearms() { + assert!(matches!( + classify_dispatch(Ok(Err(send_err())), Some(Ok(Some(PaymentStatus::Unknown)))), + DispatchVerdict::ReArm(_) + )); + } + + #[tokio::test] + async fn dispatch_rpc_error_with_no_lnd_record_rearms() { + assert!(matches!( + classify_dispatch(Ok(Err(send_err())), Some(Ok(None))), + DispatchVerdict::ReArm(_) + )); + } + + #[tokio::test] + async fn dispatch_rpc_error_with_unusable_hash_rearms() { + // No lookup was possible (hash undecodable): an in-flight payment + // cannot be confirmed, so re-arm rather than strand the payout. + assert!(matches!( + classify_dispatch(Ok(Err(send_err())), None), + DispatchVerdict::ReArm(_) + )); + } + #[tokio::test] async fn payment_success_transitions_settled_order_to_success() { // Arrange From 848238394f145c71f72a4208f6c4f0feba24ef56 Mon Sep 17 00:00:00 2001 From: Catrya <140891948+Catrya@users.noreply.github.com> Date: Mon, 17 Aug 2026 23:42:42 -0600 Subject: [PATCH 5/5] docs(db): document the wall-clock caveats of the payout claim token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The docs presented payout_claimed_at as an identity token, but it has 1-second resolution (a same-second collision is harmless: the CAS also pins the hash, and LND rejects duplicate sends) and is vulnerable to backward clock steps (bounded in practice by the queue heartbeat). Name both caveats and record the clean fix — a monotonic per-order sequence — as deliberate future work. --- src/db.rs | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/db.rs b/src/db.rs index 6f3a166e..c9af0104 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1209,6 +1209,25 @@ pub async fn find_failed_payment(pool: &SqlitePool) -> Result, Mostro /// caller from the previous attempt still loses the release CAS. It also lets /// reconciliation ignore a just-claimed payout until LND has surely registered /// it (grace window). +/// +/// # Token caveats (wall-clock, not monotonic) +/// +/// The token is `Utc::now().timestamp()`, which makes it *nearly* an +/// identity, not exactly one: +/// - **1-second resolution.** A claim released and re-claimed within the same +/// second yields an identical token, so a stale caller's release or touch +/// can match the newer claim. Not exploitable: every claim CAS also matches +/// on `payout_payment_hash`, so a same-token collision implies the same +/// invoice, and LND's duplicate rejection covers a re-send of it. +/// - **Backward clock steps.** `find_inflight_payouts` filters on +/// `payout_claimed_at <= now - grace`, so an NTP step backwards can make a +/// just-stamped claim immediately reconcilable. In practice the dispatch +/// task's queue heartbeat re-stamps the claim well inside the grace window, +/// which bounds the exposure. +/// +/// The clean fix — a monotonic per-order sequence as the identity token, with +/// wall-clock kept only for the grace window — needs a migration and a change +/// to every claim CAS, and is deliberately left as future work. pub async fn claim_order_payout( pool: &SqlitePool, order_id: Uuid, @@ -1250,6 +1269,11 @@ pub async fn claim_order_payout( /// 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). +/// +/// The token shares [`claim_order_payout`]'s wall-clock caveats (1-second +/// resolution, backward clock steps) — see the "Token caveats" section there; +/// the touch leans on the token harder than the claim does, so keep them in +/// mind before treating it as a strict identity. pub async fn touch_order_payout_claim( pool: &SqlitePool, order_id: Uuid,