Add migratable support for native node stores

enigbe · enigbe · commit d2505b4916a6 · 2026-04-27T07:21:03.000+01:00
Extend the native store plumbing to support key enumeration for store
migration by wiring `MigratableKVStore` through `DynStoreWrapper` and
implementing it for LDK-native backends, including `SqliteStore`,
`VssStore`, and tiered storage, plus the in-repo test stores needed to
keep the migration path exercised.

This lays the plumbing for backup restoration and reconciliation by
making migration a first-class capability of the stores ldk-node
natively owns and configures, ensuring they can exhaustively enumerate
their persisted keys.
diff --git a/src/builder.rs b/src/builder.rs
@@ -37,8 +37,8 @@ use lightning::routing::scoring::{
 use lightning::sign::{EntropySource, NodeSigner};
 use lightning::util::config::HTLCInterceptionFlags;
 use lightning::util::persist::{
-	KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
-	CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
+	KVStore, MigratableKVStore, CHANNEL_MANAGER_PERSISTENCE_KEY,
+	CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::ser::ReadableArgs;
 use lightning::util::sweep::OutputSweeper;
@@ -844,7 +844,7 @@ impl NodeBuilder {
 	///
 	/// [`set_ephemeral_store`]: Self::set_ephemeral_store
 	/// [`set_backup_store`]: Self::set_backup_store
-	pub fn build_with_store<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
+	pub fn build_with_store<S: SyncAndAsyncKVStore + MigratableKVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S,
 	) -> Result<Node, BuildError> {
 		let primary_store: Arc<DynStore> = Arc::new(DynStoreWrapper(kv_store));
@@ -1339,7 +1339,7 @@ impl ArcedNodeBuilder {
 	/// Builds a [`Node`] instance according to the options previously configured.
 	// Note that the generics here don't actually work for Uniffi, but we don't currently expose
 	// this so its not needed.
-	pub fn build_with_store<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
+	pub fn build_with_store<S: SyncAndAsyncKVStore + MigratableKVStore + Send + Sync + 'static>(
 		&self, node_entropy: Arc<NodeEntropy>, kv_store: S,
 	) -> Result<Arc<Node>, BuildError> {
 		self.inner.read().expect("lock").build_with_store(*node_entropy, kv_store).map(Arc::new)
diff --git a/src/io/sqlite_store/mod.rs b/src/io/sqlite_store/mod.rs
@@ -15,7 +15,8 @@ use std::sync::{Arc, Mutex};
 
 use lightning::io;
 use lightning::util::persist::{
-	KVStore, KVStoreSync, PageToken, PaginatedKVStore, PaginatedKVStoreSync, PaginatedListResponse,
+	KVStore, KVStoreSync, MigratableKVStore, PageToken, PaginatedKVStore, PaginatedKVStoreSync,
+	PaginatedListResponse,
 };
 use lightning_types::string::PrintableString;
 use rusqlite::{named_params, Connection};
@@ -255,6 +256,44 @@ impl PaginatedKVStore for SqliteStore {
 	}
 }
 
+impl MigratableKVStore for SqliteStore {
+	fn list_all_keys(&self) -> io::Result<Vec<(String, String, String)>> {
+		let locked_conn = self.inner.connection.lock().unwrap();
+
+		let sql = format!(
+			"SELECT DISTINCT primary_namespace, secondary_namespace, key FROM {}",
+			self.inner.kv_table_name
+		);
+
+		let mut stmt = locked_conn.prepare_cached(&sql).map_err(|e| {
+			let msg = format!("Failed to prepare statement: {}", e);
+			io::Error::new(io::ErrorKind::Other, msg)
+		})?;
+
+		let rows = stmt
+			.query_map([], |row| {
+				let primary_namespace: String = row.get(0)?;
+				let secondary_namespace: String = row.get(1)?;
+				let key: String = row.get(2)?;
+				Ok((primary_namespace, secondary_namespace, key))
+			})
+			.map_err(|e| {
+				let msg = format!("Failed to list all keys: {}", e);
+				io::Error::new(io::ErrorKind::Other, msg)
+			})?;
+
+		let mut keys = Vec::new();
+		for row in rows {
+			keys.push(row.map_err(|e| {
+				let msg = format!("Failed to decode row while listing all keys: {}", e);
+				io::Error::new(io::ErrorKind::Other, msg)
+			})?);
+		}
+
+		Ok(keys)
+	}
+}
+
 struct SqliteStoreInner {
 	connection: Arc<Mutex<Connection>>,
 	data_dir: PathBuf,
diff --git a/src/io/test_utils.rs b/src/io/test_utils.rs
@@ -21,7 +21,7 @@ use lightning::ln::functional_test_utils::{
 	TestChanMonCfg,
 };
 use lightning::util::persist::{
-	KVStore, KVStoreSync, MonitorUpdatingPersister, PageToken, PaginatedKVStore,
+	KVStore, KVStoreSync, MigratableKVStore, MonitorUpdatingPersister, PageToken, PaginatedKVStore,
 	PaginatedKVStoreSync, PaginatedListResponse, KVSTORE_NAMESPACE_KEY_MAX_LEN,
 };
 use lightning::util::test_utils;
@@ -245,6 +245,29 @@ impl PaginatedKVStore for InMemoryStore {
 	}
 }
 
+impl MigratableKVStore for InMemoryStore {
+	fn list_all_keys(&self) -> io::Result<Vec<(String, String, String)>> {
+		let persisted_lock = self.persisted_bytes.lock().unwrap();
+		let mut keys = Vec::new();
+
+		for (namespace, entries) in persisted_lock.iter() {
+			let mut parts = namespace.splitn(2, '/');
+			let primary_namespace = parts.next().unwrap_or_default();
+			let secondary_namespace = parts.next().unwrap_or_default();
+
+			for key in entries.keys() {
+				keys.push((
+					primary_namespace.to_string(),
+					secondary_namespace.to_string(),
+					key.clone(),
+				));
+			}
+		}
+
+		Ok(keys)
+	}
+}
+
 unsafe impl Sync for InMemoryStore {}
 unsafe impl Send for InMemoryStore {}
 
diff --git a/src/io/tier_store.rs b/src/io/tier_store.rs
@@ -12,7 +12,7 @@ use crate::types::DynStore;
 use bitcoin::io::Read;
 use lightning::ln::msgs::DecodeError;
 use lightning::util::persist::{
-	KVStore, KVStoreSync, NETWORK_GRAPH_PERSISTENCE_KEY,
+	KVStore, KVStoreSync, MigratableKVStore, NETWORK_GRAPH_PERSISTENCE_KEY,
 	NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
 	SCORER_PERSISTENCE_KEY, SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
 };
@@ -1191,6 +1191,12 @@ fn ops_match(a: &PendingBackupOp, b: &PendingBackupOp) -> bool {
 	}
 }
 
+impl MigratableKVStore for TierStore {
+	fn list_all_keys(&self) -> io::Result<Vec<(String, String, String)>> {
+		self.inner.primary_store.list_all_keys()
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use std::future::Future;
@@ -1202,7 +1208,8 @@ mod tests {
 	use bitcoin::io::ErrorKind;
 	use lightning::util::logger::Level;
 	use lightning::util::persist::{
-		CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+		MigratableKVStore, CHANNEL_MANAGER_PERSISTENCE_KEY,
+		CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
 		CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 	};
 	use lightning_persister::fs_store::v1::FilesystemStore;
@@ -1349,6 +1356,12 @@ mod tests {
 		}
 	}
 
+	impl MigratableKVStore for FailingStore {
+		fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, io::Error> {
+			Ok(Vec::new())
+		}
+	}
+
 	fn setup_tier_store(primary_store: Arc<DynStore>, logger: Arc<Logger>) -> TierStore {
 		TierStore::new(primary_store, logger)
 	}
diff --git a/src/io/vss_store.rs b/src/io/vss_store.rs
@@ -24,7 +24,7 @@ use bitcoin::Network;
 use lightning::impl_writeable_tlv_based_enum;
 use lightning::io::{self, Error, ErrorKind};
 use lightning::sign::{EntropySource as LdkEntropySource, RandomBytes};
-use lightning::util::persist::{KVStore, KVStoreSync};
+use lightning::util::persist::{KVStore, KVStoreSync, MigratableKVStore};
 use lightning::util::ser::{Readable, Writeable};
 use prost::Message;
 use vss_client::client::VssClient;
@@ -386,6 +386,27 @@ impl Drop for VssStore {
 	}
 }
 
+impl MigratableKVStore for VssStore {
+	fn list_all_keys(&self) -> io::Result<Vec<(String, String, String)>> {
+		let internal_runtime = self.internal_runtime.as_ref().ok_or_else(|| {
+			debug_assert!(false, "Failed to access internal runtime");
+			let msg = format!("Failed to access internal runtime");
+			Error::new(ErrorKind::Other, msg)
+		})?;
+		let inner = Arc::clone(&self.inner);
+		let fut = async move {
+			let stored_keys = inner.list_all_stored_keys(&inner.blocking_client).await?;
+			let mut decoded = Vec::with_capacity(stored_keys.len());
+			for stored_key in stored_keys {
+				if let Some(key_parts) = inner.decode_stored_key(&stored_key)? {
+					decoded.push(key_parts);
+				}
+			}
+			Ok(decoded)
+		};
+		tokio::task::block_in_place(move || internal_runtime.block_on(fut))
+	}
+}
 struct VssStoreInner {
 	schema_version: VssSchemaVersion,
 	blocking_client: VssClient<CustomRetryPolicy>,
@@ -507,6 +528,93 @@ impl VssStoreInner {
 		Ok(keys)
 	}
 
+	async fn list_all_stored_keys(
+		&self, client: &VssClient<CustomRetryPolicy>,
+	) -> io::Result<Vec<String>> {
+		let mut page_token = None;
+		let mut keys = Vec::new();
+
+		while page_token != Some("".to_string()) {
+			let request = ListKeyVersionsRequest {
+				store_id: self.store_id.clone(),
+				key_prefix: None,
+				page_token,
+				page_size: None,
+			};
+
+			let response = client.list_key_versions(&request).await.map_err(|e| {
+				let msg = format!("Failed to list all stored keys: {}", e);
+				Error::new(ErrorKind::Other, msg)
+			})?;
+
+			for kv in response.key_versions {
+				keys.push(kv.key);
+			}
+
+			page_token = response.next_page_token;
+		}
+
+		Ok(keys)
+	}
+
+	fn decode_stored_key(&self, stored_key: &str) -> io::Result<Option<(String, String, String)>> {
+		match self.schema_version {
+			VssSchemaVersion::V0 => {
+				if !stored_key.contains('#') {
+					let key = self.key_obfuscator.deobfuscate(stored_key)?;
+					if key == VSS_SCHEMA_VERSION_KEY {
+						return Ok(None);
+					}
+					return Ok(Some(("".to_string(), "".to_string(), key)));
+				}
+
+				let mut parts = stored_key.splitn(3, '#');
+				let primary_namespace = parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key format"))?;
+				let secondary_namespace = parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key format"))?;
+				let obfuscated_key = parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key format"))?;
+
+				let key = self.key_obfuscator.deobfuscate(obfuscated_key)?;
+				if key == VSS_SCHEMA_VERSION_KEY {
+					return Ok(None);
+				}
+
+				Ok(Some((primary_namespace.to_string(), secondary_namespace.to_string(), key)))
+			},
+			VssSchemaVersion::V1 => {
+				let mut parts = stored_key.splitn(2, '#');
+				let obfuscated_prefix = parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key format"))?;
+				let obfuscated_key = parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key format"))?;
+
+				let prefix = self.key_obfuscator.deobfuscate(obfuscated_prefix)?;
+				let key = self.key_obfuscator.deobfuscate(obfuscated_key)?;
+
+				if key == VSS_SCHEMA_VERSION_KEY {
+					return Ok(None);
+				}
+
+				let mut prefix_parts = prefix.splitn(2, '#');
+				let primary_namespace = prefix_parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key prefix"))?;
+				let secondary_namespace = prefix_parts
+					.next()
+					.ok_or_else(|| Error::new(ErrorKind::InvalidData, "Invalid VSS key prefix"))?;
+
+				Ok(Some((primary_namespace.to_string(), secondary_namespace.to_string(), key)))
+			},
+		}
+	}
+
 	async fn read_internal(
 		&self, client: &VssClient<CustomRetryPolicy>, primary_namespace: String,
 		secondary_namespace: String, key: String,
diff --git a/src/types.rs b/src/types.rs
@@ -31,7 +31,9 @@ use lightning::routing::gossip;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{CombinedScorer, ProbabilisticScoringFeeParameters};
 use lightning::sign::InMemorySigner;
-use lightning::util::persist::{KVStore, KVStoreSync, MonitorUpdatingPersisterAsync};
+use lightning::util::persist::{
+	KVStore, KVStoreSync, MigratableKVStore, MonitorUpdatingPersisterAsync,
+};
 use lightning::util::ser::{Readable, Writeable, Writer};
 use lightning::util::sweep::OutputSweeper;
 use lightning_block_sync::gossip::GossipVerifier;
@@ -85,6 +87,16 @@ pub trait DynStoreTrait: Send + Sync {
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
 	) -> Result<Vec<String>, bitcoin::io::Error>;
+
+	/// Returns all known keys as `(primary_namespace, secondary_namespace, key)` tuples.
+	///
+	/// As with [`lightning::util::persist::MigratableKVStore::list_all_keys`],
+	/// implementations must exhaustively return all entries known to the store so
+	/// migration and restoration do not miss data.
+	///
+	/// Implementations that do not support exhaustive enumeration may return an
+	/// error with [`bitcoin::io::ErrorKind::Other`].
+	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, bitcoin::io::Error>;
 }
 
 impl<'a> KVStore for dyn DynStoreTrait + 'a {
@@ -174,9 +186,13 @@ impl KVStore for DynStoreRef {
 	}
 }
 
-pub(crate) struct DynStoreWrapper<T: SyncAndAsyncKVStore + Send + Sync>(pub(crate) T);
+pub(crate) struct DynStoreWrapper<T: SyncAndAsyncKVStore + MigratableKVStore + Send + Sync>(
+	pub(crate) T,
+);
 
-impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T> {
+impl<T: SyncAndAsyncKVStore + MigratableKVStore + Send + Sync> DynStoreTrait
+	for DynStoreWrapper<T>
+{
 	fn read_async(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, bitcoin::io::Error>> + Send + 'static>> {
@@ -224,6 +240,10 @@ impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T>
 	) -> Result<Vec<String>, bitcoin::io::Error> {
 		KVStoreSync::list(&self.0, primary_namespace, secondary_namespace)
 	}
+
+	fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, bitcoin::io::Error> {
+		MigratableKVStore::list_all_keys(&self.0)
+	}
 }
 
 pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
@@ -42,7 +42,7 @@ use ldk_node::{
 use lightning::io;
 use lightning::ln::msgs::SocketAddress;
 use lightning::routing::gossip::NodeAlias;
-use lightning::util::persist::{KVStore, KVStoreSync};
+use lightning::util::persist::{KVStore, KVStoreSync, MigratableKVStore};
 use lightning::util::test_utils::TestStore;
 use lightning_invoice::{Bolt11InvoiceDescription, Description};
 use lightning_persister::fs_store::v1::FilesystemStore;
@@ -1616,6 +1616,12 @@ impl KVStoreSync for TestSyncStore {
 	}
 }
 
+impl MigratableKVStore for TestSyncStore {
+	fn list_all_keys(&self) -> lightning::io::Result<Vec<(String, String, String)>> {
+		self.inner.list_all_keys_internal()
+	}
+}
+
 struct TestSyncStoreInner {
 	serializer: RwLock<()>,
 	test_store: TestStore,
@@ -1789,4 +1795,19 @@ impl TestSyncStoreInner {
 		let _guard = self.serializer.read().unwrap();
 		self.do_list(primary_namespace, secondary_namespace)
 	}
+
+	fn list_all_keys_internal(&self) -> lightning::io::Result<Vec<(String, String, String)>> {
+		let _guard = self.serializer.read().unwrap();
+
+		let mut fs_keys = MigratableKVStore::list_all_keys(&self.fs_store)?;
+		fs_keys.sort();
+
+		let mut sqlite_keys = MigratableKVStore::list_all_keys(&self.sqlite_store)?;
+		sqlite_keys.sort();
+		assert_eq!(fs_keys, sqlite_keys);
+
+		// TODO(enigbe): Upstream MigratableKVStore implementation for TestStore
+
+		Ok(fs_keys)
+	}
 }
