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

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) {

tests/common/mod.rs

@@ -43,7 +43,10 @@ use ldk_node::config::{
 };
 use ldk_node::entropy::{generate_entropy_mnemonic, NodeEntropy};
 use ldk_node::io::sqlite_store::SqliteStore;
-use ldk_node::payment::{PaymentDirection, PaymentKind, PaymentStatus, TransactionType};
+use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY;
+use ldk_node::payment::{
+	ConfirmationStatus, PaymentDirection, PaymentKind, PaymentStatus, TransactionType,
+};
 use ldk_node::probing::ProbingConfig;
 use ldk_node::{
 	Builder, ChannelShutdownState, CustomTlvRecord, Event, LightningBalance, Node, NodeError,
@@ -409,6 +412,65 @@ pub(crate) type TestNode = Arc<Node>;
 #[cfg(not(feature = "uniffi"))]
 pub(crate) type TestNode = Node;
 
+#[derive(Clone, Copy, Debug)]
+pub(crate) enum OnchainPaymentEvent {
+	Successful,
+	Received,
+}
+
+pub(crate) trait ExpectOnchainPaymentEvent {
+	async fn expect_onchain_payment_event(&self, expected_event: OnchainPaymentEvent) -> Txid;
+}
+
+impl ExpectOnchainPaymentEvent for Node {
+	async fn expect_onchain_payment_event(&self, expected_event: OnchainPaymentEvent) -> Txid {
+		let event = tokio::time::timeout(
+			Duration::from_secs(INTEROP_TIMEOUT_SECS),
+			self.next_event_async(),
+		)
+		.await
+		.unwrap_or_else(|_| {
+			panic!("{} timed out waiting for {:?} event after 60s", self.node_id(), expected_event,)
+		});
+		let txid = match (&expected_event, &event) {
+			(OnchainPaymentEvent::Successful, Event::OnchainPaymentSuccessful { txid, .. })
+			| (OnchainPaymentEvent::Received, Event::OnchainPaymentReceived { txid, .. }) => *txid,
+			_ => panic!("Expected {:?} event, got {:?}", expected_event, event),
+		};
+		println!("{} got event {:?}", self.node_id(), event);
+		assert_onchain_payment_event_matches_payment(self, &event);
+		self.event_handled().unwrap();
+		txid
+	}
+}
+
+pub(crate) fn assert_onchain_payment_event_matches_payment(node: &Node, event: &Event) {
+	let (payment_id, txid, amount_msat, expected_direction) = match event {
+		Event::OnchainPaymentSuccessful { payment_id, txid, amount_msat, .. } => {
+			(*payment_id, *txid, *amount_msat, PaymentDirection::Outbound)
+		},
+		Event::OnchainPaymentReceived { payment_id, txid, amount_msat, .. } => {
+			(*payment_id, *txid, *amount_msat, PaymentDirection::Inbound)
+		},
+		_ => panic!("Expected on-chain payment event, got {:?}", event),
+	};
+
+	let payment = node.payment(&payment_id).unwrap();
+	assert_eq!(payment.status, PaymentStatus::Succeeded);
+	assert_eq!(payment.direction, expected_direction);
+	assert_eq!(payment.amount_msat, Some(amount_msat));
+	match payment.kind {
+		PaymentKind::Onchain {
+			txid: payment_txid,
+			status: ConfirmationStatus::Confirmed { .. },
+			tx_type: None,
+		} => {
+			assert_eq!(payment_txid, txid);
+		},
+		ref other => panic!("Expected unclassified confirmed on-chain payment, got {:?}", other),
+	}
+}
+
 fn has_onchain_tx_type<F: Fn(&TransactionType) -> bool>(node: &TestNode, predicate: F) -> bool {
 	node.list_payments().into_iter().any(|payment| {
 		matches!(
@@ -880,11 +942,12 @@ where
 
 pub(crate) async fn premine_and_distribute_funds<E: ElectrumApi>(
 	bitcoind: &BitcoindClient, electrs: &E, addrs: Vec<Address>, amount: Amount,
-) {
+) -> Txid {
 	premine_blocks(bitcoind, electrs).await;
 
-	distribute_funds_unconfirmed(bitcoind, electrs, addrs, amount).await;
+	let txid = distribute_funds_unconfirmed(bitcoind, electrs, addrs, amount).await;
 	generate_blocks_and_wait(bitcoind, electrs, 1).await;
+	txid
 }
 
 pub(crate) async fn premine_blocks<E: ElectrumApi>(bitcoind: &BitcoindClient, electrs: &E) {
@@ -1086,7 +1149,7 @@ pub(crate) async fn do_channel_full_cycle<E: ElectrumApi>(
 
 	let premine_amount_sat = if expect_anchor_channel { 2_125_000 } else { 2_100_000 };
 
-	premine_and_distribute_funds(
+	let premine_txid = premine_and_distribute_funds(
 		&bitcoind,
 		electrsd,
 		vec![addr_a, addr_b],
@@ -1132,6 +1195,18 @@ pub(crate) async fn do_channel_full_cycle<E: ElectrumApi>(
 	assert_eq!(node_a.next_event(), None);
 	assert_eq!(node_b.next_event(), None);
 
+	generate_blocks_and_wait(&bitcoind, electrsd, (ANTI_REORG_DELAY - 1) as usize).await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+	assert_eq!(
+		node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await,
+		premine_txid,
+	);
+	assert_eq!(
+		node_b.expect_onchain_payment_event(OnchainPaymentEvent::Received).await,
+		premine_txid,
+	);
+
 	println!("\nA -- open_channel -> B");
 	let funding_amount_sat = 2_080_000;
 	let push_msat = (funding_amount_sat / 2) * 1000; // balance the channel
@@ -1546,12 +1621,18 @@ pub(crate) async fn do_channel_full_cycle<E: ElectrumApi>(
 	assert!(splice_out_sat > 500_000);
 	node_b.splice_out(&user_channel_id_b, node_a.node_id(), &addr_a, splice_out_sat).unwrap();
 
-	expect_splice_negotiated_event!(node_a, node_b.node_id());
+	let splice_out_txo = expect_splice_negotiated_event!(node_a, node_b.node_id());
 	expect_splice_negotiated_event!(node_b, node_a.node_id());
 
 	generate_blocks_and_wait(&bitcoind, electrsd, 6).await;
 	node_a.sync_wallets().unwrap();
 	node_b.sync_wallets().unwrap();
+	if !allow_0conf {
+		assert_eq!(
+			node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await,
+			splice_out_txo.txid,
+		);
+	}
 
 	expect_channel_ready_event!(node_a, node_b.node_id());
 	expect_channel_ready_event!(node_b, node_a.node_id());