Add tests for chain source switching

Cargo.toml

@@ -197,3 +197,18 @@ harness = false
 #lightning-liquidity = { path = "../rust-lightning/lightning-liquidity" }
 #lightning-macros = { path = "../rust-lightning/lightning-macros" }
 #lightning-dns-resolver = { path = "../rust-lightning/lightning-dns-resolver" }
+
+[patch."https://github.com/lightningdevkit/rust-lightning"]
+lightning = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-types = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-invoice = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-net-tokio = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-persister = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-background-processor = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-rapid-gossip-sync = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-block-sync = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-transaction-sync = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-liquidity = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-macros = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+lightning-dns-resolver = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }
+possiblyrandom = { git = "https://git.rust-bitcoin.org/tnull/rust-lightning", rev = "e076821425a4aceb9006bcac6e252417b1b9f8bd" }

src/builder.rs

@@ -2012,6 +2012,7 @@ fn build_with_store_internal(
 				Arc::clone(&channel_manager),
 				Arc::clone(&om_resolver),
 				IgnoringMessageHandler {},
+				false,
 			))
 		} else {
 			Arc::new(OnionMessenger::new(

src/data_store.rs

@@ -223,7 +223,7 @@ where
 
 #[cfg(test)]
 mod tests {
-	use lightning::impl_writeable_tlv_based;
+	use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 	use lightning::io;
 	use lightning::util::persist::{PageToken, PaginatedKVStore, PaginatedListResponse};
 	use lightning::util::test_utils::TestLogger;

src/event.rs

@@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex};
 use bitcoin::blockdata::locktime::absolute::LockTime;
 use bitcoin::secp256k1::PublicKey;
 use bitcoin::{Amount, OutPoint};
+use lightning::blinded_path::message::NextMessageHop;
 use lightning::events::bump_transaction::BumpTransactionEvent;
 #[cfg(not(feature = "uniffi"))]
 use lightning::events::PaidBolt12Invoice;
@@ -29,7 +30,10 @@ use lightning::util::config::{ChannelConfigOverrides, ChannelConfigUpdate};
 use lightning::util::errors::APIError;
 use lightning::util::persist::KVStore;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
-use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum};
+use lightning::{
+	impl_ser_tlv_based as impl_writeable_tlv_based,
+	impl_ser_tlv_based_enum as impl_writeable_tlv_based_enum,
+};
 use lightning_liquidity::lsps2::utils::compute_opening_fee;
 use lightning_types::payment::{PaymentHash, PaymentPreimage};
 
@@ -1725,7 +1729,11 @@ where
 
 				self.bump_tx_event_handler.handle_event(&bte).await;
 			},
-			LdkEvent::OnionMessageIntercepted { peer_node_id, message } => {
+			LdkEvent::OnionMessageIntercepted {
+				next_hop: NextMessageHop::NodeId(peer_node_id),
+				message,
+				..
+			} => {
 				if let Some(om_mailbox) = self.om_mailbox.as_ref() {
 					om_mailbox.onion_message_intercepted(peer_node_id, message);
 				} else {
@@ -1735,6 +1743,12 @@ where
 					);
 				}
 			},
+			LdkEvent::OnionMessageIntercepted {
+				next_hop: NextMessageHop::ShortChannelId(_),
+				..
+			} => {
+				log_trace!(self.logger, "Ignoring onion message intercepted for unknown SCID");
+			},
 			LdkEvent::OnionMessagePeerConnected { peer_node_id } => {
 				if let Some(om_mailbox) = self.om_mailbox.as_ref() {
 					let messages = om_mailbox.onion_message_peer_connected(peer_node_id);

src/io/vss_store.rs

@@ -21,7 +21,7 @@ use bitcoin::bip32::{ChildNumber, Xpriv};
 use bitcoin::hashes::{sha256, Hash, HashEngine, Hmac, HmacEngine};
 use bitcoin::key::Secp256k1;
 use bitcoin::Network;
-use lightning::impl_writeable_tlv_based_enum;
+use lightning::impl_ser_tlv_based_enum as impl_writeable_tlv_based_enum;
 use lightning::io::{self, Error, ErrorKind};
 use lightning::sign::{EntropySource as LdkEntropySource, RandomBytes};
 use lightning::util::persist::{

src/lib.rs

@@ -146,7 +146,7 @@ use graph::NetworkGraph;
 use io::utils::update_and_persist_node_metrics;
 pub use lightning;
 use lightning::chain::BlockLocator;
-use lightning::impl_writeable_tlv_based;
+use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 use lightning::ln::chan_utils::FUNDING_TRANSACTION_WITNESS_WEIGHT;
 use lightning::ln::channel_state::ChannelDetails as LdkChannelDetails;
 pub use lightning::ln::channel_state::ChannelShutdownState;

src/payment/asynchronous/static_invoice_store.rs

@@ -13,7 +13,7 @@ use std::time::Duration;
 use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::hashes::Hash;
 use lightning::blinded_path::message::BlindedMessagePath;
-use lightning::impl_writeable_tlv_based;
+use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 use lightning::offers::static_invoice::StaticInvoice;
 use lightning::util::persist::KVStore;
 use lightning::util::ser::{Readable, Writeable};

src/payment/bolt11.rs

@@ -13,7 +13,7 @@ use std::sync::{Arc, RwLock};
 
 use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::hashes::Hash;
-use lightning::impl_writeable_tlv_based;
+use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 use lightning::ln::channelmanager::{
 	Bolt11InvoiceParameters, OptionalBolt11PaymentParams, PaymentId,
 };
@@ -313,7 +313,7 @@ impl Bolt11Payment {
 				let payee_pubkey = invoice.recover_payee_pub_key();
 				log_info!(
 					self.logger,
-					"Initiated sending {} msat to {}",
+					"Initiated sending {} msat to {:?}",
 					payment_amount_msat,
 					payee_pubkey
 				);

src/payment/pending_payment_store.rs

@@ -6,7 +6,7 @@
 // accordance with one or both of these licenses.
 
 use bitcoin::Txid;
-use lightning::impl_writeable_tlv_based;
+use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 use lightning::ln::channelmanager::PaymentId;
 
 use crate::data_store::{StorableObject, StorableObjectUpdate};

src/payment/store.rs

@@ -16,8 +16,8 @@ use lightning::ln::types::ChannelId;
 use lightning::offers::offer::OfferId;
 use lightning::util::ser::{Readable, Writeable};
 use lightning::{
-	_init_and_read_len_prefixed_tlv_fields, impl_writeable_tlv_based,
-	impl_writeable_tlv_based_enum, write_tlv_fields,
+	_init_and_read_len_prefixed_tlv_fields, impl_ser_tlv_based as impl_writeable_tlv_based,
+	impl_ser_tlv_based_enum as impl_writeable_tlv_based_enum, write_tlv_fields,
 };
 use lightning_types::payment::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning_types::string::UntrustedString;

src/peer_store.rs

@@ -10,7 +10,7 @@ use std::ops::Deref;
 use std::sync::{Arc, RwLock};
 
 use bitcoin::secp256k1::PublicKey;
-use lightning::impl_writeable_tlv_based;
+use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 use lightning::util::persist::KVStore;
 use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 

src/types.rs

@@ -19,7 +19,7 @@ use bitcoin_payment_instructions::hrn_resolution::{
 };
 use bitcoin_payment_instructions::onion_message_resolver::LDKOnionMessageDNSSECHrnResolver;
 use lightning::chain::chainmonitor;
-use lightning::impl_writeable_tlv_based;
+use lightning::impl_ser_tlv_based as impl_writeable_tlv_based;
 use lightning::ln::channel_state::{
 	ChannelDetails as LdkChannelDetails, ChannelShutdownState, CounterpartyForwardingInfo,
 };

tests/common/mod.rs

@@ -21,9 +21,9 @@ pub(crate) mod lnd;
 use std::collections::{HashMap, HashSet};
 use std::env;
 use std::future::Future;
+use std::net::TcpListener;
 use std::path::PathBuf;
 use std::str::FromStr;
-use std::sync::atomic::{AtomicU16, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -358,13 +358,14 @@ pub(crate) fn random_storage_path() -> PathBuf {
 	temp_path
 }
 
-static NEXT_PORT: AtomicU16 = AtomicU16::new(20000);
-
 pub(crate) fn generate_listening_addresses() -> Vec<SocketAddress> {
-	let port = NEXT_PORT.fetch_add(2, Ordering::Relaxed);
+	let listener_a = TcpListener::bind(("127.0.0.1", 0)).expect("available listener port");
+	let listener_b = TcpListener::bind(("127.0.0.1", 0)).expect("available listener port");
+	let port_a = listener_a.local_addr().expect("listener address").port();
+	let port_b = listener_b.local_addr().expect("listener address").port();
 	vec![
-		SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port },
-		SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: port + 1 },
+		SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: port_a },
+		SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: port_b },
 	]
 }
 
@@ -635,7 +636,9 @@ pub(crate) async fn generate_blocks_and_wait<E: ElectrumApi>(
 	let address = bitcoind.new_address().expect("failed to get new address");
 	// TODO: expect this Result once the WouldBlock issue is resolved upstream.
 	let _block_hashes_res = bitcoind.generate_to_address(num, &address);
-	wait_for_block(electrs, cur_height as usize + num).await;
+	let min_height = cur_height as usize + num;
+	wait_for_bitcoind_block(bitcoind, min_height).await;
+	wait_for_block(electrs, min_height).await;
 	print!(" Done!");
 	println!("\n");
 }
@@ -655,26 +658,42 @@ pub(crate) fn invalidate_blocks(bitcoind: &BitcoindClient, num_blocks: usize) {
 	assert!(new_cur_height + num_blocks == cur_height);
 }
 
+async fn wait_for_bitcoind_block(bitcoind: &BitcoindClient, min_height: usize) {
+	let mut delay = Duration::from_millis(64);
+	let mut tries = 0;
+	loop {
+		let height =
+			bitcoind.get_blockchain_info().expect("failed to get blockchain info").blocks as usize;
+		if height >= min_height {
+			return;
+		}
+		assert!(
+			tries < 120,
+			"bitcoind height did not reach {} within 60 seconds, current height {}",
+			min_height,
+			height
+		);
+		tries += 1;
+		tokio::time::sleep(delay).await;
+		if delay.as_millis() < 512 {
+			delay = delay.mul_f32(2.0);
+		}
+	}
+}
+
 pub(crate) async fn wait_for_block<E: ElectrumApi>(electrs: &E, min_height: usize) {
-	let mut header = match electrs.block_headers_subscribe() {
-		Ok(header) => header,
-		Err(_) => {
-			// While subscribing should succeed the first time around, we ran into some cases where
-			// it didn't. Since we can't proceed without subscribing, we try again after a delay
-			// and panic if it still fails.
-			tokio::time::sleep(Duration::from_secs(3)).await;
-			electrs.block_headers_subscribe().expect("failed to subscribe to block headers")
-		},
-	};
+	let mut delay = Duration::from_millis(64);
+	let mut tries = 0;
 	loop {
-		if header.height >= min_height {
-			break;
+		if electrs.block_header(min_height).is_ok() {
+			return;
+		}
+		assert!(tries < 120, "electrs did not serve block header {} within 60 seconds", min_height);
+		tries += 1;
+		tokio::time::sleep(delay).await;
+		if delay.as_millis() < 512 {
+			delay = delay.mul_f32(2.0);
 		}
-		header = exponential_backoff_poll(|| {
-			electrs.ping().expect("failed to ping electrs");
-			electrs.block_headers_pop().expect("failed to pop block header")
-		})
-		.await;
 	}
 }
 
@@ -729,6 +748,43 @@ where
 	}
 }
 
+pub(crate) async fn stop_nodes(node_a: TestNode, node_b: TestNode) {
+	let (stop_a_sender, stop_a_receiver) = tokio::sync::oneshot::channel();
+	let (stop_b_sender, stop_b_receiver) = tokio::sync::oneshot::channel();
+	std::thread::spawn(move || {
+		let _ = stop_a_sender.send(node_a.stop());
+	});
+	std::thread::spawn(move || {
+		let _ = stop_b_sender.send(node_b.stop());
+	});
+	stop_a_receiver.await.expect("node_a stop thread panicked").unwrap();
+	stop_b_receiver.await.expect("node_b stop thread panicked").unwrap();
+}
+
+pub(crate) async fn stop_nodes_concurrently(nodes: Vec<TestNode>) {
+	let stop_receivers = nodes
+		.into_iter()
+		.map(|node| {
+			let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel();
+			std::thread::spawn(move || {
+				let _ = stop_sender.send(node.stop());
+			});
+			stop_receiver
+		})
+		.collect::<Vec<_>>();
+
+	for stop_receiver in stop_receivers {
+		stop_receiver.await.expect("node stop thread panicked").unwrap();
+	}
+}
+
+pub(crate) async fn stop_node(node: TestNode) {
+	let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel();
+	std::thread::spawn(move || {
+		let _ = stop_sender.send(node.stop());
+	});
+	stop_receiver.await.expect("node stop thread panicked").unwrap();
+}
 pub(crate) async fn premine_and_distribute_funds<E: ElectrumApi>(
 	bitcoind: &BitcoindClient, electrs: &E, addrs: Vec<Address>, amount: Amount,
 ) {