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( diff --git a/src/app/release.rs b/src/app/release.rs index 272ebd60..49711f55 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, @@ -803,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; + } } } }); @@ -877,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. /// @@ -2261,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 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, 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. _ => {} } 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