Skip to content

fix: non-blocking buyer payout with bounded send_payment waits #883

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
grunch merged 10 commits into from
Aug 17, 2026
Merged

fix: non-blocking buyer payout with bounded send_payment waits #883

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
71ce09f
fix: resolve buyer pubkey before claiming the payout
Catrya Aug 17, 2026
efa3d06
fix: dispatch buyer payout off the event loop with a bounded send_pay…
Catrya Aug 17, 2026
a95ede4
fix: bound bond payout send_payment with the payout timeout
Catrya Aug 17, 2026
d74c111
fix(lightning): key the duplicate-payment guard on the real payment hash
Catrya Aug 17, 2026
9883aad
fix(bond): drain the payment status stream concurrently with send_pay…
Catrya Aug 17, 2026
bc1e1f9
fix: bound concurrent buyer-payout dispatch with a semaphore
Catrya Aug 17, 2026
05a1665
test(bond): cover the payout send-outcome classification
Catrya Aug 17, 2026
47ea73c
refactor(lightning): derive the payout timeout from LND's route timeout
Catrya Aug 17, 2026
5b8b4a9
fix: revalidate the payout claim after the dispatch queue
Catrya Aug 17, 2026
49190c4
fix: address the should-fix review notes on the payout paths
Catrya Aug 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
287 changes: 213 additions & 74 deletions src/app/bond/payout.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -664,82 +655,139 @@ async fn pay_counterparty(

// send_payment. The helper caps the fee via `routing_fee_cap_sats`,
// the same value persisted above as `payout_routing_fee_sats`.
// `send_payment` consumes LND's payment stream until a terminal state,
// which a locked-in but unresolved HTLC (hold invoice as the winner's
// payout invoice, or an HTLC stuck in route) can delay indefinitely, so
// it is bounded with the same timeout as the buyer payout; LND stops
// launching route attempts at 60s, so past 75s only an unresolved HTLC
// keeps the stream open.
//
// The bounded send and the status drain run CONCURRENTLY (`join!`):
// `send_payment` forwards every LND update through `tx` and blocks when
// the channel fills, so draining only after it returned could deadlock a
// chatty stream (>100 updates) until the timeout. Same watcher-before-send
// pattern as the buyer payout in `do_payment`, kept on this task (no
// spawn) because the scheduler job needs a single combined outcome. The
// drain always terminates: when the send future ends — normal return, RPC
// error, or dropped by the timeout — `tx` drops and `rx.recv()` yields
// `None`.
let (tx, mut rx) = channel(100);
Comment thread
grunch marked this conversation as resolved.
let send_outcome = ln_client
.send_payment(invoice, counterparty_share, tx)
.await;
if let Err(e) = send_outcome {
// The RPC call itself errored. We cannot be sure the payment did
// not partially enter LND, so treat it as indeterminate: keep
// the invoice + hash for reconciliation rather than risk a
// double payout by re-prompting.
return on_send_payment_failure(
pool,
bond,
max_retries,
claim_window_seconds,
PaymentFailureKind::Indeterminate,
&format!("{e}"),
)
.await;
}

let send_fut = timeout(
crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT,
ln_client.send_payment(invoice, counterparty_share, tx),
);

// Collect the first terminal status from the stream. Mirrors
// dev_fee::send_dev_fee_payment, but each recv is bounded by
// `PAYMENT_STATUS_RECV_TIMEOUT` so a wedged LND stream (no terminal
// update, no EOF, no InFlight churn) does not pin the scheduler
// task forever. We track *why* the stream ended: only an explicit
// `PaymentStatus::Failed` is terminal. A timeout or clean EOF leaves
// the payment outcome unknown (it may still be in flight), so it is
// routed as `Indeterminate` — `on_send_payment_failure` then keeps
// the invoice + hash for reconciliation instead of re-prompting.
let mut succeeded = false;
let mut failure: Option<(PaymentFailureKind, String)> = None;
loop {
match timeout(PAYMENT_STATUS_RECV_TIMEOUT, rx.recv()).await {
Err(_) => {
failure = Some((
PaymentFailureKind::Indeterminate,
format!(
"payment status stream timed out after {}s without a terminal update",
PAYMENT_STATUS_RECV_TIMEOUT.as_secs()
),
));
break;
}
Ok(None) => break,
Ok(Some(msg)) => {
if let Ok(status) = PaymentStatus::try_from(msg.payment.status) {
match status {
PaymentStatus::Succeeded => {
succeeded = true;
break;
}
PaymentStatus::Failed => {
failure = Some((
PaymentFailureKind::Terminal,
format!("payment failed: reason {}", msg.payment.failure_reason),
));
break;
}
_ => {}
// dev_fee::send_dev_fee_payment. The drain is bounded transitively by the
// send-side timeout above: when the send future ends — normal return, RPC
// error, or dropped at the 75s bound — `tx` drops and `recv()` yields
// `None`. We track *why* the stream ended: only an explicit
// `PaymentStatus::Failed` is terminal. A clean EOF leaves the payment
// outcome unknown (it may still be in flight), so it is routed as
// `Indeterminate` — `on_send_payment_failure` then keeps the invoice +
// hash for reconciliation instead of re-prompting.
let drain_fut = async move {
let mut succeeded = false;
let mut failure: Option<(PaymentFailureKind, String)> = None;
while let Some(msg) = rx.recv().await {
if let Ok(status) = PaymentStatus::try_from(msg.payment.status) {
match status {
PaymentStatus::Succeeded => {
succeeded = true;
break;
}
PaymentStatus::Failed => {
failure = Some((
PaymentFailureKind::Terminal,
format!("payment failed: reason {}", msg.payment.failure_reason),
));
break;
}
_ => {}
}
}
}
// Unblock a send that is still pushing updates into a channel we are
// done reading: dropping `rx` fails its next `listener.send`, so the
// send future returns immediately instead of riding out the 75s
// bound for a payment whose verdict we already hold.
drop(rx);
(succeeded, failure)
};

let (send_outcome, (succeeded, stream_failure)) = tokio::join!(send_fut, drain_fut);

@grunch grunch Aug 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

join! keeps waiting on the send future after the drain already saw Succeeded.

drain_fut breaks out of its loop on PaymentStatus::Succeeded, but rx lives in the enclosing scope and join! holds both futures alive until both complete — so breaking neither closes the channel nor ends the wait. send_payment is still listener.send(msg).await-ing every remaining LND update into a channel nobody drains any more, and the job pays up to the full 75s for a payment it already knows settled.

In practice LND closes the stream right after the terminal update and the 100-slot buffer absorbs the tail, so this is a latency/shape issue rather than a live deadlock — but it is the same "nobody is draining" shape this commit set out to remove, so it is worth closing properly. Moving rx into the drain and dropping it on the way out makes the send future finish immediately (listener.send errors → Ok(Err(..)), which classify_send_verdict already subordinates to succeeded):

let drain_fut = async move {          // take ownership of `rx`
    let mut succeeded = false;
    let mut failure: Option<(PaymentFailureKind, String)> = None;
    loop { /* unchanged */ }
    // Unblock a send that is still pushing updates into a channel we are
    // done reading.
    drop(rx);
    (succeeded, failure)
};

Separately: PAYMENT_STATUS_RECV_TIMEOUT (120s) is now unreachable. The send side is bounded at 75s, and once its future is dropped tx drops and rx.recv() returns None on the spot, so the recv timeout can never fire. The comment calls it "a second line of defense"; in this shape it is dead code. Either drop it below 75s so it can actually fire first, or say plainly that it is vestigial — classify_stream_recv_timeout_is_indeterminate currently tests a branch that production can no longer reach.


match classify_send_verdict(send_outcome, succeeded, stream_failure) {
SendVerdict::Settled => slash_after_success(pool, bond, counterparty_share).await,
SendVerdict::Failure(kind, msg) => {
on_send_payment_failure(pool, bond, max_retries, claim_window_seconds, kind, &msg).await
}
}
}

/// Combined verdict of a bounded `send_payment` and its concurrent status
/// drain (see `pay_counterparty`).
#[derive(Debug, PartialEq)]
enum SendVerdict {
/// The stream reported `Succeeded`: the payment settled — finalize the
/// slash.
Settled,
/// No settlement: route through `on_send_payment_failure` with this kind
/// and cause.
Failure(PaymentFailureKind, String),
}

/// Classify the joint outcome of the bounded send future and the status
/// drain into a single verdict.
///
/// A terminal verdict from the stream is the payment's actual outcome and
/// takes priority over however the send future ended: a `Succeeded`
/// delivered just before the timeout finalizes the slash immediately instead
/// of deferring to reconciliation, and an explicit `Failed` is safe to act
/// on regardless of the send-side result.
///
/// With no terminal verdict, classify by the most specific cause. Every such
/// branch is indeterminate — the payment may still settle, so the caller
/// keeps the invoice + hash for reconciliation and never re-prompts the
/// winner against a payment that may still succeed. (A locked-in HTLC cannot
/// be cancelled by the sender; dropping the send future on timeout closes
/// our side of the gRPC stream only.)
fn classify_send_verdict(
send_outcome: Result<Result<(), MostroError>, tokio::time::error::Elapsed>,
succeeded: bool,
stream_failure: Option<(PaymentFailureKind, String)>,
) -> SendVerdict {
if succeeded {
return slash_after_success(pool, bond, counterparty_share).await;
return SendVerdict::Settled;
}

// EOF with no terminal status (the `Ok(None)` break above) is also
// indeterminate: the stream closed without telling us the outcome.
let (kind, msg) = failure.unwrap_or((
PaymentFailureKind::Indeterminate,
"payment stream ended without terminal status".to_string(),
));
on_send_payment_failure(pool, bond, max_retries, claim_window_seconds, kind, &msg).await
let stream_failure = match stream_failure {

@grunch grunch Aug 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

classify_send_verdict admits a state its only caller can no longer produce.

49190c4 removed PAYMENT_STATUS_RECV_TIMEOUT, and with it the only producer of an Indeterminate stream failure. drain_fut can now yield exactly two things: None, or Some((Terminal, _)). The signature still takes the wider Option<(PaymentFailureKind, String)>, so:

let stream_failure = match stream_failure {
    Some((PaymentFailureKind::Terminal, msg)) => return SendVerdict::Failure(Terminal, msg),
    other => other,     // provably None from here on
};
// ...
Ok(Ok(())) => stream_failure.unwrap_or((Indeterminate, "payment stream ended…")),

Both the rebinding and the unwrap_or at L785 are dead: for every input the caller can construct, stream_failure is None by the time it is read. Some((Indeterminate, _)) is untestable and untested, and the next reader has to go two functions away and re-derive that from the drain body to know it.

The succeeded: bool + Option<…> pair has the same problem from the other direction — it can encode (true, Some(Terminal)), an impossible pairing the function nonetheless has to pick a winner for. Collapsing both into what the drain actually returns removes the guesswork:

enum StreamOutcome { Succeeded, Failed(String), Ended }

fn classify_send_verdict(
    send_outcome: Result<Result<(), MostroError>, Elapsed>,
    stream: StreamOutcome,
) -> SendVerdict

The six new tests port over almost unchanged, and the matrix they cover becomes total rather than "total over the reachable subset".

Some((PaymentFailureKind::Terminal, msg)) => {
return SendVerdict::Failure(PaymentFailureKind::Terminal, msg);
}
other => other,
};
let (kind, msg) = match send_outcome {
Err(_) => (
PaymentFailureKind::Indeterminate,
format!(
"send_payment reached no terminal state after {}s",
crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs()
),
),
Ok(Err(e)) => {
// The RPC call itself errored. We cannot be sure the payment did
// not partially enter LND.
(PaymentFailureKind::Indeterminate, format!("{e}"))
}
// EOF with no terminal status: the stream closed without telling us
// the outcome.
Ok(Ok(())) => stream_failure.unwrap_or((
PaymentFailureKind::Indeterminate,
"payment stream ended without terminal status".to_string(),
)),
};
SendVerdict::Failure(kind, msg)
}

/// Flip a `PendingPayout` row to `Slashed` after a confirmed payment.
Expand Down Expand Up @@ -2373,6 +2421,97 @@ mod tests {
assert_eq!(after.payout_payment_hash.as_deref(), Some("cafebabe"));
}

/// Produce a real `tokio::time::error::Elapsed` (it has no public
/// constructor): a zero-duration timeout over a pending future.
async fn elapsed() -> tokio::time::error::Elapsed {
timeout(std::time::Duration::ZERO, std::future::pending::<()>())
.await
.unwrap_err()
}

#[tokio::test]
async fn classify_stream_succeeded_wins_over_send_timeout() {
// A Succeeded delivered just before the 75s cutoff is the payment's
// real outcome: finalize the slash immediately instead of deferring
// to reconciliation, no matter how the send future ended.
let verdict = classify_send_verdict(Err(elapsed().await), true, None);
assert_eq!(verdict, SendVerdict::Settled);
}

#[tokio::test]
async fn classify_stream_terminal_failed_maps_to_terminal() {
let verdict = classify_send_verdict(
Ok(Ok(())),
false,
Some((
PaymentFailureKind::Terminal,
"payment failed: reason 1".to_string(),
)),
);
assert_eq!(
verdict,
SendVerdict::Failure(
PaymentFailureKind::Terminal,
"payment failed: reason 1".to_string()
)
);
}

#[tokio::test]
async fn classify_send_timeout_is_indeterminate() {
// The Elapsed branch: dropping the send future does not cancel a
// locked-in HTLC, so the verdict must be Indeterminate (keep the
// invoice + hash for reconciliation), never Terminal.
let verdict = classify_send_verdict(Err(elapsed().await), false, None);
assert_eq!(
verdict,
SendVerdict::Failure(
PaymentFailureKind::Indeterminate,
format!(
"send_payment reached no terminal state after {}s",
crate::lightning::PAYOUT_SEND_PAYMENT_TIMEOUT.as_secs()
)
)
);
}

#[tokio::test]
async fn classify_send_rpc_error_is_indeterminate() {
let rpc_err = MostroInternalErr(ServiceError::LnPaymentError("boom".to_string()));
let verdict = classify_send_verdict(Ok(Err(rpc_err)), false, None);
match verdict {
SendVerdict::Failure(PaymentFailureKind::Indeterminate, msg) => {
assert!(
msg.contains("boom"),
"cause must carry the RPC error: {msg}"
);
}
other => panic!("expected indeterminate failure, got {other:?}"),
}
}

#[tokio::test]
async fn classify_stream_succeeded_wins_over_send_error() {
// Once the drain sees Succeeded it drops `rx`, which fails the send's
// next `listener.send` — the resulting Ok(Err(..)) from the send
// future must not shadow the settled verdict.
let send_err = MostroInternalErr(ServiceError::LnNodeError("channel closed".to_string()));
let verdict = classify_send_verdict(Ok(Err(send_err)), true, None);
assert_eq!(verdict, SendVerdict::Settled);
}

#[tokio::test]
async fn classify_stream_eof_is_indeterminate() {
let verdict = classify_send_verdict(Ok(Ok(())), false, None);
assert_eq!(
verdict,
SendVerdict::Failure(
PaymentFailureKind::Indeterminate,
"payment stream ended without terminal status".to_string()
)
);
}

#[tokio::test]
async fn finalize_node_only_transitions_to_slashed() {
// `slash_node_share_pct = 1.0` style row: counterparty share is
Expand Down
Loading