Add end-to-end KV store migration test

Creates a node with LN and on-chain state and then randomly migrates
through all the KV store options and checks it still has its state.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: benthecarman

Cargo.toml

@@ -92,6 +92,7 @@ winapi = { version = "0.3", features = ["winbase"] }
 
 [dev-dependencies]
 lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3dfcc4cca1866c5e5d4d4eaf3b82e09584e2ce5c", features = ["std", "_test_utils"] }
+lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3dfcc4cca1866c5e5d4d4eaf3b82e09584e2ce5c", features = ["tokio"] }
 rand = { version = "0.9.2", default-features = false, features = ["std", "thread_rng", "os_rng"] }
 proptest = "1.0.0"
 regex = "1.5.6"

tests/common/mod.rs

@@ -1922,3 +1922,26 @@ impl TestSyncStoreInner {
 		}
 	}
 }
+
+/// 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;
+}

tests/integration_tests_migration.rs

@@ -0,0 +1,290 @@
+// 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.
+
+// The migration test exercises the filesystem, SQLite, and Postgres stores. It is gated on the
+// `postgres` feature because Postgres is the only one of the three that needs an external service.
+#![cfg(feature = "postgres")]
+
+mod common;
+
+use std::path::PathBuf;
+
+use common::{
+	drop_table, expect_channel_ready_event, expect_payment_received_event,
+	expect_payment_successful_event, test_connection_string,
+};
+use ldk_node::entropy::NodeEntropy;
+use ldk_node::io::postgres_store::PostgresStore;
+use ldk_node::io::sqlite_store::{SqliteStore, KV_TABLE_NAME, SQLITE_DB_FILE_NAME};
+use ldk_node::{Builder, Event};
+use lightning::util::persist::{migrate_kv_store_data_async, MigratableKVStore};
+use lightning_invoice::{Bolt11InvoiceDescription, Description};
+use lightning_persister::fs_store::v2::FilesystemStoreV2;
+use rand::seq::SliceRandom;
+
+async fn drop_tables<'a>(table_names: impl IntoIterator<Item = &'a String>) {
+	for table_name in table_names {
+		drop_table(table_name).await;
+	}
+}
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+enum MigrationBackend {
+	FilesystemStore,
+	Sqlite,
+	Postgres,
+}
+
+/// Everything needed to open a node's store on a particular backend: which backend, the dedicated
+/// data directory for the store (used by the filesystem and SQLite stores), and the Postgres
+/// connection string and table name (used only by the Postgres store).
+struct BackendInstance {
+	backend: MigrationBackend,
+	path: String,
+	connection_string: String,
+	table: String,
+}
+
+/// Returns the dedicated data directory for `backend`'s store under the given per-node base
+/// directory, so each backend's store lives at a distinct, clearly-named location. (The Postgres
+/// store is remote and keyed by table name, so it just uses the base directory, which holds logs.)
+fn store_dir(base_dir: &str, backend: MigrationBackend) -> String {
+	match backend {
+		MigrationBackend::FilesystemStore => format!("{base_dir}/fs_store"),
+		MigrationBackend::Sqlite => format!("{base_dir}/sqlite_store"),
+		MigrationBackend::Postgres => base_dir.to_string(),
+	}
+}
+
+fn build_migration_node(
+	instance: &BackendInstance, node_config: ldk_node::config::Config, node_entropy: NodeEntropy,
+	esplora_url: &str,
+) -> ldk_node::Node {
+	let mut builder = Builder::from_config(node_config);
+	builder.set_chain_source_esplora(esplora_url.to_string(), None);
+	// Build with the store opened at the instance's dedicated path so the node and any later
+	// migration read and write the exact same location.
+	match instance.backend {
+		MigrationBackend::FilesystemStore => {
+			builder.build_with_store(node_entropy, open_fs_store(&instance.path)).unwrap()
+		},
+		MigrationBackend::Sqlite => {
+			builder.build_with_store(node_entropy, open_sqlite_store(&instance.path)).unwrap()
+		},
+		MigrationBackend::Postgres => builder
+			.build_with_postgres_store(
+				node_entropy,
+				instance.connection_string.clone(),
+				None,
+				Some(instance.table.clone()),
+				None,
+			)
+			.unwrap(),
+	}
+}
+
+fn open_fs_store(data_dir: &str) -> FilesystemStoreV2 {
+	std::fs::create_dir_all(data_dir).unwrap();
+	FilesystemStoreV2::new(PathBuf::from(data_dir)).unwrap()
+}
+
+fn open_sqlite_store(data_dir: &str) -> SqliteStore {
+	std::fs::create_dir_all(data_dir).unwrap();
+	SqliteStore::new(
+		PathBuf::from(data_dir),
+		Some(SQLITE_DB_FILE_NAME.to_string()),
+		Some(KV_TABLE_NAME.to_string()),
+	)
+	.unwrap()
+}
+
+async fn open_postgres_store(connection_string: &str, table: &str) -> PostgresStore {
+	PostgresStore::new(connection_string.to_string(), None, Some(table.to_string()), None)
+		.await
+		.unwrap()
+}
+
+async fn migrate_into<S: MigratableKVStore>(source_store: &S, dest: &BackendInstance) {
+	match dest.backend {
+		MigrationBackend::FilesystemStore => {
+			let dest_store = open_fs_store(&dest.path);
+			migrate_kv_store_data_async(source_store, &dest_store).await.unwrap();
+		},
+		MigrationBackend::Sqlite => {
+			let dest_store = open_sqlite_store(&dest.path);
+			migrate_kv_store_data_async(source_store, &dest_store).await.unwrap();
+		},
+		MigrationBackend::Postgres => {
+			let dest_store = open_postgres_store(&dest.connection_string, &dest.table).await;
+			migrate_kv_store_data_async(source_store, &dest_store).await.unwrap();
+		},
+	}
+}
+
+/// Migrates all data from a freshly-opened handle on the `source` backend to a freshly-opened
+/// handle on the `dest` backend. The node owning the source store must be stopped beforehand.
+async fn migrate_between_backends(source: &BackendInstance, dest: &BackendInstance) {
+	match source.backend {
+		MigrationBackend::FilesystemStore => {
+			let source_store = open_fs_store(&source.path);
+			migrate_into(&source_store, dest).await;
+		},
+		MigrationBackend::Sqlite => {
+			let source_store = open_sqlite_store(&source.path);
+			migrate_into(&source_store, dest).await;
+		},
+		MigrationBackend::Postgres => {
+			let source_store = open_postgres_store(&source.connection_string, &source.table).await;
+			migrate_into(&source_store, dest).await;
+		},
+	}
+}
+
+/// Spins up a node on a KV store backend, creates some on-chain and Lightning transaction history,
+/// then migrates its data through every other backend in turn. After each migration it restarts
+/// the node on the new backend and verifies that the node identity, on-chain balance, channel, and
+/// payment history are all preserved.
+///
+/// The order in which the backends are visited is randomized.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn migrate_node_across_all_backends() {
+	let mut order =
+		[MigrationBackend::FilesystemStore, MigrationBackend::Sqlite, MigrationBackend::Postgres];
+	order.shuffle(&mut rand::rng());
+	println!("Migrating node across backends in order: {:?}", order);
+
+	// Tables we might use: one per hop plus node B's. (Only the Postgres hops actually use them.)
+	let tables: Vec<String> = (0..order.len()).map(|i| format!("migrate_chain_{i}")).collect();
+	let node_b_table = "migrate_chain_node_b".to_string();
+	drop_tables(tables.iter().chain(std::iter::once(&node_b_table))).await;
+
+	let (bitcoind, electrsd) = common::setup_bitcoind_and_electrsd();
+	let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
+	let connection_string = test_connection_string();
+
+	// Set up node B, the Lightning counterparty.
+	let config_b = common::random_config(false);
+	let node_b_instance = BackendInstance {
+		backend: MigrationBackend::Postgres,
+		path: store_dir(&config_b.node_config.storage_dir_path, MigrationBackend::Postgres),
+		connection_string: connection_string.clone(),
+		table: node_b_table.clone(),
+	};
+	let node_b = build_migration_node(
+		&node_b_instance,
+		config_b.node_config,
+		config_b.node_entropy,
+		&esplora_url,
+	);
+	node_b.start().unwrap();
+
+	// Spin up the node we'll migrate on the first backend. The same node config (storage dir,
+	// listening addresses, identity) is reused across every hop — only the backend changes — so
+	// each backend's store lives in its own subdirectory of the one storage dir.
+	let config = common::random_config(false);
+	let node_entropy = config.node_entropy;
+	let node_config = config.node_config;
+	let base_dir = node_config.storage_dir_path.clone();
+
+	let mut current = BackendInstance {
+		backend: order[0],
+		path: store_dir(&base_dir, order[0]),
+		connection_string: connection_string.clone(),
+		table: tables[0].clone(),
+	};
+	let mut node = build_migration_node(&current, node_config.clone(), node_entropy, &esplora_url);
+	node.start().unwrap();
+	let expected_node_id = node.node_id();
+
+	// On-chain receive: fund the node.
+	let addr = node.onchain_payment().new_address().unwrap();
+	common::premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![addr],
+		bitcoin::Amount::from_sat(1_000_000),
+	)
+	.await;
+	node.sync_wallets().unwrap();
+
+	// Open a channel to node B (pushing half so both sides can route) and let it confirm.
+	common::open_channel_push_amt(&node, &node_b, 200_000, Some(100_000_000), false, &electrsd)
+		.await;
+	common::generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+	expect_channel_ready_event!(node, node_b.node_id());
+	expect_channel_ready_event!(node_b, node.node_id());
+
+	// Lightning send: node -> node B.
+	let description =
+		Bolt11InvoiceDescription::Direct(Description::new("ln send".to_string()).unwrap());
+	let invoice = node_b.bolt11_payment().receive(10_000, &description.into(), 3600).unwrap();
+	let ln_send_id = node.bolt11_payment().send(&invoice, None).unwrap();
+	expect_payment_successful_event!(node, Some(ln_send_id), None);
+	expect_payment_received_event!(node_b, 10_000);
+
+	// Lightning receive: node B -> node.
+	let description =
+		Bolt11InvoiceDescription::Direct(Description::new("ln receive".to_string()).unwrap());
+	let invoice = node.bolt11_payment().receive(5_000, &description.into(), 3600).unwrap();
+	let ln_receive_id = node_b.bolt11_payment().send(&invoice, None).unwrap();
+	expect_payment_successful_event!(node_b, Some(ln_receive_id), None);
+	expect_payment_received_event!(node, 5_000);
+
+	// On-chain send: node -> a foreign address.
+	let bitcoind_addr = bitcoind.client.new_address().unwrap();
+	let txid = node.onchain_payment().send_to_address(&bitcoind_addr, 50_000, None).unwrap();
+	common::wait_for_tx(&electrsd.client, txid).await;
+	common::generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node.sync_wallets().unwrap();
+
+	// Capture the state we expect to survive every migration.
+	let expected_balance_sats = node.list_balances().total_onchain_balance_sats;
+	let expected_ln_balance_sats = node.list_balances().total_lightning_balance_sats;
+	let mut expected_payments = node.list_payments();
+	expected_payments.sort_by_key(|p| p.id.0);
+	assert!(expected_payments.len() >= 4);
+
+	for (i, &next_backend) in order.iter().enumerate().skip(1) {
+		println!("Migrating from {:?} to {:?}", current.backend, next_backend);
+
+		let next = BackendInstance {
+			backend: next_backend,
+			path: store_dir(&base_dir, next_backend),
+			connection_string: connection_string.clone(),
+			table: tables[i].clone(),
+		};
+
+		// Spin the node down so the source store is no longer being written to.
+		node.stop().unwrap();
+		drop(node);
+
+		migrate_between_backends(&current, &next).await;
+
+		// Spin the node back up on the new backend.
+		node = build_migration_node(&next, node_config.clone(), node_entropy, &esplora_url);
+		node.start().unwrap();
+		node.sync_wallets().unwrap();
+
+		// The balance, channel, and transaction history are preserved across the migration.
+		assert_eq!(node.node_id(), expected_node_id);
+		assert_eq!(node.list_balances().total_onchain_balance_sats, expected_balance_sats);
+		assert_eq!(node.list_balances().total_lightning_balance_sats, expected_ln_balance_sats);
+		assert_eq!(node.list_channels().len(), 1);
+		let mut migrated_payments = node.list_payments();
+		migrated_payments.sort_by_key(|p| p.id.0);
+		assert_eq!(migrated_payments, expected_payments);
+
+		current = next;
+	}
+
+	node.stop().unwrap();
+	node_b.stop().unwrap();
+
+	drop_tables(tables.iter().chain(std::iter::once(&node_b_table))).await;
+}

tests/integration_tests_postgres.rs

@@ -9,27 +9,11 @@
 
 mod common;
 
+use common::{drop_table, test_connection_string};
 use ldk_node::entropy::NodeEntropy;
 use ldk_node::Builder;
 use rand::RngCore;
 
-fn test_connection_string() -> String {
-	std::env::var("TEST_POSTGRES_URL")
-		.unwrap_or_else(|_| "host=localhost user=postgres password=postgres".to_string())
-}
-
-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;
-}
-
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn channel_full_cycle_with_postgres_store() {
 	drop_table("channel_cycle_a").await;