diff --git a/benches/payments.rs b/benches/payments.rs index 52769d7949..90bac68506 100644 --- a/benches/payments.rs +++ b/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, diff --git a/bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt b/bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt index 006878a4c8..33a1679b61 100644 --- a/bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt +++ b/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 diff --git a/bindings/python/src/ldk_node/test_ldk_node.py b/bindings/python/src/ldk_node/test_ldk_node.py index 304caf9c04..de3e89b2b5 100644 --- a/bindings/python/src/ldk_node/test_ldk_node.py +++ b/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) diff --git a/src/builder.rs b/src/builder.rs index 639838ff3a..421e56c638 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -1440,8 +1440,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, @@ -1455,7 +1455,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)), ) }); @@ -1742,6 +1743,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, @@ -1751,6 +1764,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), )); @@ -1853,7 +1867,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!( @@ -1866,7 +1879,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)), ) }); @@ -2212,18 +2224,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) => { diff --git a/src/data_store.rs b/src/data_store.rs index b1ed816df9..55fa5800f6 100644 --- a/src/data_store.rs +++ b/src/data_store.rs @@ -82,33 +82,43 @@ where } pub(crate) async fn insert_or_update(&self, object: SO) -> Result { + 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( + &self, object: SO, result_fn: impl FnOnce(bool, &SO) -> R, + ) -> Result { 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 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,28 @@ 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 = 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 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] }; diff --git a/src/event.rs b/src/event.rs index 91ab7b27de..0d467312fa 100644 --- a/src/event.rs +++ b/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, }, + /// 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 diff --git a/src/io/in_memory_store.rs b/src/io/in_memory_store.rs index 156fef3a38..f507776680 100644 --- a/src/io/in_memory_store.rs +++ b/src/io/in_memory_store.rs @@ -222,29 +222,3 @@ impl MigratableKVStore for InMemoryStore { unsafe impl Sync for InMemoryStore {} unsafe impl Send for InMemoryStore {} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn in_memory_store_list_all_keys() { - let store = InMemoryStore::new(); - - KVStore::write(&store, "ns_a", "sub_a", "key_a", vec![1u8]).await.unwrap(); - KVStore::write(&store, "ns_a", "sub_b", "key_b", vec![2u8]).await.unwrap(); - KVStore::write(&store, "ns_b", "", "key_c", vec![3u8]).await.unwrap(); - - let mut keys = MigratableKVStore::list_all_keys(&store).await.unwrap(); - keys.sort(); - - assert_eq!( - keys, - vec![ - ("ns_a".to_string(), "sub_a".to_string(), "key_a".to_string()), - ("ns_a".to_string(), "sub_b".to_string(), "key_b".to_string()), - ("ns_b".to_string(), "".to_string(), "key_c".to_string()), - ] - ); - } -} diff --git a/src/io/test_utils.rs b/src/io/test_utils.rs index aadb4b79a8..ea244ea978 100644 --- a/src/io/test_utils.rs +++ b/src/io/test_utils.rs @@ -161,6 +161,33 @@ const EXPECTED_UPDATES_PER_PAYMENT: u64 = 5; pub(crate) use in_memory_store::InMemoryStore; +#[cfg(test)] +mod tests { + use super::InMemoryStore; + use lightning::util::persist::{KVStore, MigratableKVStore}; + + #[tokio::test] + async fn in_memory_store_list_all_keys() { + let store = InMemoryStore::new(); + + KVStore::write(&store, "ns_a", "sub_a", "key_a", vec![1u8]).await.unwrap(); + KVStore::write(&store, "ns_a", "sub_b", "key_b", vec![2u8]).await.unwrap(); + KVStore::write(&store, "ns_b", "", "key_c", vec![3u8]).await.unwrap(); + + let mut keys = MigratableKVStore::list_all_keys(&store).await.unwrap(); + keys.sort(); + + assert_eq!( + keys, + vec![ + ("ns_a".to_string(), "sub_a".to_string(), "key_a".to_string()), + ("ns_a".to_string(), "sub_b".to_string(), "key_b".to_string()), + ("ns_b".to_string(), "".to_string(), "key_c".to_string()), + ] + ); + } +} + pub(crate) fn random_storage_path() -> PathBuf { let mut temp_path = std::env::temp_dir(); let mut rng = rng(); diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index a13019df11..7574ade729 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -54,6 +54,7 @@ use lightning_invoice::RawBolt11Invoice; use persist::KVStoreWalletPersister; use crate::config::Config; +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; @@ -91,6 +92,7 @@ pub(crate) struct Wallet { payment_store: Arc, runtime: Arc, config: Arc, + event_queue: Arc>>, logger: Arc, pending_payment_store: Arc, } @@ -101,7 +103,8 @@ impl Wallet { wallet_persister: KVStoreWalletPersister, broadcaster: Arc, fee_estimator: Arc, chain_source: Arc, payment_store: Arc, runtime: Arc, config: Arc, - logger: Arc, pending_payment_store: Arc, + event_queue: Arc>>, logger: Arc, + pending_payment_store: Arc, ) -> Self { let inner = Mutex::new(wallet); let persister = Mutex::new(wallet_persister); @@ -114,6 +117,7 @@ impl Wallet { payment_store, runtime, config, + event_queue, logger, pending_payment_store, } @@ -274,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 = @@ -300,19 +310,26 @@ impl Wallet { let mut unconfirmed_outbound_txids: Vec = 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 { @@ -320,7 +337,7 @@ impl Wallet { status: ConfirmationStatus::Unconfirmed, .. } if payment.details.direction == PaymentDirection::Outbound => { - unconfirmed_outbound_txids.push(txid); + unconfirmed_outbound_txids.push(*txid); }, _ => {}, } @@ -1443,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 { let direct_payment_id = PaymentId(target_txid.to_byte_array()); if self.pending_payment_store.contains_key(&direct_payment_id) { diff --git a/tests/common/mod.rs b/tests/common/mod.rs index deb8790ea9..fdad3c1c8d 100644 --- a/tests/common/mod.rs +++ b/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; #[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 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( bitcoind: &BitcoindClient, electrs: &E, addrs: Vec
, 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(bitcoind: &BitcoindClient, electrs: &E) { @@ -1086,7 +1149,7 @@ pub(crate) async fn do_channel_full_cycle( 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( 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( 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()); diff --git a/tests/common/scenarios/mod.rs b/tests/common/scenarios/mod.rs index 7cbf56b8e1..8ae4431008 100644 --- a/tests/common/scenarios/mod.rs +++ b/tests/common/scenarios/mod.rs @@ -20,10 +20,14 @@ use std::time::Duration; use bitcoin::Amount; use electrsd::corepc_node::Client as BitcoindClient; use electrsd::electrum_client::ElectrumApi; +use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY; use ldk_node::{Event, Node}; use super::external_node::ExternalNode; -use super::{generate_blocks_and_wait, premine_and_distribute_funds}; +use super::{ + generate_blocks_and_wait, premine_and_distribute_funds, ExpectOnchainPaymentEvent, + OnchainPaymentEvent, +}; #[derive(Debug, Clone, Copy)] pub(crate) enum Side { @@ -107,7 +111,14 @@ pub(crate) async fn setup_interop_test( ) { let ldk_address = node.onchain_payment().new_address().unwrap(); let premine_amount = Amount::from_sat(50_000_000); - premine_and_distribute_funds(bitcoind, electrs, vec![ldk_address], premine_amount).await; + let premine_txid = + premine_and_distribute_funds(bitcoind, electrs, vec![ldk_address], premine_amount).await; + generate_blocks_and_wait(bitcoind, electrs, (ANTI_REORG_DELAY - 1) as usize).await; + sync_wallets_with_retry(node).await; + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); // Fund the peer via the ldk_node_test wallet loaded by premine_and_distribute_funds. let ext_funding_addr_str = peer.get_funding_address().await.unwrap(); diff --git a/tests/integration_tests_hrn.rs b/tests/integration_tests_hrn.rs index 9102400398..b9bb80d34b 100644 --- a/tests/integration_tests_hrn.rs +++ b/tests/integration_tests_hrn.rs @@ -13,8 +13,9 @@ use bitcoin::Amount; use common::{ expect_channel_ready_event, expect_payment_successful_event, generate_blocks_and_wait, open_channel, premine_and_distribute_funds, random_chain_source, setup_bitcoind_and_electrsd, - setup_two_nodes, TestChainSource, + setup_two_nodes, ExpectOnchainPaymentEvent, OnchainPaymentEvent, TestChainSource, }; +use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY; use ldk_node::payment::UnifiedPaymentResult; use ldk_node::Event; use lightning::ln::channelmanager::PaymentId; @@ -29,7 +30,7 @@ async fn unified_send_to_hrn() { let address_a = node_a.onchain_payment().new_address().unwrap(); let premined_sats = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -38,6 +39,13 @@ async fn unified_send_to_hrn() { .await; node_a.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index b07a90629f..4de9735ea4 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -24,13 +24,14 @@ use common::{ generate_listening_addresses, invalidate_blocks, open_channel, open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds, premine_blocks, prepare_rbf, random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder, setup_node, - setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, TestChainSource, TestConfig, - TestStoreType, TestSyncStore, + setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, ExpectOnchainPaymentEvent, + OnchainPaymentEvent, TestChainSource, TestConfig, TestStoreType, TestSyncStore, }; use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; use ldk_node::config::{AsyncPaymentsRole, EsploraSyncConfig, DEFAULT_FULL_SCAN_STOP_GAP}; use ldk_node::entropy::NodeEntropy; +use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY; use ldk_node::liquidity::LSPS2ServiceConfig; use ldk_node::payment::{ ConfirmationStatus, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, @@ -140,7 +141,7 @@ async fn peer_removed_when_counterparty_force_closes_last_channel() { let address_a = node_a.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -148,6 +149,13 @@ async fn peer_removed_when_counterparty_force_closes_last_channel() { ) .await; node_a.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); // node_a opens the channel, so node_a persists node_b in its peer store. open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; @@ -305,7 +313,7 @@ async fn multi_hop_sending() { let addresses = nodes.iter().map(|n| n.onchain_payment().new_address().unwrap()).collect(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, addresses, @@ -318,6 +326,15 @@ async fn multi_hop_sending() { assert_eq!(n.list_balances().spendable_onchain_balance_sats, premine_amount_sat); assert_eq!(n.next_event(), None); } + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in &nodes { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } // Setup channel topology: // (1M:0)- N2 -(1M:0) @@ -395,7 +412,7 @@ async fn split_underpaid_bolt11_payment() { let addr_b = node_b.onchain_payment().new_address().unwrap(); let addr_c = node_c.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b, addr_c], @@ -407,6 +424,15 @@ async fn split_underpaid_bolt11_payment() { node.sync_wallets().unwrap(); assert_eq!(node.list_balances().spendable_onchain_balance_sats, premine_amount_sat); } + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&node_a, &node_b, &node_c] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } // The receiver opens both channels and pushes liquidity to both payers so each payer can send // half of the invoice back. @@ -565,7 +591,7 @@ async fn onchain_send_receive() { let addr_c = unchecked_address.assume_checked(); let premine_amount_sat = 1_100_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a.clone(), addr_b.clone()], @@ -596,6 +622,19 @@ async fn onchain_send_receive() { } } + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); + let channel_amount_sat = 1_000_000; let reserve_amount_sat = 25_000; open_channel(&node_b, &node_a, channel_amount_sat, true, &electrsd).await; @@ -673,6 +712,8 @@ async fn onchain_send_receive() { generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); + assert_eq!(node_b.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, txid,); let expected_node_a_balance = expected_node_a_balance + amount_to_send_sats; let expected_node_b_balance_lower = expected_node_b_balance_lower - amount_to_send_sats; @@ -715,6 +756,8 @@ async fn onchain_send_receive() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, txid,); + assert_eq!(node_b.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); let expected_node_b_balance_lower = expected_node_b_balance_lower + expected_node_a_balance; let expected_node_b_balance_upper = expected_node_b_balance_upper + expected_node_a_balance; @@ -738,6 +781,8 @@ async fn onchain_send_receive() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, txid,); + assert_eq!(node_b.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); let expected_node_b_balance_lower = expected_node_b_balance_lower + reserve_amount_sat; let expected_node_b_balance_upper = expected_node_b_balance_upper + reserve_amount_sat; @@ -784,6 +829,8 @@ async fn reorged_onchain_payment_returns_to_unconfirmed() { generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.next_event(), None); + assert_eq!(node_b.next_event(), None); let payment_id = PaymentId(txid.to_byte_array()); for node in [&node_a, &node_b] { @@ -837,7 +884,7 @@ async fn onchain_send_all_retains_reserve() { let premine_amount_sat = 1_000_000; let reserve_amount_sat = 25_000; let onchain_fee_buffer_sat = 1000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a.clone(), addr_b.clone()], @@ -847,6 +894,18 @@ async fn onchain_send_all_retains_reserve() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, premine_amount_sat); assert_eq!(node_b.list_balances().spendable_onchain_balance_sats, premine_amount_sat); @@ -858,6 +917,8 @@ async fn onchain_send_all_retains_reserve() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, txid,); + assert_eq!(node_b.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); // Check node a sent all and node b received it assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, 0); assert!(((premine_amount_sat * 2 - onchain_fee_buffer_sat)..=(premine_amount_sat * 2)) @@ -875,6 +936,8 @@ async fn onchain_send_all_retains_reserve() { generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); + assert_eq!(node_b.next_event(), None); assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, reserve_amount_sat); // Open a channel. @@ -899,6 +962,8 @@ async fn onchain_send_all_retains_reserve() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); + assert_eq!(node_b.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, txid,); // Check node b sent all and node a received it assert_eq!(node_b.list_balances().total_onchain_balance_sats, reserve_amount_sat); @@ -1536,7 +1601,7 @@ async fn splice_channel() { let address_a = node_a.onchain_payment().new_address().unwrap(); let address_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a, address_b], @@ -1549,6 +1614,18 @@ async fn splice_channel() { assert_eq!(node_a.list_balances().total_onchain_balance_sats, premine_amount_sat); assert_eq!(node_b.list_balances().total_onchain_balance_sats, premine_amount_sat); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; @@ -1751,7 +1828,7 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { let address_a = node_a.onchain_payment().new_address().unwrap(); let address_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a, address_b], @@ -1761,6 +1838,18 @@ async fn run_rbf_splice_channel_test(confirm_original: bool) { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; @@ -1949,7 +2038,7 @@ async fn funding_payment_graduates_without_channel_ready() { let address_a = node_a.onchain_payment().new_address().unwrap(); let address_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a, address_b], @@ -1959,6 +2048,18 @@ async fn funding_payment_graduates_without_channel_ready() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); // node_a funds the channel, so it holds the funding payment. `open_channel` drains only the // `ChannelPending` events, leaving any `ChannelReady` queued and undrained. @@ -2004,7 +2105,7 @@ async fn splice_payment_reorged_to_unconfirmed() { let address_a = node_a.onchain_payment().new_address().unwrap(); let address_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a, address_b], @@ -2014,6 +2115,18 @@ async fn splice_payment_reorged_to_unconfirmed() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; @@ -2081,7 +2194,7 @@ async fn splice_in_rbf_joins_counterparty_splice() { let address_a = node_a.onchain_payment().new_address().unwrap(); let address_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a, address_b], @@ -2091,6 +2204,18 @@ async fn splice_in_rbf_joins_counterparty_splice() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; @@ -2129,7 +2254,7 @@ async fn simple_bolt12_send_receive() { let address_a = node_a.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -2138,6 +2263,13 @@ async fn simple_bolt12_send_receive() { .await; node_a.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; @@ -2400,7 +2532,7 @@ async fn async_payment() { let address_receiver_lsp = node_receiver_lsp.onchain_payment().new_address().unwrap(); let address_receiver = node_receiver.onchain_payment().new_address().unwrap(); let premine_amount_sat = 4_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_sender, address_sender_lsp, address_receiver_lsp, address_receiver], @@ -2412,6 +2544,15 @@ async fn async_payment() { node_sender_lsp.sync_wallets().unwrap(); node_receiver_lsp.sync_wallets().unwrap(); node_receiver.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&node_sender, &node_sender_lsp, &node_receiver_lsp, &node_receiver] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } open_channel(&node_sender, &node_sender_lsp, 400_000, false, &electrsd).await; open_channel(&node_sender_lsp, &node_receiver_lsp, 400_000, true, &electrsd).await; @@ -2528,7 +2669,7 @@ async fn test_node_announcement_propagation() { let address_a = node_a.onchain_payment().new_address().unwrap(); let premine_amount_sat = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -2537,6 +2678,13 @@ async fn test_node_announcement_propagation() { .await; node_a.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); // Open an announced channel from node_a to node_b open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; @@ -2617,7 +2765,7 @@ async fn generate_bip21_uri() { assert!(initial_uni_payment.contains("lightning=")); assert!(!initial_uni_payment.contains("lno=")); // BOLT12 requires channels - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -2626,6 +2774,13 @@ async fn generate_bip21_uri() { .await; node_a.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; @@ -2669,7 +2824,7 @@ async fn unified_send_receive_bip21_uri() { let address_a = node_a.onchain_payment().new_address().unwrap(); let premined_sats = 5_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -2678,6 +2833,13 @@ async fn unified_send_receive_bip21_uri() { .await; node_a.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); open_channel(&node_a, &node_b, 4_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; @@ -2718,6 +2880,7 @@ async fn unified_send_receive_bip21_uri() { }; expect_payment_successful_event!(node_a, Some(offer_payment_id), None); + expect_payment_received_event!(node_b, expected_amount_sats * 1000); // Cut off the BOLT12 part to fallback to BOLT11. let uri_str_without_offer = uri_str.split("&lno=").next().unwrap(); @@ -2738,6 +2901,7 @@ async fn unified_send_receive_bip21_uri() { }, }; expect_payment_successful_event!(node_a, Some(invoice_payment_id), None); + expect_payment_received_event!(node_b, expected_amount_sats * 1000); let expect_onchain_amount_sats = 800_000; let onchain_uni_payment = @@ -2766,6 +2930,8 @@ async fn unified_send_receive_bip21_uri() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!(node_a.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, txid,); + assert_eq!(node_b.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, txid,); assert_eq!(node_b.list_balances().total_onchain_balance_sats, 800_000); assert_eq!(node_b.list_balances().total_lightning_balance_sats, 200_000); @@ -2830,7 +2996,7 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { let premine_amount_sat = 10_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![service_addr, client_addr, payer_addr], @@ -2840,6 +3006,15 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { service_node.sync_wallets().unwrap(); client_node.sync_wallets().unwrap(); payer_node.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&service_node, &client_node, &payer_node] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } // Open a channel payer -> service that will allow paying the JIT invoice println!("Opening channel payer_node -> service_node!"); @@ -3026,7 +3201,7 @@ async fn spontaneous_send_with_custom_preimage() { let address_a = node_a.onchain_payment().new_address().unwrap(); let premine_sat = 1_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![address_a], @@ -3035,6 +3210,13 @@ async fn spontaneous_send_with_custom_preimage() { .await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + node_a.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); open_channel(&node_a, &node_b, 500_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); @@ -3149,7 +3331,7 @@ async fn lsps2_client_trusts_lsp() { let premine_amount_sat = 10_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![service_addr_onchain, client_addr_onchain, payer_addr_onchain], @@ -3159,6 +3341,15 @@ async fn lsps2_client_trusts_lsp() { service_node.sync_wallets().unwrap(); client_node.sync_wallets().unwrap(); payer_node.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&service_node, &client_node, &payer_node] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } println!("Premine complete!"); // Open a channel payer -> service that will allow paying the JIT invoice open_channel(&payer_node, &service_node, 5_000_000, false, &electrsd).await; @@ -3326,7 +3517,7 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { let premine_amount_sat = 10_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![service_addr_onchain, client_addr_onchain, payer_addr_onchain], @@ -3336,6 +3527,15 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { service_node.sync_wallets().unwrap(); client_node.sync_wallets().unwrap(); payer_node.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&service_node, &client_node, &payer_node] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } println!("Premine complete!"); // Open a channel payer -> service that will allow paying the JIT invoice open_channel(&payer_node, &service_node, 5_000_000, false, &electrsd).await; @@ -3423,7 +3623,7 @@ async fn payment_persistence_after_restart() { // Premine sufficient funds for a large channel and many payments let premine_amount_sat = 10_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -3434,6 +3634,22 @@ async fn payment_persistence_after_restart() { node_b.sync_wallets().unwrap(); assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, premine_amount_sat); assert_eq!(node_b.list_balances().spendable_onchain_balance_sats, premine_amount_sat); + generate_blocks_and_wait( + &bitcoind.client, + &electrsd.client, + (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, + ); // Open a large channel from node_a to node_b let channel_amount_sat = 5_000_000; @@ -3704,7 +3920,7 @@ async fn onchain_fee_bump_rbf() { let addr_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 500_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a.clone(), addr_b.clone()], @@ -3714,6 +3930,18 @@ async fn onchain_fee_bump_rbf() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); // Send a transaction from node_b to node_a that we'll later bump let amount_to_send_sats = 100_000; @@ -3800,6 +4028,14 @@ async fn onchain_fee_bump_rbf() { generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + assert_eq!( + node_a.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + second_bump_txid, + ); + assert_eq!( + node_b.expect_onchain_payment_event(OnchainPaymentEvent::Successful).await, + second_bump_txid, + ); assert_eq!( Err(NodeError::InvalidPaymentId), @@ -3845,7 +4081,7 @@ async fn onchain_fee_bump_rbf_respects_anchor_reserve() { let addr_b = node_b.onchain_payment().new_address().unwrap(); let premine_amount_sat = 1_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a.clone(), addr_b], @@ -3855,6 +4091,18 @@ async fn onchain_fee_bump_rbf_respects_anchor_reserve() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); open_channel(&node_b, &node_a, 200_000, false, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; @@ -3895,7 +4143,7 @@ async fn open_channel_with_all_with_anchors() { let premine_amount_sat = 1_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -3905,6 +4153,18 @@ async fn open_channel_with_all_with_anchors() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, premine_amount_sat); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); let funding_txo = open_channel_with_all(&node_a, &node_b, false, &electrsd).await; @@ -3948,7 +4208,7 @@ async fn open_channel_with_all_without_anchors() { let premine_amount_sat = 1_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -3958,6 +4218,18 @@ async fn open_channel_with_all_without_anchors() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, premine_amount_sat); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); let funding_txo = open_channel_with_all(&node_a, &node_b, false, &electrsd).await; @@ -4000,7 +4272,7 @@ async fn splice_in_with_all_balance() { let premine_amount_sat = 5_000_000; let channel_amount_sat = 1_000_000; - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -4010,6 +4282,18 @@ async fn splice_in_with_all_balance() { node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); assert_eq!(node_a.list_balances().spendable_onchain_balance_sats, premine_amount_sat); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (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, + ); // Open a channel with a fixed amount first let funding_txo = open_channel(&node_a, &node_b, channel_amount_sat, false, &electrsd).await; diff --git a/tests/probing_tests.rs b/tests/probing_tests.rs index a26127d735..6d16ec6526 100644 --- a/tests/probing_tests.rs +++ b/tests/probing_tests.rs @@ -25,10 +25,12 @@ use std::sync::atomic::{AtomicBool, Ordering}; use common::{ expect_channel_ready_event, expect_event, generate_blocks_and_wait, open_channel, premine_and_distribute_funds, random_chain_source, random_config, setup_bitcoind_and_electrsd, - setup_node, wait_for_channel_ready_to_send, TestNode, TestStoreType, + setup_node, wait_for_channel_ready_to_send, ExpectOnchainPaymentEvent, OnchainPaymentEvent, + TestNode, TestStoreType, }; use ldk_node::bitcoin::Amount; +use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY; use ldk_node::probing::{ProbingConfigBuilder, ProbingStrategy}; use ldk_node::Event; @@ -141,7 +143,7 @@ async fn probe_budget_increments_and_decrements() { let addr_a = node_a.onchain_payment().new_address().unwrap(); let addr_b = node_b.onchain_payment().new_address().unwrap(); - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -150,6 +152,15 @@ async fn probe_budget_increments_and_decrements() { .await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&node_a, &node_b] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } open_channel(&node_a, &node_b, 1_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; @@ -235,7 +246,7 @@ async fn locked_msat_accounts_for_routing_fees() { let addr_a = node_a.onchain_payment().new_address().unwrap(); let addr_b = node_b.onchain_payment().new_address().unwrap(); - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -244,6 +255,15 @@ async fn locked_msat_accounts_for_routing_fees() { .await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&node_a, &node_b] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } open_channel(&node_a, &node_b, 1_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; @@ -326,7 +346,7 @@ async fn probing_budget_restored_after_node_restart() { let addr_a = node_a.onchain_payment().new_address().unwrap(); let addr_b = node_b.onchain_payment().new_address().unwrap(); - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -335,6 +355,15 @@ async fn probing_budget_restored_after_node_restart() { .await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&node_a, &node_b] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } open_channel(&node_a, &node_b, 1_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; @@ -425,7 +454,7 @@ async fn exhausted_probe_budget_blocks_new_probes() { let addr_a = node_a.onchain_payment().new_address().unwrap(); let addr_b = node_b.onchain_payment().new_address().unwrap(); - premine_and_distribute_funds( + let premine_txid = premine_and_distribute_funds( &bitcoind.client, &electrsd.client, vec![addr_a, addr_b], @@ -434,6 +463,15 @@ async fn exhausted_probe_budget_blocks_new_probes() { .await; node_a.sync_wallets().unwrap(); node_b.sync_wallets().unwrap(); + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, (ANTI_REORG_DELAY - 1) as usize) + .await; + for node in [&node_a, &node_b] { + node.sync_wallets().unwrap(); + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } open_channel(&node_a, &node_b, 1_000_000, true, &electrsd).await; generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 1).await; diff --git a/tests/reorg_test.rs b/tests/reorg_test.rs index 295d9fdd24..9b61775421 100644 --- a/tests/reorg_test.rs +++ b/tests/reorg_test.rs @@ -2,6 +2,7 @@ mod common; use std::collections::HashMap; use bitcoin::Amount; +use ldk_node::lightning::chain::channelmonitor::ANTI_REORG_DELAY; use ldk_node::payment::{PaymentDirection, PaymentKind}; use ldk_node::{Event, LightningBalance, PendingSweepBalance}; use proptest::prelude::prop; @@ -10,7 +11,7 @@ use proptest::proptest; use crate::common::{ expect_event, generate_blocks_and_wait, invalidate_blocks, open_channel, premine_and_distribute_funds, random_chain_source, random_config, setup_bitcoind_and_electrsd, - setup_node, wait_for_outpoint_spend, + setup_node, wait_for_outpoint_spend, ExpectOnchainPaymentEvent, OnchainPaymentEvent, }; proptest! { @@ -53,7 +54,8 @@ proptest! { let amount_sat = 2_100_000; let addr_nodes = nodes.iter().map(|node| node.onchain_payment().new_address().unwrap()).collect::>(); - premine_and_distribute_funds(bitcoind, electrs, addr_nodes, Amount::from_sat(amount_sat)).await; + let premine_txid = + premine_and_distribute_funds(bitcoind, electrs, addr_nodes, Amount::from_sat(amount_sat)).await; macro_rules! sync_wallets { () => { @@ -68,6 +70,14 @@ proptest! { assert_eq!(node.list_balances().total_onchain_balance_sats, amount_sat); }); + generate_blocks_and_wait(bitcoind, electrs, (ANTI_REORG_DELAY - 1) as usize).await; + sync_wallets!(); + for node in &nodes { + assert_eq!( + node.expect_onchain_payment_event(OnchainPaymentEvent::Received).await, + premine_txid, + ); + } let mut nodes_funding_tx = HashMap::new(); let funding_amount_sat = 2_000_000;