Emit settled on-chain payment events

Notify users when otherwise unclassified on-chain payments reach the
anti-reorg confirmation depth.

Use the wallet's effective stored payment so classified funding,
splice, close, and sweep transactions stay quiet.

Co-Authored-By: HAL 9000

Author: tnull

benches/payments.rs

@@ -9,8 +9,10 @@ use bitcoin::Amount;
 use common::{
 	expect_channel_ready_event, generate_blocks_and_wait, premine_and_distribute_funds,
 	random_chain_source, setup_bitcoind_and_electrsd, setup_two_nodes_with_store,
+	ExpectOnchainPaymentEvent, OnchainPaymentEvent,
 };
 use criterion::{criterion_group, criterion_main, Criterion};
+use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY;
 use ldk_node::{Event, Node};
 use lightning_types::payment::{PaymentHash, PaymentPreimage};
 use rand::RngCore;
@@ -142,7 +144,7 @@ fn payment_benchmark(c: &mut Criterion) {
 	runtime.block_on(async move {
 		let address_a = node_a_cloned.onchain_payment().new_address().unwrap();
 		let premine_sat = 25_000_000;
-		premine_and_distribute_funds(
+		let premine_txid = premine_and_distribute_funds(
 			&bitcoind.client,
 			&electrsd.client,
 			vec![address_a],
@@ -151,6 +153,18 @@ fn payment_benchmark(c: &mut Criterion) {
 		.await;
 		node_a_cloned.sync_wallets().unwrap();
 		node_b_cloned.sync_wallets().unwrap();
+		generate_blocks_and_wait(
+			&bitcoind.client,
+			&electrsd.client,
+			(ANTI_REORG_DELAY - 1) as usize,
+		)
+		.await;
+		node_a_cloned.sync_wallets().unwrap();
+		node_b_cloned.sync_wallets().unwrap();
+		assert_eq!(
+			node_a_cloned.expect_onchain_payment_event(OnchainPaymentEvent::Received).await,
+			premine_txid,
+		);
 		open_channel_push_amt(
 			&node_a_cloned,
 			&node_b_cloned,

bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt

@@ -226,6 +226,28 @@ class LibraryTest {
         node1.syncWallets()
         node2.syncWallets()
 
+        val onchainPaymentReceivedEvent1 = node1.waitNextEvent()
+        println("Got event: $onchainPaymentReceivedEvent1")
+        when (onchainPaymentReceivedEvent1) {
+            is Event.OnchainPaymentReceived -> {
+                assertEquals(txid1, onchainPaymentReceivedEvent1.txid)
+                assertEquals(100000000uL, onchainPaymentReceivedEvent1.amountMsat)
+            }
+            else -> error("Expected initial on-chain payment event")
+        }
+        node1.eventHandled()
+
+        val onchainPaymentReceivedEvent2 = node2.waitNextEvent()
+        println("Got event: $onchainPaymentReceivedEvent2")
+        when (onchainPaymentReceivedEvent2) {
+            is Event.OnchainPaymentReceived -> {
+                assertEquals(txid2, onchainPaymentReceivedEvent2.txid)
+                assertEquals(100000000uL, onchainPaymentReceivedEvent2.amountMsat)
+            }
+            else -> error("Expected initial on-chain payment event")
+        }
+        node2.eventHandled()
+
         val spendableBalance1 = node1.listBalances().spendableOnchainBalanceSats
         val spendableBalance2 = node2.listBalances().spendableOnchainBalanceSats
         val totalBalance1 = node1.listBalances().totalOnchainBalanceSats

bindings/python/src/ldk_node/test_ldk_node.py

@@ -158,6 +158,14 @@ def fund_nodes(node_1, node_2, esplora_endpoint, amount_sats=100000):
     node_1.sync_wallets()
     node_2.sync_wallets()
 
+    received_event_1 = expect_event(node_1, Event.ONCHAIN_PAYMENT_RECEIVED)
+    assert received_event_1.txid == txid_1
+    assert received_event_1.amount_msat == amount_sats * 1000
+
+    received_event_2 = expect_event(node_2, Event.ONCHAIN_PAYMENT_RECEIVED)
+    assert received_event_2.txid == txid_2
+    assert received_event_2.amount_msat == amount_sats * 1000
+
 def open_channel_and_wait_ready(node_1, node_2, node_id_2, listening_address_2, esplora_endpoint, channel_amount_sats=50000):
     node_1.open_channel(node_id_2, listening_address_2, channel_amount_sats, None, None)
 

src/event.rs

@@ -13,7 +13,7 @@ use std::sync::{Arc, Mutex};
 
 use bitcoin::blockdata::locktime::absolute::LockTime;
 use bitcoin::secp256k1::PublicKey;
-use bitcoin::{Amount, OutPoint};
+use bitcoin::{Amount, BlockHash, OutPoint, Txid};
 use lightning::blinded_path::message::NextMessageHop;
 use lightning::events::bump_transaction::BumpTransactionEvent;
 #[cfg(not(feature = "uniffi"))]
@@ -271,6 +271,46 @@ pub enum Event {
 		/// This will be `None` for events serialized by LDK Node v0.2.1 and prior.
 		reason: Option<ClosureReason>,
 	},
+	/// A sent on-chain payment was successful.
+	///
+	/// This is only emitted for wallet transactions which were not classified as channel
+	/// funding, splices, closes, sweeps, or other LDK-driven chain activity.
+	///
+	/// It's guaranteed to have reached at least [`ANTI_REORG_DELAY`] confirmations.
+	///
+	/// [`ANTI_REORG_DELAY`]: lightning::chain::channelmonitor::ANTI_REORG_DELAY
+	OnchainPaymentSuccessful {
+		/// A local identifier used to track the payment.
+		payment_id: PaymentId,
+		/// The transaction identifier.
+		txid: Txid,
+		/// The value, in thousandths of a satoshi, that was sent.
+		amount_msat: u64,
+		/// The hash of the block in which the transaction was confirmed.
+		block_hash: BlockHash,
+		/// The height at which the block was confirmed.
+		block_height: u32,
+	},
+	/// An on-chain payment has been received.
+	///
+	/// This is only emitted for wallet transactions which were not classified as channel
+	/// funding, splices, closes, sweeps, or other LDK-driven chain activity.
+	///
+	/// It's guaranteed to have reached at least [`ANTI_REORG_DELAY`] confirmations.
+	///
+	/// [`ANTI_REORG_DELAY`]: lightning::chain::channelmonitor::ANTI_REORG_DELAY
+	OnchainPaymentReceived {
+		/// A local identifier used to track the payment.
+		payment_id: PaymentId,
+		/// The transaction identifier.
+		txid: Txid,
+		/// The value, in thousandths of a satoshi, that has been received.
+		amount_msat: u64,
+		/// The hash of the block in which the transaction was confirmed.
+		block_hash: BlockHash,
+		/// The height at which the block was confirmed.
+		block_height: u32,
+	},
 	/// A channel splice has been negotiated and the funding transaction is pending
 	/// confirmation on-chain.
 	SpliceNegotiated {
@@ -374,6 +414,20 @@ impl_writeable_tlv_based_enum!(Event,
 		(5, user_channel_id, required),
 		// TLV 7 (abandoned_funding_txo) may be set for LDK Node v0.7.
 	},
+	(10, OnchainPaymentSuccessful) => {
+		(0, payment_id, required),
+		(2, txid, required),
+		(4, amount_msat, required),
+		(6, block_hash, required),
+		(8, block_height, required),
+	},
+	(11, OnchainPaymentReceived) => {
+		(0, payment_id, required),
+		(2, txid, required),
+		(4, amount_msat, required),
+		(6, block_hash, required),
+		(8, block_height, required),
+	},
 );
 
 pub struct EventQueue<L: Deref>

src/wallet/mod.rs

@@ -54,7 +54,7 @@ use lightning_invoice::RawBolt11Invoice;
 use persist::KVStoreWalletPersister;
 
 use crate::config::Config;
-use crate::event::EventQueue;
+use crate::event::{Event, EventQueue};
 use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator};
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::store::ConfirmationStatus;
@@ -278,16 +278,22 @@ impl Wallet {
 						confirmation_status,
 					);
 
-					self.runtime.block_on(self.payment_store.insert_or_update(payment.clone()))?;
+					self.runtime.block_on(async {
+						let (updated, stored_payment) =
+							self.payment_store.insert_or_update_and_get(payment).await?;
 
-					if payment_status == PaymentStatus::Pending {
-						let pending_payment =
-							self.create_pending_payment_from_tx(payment, Vec::new());
+						if updated && payment_status == PaymentStatus::Succeeded {
+							self.emit_onchain_payment_event(&stored_payment).await?;
+						}
 
-						self.runtime.block_on(
-							self.pending_payment_store.insert_or_update(pending_payment),
-						)?;
-					}
+						if payment_status == PaymentStatus::Pending {
+							let pending_payment =
+								self.create_pending_payment_from_tx(stored_payment, Vec::new());
+							self.pending_payment_store.insert_or_update(pending_payment).await?;
+						}
+
+						Ok::<(), Error>(())
+					})?;
 				},
 				WalletEvent::ChainTipChanged { new_tip, .. } => {
 					let pending_payments: Vec<PendingPaymentDetails> =
@@ -304,27 +310,34 @@ impl Wallet {
 					let mut unconfirmed_outbound_txids: Vec<Txid> = Vec::new();
 
 					for mut payment in pending_payments {
-						match payment.details.kind {
+						match &payment.details.kind {
 							PaymentKind::Onchain {
 								status: ConfirmationStatus::Confirmed { height, .. },
 								..
 							} => {
 								let payment_id = payment.details.id;
-								if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
+								if new_tip.height >= *height + ANTI_REORG_DELAY - 1 {
 									payment.details.status = PaymentStatus::Succeeded;
-									self.runtime.block_on(
-										self.payment_store.insert_or_update(payment.details),
-									)?;
-									self.runtime
-										.block_on(self.pending_payment_store.remove(&payment_id))?;
+									self.runtime.block_on(async {
+										let (updated, stored_payment) = self
+											.payment_store
+											.insert_or_update_and_get(payment.details)
+											.await?;
+										if updated {
+											self.emit_onchain_payment_event(&stored_payment)
+												.await?;
+										}
+										self.pending_payment_store.remove(&payment_id).await?;
+										Ok::<(), Error>(())
+									})?;
 								}
 							},
 							PaymentKind::Onchain {
 								txid,
 								status: ConfirmationStatus::Unconfirmed,
 								..
 							} if payment.details.direction == PaymentDirection::Outbound => {
-								unconfirmed_outbound_txids.push(txid);
+								unconfirmed_outbound_txids.push(*txid);
 							},
 							_ => {},
 						}
@@ -1447,6 +1460,52 @@ impl Wallet {
 		PendingPaymentDetails::new(payment, conflicting_txids, Vec::new())
 	}
 
+	async fn emit_onchain_payment_event(&self, payment: &PaymentDetails) -> Result<(), Error> {
+		if payment.status != PaymentStatus::Succeeded {
+			return Ok(());
+		}
+
+		let (txid, block_hash, block_height) = match &payment.kind {
+			PaymentKind::Onchain {
+				txid,
+				status: ConfirmationStatus::Confirmed { block_hash, height, .. },
+				tx_type: None,
+			} => (*txid, *block_hash, *height),
+			_ => return Ok(()),
+		};
+
+		let Some(amount_msat) = payment.amount_msat else {
+			log_error!(
+				self.logger,
+				"Skipping on-chain payment event for {} due to missing amount",
+				payment.id
+			);
+			return Ok(());
+		};
+
+		let event = match payment.direction {
+			PaymentDirection::Outbound => Event::OnchainPaymentSuccessful {
+				payment_id: payment.id,
+				txid,
+				amount_msat,
+				block_hash,
+				block_height,
+			},
+			PaymentDirection::Inbound => Event::OnchainPaymentReceived {
+				payment_id: payment.id,
+				txid,
+				amount_msat,
+				block_hash,
+				block_height,
+			},
+		};
+
+		self.event_queue.add_event(event).await.map_err(|e| {
+			log_error!(self.logger, "Failed to push on-chain payment event: {}", e);
+			Error::PersistenceFailed
+		})
+	}
+
 	fn find_payment_by_txid(&self, target_txid: Txid) -> Option<PaymentId> {
 		let direct_payment_id = PaymentId(target_txid.to_byte_array());
 		if self.pending_payment_store.contains_key(&direct_payment_id) {