diff --git a/migrations/20260815120000_order_payout_inflight.sql b/migrations/20260815120000_order_payout_inflight.sql new file mode 100644 index 00000000..cae4e50f --- /dev/null +++ b/migrations/20260815120000_order_payout_inflight.sql @@ -0,0 +1,11 @@ +-- Track the payment hash of an in-flight buyer payout so a +-- `settled-hold-invoice` order is paid at most once. +-- +-- While this column is non-NULL a payout for the order is considered in +-- flight: the failed-payment retry job skips the order and `AddInvoice` +-- swaps are rejected, so no second payout can be dispatched against the +-- same settled escrow. A reconciliation job resolves the marker against +-- LND (Succeeded -> finalize, Failed/Unknown -> clear & re-arm retry, +-- InFlight -> wait) so a stranded payout neither drains funds nor blocks +-- the order forever. +ALTER TABLE orders ADD COLUMN payout_payment_hash char(64); diff --git a/migrations/20260815120100_order_payout_claimed_at.sql b/migrations/20260815120100_order_payout_claimed_at.sql new file mode 100644 index 00000000..f0a9f5ec --- /dev/null +++ b/migrations/20260815120100_order_payout_claimed_at.sql @@ -0,0 +1,9 @@ +-- Timestamp (unix seconds) when a buyer payout was claimed, sealed together +-- with `payout_payment_hash` in the same CAS. +-- +-- Reconciliation ignores a marker younger than a grace window so a just-claimed +-- payout is never misread as "unknown to LND" during the brief window between +-- the claim and LND registering the payment. Without it, a reconciliation tick +-- landing in that window could clear the marker and let a second payout be +-- dispatched for the same escrow (a scriptable timing race). +ALTER TABLE orders ADD COLUMN payout_claimed_at integer; diff --git a/src/app/add_invoice.rs b/src/app/add_invoice.rs index 81bbcb17..9aa21c98 100644 --- a/src/app/add_invoice.rs +++ b/src/app/add_invoice.rs @@ -13,7 +13,11 @@ use sqlx::{Pool, Sqlite}; /// /// Uses a status-guarded targeted `UPDATE` (`WHERE status = settled-hold-invoice`) /// so a stale full-row write cannot resurrect payment state after -/// `payment_success` has already moved the order to `Success`. When the CAS +/// `payment_success` has already moved the order to `Success`. The +/// `payout_payment_hash IS NULL` guard additionally rejects the swap while a +/// prior payout for the order is still in flight: without it, swapping to a +/// fresh invoice and resetting `payment_attempts` re-arms a new payout on top of +/// an already-dispatched one (the invoice-swap re-arm drain). When the CAS /// misses, returns `CantDo(NotAllowedByStatus)` and does not enqueue /// `InvoiceUpdated`. pub async fn pay_new_invoice( @@ -22,7 +26,8 @@ pub async fn pay_new_invoice( msg: &Message, ) -> Result<(), MostroError> { let result = sqlx::query( - "UPDATE orders SET buyer_invoice = ?, payment_attempts = 0 WHERE id = ? AND status = ?", + "UPDATE orders SET buyer_invoice = ?, payment_attempts = 0 \ + WHERE id = ? AND status = ? AND payout_payment_hash IS NULL", ) .bind(&order.buyer_invoice) .bind(order.id) @@ -33,7 +38,7 @@ pub async fn pay_new_invoice( if result.rows_affected() == 0 { tracing::warn!( - "Ignoring stale buyer invoice update for order {}: row no longer in settled-hold-invoice", + "Ignoring buyer invoice update for order {}: not in settled-hold-invoice or a payout is already in flight", order.id ); return Err(MostroCantDo(CantDoReason::NotAllowedByStatus)); @@ -332,6 +337,48 @@ mod tests { .contains(&Action::InvoiceUpdated)); } + /// A swap must be rejected while a payout for the order is already in + /// flight (`payout_payment_hash` set): otherwise a fresh invoice would reset + /// `payment_attempts` on top of a pending payout and re-arm a second one. + #[tokio::test] + async fn pay_new_invoice_rejects_swap_while_payout_in_flight() { + let pool = setup_pool().await; + let seller = Keys::generate().public_key(); + let buyer = Keys::generate().public_key(); + + let mut order = waiting_invoice_sell_order(seller, buyer); + order.status = Status::SettledHoldInvoice.to_string(); + order.payment_attempts = 2; + order.buyer_invoice = Some("lnbc-current".to_string()); + let order = order.create(&pool).await.unwrap(); + + // A payout is already in flight for this order. + crate::db::claim_order_payout(&pool, order.id, &"a".repeat(64)) + .await + .unwrap(); + + let mut swap = order.clone(); + swap.buyer_invoice = Some("lnbc-new".to_string()); + let result = pay_new_invoice( + &mut swap, + &pool, + &Message::new_order(Some(order.id), Some(1), None, Action::AddInvoice, None), + ) + .await; + + assert!( + matches!(result, Err(MostroCantDo(CantDoReason::NotAllowedByStatus))), + "swap must be rejected while a payout is in flight: {result:?}" + ); + // Neither the invoice nor the attempts counter changed. + let stored = Order::by_id(&pool, order.id).await.unwrap().unwrap(); + assert_eq!(stored.buyer_invoice.as_deref(), Some("lnbc-current")); + assert_eq!(stored.payment_attempts, 2); + assert!(!queued_actions_for(buyer) + .await + .contains(&Action::InvoiceUpdated)); + } + /// A `SettledHoldInvoice` order routes through `pay_new_invoice`: the /// payment-attempts counter is reset and the buyer is told the invoice /// was updated. No LND is involved so the handler returns `Ok`. diff --git a/src/app/release.rs b/src/app/release.rs index 61c3a7c1..3b017d8e 100644 --- a/src/app/release.rs +++ b/src/app/release.rs @@ -7,10 +7,11 @@ use crate::lightning::LndConnector; use crate::lnurl::resolv_ln_address; use crate::nip33::{new_order_event_with_created_at, order_to_tags}; use crate::util::{ - enqueue_order_msg, get_order, mark_orderbook_publish_failed, monotonic_order_event_timestamp, - settle_seller_hold_invoice, update_order_event, + bytes_to_string, enqueue_order_msg, get_order, mark_orderbook_publish_failed, + monotonic_order_event_timestamp, settle_seller_hold_invoice, update_order_event, }; use crate::Result; +use bitcoin::hashes::hex::FromHex; use fedimint_tonic_lnd::lnrpc::payment::PaymentStatus; use lnurl::lightning_address::LightningAddress; @@ -607,15 +608,71 @@ pub async fn do_payment( }, None => payment_request, }; + + // 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. let mut ln_client_payment = LndConnector::new().await?; + + // Idempotency claim: persist the payout invoice's `payment_hash` (and the + // claim timestamp) immediately before dispatch. While the marker is set, + // `find_failed_payment` skips this order and `pay_new_invoice` rejects + // invoice swaps, so no second payout can be dispatched for the same settled + // escrow. This CAS also loses to a concurrent claim (two scheduler ticks + // 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. + let payout_hash = decode_invoice(&payment_request) + .map(|inv| bytes_to_string(inv.payment_hash().as_ref())) + .map_err(|_| MostroInternalErr(ServiceError::InvoiceInvalidError))?; + let Some(payout_claimed_at) = + crate::db::claim_order_payout(ctx.pool(), order.id, &payout_hash).await? + else { + warn!( + "Order {}: a payout is already in flight (or status changed); skipping duplicate send_payment", + order.id + ); + 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); - check_failure_retries_or_log(ctx, &order, request_id).await; - // Do not spawn the status watcher or report Ok: nothing was submitted - // to LND (or the attempt aborted before a usable status stream). + // `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); } @@ -640,14 +697,22 @@ pub async fn do_payment( "Order Id {}: Invoice with hash: {} paid!", order.id, msg.payment.payment_hash ); - let _ = payment_success( - &ctx, - &mut order, - buyer_pubkey, - &my_keys, - request_id, - ) - .await; + // 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( + ctx.pool(), + order.id, + &payout_hash, + Some(payout_claimed_at), + ) + .await; + } } PaymentStatus::Failed => { warn!( @@ -655,8 +720,24 @@ pub async fn do_payment( order.id, msg.payment.payment_hash ); - // Mark payment as failed - check_failure_retries_or_log(&ctx, &order, request_id).await; + // 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; + } } _ => {} } @@ -668,59 +749,216 @@ pub async fn do_payment( Ok(()) } +/// Finalize a paid order: transition `settled-hold-invoice` → `Success` and +/// notify the buyer, but only after the status CAS actually commits. +/// +/// Returns `Ok(true)` when the order is now terminal — this call committed the +/// transition, or a concurrent task already did — meaning the caller may safely +/// release the payout marker. Returns `Ok(false)` when the transition could not +/// be built/persisted (`update_order_event` failed): the caller must KEEP the +/// marker so reconciliation retries finalization, otherwise the buyer would be +/// paid on an order stranded in `settled-hold-invoice` with no recovery hook. +/// +/// Buyer notifications (`PurchaseCompleted`, `Rate`) are enqueued only after a +/// successful commit, so a retried finalization never spams the buyer. async fn payment_success( ctx: &AppContext, order: &mut Order, buyer_pubkey: PublicKey, my_keys: &Keys, request_id: Option, -) -> Result<()> { - // Purchase completed message to buyer +) -> Result { + let pool = ctx.pool(); + + let order_updated = match update_order_event(my_keys, Status::Success, order).await { + Ok(updated) => updated, + // Could not build/publish the Success event: leave the order in + // settled-hold-invoice and signal "not finalized" so the caller keeps + // the marker for reconciliation. + Err(_) => return Ok(false), + }; + + // Only update status and event_id to avoid overwriting fields modified by + // concurrent processes (dev_fee_paid, dev_fee_payment_hash, etc.) + // The WHERE guard prevents double success transitions from concurrent tasks. + let result = + sqlx::query("UPDATE orders SET status = ?, event_id = ? WHERE id = ? AND status = ?") + .bind(&order_updated.status) + .bind(&order_updated.event_id) + .bind(order_updated.id) + .bind(Status::SettledHoldInvoice.to_string()) + .execute(pool) + .await + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + + if result.rows_affected() == 0 { + // Another task already finalized this order: it is terminal, so the + // caller may release the marker, but the notifications were already + // sent by that task — do not duplicate them. + tracing::warn!( + "Order {} not transitioned to success: already processed by another task", + order_updated.id + ); + return Ok(true); + } + + // Committed by us — notify the buyer now. enqueue_order_msg( None, - Some(order.id), + Some(order_updated.id), Action::PurchaseCompleted, None, buyer_pubkey, None, ) .await; + enqueue_order_msg( + request_id, + Some(order_updated.id), + Action::Rate, + None, + buyer_pubkey, + None, + ) + .await; + Ok(true) +} - let pool = ctx.pool(); +/// The one LND capability `reconcile_inflight_payout` needs: query a payment's +/// status by hash. Behind a trait (mirroring [`crate::app::cancel`]'s +/// `CancelLightning`) so the reconcile branches are unit-testable with a stub +/// instead of a live node. +pub trait PayoutStatusLookup { + fn lookup_payment_status<'a>( + &'a mut self, + payment_hash: &'a [u8], + ) -> std::pin::Pin< + Box< + dyn std::future::Future, MostroError>> + + Send + + 'a, + >, + >; +} - if let Ok(order_updated) = update_order_event(my_keys, Status::Success, order).await { - // Only update status and event_id to avoid overwriting fields modified by - // concurrent processes (dev_fee_paid, dev_fee_payment_hash, etc.) - // The WHERE guard prevents double success transitions from concurrent tasks. - let result = - sqlx::query("UPDATE orders SET status = ?, event_id = ? WHERE id = ? AND status = ?") - .bind(&order_updated.status) - .bind(&order_updated.event_id) - .bind(order_updated.id) - .bind(Status::SettledHoldInvoice.to_string()) - .execute(pool) - .await - .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; +impl PayoutStatusLookup for LndConnector { + fn lookup_payment_status<'a>( + &'a mut self, + payment_hash: &'a [u8], + ) -> std::pin::Pin< + Box< + dyn std::future::Future, MostroError>> + + Send + + 'a, + >, + > { + Box::pin(async move { LndConnector::lookup_payment_status(self, payment_hash).await }) + } +} - if result.rows_affected() == 0 { - tracing::warn!( - "Order {} not transitioned to success: already processed by another task", - order_updated.id - ); +/// Reconcile a single in-flight buyer payout against LND. +/// +/// Called by the scheduler for every order whose `payout_payment_hash` is set. +/// This is the counterpart to `do_payment`'s in-process status watcher: it +/// resolves the durable marker for payouts whose watcher never delivered a +/// terminal update — a held/stranded HTLC, or a payout whose watcher task was +/// lost across a restart. Without it, such an order would stay locked forever +/// (regressing the do-payment-stuck bug); with it, the marker is authoritative +/// only until LND confirms the real outcome: +/// +/// - `Succeeded` → finalize as `Success` (idempotent via the status CAS) and +/// clear the marker. +/// - `Failed` / `Unknown` / not found → clear the marker, re-arm retry, and run +/// the same failure bookkeeping as the in-process watcher (advance +/// `payment_attempts`, notify the buyer). Re-arming is safe because LND itself +/// rejects a second `SendPaymentV2` for a payment hash it already has in +/// flight or settled, so a fresh dispatch of the same invoice cannot +/// double-pay. +/// - `InFlight` → leave as is; the payout is genuinely pending. +/// +/// `payout_claimed_at` is the per-claim token observed for this marker; every +/// release is scoped to it so a claim replaced between the snapshot and here is +/// never clobbered. +pub async fn reconcile_inflight_payout( + ctx: &AppContext, + ln_client: &mut impl PayoutStatusLookup, + order_id: uuid::Uuid, + payout_payment_hash: &str, + payout_claimed_at: Option, +) -> Result<(), MostroError> { + let pool = ctx.pool(); + + // A payment hash is exactly 32 bytes (64 hex chars). Decode with the same + // `FromHex` used across the codebase and length-check it; a bad-hex or + // wrong-length marker is corrupt, so treat it as malformed and re-arm rather + // than sending a truncated hash to LND. + let hash_bytes = match Vec::::from_hex(payout_payment_hash) { + Ok(bytes) if bytes.len() == 32 => bytes, + _ => { + warn!("Order {order_id}: malformed payout_payment_hash; clearing and re-arming retry"); + crate::db::fail_order_payout(pool, order_id, payout_payment_hash, payout_claimed_at) + .await?; return Ok(()); } + }; - // Send dm to buyer to rate counterpart - enqueue_order_msg( - request_id, - Some(order_updated.id), - Action::Rate, - None, - buyer_pubkey, - None, - ) - .await; + match ln_client.lookup_payment_status(&hash_bytes).await { + Ok(Some(PaymentStatus::Succeeded)) => { + // Clear the marker only once the order is actually finalized. If + // finalization fails (or the buyer key is unreadable), keep the + // marker so a later tick retries it instead of stranding a paid + // order in settled-hold-invoice. + let finalized = match Order::by_id(pool, order_id).await { + Ok(Some(mut order)) => { + let my_keys = ctx.keys().clone(); + match order.get_buyer_pubkey() { + Ok(buyer_pubkey) => { + payment_success(ctx, &mut order, buyer_pubkey, &my_keys, None) + .await + .unwrap_or(false) + } + Err(_) => false, + } + } + _ => false, + }; + if finalized { + crate::db::clear_order_payout( + pool, + order_id, + payout_payment_hash, + payout_claimed_at, + ) + .await?; + } + } + Ok(Some(PaymentStatus::Failed)) | Ok(Some(PaymentStatus::Unknown)) | Ok(None) => { + // Snapshot the order *before* re-arming so the bookkeeping sees the + // pre-failure state: `fail_order_payout` sets failed_payment = true, + // and `count_failed_payment` treats an already-failed order as a + // subsequent failure (no first-failure notice / attempt bump). This + // mirrors the in-process watcher, which passes its pre-failure copy. + let pre = Order::by_id(pool, order_id).await.ok().flatten(); + // Re-arm retry, and — only if we still owned this claim — run the + // same failure bookkeeping the in-process watcher does, so a payout + // that resolves only through reconciliation (watcher lost across a + // restart) still advances payment_attempts and notifies the buyer. + if crate::db::fail_order_payout(pool, order_id, payout_payment_hash, payout_claimed_at) + .await? + { + if let Some(order) = pre { + check_failure_retries_or_log(ctx, &order, None).await; + } + } + } + Ok(Some(PaymentStatus::InFlight)) => { + // Still pending — do not re-dispatch; a later tick will reconcile. + } + Err(e) => { + warn!("Order {order_id}: payout reconciliation lookup failed: {e}"); + } } + Ok(()) } @@ -1911,8 +2149,8 @@ mod tests { // Act let result = payment_success(&ctx, &mut order, buyer, &my_keys, None).await; - // Assert - assert!(result.is_ok()); + // Assert: committed the transition (returns true) and notified the buyer. + assert!(result.unwrap(), "a committed finalization returns true"); let db_order = Order::by_id(&pool, order.id).await.unwrap().unwrap(); assert_eq!(db_order.status, Status::Success.to_string()); let actions = queued_actions_for(order.id).await; @@ -1936,12 +2174,195 @@ mod tests { // Act let result = payment_success(&ctx, &mut order, buyer, &my_keys, None).await; - // Assert: early return — status untouched, no Rate message queued. - assert!(result.is_ok()); + // Assert: the guarded UPDATE matched no rows (already finalized + // elsewhere), so the call reports terminal (`true`) — the caller may + // release the marker — but sends no duplicate notifications, and the + // status is left untouched. + assert!( + result.unwrap(), + "a no-op CAS (already processed) is terminal and returns true" + ); let db_order = Order::by_id(&pool, order.id).await.unwrap().unwrap(); assert_eq!(db_order.status, Status::Active.to_string()); let actions = queued_actions_for(order.id).await; - assert!(actions.contains(&Action::PurchaseCompleted)); + assert!(!actions.contains(&Action::PurchaseCompleted)); assert!(!actions.contains(&Action::Rate)); } + + // --- reconcile_inflight_payout branch coverage (stubbed LND) --- + + /// Configurable LND stub: returns a fixed `PaymentStatus` (or an error) for + /// every `lookup_payment_status`, so the four reconcile branches can be + /// exercised without a live node. + struct StubLnClient { + status: Option, + error: bool, + } + + impl PayoutStatusLookup for StubLnClient { + fn lookup_payment_status<'a>( + &'a mut self, + _payment_hash: &'a [u8], + ) -> std::pin::Pin< + Box< + dyn std::future::Future, MostroError>> + + Send + + 'a, + >, + > { + let status = self.status; + let error = self.error; + Box::pin(async move { + if error { + Err(MostroInternalErr(ServiceError::LnPaymentError( + "stub".to_string(), + ))) + } else { + Ok(status) + } + }) + } + } + + /// Create a `settled-hold-invoice` order and claim a payout marker on it, + /// returning `(order_id, payout_hash, claim_token)`. + async fn settled_order_with_marker(pool: &SqlitePool) -> (uuid::Uuid, String, i64) { + let seller = Keys::generate().public_key(); + let buyer = Keys::generate().public_key(); + let mut order = fiat_sent_sell_order(seller, buyer); + order.status = Status::SettledHoldInvoice.to_string(); + let order = order.create(pool).await.unwrap(); + let hash = "a".repeat(64); + let token = crate::db::claim_order_payout(pool, order.id, &hash) + .await + .unwrap() + .expect("claim must win on a fresh order"); + (order.id, hash, token) + } + + async fn marker_of(pool: &SqlitePool, id: uuid::Uuid) -> Option { + sqlx::query_scalar::<_, Option>( + "SELECT payout_payment_hash FROM orders WHERE id = ?", + ) + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn reconcile_succeeded_finalizes_and_clears_marker() { + init_global_config(); + let pool = create_test_pool().await; + let ctx = build_ctx(&pool); + let (id, hash, token) = settled_order_with_marker(&pool).await; + let mut ln = StubLnClient { + status: Some(PaymentStatus::Succeeded), + error: false, + }; + + reconcile_inflight_payout(&ctx, &mut ln, id, &hash, Some(token)) + .await + .unwrap(); + + let db_order = Order::by_id(&pool, id).await.unwrap().unwrap(); + assert_eq!(db_order.status, Status::Success.to_string()); + assert!( + marker_of(&pool, id).await.is_none(), + "marker released after finalize" + ); + } + + #[tokio::test] + async fn reconcile_failed_rearms_and_runs_bookkeeping() { + init_global_config(); + let pool = create_test_pool().await; + let ctx = build_ctx(&pool); + let (id, hash, token) = settled_order_with_marker(&pool).await; + let mut ln = StubLnClient { + status: Some(PaymentStatus::Failed), + error: false, + }; + + reconcile_inflight_payout(&ctx, &mut ln, id, &hash, Some(token)) + .await + .unwrap(); + + assert!(marker_of(&pool, id).await.is_none(), "marker released"); + let db_order = Order::by_id(&pool, id).await.unwrap().unwrap(); + assert!(db_order.failed_payment, "retry re-armed"); + assert_eq!( + db_order.payment_attempts, 1, + "bookkeeping advanced payment_attempts" + ); + assert!( + queued_actions_for(id) + .await + .contains(&Action::PaymentFailed), + "buyer notified on first failure" + ); + } + + #[tokio::test] + async fn reconcile_inflight_is_a_noop() { + init_global_config(); + let pool = create_test_pool().await; + let ctx = build_ctx(&pool); + let (id, hash, token) = settled_order_with_marker(&pool).await; + let mut ln = StubLnClient { + status: Some(PaymentStatus::InFlight), + error: false, + }; + + reconcile_inflight_payout(&ctx, &mut ln, id, &hash, Some(token)) + .await + .unwrap(); + + assert_eq!( + marker_of(&pool, id).await.as_deref(), + Some(hash.as_str()), + "an in-flight payout keeps its marker" + ); + let db_order = Order::by_id(&pool, id).await.unwrap().unwrap(); + assert_eq!(db_order.status, Status::SettledHoldInvoice.to_string()); + assert!(!db_order.failed_payment, "in-flight must not re-arm retry"); + } + + #[tokio::test] + async fn reconcile_malformed_hash_rearms_without_lookup() { + init_global_config(); + let pool = create_test_pool().await; + let ctx = build_ctx(&pool); + let seller = Keys::generate().public_key(); + let buyer = Keys::generate().public_key(); + let mut order = fiat_sent_sell_order(seller, buyer); + order.status = Status::SettledHoldInvoice.to_string(); + let order = order.create(&pool).await.unwrap(); + // Force a corrupt (non-32-byte) marker directly. + sqlx::query( + "UPDATE orders SET payout_payment_hash = ?, payout_claimed_at = ? WHERE id = ?", + ) + .bind("abc") + .bind(1000_i64) + .bind(order.id) + .execute(&pool) + .await + .unwrap(); + + // Stub set to error to prove it is never consulted for a malformed hash. + let mut ln = StubLnClient { + status: None, + error: true, + }; + reconcile_inflight_payout(&ctx, &mut ln, order.id, "abc", Some(1000)) + .await + .unwrap(); + + assert!( + marker_of(&pool, order.id).await.is_none(), + "malformed marker cleared" + ); + let db_order = Order::by_id(&pool, order.id).await.unwrap().unwrap(); + assert!(db_order.failed_payment, "malformed marker re-arms retry"); + } } diff --git a/src/db.rs b/src/db.rs index 6c9bdfd1..f46e0723 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1176,11 +1176,16 @@ pub async fn find_escrow_deadline_orders( } pub async fn find_failed_payment(pool: &SqlitePool) -> Result, MostroError> { + // `payout_payment_hash IS NULL` excludes orders that already have a payout + // in flight: retrying them would dispatch a second payment against the same + // settled escrow. Those are resolved by `find_inflight_payouts` / + // reconciliation instead. let order = sqlx::query_as::<_, Order>( r#" SELECT * FROM orders - WHERE failed_payment == true AND status == 'settled-hold-invoice' + WHERE failed_payment == true AND status == 'settled-hold-invoice' + AND payout_payment_hash IS NULL "#, ) .fetch_all(pool) @@ -1190,6 +1195,147 @@ pub async fn find_failed_payment(pool: &SqlitePool) -> Result, Mostro Ok(order) } +/// Atomically claim the buyer payout for `order_id` by persisting the payout +/// invoice's `payment_hash` (and a claim timestamp), but only while no payout is +/// already in flight and the order is still `settled-hold-invoice`. Returns +/// `Some(claimed_at)` — the unix-second timestamp sealed into the claim — when +/// this caller won, or `None` when it lost the CAS. +/// +/// The CAS is the idempotency anchor: once set, `find_failed_payment` skips the +/// order and `pay_new_invoice` rejects invoice swaps, so a concurrent tick (or a +/// re-arm attempt) can never dispatch a second payout for the same escrow. The +/// returned `claimed_at` is a per-claim token: releases are scoped to it (not +/// just the hash), so when a retry re-claims the *same* invoice/hash a stale +/// 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). +pub async fn claim_order_payout( + pool: &SqlitePool, + order_id: Uuid, + payment_hash: &str, +) -> Result, MostroError> { + let claimed_at = chrono::Utc::now().timestamp(); + let result = sqlx::query( + "UPDATE orders SET payout_payment_hash = ?1, payout_claimed_at = ?2 \ + WHERE id = ?3 AND payout_payment_hash IS NULL AND status = 'settled-hold-invoice'", + ) + .bind(payment_hash) + .bind(claimed_at) + .bind(order_id) + .execute(pool) + .await + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + + Ok((result.rows_affected() > 0).then_some(claimed_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`). +/// +/// Scoped to both `payment_hash` and `claimed_at` (the per-claim token): the +/// release only fires when the marker still holds *this* caller's claim. A +/// status watcher can outlive its own claim (its stream stalls, reconciliation +/// resolves the payout, and a new payout — possibly with the same reused +/// invoice/hash — is claimed for the same order); scoping the CAS to the claim +/// timestamp stops such a stale watcher from erasing a *newer* claim and letting +/// a second payment be dispatched. `claimed_at` uses `IS` so a legacy NULL +/// timestamp still matches. +/// +/// Returns `true` when this caller still owned the claim (the row was cleared), +/// so callers can gate their own terminal side-effects on claim ownership. +pub async fn clear_order_payout( + pool: &SqlitePool, + order_id: Uuid, + payment_hash: &str, + claimed_at: Option, +) -> Result { + let result = sqlx::query( + "UPDATE orders SET payout_payment_hash = NULL, payout_claimed_at = NULL \ + WHERE id = ?1 AND payout_payment_hash = ?2 AND payout_claimed_at IS ?3", + ) + .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) +} + +/// Release the in-flight marker AND re-arm retry after a failed / unknown +/// terminal outcome: the next scheduler tick may dispatch a fresh payout once +/// the buyer supplies a new invoice. +/// +/// Scoped to `payment_hash` and `claimed_at` for the same reason as +/// [`clear_order_payout`]: a stale caller must not re-arm retry against a newer +/// in-flight claim. Returns `true` when this caller still owned the claim, so +/// failure bookkeeping and buyer notifications can be gated on ownership. +pub async fn fail_order_payout( + pool: &SqlitePool, + order_id: Uuid, + payment_hash: &str, + claimed_at: Option, +) -> Result { + let result = sqlx::query( + "UPDATE orders SET payout_payment_hash = NULL, payout_claimed_at = NULL, failed_payment = true \ + WHERE id = ?1 AND payout_payment_hash = ?2 AND payout_claimed_at IS ?3", + ) + .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) +} + +/// Orders with a buyer payout in flight (marker set) that are old enough to +/// reconcile: only those claimed at or before `claimed_before` (unix seconds) +/// are returned, so a payout still inside its grace window — where LND may not +/// have registered it yet — is never reconciled and mistaken for lost. Rows +/// predating the `payout_claimed_at` column (NULL) are always eligible. +/// Returns `(order_id, payout_payment_hash, payout_claimed_at)` tuples so the +/// caller can scope its release to the exact claim it observed. +/// +/// Note: the `status = 'settled-hold-invoice'` filter means a marker left on an +/// order that has since moved to a terminal status (e.g. `Success` when the +/// post-finalize `clear_order_payout` lost to a DB blip) is never returned here. +/// Such a marker is inert residue — `find_failed_payment` and `pay_new_invoice` +/// also require `settled-hold-invoice`, so nothing ever acts on it — but it can +/// linger; treat a marker on a non-`settled-hold-invoice` order as stale. +pub async fn find_inflight_payouts( + pool: &SqlitePool, + claimed_before: i64, +) -> Result)>, MostroError> { + let rows = sqlx::query( + r#" + SELECT id, payout_payment_hash, payout_claimed_at + FROM orders + WHERE payout_payment_hash IS NOT NULL AND status == 'settled-hold-invoice' + AND (payout_claimed_at IS NULL OR payout_claimed_at <= ?1) + "#, + ) + .bind(claimed_before) + .fetch_all(pool) + .await + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + + let mut out = Vec::with_capacity(rows.len()); + for row in rows { + let id: Uuid = row + .try_get("id") + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + let hash: String = row + .try_get("payout_payment_hash") + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + let claimed_at: Option = row + .try_get("payout_claimed_at") + .map_err(|e| MostroInternalErr(ServiceError::DbAccessError(e.to_string())))?; + out.push((id, hash, claimed_at)); + } + Ok(out) +} + pub async fn find_unpaid_dev_fees(pool: &SqlitePool) -> Result, MostroError> { let orders = sqlx::query_as::<_, Order>( r#" @@ -1843,7 +1989,9 @@ mod tests { dev_fee_payment_hash char(64), cashu_mint_url text, cashu_escrow_token text, - cashu_escrow_locked_at integer + cashu_escrow_locked_at integer, + payout_payment_hash char(64), + payout_claimed_at integer ) "#, ) @@ -2809,6 +2957,290 @@ mod tests { ); } + #[tokio::test] + async fn test_find_failed_payment_ignores_inflight_payout() { + let pool = setup_orders_db().await.unwrap(); + + // failed_payment = true and settled-hold-invoice, but a payout is + // already in flight (payout_payment_hash set) → must be skipped so the + // retry job never dispatches a second payment for the same escrow. + sqlx::query( + r#"INSERT INTO orders (id, kind, event_id, status, premium, payment_method, + amount, fiat_code, fiat_amount, created_at, expires_at, + failed_payment, payment_attempts, dev_fee, dev_fee_paid, payout_payment_hash) + VALUES (?1, 'buy', 'ev1', 'settled-hold-invoice', 0, 'lightning', + 100000, 'USD', 100, 1700000000, 1700086400, + 1, 0, 0, 0, ?2)"#, + ) + .bind(uuid::Uuid::new_v4()) + .bind("a".repeat(64)) + .execute(&pool) + .await + .unwrap(); + + let result = super::find_failed_payment(&pool).await.unwrap(); + assert!( + result.is_empty(), + "Orders with a payout in flight must be excluded from retry" + ); + } + + // -- Tests for the in-flight payout marker -- + + async fn insert_settled_order(pool: &SqlitePool, id: uuid::Uuid, status: &str) { + sqlx::query( + r#"INSERT INTO orders (id, kind, event_id, status, premium, payment_method, + amount, fiat_code, fiat_amount, created_at, expires_at, + failed_payment, payment_attempts, dev_fee, dev_fee_paid) + VALUES (?1, 'buy', 'ev1', ?2, 0, 'lightning', + 100000, 'USD', 100, 1700000000, 1700086400, + 0, 0, 0, 0)"#, + ) + .bind(id) + .bind(status) + .execute(pool) + .await + .unwrap(); + } + + async fn payout_hash_of(pool: &SqlitePool, id: uuid::Uuid) -> Option { + sqlx::query_scalar::<_, Option>( + "SELECT payout_payment_hash FROM orders WHERE id = ?", + ) + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn test_claim_order_payout_is_atomic() { + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + insert_settled_order(&pool, id, "settled-hold-invoice").await; + + let hash = "b".repeat(64); + // First claim wins and returns its timestamp token. + assert!(super::claim_order_payout(&pool, id, &hash) + .await + .unwrap() + .is_some()); + assert_eq!( + payout_hash_of(&pool, id).await.as_deref(), + Some(hash.as_str()) + ); + + // A second claim (concurrent tick / re-arm) loses — the marker is set. + assert!( + super::claim_order_payout(&pool, id, &"c".repeat(64)) + .await + .unwrap() + .is_none(), + "second claim must lose the CAS" + ); + // The original hash is untouched. + assert_eq!( + payout_hash_of(&pool, id).await.as_deref(), + Some(hash.as_str()) + ); + } + + #[tokio::test] + async fn test_claim_order_payout_rejects_wrong_status() { + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + insert_settled_order(&pool, id, "active").await; + + assert!( + super::claim_order_payout(&pool, id, &"d".repeat(64)) + .await + .unwrap() + .is_none(), + "must not claim a payout on a non settled-hold-invoice order" + ); + assert!(payout_hash_of(&pool, id).await.is_none()); + } + + #[tokio::test] + async fn test_clear_order_payout_releases_marker() { + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let hash = "e".repeat(64); + insert_settled_order(&pool, id, "settled-hold-invoice").await; + let claimed_at = super::claim_order_payout(&pool, id, &hash).await.unwrap(); + + let owned = super::clear_order_payout(&pool, id, &hash, claimed_at) + .await + .unwrap(); + assert!(owned, "clearing an owned claim must report ownership"); + assert!(payout_hash_of(&pool, id).await.is_none()); + } + + #[tokio::test] + async fn test_fail_order_payout_clears_marker_and_rearms_retry() { + let pool = setup_orders_db().await.unwrap(); + let id = uuid::Uuid::new_v4(); + let hash = "f".repeat(64); + insert_settled_order(&pool, id, "settled-hold-invoice").await; + let claimed_at = super::claim_order_payout(&pool, id, &hash).await.unwrap(); + + let owned = super::fail_order_payout(&pool, id, &hash, claimed_at) + .await + .unwrap(); + assert!(owned, "failing an owned claim must report ownership"); + + assert!(payout_hash_of(&pool, id).await.is_none()); + let failed: i64 = sqlx::query_scalar("SELECT failed_payment FROM orders WHERE id = ?") + .bind(id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(failed, 1, "failed_payment must be re-armed for retry"); + // Now that the marker is clear, the order is retry-eligible again. + let matching = super::find_failed_payment(&pool).await.unwrap(); + assert_eq!(matching.len(), 1); + } + + /// A stale caller (e.g. a watcher that outlived its own claim) must not + /// release a *newer* claim. The release is scoped to both the hash and the + /// per-claim timestamp token, so neither a different hash nor — the case a + /// hash alone misses when a retry reuses the same invoice — a different + /// timestamp against the same hash can erase the current marker. + #[tokio::test] + async fn test_clear_and_fail_order_payout_ignore_stale_hash() { + let pool = setup_orders_db().await.unwrap(); + let current = "1".repeat(64); + let stale = "2".repeat(64); + + // clear_order_payout with the wrong hash leaves the current marker. + let id_a = uuid::Uuid::new_v4(); + insert_settled_order(&pool, id_a, "settled-hold-invoice").await; + let token_a = super::claim_order_payout(&pool, id_a, ¤t) + .await + .unwrap(); + let owned = super::clear_order_payout(&pool, id_a, &stale, token_a) + .await + .unwrap(); + assert!( + !owned, + "clearing with a stale hash must report no ownership" + ); + assert_eq!( + payout_hash_of(&pool, id_a).await.as_deref(), + Some(current.as_str()), + "clear with a stale hash must not erase the current marker" + ); + + // fail_order_payout with the wrong hash likewise leaves it intact and + // does not re-arm retry. + let id_b = uuid::Uuid::new_v4(); + insert_settled_order(&pool, id_b, "settled-hold-invoice").await; + let token_b = super::claim_order_payout(&pool, id_b, ¤t) + .await + .unwrap(); + let owned = super::fail_order_payout(&pool, id_b, &stale, token_b) + .await + .unwrap(); + assert!(!owned, "failing with a stale hash must report no ownership"); + assert_eq!( + payout_hash_of(&pool, id_b).await.as_deref(), + Some(current.as_str()), + "fail with a stale hash must not erase the current marker" + ); + let failed: i64 = sqlx::query_scalar("SELECT failed_payment FROM orders WHERE id = ?") + .bind(id_b) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(failed, 0, "a stale fail must not re-arm retry"); + + // Correct hash but a STALE timestamp token (the same-invoice retry + // case): a watcher from a previous attempt must still lose the release + // even though the hash matches the live claim. + let id_c = uuid::Uuid::new_v4(); + insert_settled_order(&pool, id_c, "settled-hold-invoice").await; + let live_token = super::claim_order_payout(&pool, id_c, ¤t) + .await + .unwrap() + .unwrap(); + let stale_token = Some(live_token - 1); + let owned = super::clear_order_payout(&pool, id_c, ¤t, stale_token) + .await + .unwrap(); + assert!( + !owned, + "a stale timestamp token must lose even when the hash matches" + ); + assert_eq!( + payout_hash_of(&pool, id_c).await.as_deref(), + Some(current.as_str()), + "a stale-token release must not erase the live claim" + ); + } + + async fn insert_inflight_order( + pool: &SqlitePool, + id: uuid::Uuid, + hash: &str, + claimed_at: Option, + ) { + insert_settled_order(pool, id, "settled-hold-invoice").await; + sqlx::query( + "UPDATE orders SET payout_payment_hash = ?1, payout_claimed_at = ?2 WHERE id = ?3", + ) + .bind(hash) + .bind(claimed_at) + .bind(id) + .execute(pool) + .await + .unwrap(); + } + + #[tokio::test] + async fn test_find_inflight_payouts_returns_only_marked() { + let pool = setup_orders_db().await.unwrap(); + let marked = uuid::Uuid::new_v4(); + let unmarked = uuid::Uuid::new_v4(); + let hash = "1".repeat(64); + insert_inflight_order(&pool, marked, &hash, Some(1000)).await; + insert_settled_order(&pool, unmarked, "settled-hold-invoice").await; + + // cutoff well after the claim → the marked order is old enough. + let inflight = super::find_inflight_payouts(&pool, 2000).await.unwrap(); + assert_eq!(inflight.len(), 1); + assert_eq!(inflight[0].0, marked); + assert_eq!(inflight[0].1, hash); + assert_eq!(inflight[0].2, Some(1000), "the claim token is returned"); + } + + #[tokio::test] + async fn test_find_inflight_payouts_respects_grace_window() { + let pool = setup_orders_db().await.unwrap(); + let fresh = uuid::Uuid::new_v4(); + let aged = uuid::Uuid::new_v4(); + let legacy = uuid::Uuid::new_v4(); + insert_inflight_order(&pool, fresh, &"a".repeat(64), Some(2000)).await; + insert_inflight_order(&pool, aged, &"b".repeat(64), Some(500)).await; + // Row predating the payout_claimed_at column (NULL) is always eligible. + insert_inflight_order(&pool, legacy, &"c".repeat(64), None).await; + + // Grace cutoff = 1000: the fresh claim (2000) is skipped; the aged (500) + // and legacy (NULL) ones are eligible for reconciliation. + let inflight = super::find_inflight_payouts(&pool, 1000).await.unwrap(); + let ids: std::collections::HashSet = + inflight.iter().map(|(id, _, _)| *id).collect(); + assert!( + !ids.contains(&fresh), + "a just-claimed payout must not reconcile" + ); + assert!(ids.contains(&aged), "an aged payout must reconcile"); + assert!( + ids.contains(&legacy), + "a NULL-timestamp payout must reconcile" + ); + assert_eq!(ids.len(), 2); + } + // -- Tests for find_order_by_hash -- #[tokio::test] diff --git a/src/scheduler.rs b/src/scheduler.rs index 7b006a8c..fcbb4d59 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -1,7 +1,7 @@ use crate::app::bond; use crate::app::context::AppContext; use crate::app::dev_fee::run_dev_fee_cycle; -use crate::app::release::do_payment; +use crate::app::release::{do_payment, reconcile_inflight_payout}; use crate::config; use crate::db::*; use crate::escrow::EscrowBackend; @@ -41,6 +41,7 @@ pub async fn start_scheduler(ctx: AppContext) { job_cancel_orders(ctx.clone()).await; job_enforce_escrow_deadline(ctx.clone()).await; job_retry_failed_payments(ctx.clone()).await; + job_reconcile_inflight_payouts(ctx.clone()).await; job_process_dev_fee_payment(ctx.clone()).await; job_process_bond_payouts(ctx.clone()).await; job_reconcile_stranded_maker_bonds(ctx.clone()).await; @@ -243,6 +244,74 @@ async fn job_retry_failed_payments(ctx: AppContext) { }); } +/// 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 +/// held/stranded payout — or one whose in-process watcher was lost across a +/// restart — is finalized, re-armed, or left pending based on LND's real state, +/// instead of blocking the order forever. Runs at startup and every tick. +async fn job_reconcile_inflight_payouts(ctx: AppContext) { + // Reconcile poll cadence. Deliberately a fixed value and NOT derived from + // `payment_retries_interval`: it only bounds how quickly a stranded payout + // is noticed (a liveness knob), not correctness, so it stays independent of + // the operator's retry tuning. + const RECONCILE_INTERVAL_SECS: u64 = 60; + // Grace window: a payout claimed less than this many seconds ago is not + // 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 + .payment_retries_interval + .max(MIN_GRACE_SECS) as i64; + + tokio::spawn(async move { + // Same capped-backoff LndConnector bootstrap as the bond payout job: a + // transient LND outage at boot must not permanently halt reconciliation. + let mut backoff_secs: u64 = 2; + let mut ln_client = loop { + match LndConnector::new().await { + Ok(client) => break client, + Err(e) => { + error!("payout reconcile: LndConnector::new failed: {e} — retrying in {backoff_secs}s"); + tokio::time::sleep(tokio::time::Duration::from_secs(backoff_secs)).await; + backoff_secs = (backoff_secs * 2).min(60); + } + } + }; + + let pool = ctx.pool(); + loop { + let claimed_before = Utc::now().timestamp() - grace_secs; + match crate::db::find_inflight_payouts(pool, claimed_before).await { + Ok(inflight) => { + for (order_id, payout_hash, payout_claimed_at) in inflight.into_iter() { + if let Err(e) = reconcile_inflight_payout( + &ctx, + &mut ln_client, + order_id, + &payout_hash, + payout_claimed_at, + ) + .await + { + error!("payout reconcile for order {order_id}: {e}"); + } + } + } + Err(e) => error!("payout reconcile: find_inflight_payouts failed: {e}"), + } + tokio::time::sleep(tokio::time::Duration::from_secs(RECONCILE_INTERVAL_SECS)).await; + } + }); +} + async fn job_update_rate_events(ctx: AppContext) { // Clone for closure owning with Arc let queue_order_rate = MESSAGE_QUEUES.queue_order_rate.clone();