Include onchain transactions in events

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;
@@ -137,7 +139,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],
@@ -146,6 +148,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
@@ -329,6 +351,24 @@ class LibraryTest {
         assert(spendableBalance1AfterClose < 100000u)
         assertEquals(102500uL, spendableBalance2AfterClose)
 
+        val externalAddress = bitcoinCli("getnewaddress")
+        val onchainTxid = node1.onchainPayment().sendToAddress(externalAddress, 10000u, null)
+        waitForTx(esploraEndpoint, onchainTxid)
+        mineAndWait(esploraEndpoint, 6u)
+        node1.syncWallets()
+
+        val onchainPaymentSuccessfulEvent = node1.waitNextEvent()
+        println("Got event: $onchainPaymentSuccessfulEvent")
+        when (onchainPaymentSuccessfulEvent) {
+            is Event.OnchainPaymentSuccessful -> {
+                assertEquals(onchainTxid, onchainPaymentSuccessfulEvent.txid)
+                assertEquals(10000000uL, onchainPaymentSuccessfulEvent.amountMsat)
+                assertTrue(onchainPaymentSuccessfulEvent.feePaidMsat != null)
+            }
+            else -> error("Expected successful on-chain payment event")
+        }
+        node1.eventHandled()
+
         assertTrue(logWriter1.getLogMessages().isNotEmpty())
         assertTrue(logWriter2.getLogMessages().isNotEmpty())
 

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/builder.rs

@@ -1439,8 +1439,8 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res) =
-		runtime.block_on(async move {
+	let (payment_store_res, node_metris_res, pending_payment_store_res, event_queue_res) = runtime
+		.block_on(async move {
 			tokio::join!(
 				read_all_objects(
 					&*kv_store_ref,
@@ -1454,7 +1454,8 @@ fn build_with_store_internal(
 					PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
 					PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
 					Arc::clone(&logger_ref),
-				)
+				),
+				read_event_queue(Arc::clone(&kv_store_ref), Arc::clone(&logger_ref)),
 			)
 		});
 
@@ -1741,6 +1742,18 @@ fn build_with_store_internal(
 		},
 	};
 
+	let event_queue = match event_queue_res {
+		Ok(event_queue) => Arc::new(event_queue),
+		Err(e) => {
+			if e.kind() == std::io::ErrorKind::NotFound {
+				Arc::new(EventQueue::new(Arc::clone(&kv_store), Arc::clone(&logger)))
+			} else {
+				log_error!(logger, "Failed to read event queue from store: {}", e);
+				return Err(BuildError::ReadFailed);
+			}
+		},
+	};
+
 	let wallet = Arc::new(Wallet::new(
 		bdk_wallet,
 		wallet_persister,
@@ -1750,6 +1763,7 @@ fn build_with_store_internal(
 		Arc::clone(&payment_store),
 		Arc::clone(&runtime),
 		Arc::clone(&config),
+		Arc::clone(&event_queue),
 		Arc::clone(&logger),
 		Arc::clone(&pending_payment_store),
 	));
@@ -1852,7 +1866,6 @@ fn build_with_store_internal(
 		external_scores_res,
 		channel_manager_bytes_res,
 		sweeper_bytes_res,
-		event_queue_res,
 		peer_info_res,
 	) = runtime.block_on(async move {
 		tokio::join!(
@@ -1865,7 +1878,6 @@ fn build_with_store_internal(
 				CHANNEL_MANAGER_PERSISTENCE_KEY,
 			),
 			output_sweeper_future,
-			read_event_queue(Arc::clone(&kv_store_ref), Arc::clone(&logger_ref)),
 			read_peer_info(Arc::clone(&kv_store_ref), Arc::clone(&logger_ref)),
 		)
 	});
@@ -2211,18 +2223,6 @@ fn build_with_store_internal(
 		},
 	};
 
-	let event_queue = match event_queue_res {
-		Ok(event_queue) => Arc::new(event_queue),
-		Err(e) => {
-			if e.kind() == std::io::ErrorKind::NotFound {
-				Arc::new(EventQueue::new(Arc::clone(&kv_store), Arc::clone(&logger)))
-			} else {
-				log_error!(logger, "Failed to read event queue from store: {}", e);
-				return Err(BuildError::ReadFailed);
-			}
-		},
-	};
-
 	let peer_store = match peer_info_res {
 		Ok(peer_store) => Arc::new(peer_store),
 		Err(e) => {

src/data_store.rs

@@ -82,33 +82,43 @@ where
 	}
 
 	pub(crate) async fn insert_or_update(&self, object: SO) -> Result<bool, Error> {
+		self.insert_or_update_with(object, |updated, _| updated).await
+	}
+
+	/// Inserts `object` or merges it into an existing object, returning whether the store changed
+	/// and the effective object after the merge.
+	pub(crate) async fn insert_or_update_and_get(&self, object: SO) -> Result<(bool, SO), Error> {
+		self.insert_or_update_with(object, |updated, stored_object| {
+			(updated, stored_object.clone())
+		})
+		.await
+	}
+
+	async fn insert_or_update_with<R>(
+		&self, object: SO, result_fn: impl FnOnce(bool, &SO) -> R,
+	) -> Result<R, Error> {
 		let _guard = self.mutation_lock.lock().await;
 
 		let id = object.id();
-		let data_to_persist = {
+		let updated_object = {
 			let locked_objects = self.objects.lock().expect("lock");
 			if let Some(existing_object) = locked_objects.get(&id) {
 				let mut updated_object = existing_object.clone();
 				let updated = updated_object.update(object.to_update());
 				if updated {
-					Some(updated_object)
+					updated_object
 				} else {
-					None
+					return Ok(result_fn(false, existing_object));
 				}
 			} else {
-				Some(object)
+				object
 			}
 		};
 
-		match data_to_persist {
-			Some(updated_object) => {
-				self.persist(&updated_object).await?;
-				let mut locked_objects = self.objects.lock().expect("lock");
-				locked_objects.insert(id, updated_object);
-				Ok(true)
-			},
-			None => Ok(false),
-		}
+		self.persist(&updated_object).await?;
+		let mut locked_objects = self.objects.lock().expect("lock");
+		let stored_object = locked_objects.entry(id).insert_entry(updated_object).into_mut();
+		Ok(result_fn(true, stored_object))
 	}
 
 	pub(crate) async fn remove(&self, id: &SO::Id) -> Result<(), Error> {
@@ -287,6 +297,52 @@ mod tests {
 		(2, data, required),
 	});
 
+	struct MergingTestObjectUpdate {
+		id: TestObjectId,
+		data: [u8; 3],
+	}
+
+	impl StorableObjectUpdate<MergingTestObject> for MergingTestObjectUpdate {
+		fn id(&self) -> TestObjectId {
+			self.id
+		}
+	}
+
+	#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+	struct MergingTestObject {
+		id: TestObjectId,
+		data: [u8; 3],
+		preserved_data: [u8; 3],
+	}
+
+	impl StorableObject for MergingTestObject {
+		type Id = TestObjectId;
+		type Update = MergingTestObjectUpdate;
+
+		fn id(&self) -> Self::Id {
+			self.id
+		}
+
+		fn update(&mut self, update: Self::Update) -> bool {
+			if self.data != update.data {
+				self.data = update.data;
+				true
+			} else {
+				false
+			}
+		}
+
+		fn to_update(&self) -> Self::Update {
+			Self::Update { id: self.id, data: self.data }
+		}
+	}
+
+	impl_writeable_tlv_based!(MergingTestObject, {
+		(0, id, required),
+		(2, data, required),
+		(4, preserved_data, required),
+	});
+
 	struct FailingStore;
 
 	impl KVStore for FailingStore {
@@ -403,6 +459,35 @@ mod tests {
 		assert_eq!(Ok(true), data_store.insert_or_update(new_iou_object).await);
 	}
 
+	#[tokio::test]
+	async fn insert_or_update_and_get_returns_merged_object() {
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(InMemoryStore::new()));
+		let logger = Arc::new(TestLogger::new());
+		let id = TestObjectId { id: [42u8; 4] };
+		let existing = MergingTestObject { id, data: [23u8; 3], preserved_data: [24u8; 3] };
+		let data_store = DataStore::new(
+			vec![existing],
+			"datastore_test_primary".to_string(),
+			"datastore_test_secondary".to_string(),
+			store,
+			logger,
+		);
+
+		let inserted = MergingTestObject {
+			id: TestObjectId { id: [43u8; 4] },
+			data: [44u8; 3],
+			preserved_data: [45u8; 3],
+		};
+		assert_eq!(Ok((true, inserted)), data_store.insert_or_update_and_get(inserted).await);
+
+		let supplied = MergingTestObject { id, data: [25u8; 3], preserved_data: [26u8; 3] };
+		let expected = MergingTestObject { data: supplied.data, ..existing };
+		assert_eq!(Ok((true, expected)), data_store.insert_or_update_and_get(supplied).await);
+
+		let unchanged = MergingTestObject { preserved_data: [27u8; 3], ..expected };
+		assert_eq!(Ok((false, expected)), data_store.insert_or_update_and_get(unchanged).await);
+	}
+
 	#[tokio::test]
 	async fn insert_or_update_does_not_mutate_memory_if_persist_fails() {
 		let existing_id = TestObjectId { id: [42u8; 4] };