mod.rs

// This file is Copyright its original authors, visible in version control history.
//
// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
// http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
// accordance with one or both of these licenses.

#![cfg(any(test, cln_test, lnd_test, eclair_test, vss_test))]
#![allow(dead_code)]

pub(crate) mod external_node;
pub(crate) mod logging;

#[cfg(cln_test)]
pub(crate) mod cln;
#[cfg(eclair_test)]
pub(crate) mod eclair;
#[cfg(lnd_test)]
pub(crate) mod lnd;

use std::collections::{HashMap, HashSet};
use std::env;
use std::future::Future;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::atomic::{AtomicU16, Ordering};
use std::sync::Arc;
use std::time::Duration;

use bitcoin::hashes::hex::FromHex;
use bitcoin::hashes::sha256::Hash as Sha256;
use bitcoin::hashes::Hash;
use bitcoin::secp256k1::PublicKey;
use bitcoin::{
	Address, Amount, Network, OutPoint, ScriptBuf, Sequence, Transaction, Txid, Witness,
};
use electrsd::corepc_node::{Client as BitcoindClient, Node as BitcoinD};
use electrsd::electrum_client::ElectrumApi;
use electrsd::{corepc_node, ElectrsD};
use ldk_node::config::{
	AsyncPaymentsRole, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig,
	HumanReadableNamesConfig,
};
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::probing::ProbingConfig;
use ldk_node::{
	Builder, ChannelShutdownState, CustomTlvRecord, Event, LightningBalance, Node, NodeError,
	PendingSweepBalance, UserChannelId,
};
use lightning::io;
use lightning::ln::msgs::SocketAddress;
use lightning::routing::gossip::NodeAlias;
use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse};
use lightning_invoice::{Bolt11InvoiceDescription, Description};
use lightning_persister::fs_store::v1::FilesystemStore;
use lightning_types::payment::{PaymentHash, PaymentPreimage};
use logging::TestLogWriter;
use rand::distr::Alphanumeric;
use rand::{rng, Rng};
use serde_json::{json, Value};

#[path = "../../src/io/in_memory_store.rs"]
mod in_memory_store;
pub(crate) use in_memory_store::InMemoryStore;

/// Shared timeout (in seconds) for waiting on LDK events and external node operations.
pub(crate) const INTEROP_TIMEOUT_SECS: u64 = 60;

macro_rules! expect_event {
	($node:expr, $event_type:ident) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!(
				"{} timed out waiting for {} event after 60s",
				$node.node_id(),
				std::stringify!($event_type)
			)
		});
		match event {
			ref e @ Event::$event_type { .. } => {
				println!("{} got event {:?}", $node.node_id(), e);
				$node.event_handled().unwrap();
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
			},
		}
	}};
}

pub(crate) use expect_event;

macro_rules! expect_channel_pending_event {
	($node:expr, $counterparty_node_id:expr) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!("{} timed out waiting for ChannelPending event after 60s", $node.node_id())
		});
		match event {
			ref e @ Event::ChannelPending { funding_txo, counterparty_node_id, .. } => {
				println!("{} got event {:?}", $node.node_id(), e);
				assert_eq!(counterparty_node_id, $counterparty_node_id);
				$node.event_handled().unwrap();
				funding_txo
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
			},
		}
	}};
}

pub(crate) use expect_channel_pending_event;

macro_rules! expect_channel_ready_event {
	($node:expr, $counterparty_node_id:expr) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!("{} timed out waiting for ChannelReady event after 60s", $node.node_id())
		});
		match event {
			ref e @ Event::ChannelReady { user_channel_id, counterparty_node_id, .. } => {
				println!("{} got event {:?}", $node.node_id(), e);
				assert_eq!(counterparty_node_id, Some($counterparty_node_id));
				$node.event_handled().unwrap();
				user_channel_id
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
			},
		}
	}};
}

pub(crate) use expect_channel_ready_event;

macro_rules! expect_channel_ready_events {
	($node:expr, $counterparty_node_id_a:expr, $counterparty_node_id_b:expr) => {{
		let mut ids = Vec::new();
		for _ in 0..2 {
			let event = tokio::time::timeout(
				std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
				$node.next_event_async(),
			)
			.await
			.unwrap_or_else(|_| {
				panic!("{} timed out waiting for ChannelReady event after 60s", $node.node_id())
			});
			match event {
				ref e @ Event::ChannelReady { counterparty_node_id, .. } => {
					println!("{} got event {:?}", $node.node_id(), e);
					ids.push(counterparty_node_id);
					$node.event_handled().unwrap();
				},
				ref e => {
					panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
				},
			}
		}
		assert!(
			ids.contains(&Some($counterparty_node_id_a))
				&& ids.contains(&Some($counterparty_node_id_b)),
			"Expected ChannelReady events from {:?} and {:?}, but got {:?}",
			$counterparty_node_id_a,
			$counterparty_node_id_b,
			ids
		);
	}};
}

pub(crate) use expect_channel_ready_events;

macro_rules! expect_splice_negotiated_event {
	($node:expr, $counterparty_node_id:expr) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!("{} timed out waiting for SpliceNegotiated event after 60s", $node.node_id())
		});
		match event {
			ref e @ Event::SpliceNegotiated { new_funding_txo, counterparty_node_id, .. } => {
				println!("{} got event {:?}", $node.node_id(), e);
				assert_eq!(counterparty_node_id, $counterparty_node_id);
				$node.event_handled().unwrap();
				new_funding_txo
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
			},
		}
	}};
}

pub(crate) use expect_splice_negotiated_event;

macro_rules! expect_payment_received_event {
	($node:expr, $amount_msat:expr) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!("{} timed out waiting for PaymentReceived event after 60s", $node.node_id())
		});
		match event {
			ref e @ Event::PaymentReceived { payment_id, amount_msat, .. } => {
				println!("{} got event {:?}", $node.node_id(), e);
				assert_eq!(amount_msat, $amount_msat);
				let payment = $node.payment(&payment_id.unwrap()).unwrap();
				if !matches!(payment.kind, ldk_node::payment::PaymentKind::Onchain { .. }) {
					assert_eq!(payment.fee_paid_msat, None);
				}
				$node.event_handled().unwrap();
				payment_id
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
			},
		}
	}};
}

pub(crate) use expect_payment_received_event;

macro_rules! expect_payment_claimable_event {
	($node:expr, $payment_id:expr, $payment_hash:expr, $claimable_amount_msat:expr) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!(
				"{} timed out waiting for PaymentClaimable event after 60s",
				std::stringify!($node)
			)
		});
		match event {
			ref e @ Event::PaymentClaimable {
				payment_id,
				payment_hash,
				claimable_amount_msat,
				..
			} => {
				println!("{} got event {:?}", std::stringify!($node), e);
				assert_eq!(payment_hash, $payment_hash);
				assert_eq!(payment_id, $payment_id);
				assert_eq!(claimable_amount_msat, $claimable_amount_msat);
				$node.event_handled().unwrap();
				claimable_amount_msat
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!($node), e);
			},
		}
	}};
}

pub(crate) use expect_payment_claimable_event;

macro_rules! expect_payment_successful_event {
	($node:expr, $payment_id:expr, $fee_paid_msat:expr) => {{
		let event = tokio::time::timeout(
			std::time::Duration::from_secs(crate::common::INTEROP_TIMEOUT_SECS),
			$node.next_event_async(),
		)
		.await
		.unwrap_or_else(|_| {
			panic!("{} timed out waiting for PaymentSuccessful event after 60s", $node.node_id())
		});
		match event {
			ref e @ Event::PaymentSuccessful { payment_id, fee_paid_msat, .. } => {
				println!("{} got event {:?}", $node.node_id(), e);
				if let Some(fee_msat) = $fee_paid_msat {
					assert_eq!(fee_paid_msat, fee_msat);
				}
				let payment = $node.payment(&$payment_id.unwrap()).unwrap();
				assert_eq!(payment.fee_paid_msat, fee_paid_msat);
				assert_eq!(payment_id, $payment_id);
				$node.event_handled().unwrap();
			},
			ref e => {
				panic!("{} got unexpected event!: {:?}", std::stringify!(node_b), e);
			},
		}
	}};
}

pub(crate) use expect_payment_successful_event;

pub(crate) fn setup_bitcoind_and_electrsd() -> (BitcoinD, ElectrsD) {
	let bitcoind_exe =
		env::var("BITCOIND_EXE").ok().or_else(|| corepc_node::downloaded_exe_path().ok()).expect(
			"you need to provide an env var BITCOIND_EXE or specify a bitcoind version feature",
		);
	let mut bitcoind_conf = corepc_node::Conf::default();
	bitcoind_conf.network = "regtest";
	bitcoind_conf.args.push("-rest");
	let bitcoind = BitcoinD::with_conf(bitcoind_exe, &bitcoind_conf).unwrap();

	let electrs_exe = env::var("ELECTRS_EXE")
		.ok()
		.or_else(electrsd::downloaded_exe_path)
		.expect("you need to provide env var ELECTRS_EXE or specify an electrsd version feature");
	let mut electrsd_conf = electrsd::Conf::default();
	electrsd_conf.http_enabled = true;
	electrsd_conf.network = "regtest";
	let electrsd = ElectrsD::with_conf(electrs_exe, &bitcoind, &electrsd_conf).unwrap();
	(bitcoind, electrsd)
}

pub(crate) fn random_chain_source<'a>(
	bitcoind: &'a BitcoinD, electrsd: &'a ElectrsD,
) -> TestChainSource<'a> {
	let r = rand::random_range(0..3);
	match r {
		0 => {
			println!("Randomly setting up Esplora chain syncing...");
			TestChainSource::Esplora(electrsd)
		},
		1 => {
			println!("Randomly setting up Electrum chain syncing...");
			TestChainSource::Electrum(electrsd)
		},
		2 => {
			println!("Randomly setting up Bitcoind RPC chain syncing...");
			TestChainSource::BitcoindRpcSync(bitcoind)
		},
		3 => {
			println!("Randomly setting up Bitcoind REST chain syncing...");
			TestChainSource::BitcoindRestSync(bitcoind)
		},
		_ => unreachable!(),
	}
}

pub(crate) fn random_storage_path() -> PathBuf {
	let mut temp_path = std::env::temp_dir();
	let mut rng = rng();
	let rand_dir: String = (0..7).map(|_| rng.sample(Alphanumeric) as char).collect();
	temp_path.push(rand_dir);
	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);
	vec![
		SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port },
		SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: port + 1 },
	]
}

pub(crate) fn random_node_alias() -> Option<NodeAlias> {
	let mut rng = rng();
	let rand_val = rng.random_range(0..1000);
	let alias = format!("ldk-node-{}", rand_val);
	let mut bytes = [0u8; 32];
	bytes[..alias.as_bytes().len()].copy_from_slice(alias.as_bytes());
	Some(NodeAlias(bytes))
}

pub(crate) fn random_config() -> TestConfig {
	let mut node_config = Config::default();

	#[cfg(zero_fee_commitment_tests)]
	{
		node_config.anchor_channels_config.enable_zero_fee_commitments = true;
	}

	node_config.network = Network::Regtest;
	println!("Setting network: {}", node_config.network);

	let rand_dir = random_storage_path();
	println!("Setting random LDK storage dir: {}", rand_dir.display());
	node_config.storage_dir_path = rand_dir.to_str().unwrap().to_owned();

	let listening_addresses = generate_listening_addresses();
	println!("Setting LDK listening addresses: {:?}", listening_addresses);
	node_config.listening_addresses = Some(listening_addresses);

	let alias = random_node_alias();
	println!("Setting random LDK node alias: {:?}", alias);
	node_config.node_alias = alias;

	TestConfig { node_config, ..Default::default() }
}

#[cfg(feature = "uniffi")]
pub(crate) type TestNode = Arc<Node>;
#[cfg(not(feature = "uniffi"))]
pub(crate) type TestNode = Node;

fn has_onchain_tx_type<F: Fn(&TransactionType) -> bool>(node: &TestNode, predicate: F) -> bool {
	node.list_payments().into_iter().any(|payment| {
		matches!(
			payment.kind,
			PaymentKind::Onchain { tx_type: Some(ref tx_type), .. } if predicate(tx_type)
		)
	})
}

fn assert_any_node_has_onchain_tx_type<F: Fn(&TransactionType) -> bool + Copy>(
	nodes: &[(&str, &TestNode)], tx_type_name: &str, predicate: F,
) {
	if nodes.iter().any(|(_, node)| has_onchain_tx_type(node, predicate)) {
		return;
	}

	let observed: Vec<String> = nodes
		.iter()
		.flat_map(|(name, node)| {
			node.list_payments().into_iter().filter_map(move |payment| match payment.kind {
				PaymentKind::Onchain { tx_type, .. } => Some(format!("{}:{:?}", name, tx_type)),
				_ => None,
			})
		})
		.collect();
	panic!("Expected on-chain payment with tx_type {}; observed {:?}", tx_type_name, observed);
}

fn assert_all_nodes_have_onchain_tx_type<F: Fn(&TransactionType) -> bool + Copy>(
	nodes: &[(&str, &TestNode)], panic_msg: &str, tx_type_name: &str, predicate: F,
) {
	if nodes.iter().all(|(_, node)| has_onchain_tx_type(node, predicate)) {
		return;
	}

	let observed: Vec<String> = nodes
		.iter()
		.flat_map(|(name, node)| {
			node.list_payments().into_iter().filter_map(move |payment| match payment.kind {
				PaymentKind::Onchain { tx_type, .. } => Some(format!("{}:{:?}", name, tx_type)),
				_ => None,
			})
		})
		.collect();
	panic!(
		"Expected {}nodes to have on-chain payment with tx_type {}; observed {:?}",
		panic_msg, tx_type_name, observed
	);
}

async fn settle_force_close_balance<E: ElectrumApi>(
	node: &TestNode, counterparty_node_id: PublicKey, peer_node: &TestNode,
	bitcoind: &BitcoindClient, electrsd: &E,
) {
	let balances = node.list_balances();
	if balances.lightning_balances.len() == 1 {
		match balances.lightning_balances[0] {
			LightningBalance::ClaimableAwaitingConfirmations {
				counterparty_node_id: actual_counterparty_node_id,
				confirmation_height,
				..
			} => {
				assert_eq!(actual_counterparty_node_id, counterparty_node_id);
				let cur_height = node.status().current_best_block.height;
				let blocks_to_go = confirmation_height - cur_height;
				generate_blocks_and_wait(bitcoind, electrsd, blocks_to_go as usize).await;
				node.sync_wallets().unwrap();
				peer_node.sync_wallets().unwrap();
			},
			_ => panic!("Unexpected balance state!"),
		}
	} else {
		assert!(balances.lightning_balances.is_empty(), "Unexpected balance state: {:?}", balances);
		assert_eq!(balances.pending_balances_from_channel_closures.len(), 1);
	}

	for _ in 0..6 {
		if node.list_balances().lightning_balances.is_empty() {
			break;
		}
		generate_blocks_and_wait(bitcoind, electrsd, 1).await;
		node.sync_wallets().unwrap();
		peer_node.sync_wallets().unwrap();
	}

	let balances = node.list_balances();
	assert!(balances.lightning_balances.is_empty(), "Unexpected balance state: {:?}", balances);
	assert_eq!(balances.pending_balances_from_channel_closures.len(), 1);
	match balances.pending_balances_from_channel_closures[0] {
		PendingSweepBalance::BroadcastAwaitingConfirmation { .. } => {
			generate_blocks_and_wait(bitcoind, electrsd, 1).await;
			node.sync_wallets().unwrap();
			peer_node.sync_wallets().unwrap();

			assert!(node.list_balances().lightning_balances.is_empty());
			assert_eq!(node.list_balances().pending_balances_from_channel_closures.len(), 1);
			match node.list_balances().pending_balances_from_channel_closures[0] {
				PendingSweepBalance::AwaitingThresholdConfirmations { .. } => {},
				_ => panic!("Unexpected balance state!"),
			}
		},
		PendingSweepBalance::AwaitingThresholdConfirmations { .. } => {},
		_ => panic!("Unexpected balance state!"),
	}

	generate_blocks_and_wait(bitcoind, electrsd, 5).await;
	node.sync_wallets().unwrap();
	peer_node.sync_wallets().unwrap();
}

#[derive(Clone)]
pub(crate) enum TestChainSource<'a> {
	Esplora(&'a ElectrsD),
	Electrum(&'a ElectrsD),
	BitcoindRpcSync(&'a BitcoinD),
	BitcoindRestSync(&'a BitcoinD),
}

#[derive(Clone, Copy)]
pub(crate) enum TestStoreType {
	TestSyncStore,
	Sqlite,
	FilesystemStore,
}

impl Default for TestStoreType {
	fn default() -> Self {
		TestStoreType::TestSyncStore
	}
}

#[derive(Clone)]
pub(crate) struct TestConfig {
	pub node_config: Config,
	pub log_writer: TestLogWriter,
	pub store_type: TestStoreType,
	pub node_entropy: NodeEntropy,
	pub async_payments_role: Option<AsyncPaymentsRole>,
	pub wallet_rescan_from_height: Option<u32>,
	pub force_wallet_full_scan: bool,
	pub full_scan_stop_gap: Option<u32>,
	pub probing: Option<ProbingConfig>,
}

impl Default for TestConfig {
	fn default() -> Self {
		let node_config = Default::default();
		let log_writer = Default::default();
		let store_type = Default::default();

		let mnemonic = generate_entropy_mnemonic(None);
		let node_entropy = NodeEntropy::from_bip39_mnemonic(mnemonic, None);
		let async_payments_role = None;
		let wallet_rescan_from_height = None;
		let force_wallet_full_scan = false;
		let full_scan_stop_gap = None;
		TestConfig {
			node_config,
			log_writer,
			store_type,
			node_entropy,
			async_payments_role,
			wallet_rescan_from_height,
			force_wallet_full_scan,
			full_scan_stop_gap,
			probing: None,
		}
	}
}

macro_rules! setup_builder {
	($builder:ident, $config:expr) => {
		#[cfg(feature = "uniffi")]
		let $builder = Builder::from_config($config.clone());
		#[cfg(not(feature = "uniffi"))]
		let mut $builder = Builder::from_config($config.clone());
	};
}

pub(crate) use setup_builder;

#[cfg(any(cln_test, lnd_test, eclair_test))]
pub(crate) mod scenarios;

pub(crate) fn setup_two_nodes(
	chain_source: &TestChainSource, allow_0conf: bool, anchors_trusted_no_reserve: bool,
) -> (TestNode, TestNode) {
	setup_two_nodes_with_store(
		chain_source,
		allow_0conf,
		anchors_trusted_no_reserve,
		TestStoreType::TestSyncStore,
	)
}

pub(crate) fn setup_two_nodes_with_store(
	chain_source: &TestChainSource, allow_0conf: bool, anchors_trusted_no_reserve: bool,
	store_type: TestStoreType,
) -> (TestNode, TestNode) {
	println!("== Node A ==");
	let mut config_a = random_config();
	config_a.store_type = store_type;

	if cfg!(hrn_tests) {
		config_a.node_config.hrn_config =
			HumanReadableNamesConfig { resolution_config: HRNResolverConfig::Blip32 };
	}

	let node_a = setup_node(chain_source, config_a);

	println!("\n== Node B ==");
	let mut config_b = random_config();
	config_b.store_type = store_type;

	if cfg!(hrn_tests) {
		config_b.node_config.hrn_config = HumanReadableNamesConfig {
			resolution_config: HRNResolverConfig::Dns {
				dns_server_address: SocketAddress::from_str("8.8.8.8:53").unwrap(),
				enable_hrn_resolution_service: true,
			},
		};
	}

	if allow_0conf {
		config_b.node_config.trusted_peers_0conf.push(node_a.node_id());
	}
	if anchors_trusted_no_reserve {
		config_b.node_config.anchor_channels_config.trusted_peers_no_reserve.push(node_a.node_id());
	}
	let node_b = setup_node(chain_source, config_b);
	(node_a, node_b)
}

pub(crate) fn setup_node(chain_source: &TestChainSource, config: TestConfig) -> TestNode {
	setup_builder!(builder, config.node_config);
	match chain_source {
		TestChainSource::Esplora(electrsd) => {
			let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
			let mut sync_config = EsploraSyncConfig::default();
			sync_config.background_sync_config = None;
			sync_config.force_wallet_full_scan = config.force_wallet_full_scan;
			if let Some(full_scan_stop_gap) = config.full_scan_stop_gap {
				sync_config.full_scan_stop_gap = full_scan_stop_gap;
			}
			builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config));
		},
		TestChainSource::Electrum(electrsd) => {
			let electrum_url = format!("tcp://{}", electrsd.electrum_url);
			let mut sync_config = ElectrumSyncConfig::default();
			sync_config.background_sync_config = None;
			sync_config.force_wallet_full_scan = config.force_wallet_full_scan;
			if let Some(full_scan_stop_gap) = config.full_scan_stop_gap {
				sync_config.full_scan_stop_gap = full_scan_stop_gap;
			}
			builder.set_chain_source_electrum(electrum_url.clone(), Some(sync_config));
		},
		TestChainSource::BitcoindRpcSync(bitcoind) => {
			let rpc_host = bitcoind.params.rpc_socket.ip().to_string();
			let rpc_port = bitcoind.params.rpc_socket.port();
			let values = bitcoind.params.get_cookie_values().unwrap().unwrap();
			let rpc_user = values.user;
			let rpc_password = values.password;
			builder.set_chain_source_bitcoind_rpc(
				rpc_host,
				rpc_port,
				rpc_user,
				rpc_password,
				config.wallet_rescan_from_height,
			);
		},
		TestChainSource::BitcoindRestSync(bitcoind) => {
			let rpc_host = bitcoind.params.rpc_socket.ip().to_string();
			let rpc_port = bitcoind.params.rpc_socket.port();
			let values = bitcoind.params.get_cookie_values().unwrap().unwrap();
			let rpc_user = values.user;
			let rpc_password = values.password;
			let rest_host = bitcoind.params.rpc_socket.ip().to_string();
			let rest_port = bitcoind.params.rpc_socket.port();
			builder.set_chain_source_bitcoind_rest(
				rest_host,
				rest_port,
				rpc_host,
				rpc_port,
				rpc_user,
				rpc_password,
				config.wallet_rescan_from_height,
			);
		},
	}

	match &config.log_writer {
		TestLogWriter::FileWriter => {
			builder.set_filesystem_logger(None, None);
		},
		TestLogWriter::LogFacade => {
			builder.set_log_facade_logger();
		},
		TestLogWriter::Custom(custom_log_writer) => {
			builder.set_custom_logger(Arc::clone(custom_log_writer));
		},
	}

	builder.set_async_payments_role(config.async_payments_role).unwrap();

	if let Some(probing) = config.probing {
		builder.set_probing_config(probing.into());
	}

	let node = match config.store_type {
		TestStoreType::TestSyncStore => {
			let kv_store = TestSyncStore::new(config.node_config.storage_dir_path.into());
			builder.build_with_store(config.node_entropy.into(), kv_store).unwrap()
		},
		TestStoreType::Sqlite => builder.build(config.node_entropy.into()).unwrap(),
		TestStoreType::FilesystemStore => {
			builder.build_with_fs_store(config.node_entropy.into()).unwrap()
		},
	};

	node.start().unwrap();
	assert!(node.status().is_running);
	assert!(node.status().latest_fee_rate_cache_update_timestamp.is_some());
	node
}

pub(crate) async fn generate_blocks_and_wait<E: ElectrumApi>(
	bitcoind: &BitcoindClient, electrs: &E, num: usize,
) {
	let _ = bitcoind.create_wallet("ldk_node_test");
	let _ = bitcoind.load_wallet("ldk_node_test");
	print!("Generating {} blocks...", num);
	let blockchain_info = bitcoind.get_blockchain_info().expect("failed to get blockchain info");
	let cur_height = blockchain_info.blocks;
	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(bitcoind, electrs, cur_height as usize + num).await;
	print!(" Done!");
	println!("\n");
}

pub(crate) fn invalidate_blocks(bitcoind: &BitcoindClient, num_blocks: usize) {
	let blockchain_info = bitcoind.get_blockchain_info().expect("failed to get blockchain info");
	let cur_height = blockchain_info.blocks as usize;
	let target_height = cur_height - num_blocks + 1;
	let block_hash = bitcoind
		.get_block_hash(target_height as u64)
		.expect("failed to get block hash")
		.block_hash()
		.expect("block hash should be present");
	bitcoind.invalidate_block(block_hash).expect("failed to invalidate block");
	let blockchain_info = bitcoind.get_blockchain_info().expect("failed to get blockchain info");
	let new_cur_height = blockchain_info.blocks as usize;
	assert!(new_cur_height + num_blocks == cur_height);
}

pub(crate) async fn wait_for_block<E: ElectrumApi>(
	bitcoind: &BitcoindClient, electrs: &E, min_height: usize,
) {
	let expected_block_hash = exponential_backoff_poll(|| {
		let bitcoind_height =
			bitcoind.get_blockchain_info().expect("failed to get blockchain info").blocks as usize;
		if bitcoind_height < min_height {
			return None;
		}
		bitcoind.get_block_hash(min_height as u64).ok()?.block_hash().ok()
	})
	.await;
	// A height-only wait can return the old header during a same-height reorg. Require the
	// replacement hash so callers cannot sync against the stale chain by mistake.
	exponential_backoff_poll(|| {
		let header = electrs.block_header(min_height).ok()?;
		(header.block_hash() == expected_block_hash).then_some(())
	})
	.await;
}

pub(crate) async fn wait_for_tx<E: ElectrumApi>(electrs: &E, txid: Txid) {
	if electrs.transaction_get(&txid).is_ok() {
		return;
	}

	exponential_backoff_poll(|| {
		electrs.ping().unwrap();
		electrs.transaction_get(&txid).ok()
	})
	.await;
}

pub(crate) async fn wait_for_outpoint_spend<E: ElectrumApi>(
	electrs: &E, outpoint: OutPoint,
) -> Transaction {
	let tx = electrs.transaction_get(&outpoint.txid).unwrap();
	let txout_script = tx.output.get(outpoint.vout as usize).unwrap().clone().script_pubkey;

	// Script history already contains the funding transaction itself and may contain unrelated
	// transactions for reused scripts, so only return a transaction spending the exact outpoint.
	exponential_backoff_poll(|| {
		electrs.ping().unwrap();

		electrs.script_get_history(&txout_script).unwrap().into_iter().find_map(|entry| {
			if entry.tx_hash == outpoint.txid {
				return None;
			}
			let transaction = electrs.transaction_get(&entry.tx_hash).ok()?;
			transaction
				.input
				.iter()
				.any(|input| input.previous_output == outpoint)
				.then_some(transaction)
		})
	})
	.await
}

/// Polls the channel from `source_node` to `counterparty_node` until it reports `is_usable`
/// and can carry an HTLC of `min_amount_msat` from `source_node`'s side.
///
/// After `ChannelReady`, channel-monitor persistence can lag for tens of seconds on slow
/// CI runners; during that window `send_probe`/`send_payment` reject with
/// `ParameterError("...monitor update is in progress...")`. This helper gives tests a
/// deterministic readiness gate instead of racing the monitor-update pipeline.
pub(crate) async fn wait_for_channel_ready_to_send(
	source_node: &TestNode, counterparty_node: &TestNode, min_amount_msat: u64,
) {
	let counterparty = counterparty_node.node_id();
	let deadline = tokio::time::Instant::now() + Duration::from_secs(180);
	while tokio::time::Instant::now() < deadline {
		let ready = source_node.list_channels().iter().any(|c| {
			c.counterparty.node_id == counterparty
				&& c.is_usable
				&& c.next_outbound_htlc_limit_msat >= min_amount_msat
		});
		if ready {
			return;
		}
		tokio::time::sleep(Duration::from_millis(100)).await;
	}
	panic!(
		"channel from {} to {} not ready to send {} msat within 180s",
		source_node.node_id(),
		counterparty,
		min_amount_msat,
	);
}

pub(crate) async fn exponential_backoff_poll<T, F>(mut poll: F) -> T
where
	F: FnMut() -> Option<T>,
{
	let mut delay = Duration::from_millis(64);
	let mut tries = 0;
	loop {
		match poll() {
			Some(data) => break data,
			None if delay.as_millis() < 512 => {
				delay = delay.mul_f32(2.0);
			},

			None => {},
		}
		assert!(tries < 20, "Reached max tries.");
		tries += 1;
		tokio::time::sleep(delay).await;
	}
}

pub(crate) async fn premine_and_distribute_funds<E: ElectrumApi>(
	bitcoind: &BitcoindClient, electrs: &E, addrs: Vec<Address>, amount: Amount,
) {
	premine_blocks(bitcoind, electrs).await;

	distribute_funds_unconfirmed(bitcoind, electrs, addrs, amount).await;
	generate_blocks_and_wait(bitcoind, electrs, 1).await;
}

pub(crate) async fn premine_blocks<E: ElectrumApi>(bitcoind: &BitcoindClient, electrs: &E) {
	let _ = bitcoind.create_wallet("ldk_node_test");
	let _ = bitcoind.load_wallet("ldk_node_test");
	generate_blocks_and_wait(bitcoind, electrs, 101).await;
}

pub(crate) async fn distribute_funds_unconfirmed<E: ElectrumApi>(
	bitcoind: &BitcoindClient, electrs: &E, addrs: Vec<Address>, amount: Amount,
) -> Txid {
	let mut amounts = HashMap::<String, f64>::new();
	for addr in &addrs {
		amounts.insert(addr.to_string(), amount.to_btc());
	}

	let empty_account = json!("");
	let amounts_json = json!(amounts);
	let txid = bitcoind
		.call::<Value>("sendmany", &[empty_account, amounts_json])
		.unwrap()
		.as_str()
		.unwrap()
		.parse()
		.unwrap();

	wait_for_tx(electrs, txid).await;

	txid
}

pub(crate) fn prepare_rbf<E: ElectrumApi>(
	electrs: &E, txid: Txid, scripts_buf: &HashSet<ScriptBuf>,
) -> (Transaction, usize) {
	let tx = electrs.transaction_get(&txid).unwrap();

	let fee_output_index = tx
		.output
		.iter()
		.position(|output| !scripts_buf.contains(&output.script_pubkey))
		.expect("No output available for fee bumping");

	(tx, fee_output_index)
}

pub(crate) async fn bump_fee_and_broadcast<E: ElectrumApi>(
	bitcoind: &BitcoindClient, electrs: &E, mut tx: Transaction, fee_output_index: usize,
	is_insert_block: bool,
) -> Transaction {
	let mut bump_fee_amount_sat = tx.vsize() as u64;
	let attempts = 5;

	for _ in 0..attempts {
		let fee_output = &mut tx.output[fee_output_index];
		let new_fee_value = fee_output.value.to_sat().saturating_sub(bump_fee_amount_sat);
		if new_fee_value < 546 {
			panic!("Warning: Fee output approaching dust limit ({} sats)", new_fee_value);
		}
		fee_output.value = Amount::from_sat(new_fee_value);

		for input in &mut tx.input {
			input.sequence = Sequence::ENABLE_RBF_NO_LOCKTIME;
			input.script_sig = ScriptBuf::new();
			input.witness = Witness::new();
		}

		let signed_result = bitcoind.sign_raw_transaction_with_wallet(&tx).unwrap();
		assert!(signed_result.complete, "Failed to sign RBF transaction");

		let tx_bytes = Vec::<u8>::from_hex(&signed_result.hex).unwrap();
		tx = bitcoin::consensus::encode::deserialize::<Transaction>(&tx_bytes).unwrap();

		match bitcoind.send_raw_transaction(&tx) {
			Ok(res) => {
				if is_insert_block {
					generate_blocks_and_wait(bitcoind, electrs, 1).await;
				}
				let new_txid: Txid = res.0.parse().unwrap();
				wait_for_tx(electrs, new_txid).await;
				return tx;
			},
			Err(_) => {
				bump_fee_amount_sat += bump_fee_amount_sat * 5;
				if tx.output[fee_output_index].value.to_sat() < bump_fee_amount_sat {
					panic!("Insufficient funds to increase fee");
				}
			},
		}
	}

	panic!("Failed to bump fee after {} attempts", attempts);
}

pub async fn open_channel(
	node_a: &TestNode, node_b: &TestNode, funding_amount_sat: u64, should_announce: bool,
	electrsd: &ElectrsD,
) -> OutPoint {
	let funding_txo =
		open_channel_no_wait(node_a, node_b, funding_amount_sat, None, should_announce).await;
	wait_for_tx(&electrsd.client, funding_txo.txid).await;
	funding_txo
}

/// Like [`open_channel`] but skips the `wait_for_tx` electrum check so that
/// multiple channels can be opened back-to-back before any blocks are mined.
/// The caller is responsible for mining blocks and confirming the funding txs.
pub async fn open_channel_no_wait(
	node_a: &TestNode, node_b: &TestNode, funding_amount_sat: u64, push_amount_msat: Option<u64>,
	should_announce: bool,
) -> OutPoint {
	if should_announce {
		node_a
			.open_announced_channel(
				node_b.node_id(),
				node_b.listening_addresses().unwrap().first().unwrap().clone(),
				funding_amount_sat,
				push_amount_msat,
				None,
			)
			.unwrap();
	} else {
		node_a
			.open_channel(
				node_b.node_id(),
				node_b.listening_addresses().unwrap().first().unwrap().clone(),
				funding_amount_sat,
				push_amount_msat,
				None,
			)
			.unwrap();
	}
	assert!(node_a.list_peers().iter().find(|c| { c.node_id == node_b.node_id() }).is_some());

	let funding_txo_a = expect_channel_pending_event!(node_a, node_b.node_id());
	let funding_txo_b = expect_channel_pending_event!(node_b, node_a.node_id());
	assert_eq!(funding_txo_a, funding_txo_b);
	funding_txo_a
}

pub async fn open_channel_push_amt(
	node_a: &TestNode, node_b: &TestNode, funding_amount_sat: u64, push_amount_msat: Option<u64>,
	should_announce: bool, electrsd: &ElectrsD,
) -> OutPoint {
	let funding_txo =
		open_channel_no_wait(node_a, node_b, funding_amount_sat, push_amount_msat, should_announce)
			.await;
	wait_for_tx(&electrsd.client, funding_txo.txid).await;
	funding_txo
}

pub async fn open_channel_with_all(
	node_a: &TestNode, node_b: &TestNode, should_announce: bool, electrsd: &ElectrsD,
) -> OutPoint {
	if should_announce {
		node_a
			.open_announced_channel_with_all(
				node_b.node_id(),
				node_b.listening_addresses().unwrap().first().unwrap().clone(),
				None,
				None,
			)
			.unwrap();
	} else {
		node_a
			.open_channel_with_all(
				node_b.node_id(),
				node_b.listening_addresses().unwrap().first().unwrap().clone(),
				None,
				None,
			)
			.unwrap();
	}
	assert!(node_a.list_peers().iter().find(|c| { c.node_id == node_b.node_id() }).is_some());

	let funding_txo_a = expect_channel_pending_event!(node_a, node_b.node_id());
	let funding_txo_b = expect_channel_pending_event!(node_b, node_a.node_id());
	assert_eq!(funding_txo_a, funding_txo_b);
	wait_for_tx(&electrsd.client, funding_txo_a.txid).await;

	funding_txo_a
}

pub async fn splice_in_with_all(
	node_a: &TestNode, node_b: &TestNode, user_channel_id: &UserChannelId, electrsd: &ElectrsD,
) {
	node_a.splice_in_with_all(user_channel_id, node_b.node_id()).unwrap();

	let splice_txo = expect_splice_negotiated_event!(node_a, node_b.node_id());
	expect_splice_negotiated_event!(node_b, node_a.node_id());
	wait_for_tx(&electrsd.client, splice_txo.txid).await;
}

pub(crate) async fn do_channel_full_cycle<E: ElectrumApi>(
	node_a: TestNode, node_b: TestNode, bitcoind: &BitcoindClient, electrsd: &E, allow_0conf: bool,
	disable_node_b_reserve: bool, expect_anchor_channel: bool, force_close: bool,
) {
	let addr_a = node_a.onchain_payment().new_address().unwrap();
	let addr_b = node_b.onchain_payment().new_address().unwrap();

	let premine_amount_sat = if expect_anchor_channel { 2_125_000 } else { 2_100_000 };

	premine_and_distribute_funds(
		&bitcoind,
		electrsd,
		vec![addr_a, addr_b],
		Amount::from_sat(premine_amount_sat),
	)
	.await;
	node_a.sync_wallets().unwrap();
	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);

	// Check we saw the node funding transactions.
	assert_eq!(
		node_a
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Inbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		1
	);
	assert_eq!(
		node_a
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Outbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		0
	);
	assert_eq!(
		node_b
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Inbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		1
	);
	assert_eq!(
		node_b
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Outbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		0
	);

	// Check we haven't got any events yet
	assert_eq!(node_a.next_event(), None);
	assert_eq!(node_b.next_event(), None);

	println!("\nA -- open_channel -> B");
	let funding_amount_sat = 2_080_000;
	let push_msat = (funding_amount_sat / 2) * 1000; // balance the channel
	if disable_node_b_reserve {
		node_a
			.open_0reserve_channel(
				node_b.node_id(),
				node_b.listening_addresses().unwrap().first().unwrap().clone(),
				funding_amount_sat,
				Some(push_msat),
				None,
			)
			.unwrap();
	} else {
		node_a
			.open_announced_channel(
				node_b.node_id(),
				node_b.listening_addresses().unwrap().first().unwrap().clone(),
				funding_amount_sat,
				Some(push_msat),
				None,
			)
			.unwrap();
	}

	assert_eq!(node_a.list_peers().first().unwrap().node_id, node_b.node_id());
	assert!(node_a.list_peers().first().unwrap().is_persisted);
	let funding_txo_a = expect_channel_pending_event!(node_a, node_b.node_id());
	let funding_txo_b = expect_channel_pending_event!(node_b, node_a.node_id());
	assert_eq!(funding_txo_a, funding_txo_b);

	wait_for_tx(electrsd, funding_txo_a.txid).await;

	if !allow_0conf {
		generate_blocks_and_wait(&bitcoind, electrsd, 6).await;
	}

	node_a.sync_wallets().unwrap();
	node_b.sync_wallets().unwrap();

	// Check we now see the channel funding transaction as outbound.
	assert_eq!(
		node_a
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Outbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		1
	);

	let onchain_fee_buffer_sat = 5000;
	let node_a_anchor_reserve_sat = if expect_anchor_channel { 25_000 } else { 0 };
	let node_a_upper_bound_sat =
		premine_amount_sat - node_a_anchor_reserve_sat - funding_amount_sat;
	let node_a_lower_bound_sat = premine_amount_sat
		- node_a_anchor_reserve_sat
		- funding_amount_sat
		- onchain_fee_buffer_sat;
	assert!(node_a.list_balances().spendable_onchain_balance_sats < node_a_upper_bound_sat);
	assert!(node_a.list_balances().spendable_onchain_balance_sats > node_a_lower_bound_sat);
	assert_eq!(
		node_a.list_balances().total_anchor_channels_reserve_sats,
		node_a_anchor_reserve_sat
	);

	let node_b_anchor_reserve_sat = if node_b
		.config()
		.anchor_channels_config
		.trusted_peers_no_reserve
		.contains(&node_a.node_id())
	{
		0
	} else {
		25_000
	};
	assert_eq!(
		node_b.list_balances().spendable_onchain_balance_sats,
		premine_amount_sat - node_b_anchor_reserve_sat
	);
	assert_eq!(
		node_b.list_balances().total_anchor_channels_reserve_sats,
		node_b_anchor_reserve_sat
	);

	// Note that only node B has 0-reserve, we don't yet have an API to allow the opener of the
	// channel to have 0-reserve.
	if disable_node_b_reserve {
		assert_eq!(node_b.list_channels()[0].unspendable_punishment_reserve, Some(0));
		assert_eq!(node_b.list_channels()[0].outbound_capacity_msat, push_msat);
		assert_eq!(node_b.list_channels()[0].next_outbound_htlc_limit_msat, push_msat);

		assert_eq!(node_b.list_balances().total_lightning_balance_sats * 1000, push_msat);
		let LightningBalance::ClaimableOnChannelClose { amount_satoshis, .. } =
			node_b.list_balances().lightning_balances[0]
		else {
			panic!("Unexpected `LightningBalance` variant");
		};
		assert_eq!(amount_satoshis * 1000, push_msat);
	}

	let user_channel_id_a = expect_channel_ready_event!(node_a, node_b.node_id());
	let user_channel_id_b = expect_channel_ready_event!(node_b, node_a.node_id());

	// After channel_ready, no shutdown should be in progress.
	assert!(node_a.list_channels().iter().all(|c| matches!(
		c.channel_shutdown_state,
		None | Some(ChannelShutdownState::NotShuttingDown)
	)));

	println!("\nB receive");
	let invoice_amount_1_msat = 2500_000;
	let invoice_description: Bolt11InvoiceDescription =
		Bolt11InvoiceDescription::Direct(Description::new(String::from("asdf")).unwrap());
	let invoice = node_b
		.bolt11_payment()
		.receive(invoice_amount_1_msat, &invoice_description.clone().into(), 9217)
		.unwrap();

	println!("\nA send");
	let payment_id = node_a.bolt11_payment().send(&invoice, None).unwrap();
	assert_eq!(node_a.bolt11_payment().send(&invoice, None), Err(NodeError::DuplicatePayment));

	assert!(!node_a.list_payments_with_filter(|p| p.id == payment_id).is_empty());

	let outbound_payments_a = node_a.list_payments_with_filter(|p| {
		p.direction == PaymentDirection::Outbound && matches!(p.kind, PaymentKind::Bolt11 { .. })
	});
	assert_eq!(outbound_payments_a.len(), 1);

	let inbound_payments_a = node_a.list_payments_with_filter(|p| {
		p.direction == PaymentDirection::Inbound && matches!(p.kind, PaymentKind::Bolt11 { .. })
	});
	assert_eq!(inbound_payments_a.len(), 0);

	let outbound_payments_b = node_b.list_payments_with_filter(|p| {
		p.direction == PaymentDirection::Outbound && matches!(p.kind, PaymentKind::Bolt11 { .. })
	});
	assert_eq!(outbound_payments_b.len(), 0);

	let inbound_payments_b = node_b.list_payments_with_filter(|p| {
		p.direction == PaymentDirection::Inbound && matches!(p.kind, PaymentKind::Bolt11 { .. })
	});
	assert_eq!(inbound_payments_b.len(), 1);

	// Verify bolt12_invoice is None for BOLT11 payments
	match node_a.next_event_async().await {
		ref e @ Event::PaymentSuccessful { ref bolt12_invoice, .. } => {
			println!("{} got event {:?}", node_a.node_id(), e);
			assert!(bolt12_invoice.is_none(), "bolt12_invoice should be None for BOLT11 payments");
			node_a.event_handled().unwrap();
		},
		ref e => {
			panic!("{} got unexpected event!: {:?}", std::stringify!(node_a), e);
		},
	}
	expect_event!(node_b, PaymentReceived);
	assert_eq!(node_a.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_a.payment(&payment_id).unwrap().direction, PaymentDirection::Outbound);
	assert_eq!(node_a.payment(&payment_id).unwrap().amount_msat, Some(invoice_amount_1_msat));
	assert!(matches!(node_a.payment(&payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));
	assert_eq!(node_b.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_b.payment(&payment_id).unwrap().direction, PaymentDirection::Inbound);
	assert_eq!(node_b.payment(&payment_id).unwrap().amount_msat, Some(invoice_amount_1_msat));
	assert!(matches!(node_b.payment(&payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));

	// Assert we fail duplicate outbound payments and check the status hasn't changed.
	assert_eq!(Err(NodeError::DuplicatePayment), node_a.bolt11_payment().send(&invoice, None));
	assert_eq!(node_a.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_a.payment(&payment_id).unwrap().direction, PaymentDirection::Outbound);
	assert_eq!(node_a.payment(&payment_id).unwrap().amount_msat, Some(invoice_amount_1_msat));
	assert_eq!(node_b.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_b.payment(&payment_id).unwrap().direction, PaymentDirection::Inbound);
	assert_eq!(node_b.payment(&payment_id).unwrap().amount_msat, Some(invoice_amount_1_msat));

	// Test under-/overpayment
	let invoice_amount_2_msat = 2500_000;
	let invoice = node_b
		.bolt11_payment()
		.receive(invoice_amount_2_msat, &invoice_description.clone().into(), 9217)
		.unwrap();

	let underpaid_amount = invoice_amount_2_msat - 1;
	assert_eq!(
		Err(NodeError::InvalidAmount),
		node_a.bolt11_payment().send_using_amount(&invoice, underpaid_amount, None)
	);

	println!("\nB overpaid receive");
	let invoice = node_b
		.bolt11_payment()
		.receive(invoice_amount_2_msat, &invoice_description.clone().into(), 9217)
		.unwrap();
	let overpaid_amount_msat = invoice_amount_2_msat + 100;

	println!("\nA overpaid send");
	let payment_id =
		node_a.bolt11_payment().send_using_amount(&invoice, overpaid_amount_msat, None).unwrap();
	expect_event!(node_a, PaymentSuccessful);
	let received_amount = match node_b.next_event_async().await {
		ref e @ Event::PaymentReceived { amount_msat, .. } => {
			println!("{} got event {:?}", std::stringify!(node_b), e);
			node_b.event_handled().unwrap();
			amount_msat
		},
		ref e => {
			panic!("{} got unexpected event!: {:?}", std::stringify!(node_b), e);
		},
	};
	assert_eq!(received_amount, overpaid_amount_msat);
	assert_eq!(node_a.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_a.payment(&payment_id).unwrap().direction, PaymentDirection::Outbound);
	assert_eq!(node_a.payment(&payment_id).unwrap().amount_msat, Some(overpaid_amount_msat));
	assert!(matches!(node_a.payment(&payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));
	assert_eq!(node_b.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_b.payment(&payment_id).unwrap().direction, PaymentDirection::Inbound);
	assert_eq!(node_b.payment(&payment_id).unwrap().amount_msat, Some(overpaid_amount_msat));
	assert!(matches!(node_b.payment(&payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));

	// Test "zero-amount" invoice payment
	println!("\nB receive_variable_amount_payment");
	let variable_amount_invoice = node_b
		.bolt11_payment()
		.receive_variable_amount(&invoice_description.clone().into(), 9217)
		.unwrap();
	let determined_amount_msat = 2345_678;
	assert_eq!(
		Err(NodeError::InvalidInvoice),
		node_a.bolt11_payment().send(&variable_amount_invoice, None)
	);
	println!("\nA send_using_amount");
	let payment_id = node_a
		.bolt11_payment()
		.send_using_amount(&variable_amount_invoice, determined_amount_msat, None)
		.unwrap();

	expect_event!(node_a, PaymentSuccessful);
	let received_amount = match node_b.next_event_async().await {
		ref e @ Event::PaymentReceived { amount_msat, .. } => {
			println!("{} got event {:?}", std::stringify!(node_b), e);
			node_b.event_handled().unwrap();
			amount_msat
		},
		ref e => {
			panic!("{} got unexpected event!: {:?}", std::stringify!(node_b), e);
		},
	};
	assert_eq!(received_amount, determined_amount_msat);
	assert_eq!(node_a.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_a.payment(&payment_id).unwrap().direction, PaymentDirection::Outbound);
	assert_eq!(node_a.payment(&payment_id).unwrap().amount_msat, Some(determined_amount_msat));
	assert!(matches!(node_a.payment(&payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));
	assert_eq!(node_b.payment(&payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_b.payment(&payment_id).unwrap().direction, PaymentDirection::Inbound);
	assert_eq!(node_b.payment(&payment_id).unwrap().amount_msat, Some(determined_amount_msat));
	assert!(matches!(node_b.payment(&payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));

	// Test claiming manually registered payments.
	let invoice_amount_3_msat = 5_532_000;
	let manual_preimage = PaymentPreimage([42u8; 32]);
	let manual_payment_hash = PaymentHash(Sha256::hash(&manual_preimage.0).to_byte_array());
	let manual_invoice = node_b
		.bolt11_payment()
		.receive_for_hash(
			invoice_amount_3_msat,
			&invoice_description.clone().into(),
			9217,
			manual_payment_hash,
		)
		.unwrap();
	let manual_payment_id = node_a.bolt11_payment().send(&manual_invoice, None).unwrap();

	let claimable_amount_msat = expect_payment_claimable_event!(
		node_b,
		manual_payment_id,
		manual_payment_hash,
		invoice_amount_3_msat
	);
	node_b
		.bolt11_payment()
		.claim_for_hash(manual_payment_hash, claimable_amount_msat, manual_preimage)
		.unwrap();
	expect_payment_received_event!(node_b, claimable_amount_msat);
	expect_payment_successful_event!(node_a, Some(manual_payment_id), None);
	assert_eq!(node_a.payment(&manual_payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_a.payment(&manual_payment_id).unwrap().direction, PaymentDirection::Outbound);
	assert_eq!(
		node_a.payment(&manual_payment_id).unwrap().amount_msat,
		Some(invoice_amount_3_msat)
	);
	assert!(matches!(node_a.payment(&manual_payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));
	assert_eq!(node_b.payment(&manual_payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_b.payment(&manual_payment_id).unwrap().direction, PaymentDirection::Inbound);
	assert_eq!(
		node_b.payment(&manual_payment_id).unwrap().amount_msat,
		Some(invoice_amount_3_msat)
	);
	assert!(matches!(node_b.payment(&manual_payment_id).unwrap().kind, PaymentKind::Bolt11 { .. }));

	// Test failing manually registered payments.
	let invoice_amount_4_msat = 5_532_000;
	let manual_fail_preimage = PaymentPreimage([43u8; 32]);
	let manual_fail_payment_hash =
		PaymentHash(Sha256::hash(&manual_fail_preimage.0).to_byte_array());
	let manual_fail_invoice = node_b
		.bolt11_payment()
		.receive_for_hash(
			invoice_amount_3_msat,
			&invoice_description.into(),
			9217,
			manual_fail_payment_hash,
		)
		.unwrap();
	let manual_fail_payment_id = node_a.bolt11_payment().send(&manual_fail_invoice, None).unwrap();

	expect_payment_claimable_event!(
		node_b,
		manual_fail_payment_id,
		manual_fail_payment_hash,
		invoice_amount_4_msat
	);
	node_b.bolt11_payment().fail_for_hash(manual_fail_payment_hash).unwrap();
	expect_event!(node_a, PaymentFailed);
	assert_eq!(node_a.payment(&manual_fail_payment_id).unwrap().status, PaymentStatus::Failed);
	assert_eq!(
		node_a.payment(&manual_fail_payment_id).unwrap().direction,
		PaymentDirection::Outbound
	);
	assert_eq!(
		node_a.payment(&manual_fail_payment_id).unwrap().amount_msat,
		Some(invoice_amount_4_msat)
	);
	assert!(matches!(
		node_a.payment(&manual_fail_payment_id).unwrap().kind,
		PaymentKind::Bolt11 { .. }
	));
	assert_eq!(node_b.payment(&manual_fail_payment_id).unwrap().status, PaymentStatus::Failed);
	assert_eq!(
		node_b.payment(&manual_fail_payment_id).unwrap().direction,
		PaymentDirection::Inbound
	);
	assert_eq!(
		node_b.payment(&manual_fail_payment_id).unwrap().amount_msat,
		Some(invoice_amount_4_msat)
	);
	assert!(matches!(
		node_b.payment(&manual_fail_payment_id).unwrap().kind,
		PaymentKind::Bolt11 { .. }
	));

	// Test spontaneous/keysend payments
	println!("\nA send_spontaneous_payment");
	let keysend_amount_msat = 2500_000;
	let custom_tlvs = vec![CustomTlvRecord { type_num: 13377331, value: vec![1, 2, 3] }];
	let keysend_payment_id = node_a
		.spontaneous_payment()
		.send_with_custom_tlvs(keysend_amount_msat, node_b.node_id(), None, custom_tlvs.clone())
		.unwrap();
	expect_event!(node_a, PaymentSuccessful);
	let next_event = node_b.next_event_async().await;
	let (received_keysend_amount, received_custom_records) = match next_event {
		ref e @ Event::PaymentReceived { amount_msat, ref custom_records, .. } => {
			println!("{} got event {:?}", std::stringify!(node_b), e);
			node_b.event_handled().unwrap();
			(amount_msat, custom_records)
		},
		ref e => {
			panic!("{} got unexpected event!: {:?}", std::stringify!(node_b), e);
		},
	};
	assert_eq!(received_keysend_amount, keysend_amount_msat);
	assert_eq!(node_a.payment(&keysend_payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_a.payment(&keysend_payment_id).unwrap().direction, PaymentDirection::Outbound);
	assert_eq!(node_a.payment(&keysend_payment_id).unwrap().amount_msat, Some(keysend_amount_msat));
	assert!(matches!(
		node_a.payment(&keysend_payment_id).unwrap().kind,
		PaymentKind::Spontaneous { .. }
	));
	assert_eq!(received_custom_records, &custom_tlvs);
	assert_eq!(node_b.payment(&keysend_payment_id).unwrap().status, PaymentStatus::Succeeded);
	assert_eq!(node_b.payment(&keysend_payment_id).unwrap().direction, PaymentDirection::Inbound);
	assert_eq!(node_b.payment(&keysend_payment_id).unwrap().amount_msat, Some(keysend_amount_msat));
	assert!(matches!(
		node_b.payment(&keysend_payment_id).unwrap().kind,
		PaymentKind::Spontaneous { .. }
	));
	assert_eq!(
		node_a.list_payments_with_filter(|p| matches!(p.kind, PaymentKind::Bolt11 { .. })).len(),
		5
	);
	assert_eq!(
		node_b.list_payments_with_filter(|p| matches!(p.kind, PaymentKind::Bolt11 { .. })).len(),
		6
	);
	assert_eq!(
		node_a
			.list_payments_with_filter(|p| matches!(p.kind, PaymentKind::Spontaneous { .. }))
			.len(),
		1
	);
	assert_eq!(
		node_b
			.list_payments_with_filter(|p| matches!(p.kind, PaymentKind::Spontaneous { .. }))
			.len(),
		1
	);

	// Mine a block to give time for the HTLC to resolve
	generate_blocks_and_wait(&bitcoind, electrsd, 1).await;

	println!("\nB splices out to pay A");
	let addr_a = node_a.onchain_payment().new_address().unwrap();
	let available_splice_out_sat = node_b.list_channels()[0].outbound_capacity_msat / 1000;
	let splice_out_sat = available_splice_out_sat / 2;
	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());
	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();

	expect_channel_ready_event!(node_a, node_b.node_id());
	expect_channel_ready_event!(node_b, node_a.node_id());

	assert_eq!(
		node_a
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Inbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		2
	);

	println!("\nA splices in the splice-out payment from B");
	let splice_in_sat = splice_out_sat;
	node_a.splice_in(&user_channel_id_a, node_b.node_id(), splice_in_sat).unwrap();

	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();

	expect_channel_ready_event!(node_a, node_b.node_id());
	expect_channel_ready_event!(node_b, node_a.node_id());

	assert_eq!(
		node_a
			.list_payments_with_filter(|p| p.direction == PaymentDirection::Outbound
				&& matches!(p.kind, PaymentKind::Onchain { .. }))
			.len(),
		2
	);

	if disable_node_b_reserve {
		let node_a_outbound_capacity_msat = node_a.list_channels()[0].outbound_capacity_msat;
		let node_a_reserve_msat =
			node_a.list_channels()[0].unspendable_punishment_reserve.unwrap() * 1000;
		let zero_fee_commitments = node_a.list_channels()[0].feerate_sat_per_1000_weight == 0;
		let node_a_anchors_msat = if zero_fee_commitments { 0 } else { 2 * 330 * 1000 };
		let funding_amount_msat = node_a.list_channels()[0].channel_value_sats * 1000;
		// Node B does not have any reserve, so we only subtract a few items on node A's
		// side to arrive at node B's capacity
		let node_b_capacity_msat = funding_amount_msat
			- node_a_outbound_capacity_msat
			- node_a_reserve_msat
			- node_a_anchors_msat;
		let got_capacity_msat = node_b.list_channels()[0].outbound_capacity_msat;
		assert_eq!(got_capacity_msat, node_b_capacity_msat);
		assert_ne!(got_capacity_msat, 0);
		// Sanity check to make sure this is a non-trivial amount
		assert!(got_capacity_msat > 15_000_000);

		// This is a private channel, so node B can send 100% of the value over
		assert_eq!(node_b.list_channels()[0].next_outbound_htlc_limit_msat, node_b_capacity_msat);

		node_b.spontaneous_payment().send(node_b_capacity_msat, node_a.node_id(), None).unwrap();
		expect_event!(node_b, PaymentSuccessful);
		expect_event!(node_a, PaymentReceived);

		node_a.spontaneous_payment().send(node_b_capacity_msat, node_b.node_id(), None).unwrap();
		expect_event!(node_a, PaymentSuccessful);
		expect_event!(node_b, PaymentReceived);
	}

	println!("\nB close_channel (force: {})", force_close);
	tokio::time::sleep(Duration::from_secs(1)).await;
	if force_close {
		node_a.force_close_channel(&user_channel_id_a, node_b.node_id(), None).unwrap();
	} else {
		node_a.close_channel(&user_channel_id_a, node_b.node_id()).unwrap();
		// The cooperative shutdown may complete before we get to check, but if the channel
		// is still visible it must already be in a shutdown state.
		if let Some(channel) =
			node_a.list_channels().into_iter().find(|c| c.user_channel_id == user_channel_id_a)
		{
			assert!(
				!matches!(
					channel.channel_shutdown_state,
					None | Some(ChannelShutdownState::NotShuttingDown)
				),
				"Expected shutdown in progress on node_a, got {:?}",
				channel.channel_shutdown_state,
			);
		}
	}

	expect_event!(node_a, ChannelClosed);
	expect_event!(node_b, ChannelClosed);

	wait_for_outpoint_spend(electrsd, funding_txo_b).await;

	generate_blocks_and_wait(&bitcoind, electrsd, 1).await;
	node_a.sync_wallets().unwrap();
	node_b.sync_wallets().unwrap();

	if force_close {
		settle_force_close_balance(&node_b, node_a.node_id(), &node_a, &bitcoind, electrsd).await;
		assert!(node_b.list_balances().lightning_balances.is_empty());
		assert_eq!(node_b.list_balances().pending_balances_from_channel_closures.len(), 1);

		settle_force_close_balance(&node_a, node_b.node_id(), &node_b, &bitcoind, electrsd).await;
	} else {
		assert_eq!(node_a.list_balances().lightning_balances.len(), 1);
		assert!(node_a.list_balances().pending_balances_from_channel_closures.is_empty());
		let node_a_blocks_to_go = match node_a.list_balances().lightning_balances[0] {
			LightningBalance::ClaimableAwaitingConfirmations {
				counterparty_node_id,
				confirmation_height,
				..
			} => {
				assert_eq!(counterparty_node_id, node_b.node_id());
				let cur_height = node_a.status().current_best_block.height;
				let blocks_to_go = confirmation_height - cur_height;
				blocks_to_go
			},
			_ => panic!("Unexpected balance state!"),
		};

		assert_eq!(node_b.list_balances().lightning_balances.len(), 1);
		assert!(node_b.list_balances().pending_balances_from_channel_closures.is_empty());
		let node_b_blocks_to_go = match node_b.list_balances().lightning_balances[0] {
			LightningBalance::ClaimableAwaitingConfirmations {
				counterparty_node_id,
				confirmation_height,
				..
			} => {
				assert_eq!(counterparty_node_id, node_a.node_id());
				let cur_height = node_b.status().current_best_block.height;
				let blocks_to_go = confirmation_height - cur_height;
				blocks_to_go
			},
			_ => panic!("Unexpected balance state!"),
		};

		assert_eq!(node_a_blocks_to_go, node_b_blocks_to_go);

		generate_blocks_and_wait(&bitcoind, electrsd, node_a_blocks_to_go as usize).await;
		node_a.sync_wallets().unwrap();
		node_b.sync_wallets().unwrap();

		assert!(node_a.list_balances().lightning_balances.is_empty());
		assert!(node_a.list_balances().pending_balances_from_channel_closures.is_empty());
		assert!(node_b.list_balances().lightning_balances.is_empty());
		assert!(node_b.list_balances().pending_balances_from_channel_closures.is_empty());
	}

	if force_close {
		// The recovery reconnect completed while the force-close settled, so the peer no longer
		// needs to remain persisted.
		assert!(
			!node_a.list_peers().iter().any(|p| p.node_id == node_b.node_id() && p.is_persisted),
			"node_b should be removed from node_a peer store after the recovery reconnect"
		);
		assert_all_nodes_have_onchain_tx_type(
			&[("node_a", &node_a), ("node_b", &node_b)],
			"no ",
			"UnilateralClose",
			|tx_type| !matches!(tx_type, TransactionType::UnilateralClose { .. }),
		);
		assert_any_node_has_onchain_tx_type(
			&[("node_a", &node_a), ("node_b", &node_b)],
			"Sweep",
			|tx_type| matches!(tx_type, TransactionType::Sweep { .. }),
		);
	} else {
		assert_all_nodes_have_onchain_tx_type(
			&[("node_a", &node_a), ("node_b", &node_b)],
			"all ",
			"CooperativeClose",
			|tx_type| matches!(tx_type, TransactionType::CooperativeClose { .. }),
		);
		// Peer removed after cooperative close — no further reason to reconnect.
		assert!(
			!node_a.list_peers().iter().any(|p| p.node_id == node_b.node_id() && p.is_persisted),
			"node_b should be removed from node_a peer store after cooperative close"
		);
	}

	let sum_of_all_payments_sat = (push_msat
		+ invoice_amount_1_msat
		+ overpaid_amount_msat
		+ invoice_amount_3_msat
		+ determined_amount_msat
		+ keysend_amount_msat)
		/ 1000 - splice_out_sat;
	let node_a_upper_bound_sat =
		(premine_amount_sat - funding_amount_sat) + (funding_amount_sat - sum_of_all_payments_sat);
	let node_a_lower_bound_sat = node_a_upper_bound_sat - onchain_fee_buffer_sat;
	assert!(node_a.list_balances().spendable_onchain_balance_sats > node_a_lower_bound_sat);
	assert!(node_a.list_balances().spendable_onchain_balance_sats < node_a_upper_bound_sat);

	let node_b_upper_bound_sat = premine_amount_sat + sum_of_all_payments_sat;
	let node_b_lower_bound_sat = node_b_upper_bound_sat - onchain_fee_buffer_sat;
	assert!(node_b.list_balances().spendable_onchain_balance_sats > node_b_lower_bound_sat);
	assert!(node_b.list_balances().spendable_onchain_balance_sats <= node_b_upper_bound_sat);

	assert_eq!(node_a.list_balances().total_anchor_channels_reserve_sats, 0);
	assert_eq!(node_b.list_balances().total_anchor_channels_reserve_sats, 0);

	// Now we should have seen the channel closing transaction on-chain.
	let node_a_inbound_onchain_count = node_a
		.list_payments_with_filter(|p| {
			p.direction == PaymentDirection::Inbound
				&& matches!(p.kind, PaymentKind::Onchain { .. })
		})
		.len();
	let node_b_inbound_onchain_count = node_b
		.list_payments_with_filter(|p| {
			p.direction == PaymentDirection::Inbound
				&& matches!(p.kind, PaymentKind::Onchain { .. })
		})
		.len();
	assert!(node_a_inbound_onchain_count >= 3);
	assert!(node_b_inbound_onchain_count >= 2);

	// Check we handled all events
	assert_eq!(node_a.next_event(), None);
	assert_eq!(node_b.next_event(), None);

	node_a.stop().unwrap();
	println!("\nA stopped");
	node_b.stop().unwrap();
	println!("\nB stopped");
}

// A `KVStore` impl for testing purposes that wraps all our `KVStore`s and asserts their synchronicity.
#[derive(Clone)]
pub(crate) struct TestSyncStore {
	inner: Arc<TestSyncStoreInner>,
}

impl TestSyncStore {
	pub(crate) fn new(dest_dir: PathBuf) -> Self {
		let inner = Arc::new(TestSyncStoreInner::new(dest_dir));
		Self { inner }
	}
}

impl KVStore for TestSyncStore {
	fn read(
		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
	) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send {
		let primary_namespace = primary_namespace.to_string();
		let secondary_namespace = secondary_namespace.to_string();
		let key = key.to_string();
		let inner = Arc::clone(&self.inner);
		async move { inner.read_internal_async(&primary_namespace, &secondary_namespace, &key).await }
	}

	fn write(
		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
	) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
		let primary_namespace = primary_namespace.to_string();
		let secondary_namespace = secondary_namespace.to_string();
		let key = key.to_string();
		let inner = Arc::clone(&self.inner);
		async move {
			inner.write_internal_async(&primary_namespace, &secondary_namespace, &key, buf).await
		}
	}

	fn remove(
		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
	) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
		let primary_namespace = primary_namespace.to_string();
		let secondary_namespace = secondary_namespace.to_string();
		let key = key.to_string();
		let inner = Arc::clone(&self.inner);
		async move {
			inner.remove_internal_async(&primary_namespace, &secondary_namespace, &key, lazy).await
		}
	}

	fn list(
		&self, primary_namespace: &str, secondary_namespace: &str,
	) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + Send {
		let primary_namespace = primary_namespace.to_string();
		let secondary_namespace = secondary_namespace.to_string();
		let inner = Arc::clone(&self.inner);
		async move { inner.list_internal_async(&primary_namespace, &secondary_namespace).await }
	}
}

impl PaginatedKVStore for TestSyncStore {
	fn list_paginated(
		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
	) -> impl Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + Send {
		let primary_namespace = primary_namespace.to_string();
		let secondary_namespace = secondary_namespace.to_string();
		let inner = Arc::clone(&self.inner);
		async move {
			inner
				.list_paginated_internal_async(&primary_namespace, &secondary_namespace, page_token)
				.await
		}
	}
}

struct TestSyncStoreInner {
	serializer: tokio::sync::RwLock<()>,
	test_store: InMemoryStore,
	fs_store: FilesystemStore,
	sqlite_store: SqliteStore,
}

impl TestSyncStoreInner {
	fn new(dest_dir: PathBuf) -> Self {
		let serializer = tokio::sync::RwLock::new(());
		let mut fs_dir = dest_dir.clone();
		fs_dir.push("fs_store");
		let fs_store = FilesystemStore::new(fs_dir);
		let mut sql_dir = dest_dir.clone();
		sql_dir.push("sqlite_store");
		let sqlite_store = SqliteStore::new(
			sql_dir,
			Some("test_sync_db".to_string()),
			Some("test_sync_table".to_string()),
		)
		.unwrap();
		let test_store = InMemoryStore::new();
		Self { serializer, fs_store, sqlite_store, test_store }
	}

	async fn do_list_async(
		&self, primary_namespace: &str, secondary_namespace: &str,
	) -> lightning::io::Result<Vec<String>> {
		let fs_res = KVStore::list(&self.fs_store, primary_namespace, secondary_namespace).await;
		let sqlite_res =
			KVStore::list(&self.sqlite_store, primary_namespace, secondary_namespace).await;
		let test_res =
			KVStore::list(&self.test_store, primary_namespace, secondary_namespace).await;

		match fs_res {
			Ok(mut list) => {
				list.sort();

				let mut sqlite_list = sqlite_res.unwrap();
				sqlite_list.sort();
				assert_eq!(list, sqlite_list);

				let mut test_list = test_res.unwrap();
				test_list.sort();
				assert_eq!(list, test_list);

				Ok(list)
			},
			Err(e) => {
				assert!(sqlite_res.is_err());
				assert!(test_res.is_err());
				Err(e)
			},
		}
	}

	async fn list_internal_async(
		&self, primary_namespace: &str, secondary_namespace: &str,
	) -> lightning::io::Result<Vec<String>> {
		let _guard = self.serializer.read().await;
		self.do_list_async(primary_namespace, secondary_namespace).await
	}

	async fn list_paginated_internal_async(
		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
	) -> lightning::io::Result<PaginatedListResponse> {
		let _guard = self.serializer.read().await;
		let sqlite_res = PaginatedKVStore::list_paginated(
			&self.sqlite_store,
			primary_namespace,
			secondary_namespace,
			page_token.clone(),
		)
		.await;
		let test_res = PaginatedKVStore::list_paginated(
			&self.test_store,
			primary_namespace,
			secondary_namespace,
			page_token,
		)
		.await;

		match sqlite_res {
			Ok(sqlite_response) => {
				assert_eq!(sqlite_response, test_res.unwrap());
				Ok(sqlite_response)
			},
			Err(e) => {
				assert!(test_res.is_err());
				Err(e)
			},
		}
	}

	async fn read_internal_async(
		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
	) -> lightning::io::Result<Vec<u8>> {
		let _guard = self.serializer.read().await;

		let fs_res =
			KVStore::read(&self.fs_store, primary_namespace, secondary_namespace, key).await;
		let sqlite_res =
			KVStore::read(&self.sqlite_store, primary_namespace, secondary_namespace, key).await;
		let test_res =
			KVStore::read(&self.test_store, primary_namespace, secondary_namespace, key).await;

		match fs_res {
			Ok(read) => {
				assert_eq!(read, sqlite_res.unwrap());
				assert_eq!(read, test_res.unwrap());
				Ok(read)
			},
			Err(e) => {
				assert!(sqlite_res.is_err());
				assert_eq!(e.kind(), unsafe { sqlite_res.unwrap_err_unchecked().kind() });
				assert!(test_res.is_err());
				assert_eq!(e.kind(), unsafe { test_res.unwrap_err_unchecked().kind() });
				Err(e)
			},
		}
	}

	async fn write_internal_async(
		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
	) -> lightning::io::Result<()> {
		let _guard = self.serializer.write().await;
		let fs_res = KVStore::write(
			&self.fs_store,
			primary_namespace,
			secondary_namespace,
			key,
			buf.clone(),
		)
		.await;
		let sqlite_res = KVStore::write(
			&self.sqlite_store,
			primary_namespace,
			secondary_namespace,
			key,
			buf.clone(),
		)
		.await;
		let test_res = KVStore::write(
			&self.test_store,
			primary_namespace,
			secondary_namespace,
			key,
			buf.clone(),
		)
		.await;

		assert!(self
			.do_list_async(primary_namespace, secondary_namespace)
			.await
			.unwrap()
			.contains(&key.to_string()));

		match fs_res {
			Ok(()) => {
				assert!(sqlite_res.is_ok());
				assert!(test_res.is_ok());
				Ok(())
			},
			Err(e) => {
				assert!(sqlite_res.is_err());
				assert!(test_res.is_err());
				Err(e)
			},
		}
	}

	async fn remove_internal_async(
		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
	) -> lightning::io::Result<()> {
		let _guard = self.serializer.write().await;
		let fs_res =
			KVStore::remove(&self.fs_store, primary_namespace, secondary_namespace, key, lazy)
				.await;
		let sqlite_res =
			KVStore::remove(&self.sqlite_store, primary_namespace, secondary_namespace, key, lazy)
				.await;
		let test_res =
			KVStore::remove(&self.test_store, primary_namespace, secondary_namespace, key, lazy)
				.await;

		assert!(!self
			.do_list_async(primary_namespace, secondary_namespace)
			.await
			.unwrap()
			.contains(&key.to_string()));

		match fs_res {
			Ok(()) => {
				assert!(sqlite_res.is_ok());
				assert!(test_res.is_ok());
				Ok(())
			},
			Err(e) => {
				assert!(sqlite_res.is_err());
				assert!(test_res.is_err());
				Err(e)
			},
		}
	}
}

/// The PostgreSQL connection string used by the Postgres-backed tests, overridable via the
/// `TEST_POSTGRES_URL` environment variable.
#[cfg(feature = "postgres")]
pub(crate) fn test_connection_string() -> String {
	std::env::var("TEST_POSTGRES_URL")
		.unwrap_or_else(|_| "host=localhost user=postgres password=postgres".to_string())
}

/// Drops the given table from the `ldk_db` database, ignoring the case where the database doesn't
/// exist yet. Used to ensure a clean slate before and after Postgres-backed tests.
#[cfg(feature = "postgres")]
pub(crate) async fn drop_table(table_name: &str) {
	let connection_string = format!("{} dbname=ldk_db", test_connection_string());
	let Ok((client, connection)) =
		tokio_postgres::connect(&connection_string, tokio_postgres::NoTls).await
	else {
		// Database doesn't exist yet — nothing to drop.
		return;
	};
	tokio::spawn(connection);
	let _ = client.execute(&format!("DROP TABLE IF EXISTS {table_name}"), &[]).await;
}