Synchronize TierStore backups before node startup

A newly configured backup starts empty, while a backup restored after
primary-only operation may contain missing or stale values. Treating either
backup as current could leave it unusable for recovery.

In this commit, we:

- Add exhaustive key enumeration to the dynamic store interface when tiered
  storage is enabled, while preserving the existing requirements for builds
  without tiered storage.

- Recover pending journal operations before taking the primary-store snapshot.

- Copy all durable primary values into the backup and remove values that no
  longer exist in primary.

- Exclude ephemeral cache values from the backup and preserve TierStore's
  synchronization metadata during stale-value cleanup.

- Write the backup completion record only after every synchronization step
  succeeds, ensuring interrupted attempts are retried safely.

- Run required backup synchronization during node construction and fail the
  build if synchronization cannot complete.

- Implement `MigratableKVStore` for existing test stores so the tiered-storage
  feature continues compiling and exercising their original behavior.

- Add coverage for successful synchronization, backup-only configuration,
  journal recovery, stale-value removal, metadata preservation, and retries
  after failures at each stage.

Assisted-by: Amp (AI coding agent)

Author: enigbe

src/builder.rs

@@ -41,6 +41,8 @@ use lightning::routing::scoring::{
 };
 use lightning::sign::{EntropySource, NodeSigner};
 use lightning::util::config::HTLCInterceptionFlags;
+#[cfg(feature = "storage-tier")]
+use lightning::util::persist::MigratableKVStore;
 use lightning::util::persist::{
 	KVStore, PaginatedKVStore, CHANNEL_MANAGER_PERSISTENCE_KEY,
 	CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -73,7 +75,7 @@ use crate::io::fs_store::open_or_migrate_fs_store;
 #[cfg(feature = "storage-sqlite")]
 use crate::io::sqlite_store::SqliteStore;
 #[cfg(feature = "storage-tier")]
-use crate::io::tier_store::{setup_index_store, TierStore};
+use crate::io::tier_store::{setup_index_store, BackupSyncStatus, TierStore};
 use crate::io::utils::{
 	read_all_objects, read_event_queue, read_external_pathfinding_scores_from_cache,
 	read_n_objects, read_network_graph, read_node_metrics, read_output_sweeper, read_peer_info,
@@ -959,13 +961,34 @@ impl NodeBuilder {
 	///
 	/// [`set_ephemeral_storage_dir_path`]: Self::set_ephemeral_storage_dir_path
 	/// [`set_backup_storage_dir_path`]: Self::set_backup_storage_dir_path
+	#[cfg(not(feature = "storage-tier"))]
 	pub fn build_with_store<S: PaginatedKVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S,
 	) -> Result<Node, BuildError> {
 		let logger = setup_logger(&self.log_writer_config, &self.config)?;
 		self.build_with_store_and_logger(node_entropy, kv_store, logger)
 	}
 
+	/// Builds a [`Node`] instance according to the options previously configured.
+	///
+	/// The provided `kv_store` will be used as the primary storage backend. Optionally,
+	/// an ephemeral store for frequently-accessed non-critical data (e.g., network graph, scorer)
+	/// and a local SQLite backup store for disaster recovery can be configured via
+	/// [`set_ephemeral_storage_dir_path`] and [`set_backup_storage_dir_path`].
+	///
+	/// The store must implement [`MigratableKVStore`] so a configured backup can be backfilled or
+	/// resilvered from the complete set of primary-store keys before the node starts.
+	///
+	/// [`set_ephemeral_storage_dir_path`]: Self::set_ephemeral_storage_dir_path
+	/// [`set_backup_storage_dir_path`]: Self::set_backup_storage_dir_path
+	#[cfg(feature = "storage-tier")]
+	pub fn build_with_store<S: PaginatedKVStore + MigratableKVStore + Send + Sync + 'static>(
+		&self, node_entropy: NodeEntropy, kv_store: S,
+	) -> Result<Node, BuildError> {
+		let logger = setup_logger(&self.log_writer_config, &self.config)?;
+		self.build_with_store_and_logger(node_entropy, kv_store, logger)
+	}
+
 	fn setup_runtime(&self, logger: &Arc<Logger>) -> Result<Arc<Runtime>, BuildError> {
 		if let Some(handle) = self.runtime_handle.as_ref() {
 			Ok(Arc::new(Runtime::with_handle(handle.clone(), Arc::clone(logger))))
@@ -977,17 +1000,38 @@ impl NodeBuilder {
 		}
 	}
 
+	#[cfg(not(feature = "storage-tier"))]
 	fn build_with_store_and_logger<S: PaginatedKVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S, logger: Arc<Logger>,
 	) -> Result<Node, BuildError> {
 		let runtime = self.setup_runtime(&logger)?;
 		self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)
 	}
 
+	#[cfg(feature = "storage-tier")]
+	fn build_with_store_and_logger<
+		S: PaginatedKVStore + MigratableKVStore + Send + Sync + 'static,
+	>(
+		&self, node_entropy: NodeEntropy, kv_store: S, logger: Arc<Logger>,
+	) -> Result<Node, BuildError> {
+		let runtime = self.setup_runtime(&logger)?;
+		self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)
+	}
+
+	#[cfg(not(feature = "storage-tier"))]
 	fn build_with_store_runtime_and_logger<S: PaginatedKVStore + Send + Sync + 'static>(
 		&self, node_entropy: NodeEntropy, kv_store: S, runtime: Arc<Runtime>, logger: Arc<Logger>,
 	) -> Result<Node, BuildError> {
-		#[cfg(feature = "storage-tier")]
+		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(kv_store));
+		self.build_with_dyn_store(node_entropy, store, runtime, logger)
+	}
+
+	#[cfg(feature = "storage-tier")]
+	fn build_with_store_runtime_and_logger<
+		S: PaginatedKVStore + MigratableKVStore + Send + Sync + 'static,
+	>(
+		&self, node_entropy: NodeEntropy, kv_store: S, runtime: Arc<Runtime>, logger: Arc<Logger>,
+	) -> Result<Node, BuildError> {
 		let store: Arc<DynStore> = {
 			let ts_config = self.tier_store_config.as_ref();
 			let primary_store = Arc::new(DynStoreWrapper(kv_store));
@@ -1040,15 +1084,27 @@ impl NodeBuilder {
 					tier_store.set_backup_store(backup_store);
 				}
 			}
-			runtime.block_on(tier_store.initialize_backup_synchronization()).map_err(|e| {
-				log_error!(logger, "Failed to initialize tier-store backup synchronization: {}", e);
-				BuildError::KVStoreSetupFailed
-			})?;
+			runtime
+				.block_on(async {
+					let status = tier_store.initialize_backup_synchronization().await?;
+					if status == BackupSyncStatus::Required {
+						tier_store.synchronize_backup().await?;
+					}
+					Ok::<(), bitcoin::io::Error>(())
+				})
+				.map_err(|e| {
+					log_error!(logger, "Failed to prepare or synchronize tier-store backup: {}", e);
+					BuildError::KVStoreSetupFailed
+				})?;
 			Arc::new(DynStoreWrapper(tier_store))
 		};
-		#[cfg(not(feature = "storage-tier"))]
-		let store: Arc<DynStore> = Arc::new(DynStoreWrapper(kv_store));
+		self.build_with_dyn_store(node_entropy, store, runtime, logger)
+	}
 
+	fn build_with_dyn_store(
+		&self, node_entropy: NodeEntropy, store: Arc<DynStore>, runtime: Arc<Runtime>,
+		logger: Arc<Logger>,
+	) -> Result<Node, BuildError> {
 		let seed_bytes = node_entropy.to_seed_bytes();
 		let config = Arc::new(self.config.clone());
 
@@ -1594,11 +1650,19 @@ 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.
+	#[cfg(not(feature = "storage-tier"))]
 	pub fn build_with_store<S: PaginatedKVStore + 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)
 	}
+
+	#[cfg(feature = "storage-tier")]
+	pub fn build_with_store<S: PaginatedKVStore + 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)
+	}
 }
 
 /// Builds a [`Node`] instance according to the options previously configured.
@@ -2698,30 +2762,66 @@ pub(crate) fn sanitize_alias(alias_str: &str) -> Result<NodeAlias, BuildError> {
 #[cfg(test)]
 mod tests {
 	use std::future::Future;
+	#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+	use std::path::PathBuf;
 	use std::sync::Arc;
 
 	use lightning::io;
 	use lightning::util::persist::{
-		KVStore, PageToken, PaginatedKVStore, PaginatedListResponse,
+		KVStore, MigratableKVStore, PageToken, PaginatedKVStore, PaginatedListResponse,
 		CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
 		CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
 	};
 
 	use super::{sanitize_alias, BuildError, NodeAlias, NodeBuilder};
 	use crate::entropy::NodeEntropy;
 	use crate::io::test_utils::InMemoryStore;
+	#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+	use crate::io::{
+		sqlite_store::{SqliteStore, KV_TABLE_NAME, SQLITE_BACKUP_DB_FILE_NAME},
+		test_utils::random_storage_path,
+	};
 	use crate::logger::Logger;
 
-	struct ChannelManagerReadFailingStore(InMemoryStore);
+	#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+	struct CleanupDir(PathBuf);
+
+	#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+	impl Drop for CleanupDir {
+		fn drop(&mut self) {
+			let _ = std::fs::remove_dir_all(&self.0);
+		}
+	}
+
+	#[derive(Clone, Copy)]
+	enum BuilderStoreBehavior {
+		#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+		Normal,
+		FailChannelManagerRead,
+		#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+		FailListAllKeys,
+	}
+
+	struct BuilderTestStore {
+		inner: InMemoryStore,
+		behavior: BuilderStoreBehavior,
+	}
+
+	impl BuilderTestStore {
+		fn new(behavior: BuilderStoreBehavior) -> Self {
+			Self { inner: InMemoryStore::new(), behavior }
+		}
+	}
 
-	impl KVStore for ChannelManagerReadFailingStore {
+	impl KVStore for BuilderTestStore {
 		fn read(
 			&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 		) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send {
-			let fail_read = primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE
+			let fail_read = matches!(self.behavior, BuilderStoreBehavior::FailChannelManagerRead)
+				&& primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE
 				&& secondary_namespace == CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE
 				&& key == CHANNEL_MANAGER_PERSISTENCE_KEY;
-			let read = KVStore::read(&self.0, primary_namespace, secondary_namespace, key);
+			let read = KVStore::read(&self.inner, primary_namespace, secondary_namespace, key);
 			async move {
 				if fail_read {
 					Err(io::Error::new(io::ErrorKind::Other, "channel manager read failed"))
@@ -2734,36 +2834,56 @@ mod tests {
 		fn write(
 			&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
 		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
-			KVStore::write(&self.0, primary_namespace, secondary_namespace, key, buf)
+			KVStore::write(&self.inner, primary_namespace, secondary_namespace, key, buf)
 		}
 
 		fn remove(
 			&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
 		) -> impl Future<Output = Result<(), io::Error>> + 'static + Send {
-			KVStore::remove(&self.0, primary_namespace, secondary_namespace, key, lazy)
+			KVStore::remove(&self.inner, primary_namespace, secondary_namespace, key, lazy)
 		}
 
 		fn list(
 			&self, primary_namespace: &str, secondary_namespace: &str,
 		) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + Send {
-			KVStore::list(&self.0, primary_namespace, secondary_namespace)
+			KVStore::list(&self.inner, primary_namespace, secondary_namespace)
 		}
 	}
 
-	impl PaginatedKVStore for ChannelManagerReadFailingStore {
+	impl PaginatedKVStore for BuilderTestStore {
 		fn list_paginated(
 			&self, primary_namespace: &str, secondary_namespace: &str,
 			page_token: Option<PageToken>,
 		) -> impl Future<Output = Result<PaginatedListResponse, io::Error>> + 'static + Send {
 			PaginatedKVStore::list_paginated(
-				&self.0,
+				&self.inner,
 				primary_namespace,
 				secondary_namespace,
 				page_token,
 			)
 		}
 	}
 
+	impl MigratableKVStore for BuilderTestStore {
+		fn list_all_keys(
+			&self,
+		) -> impl Future<Output = Result<Vec<(String, String, String)>, io::Error>> + 'static + Send
+		{
+			#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+			let fail = matches!(self.behavior, BuilderStoreBehavior::FailListAllKeys);
+			#[cfg(any(not(feature = "storage-tier"), feature = "uniffi"))]
+			let fail = false;
+			let list = MigratableKVStore::list_all_keys(&self.inner);
+			async move {
+				if fail {
+					Err(io::Error::new(io::ErrorKind::Other, "list_all_keys failed"))
+				} else {
+					list.await
+				}
+			}
+		}
+	}
+
 	#[test]
 	fn channel_manager_read_failure_fails_build() {
 		let builder = NodeBuilder::new();
@@ -2775,13 +2895,80 @@ mod tests {
 
 		let result = builder.build_with_store_and_logger(
 			node_entropy,
-			ChannelManagerReadFailingStore(InMemoryStore::new()),
+			BuilderTestStore::new(BuilderStoreBehavior::FailChannelManagerRead),
 			logger,
 		);
 
 		assert!(matches!(result, Err(BuildError::ReadFailed)));
 	}
 
+	#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+	#[test]
+	fn builder_synchronizes_a_configured_backup_before_returning() {
+		let base_dir = random_storage_path();
+		let _cleanup = CleanupDir(base_dir.clone());
+		let storage_dir = base_dir.join("node");
+		let backup_dir = base_dir.join("backup");
+		let runtime = tokio::runtime::Runtime::new().unwrap();
+		let primary_store = BuilderTestStore::new(BuilderStoreBehavior::Normal);
+		runtime.block_on(primary_store.write("namespace", "", "current", vec![1])).unwrap();
+		let backup_store = SqliteStore::new(
+			backup_dir.clone(),
+			Some(SQLITE_BACKUP_DB_FILE_NAME.to_string()),
+			Some(KV_TABLE_NAME.to_string()),
+		)
+		.unwrap();
+		runtime
+			.block_on(async { backup_store.write("namespace", "", "stale", vec![2]).await })
+			.unwrap();
+		drop(backup_store);
+		drop(runtime);
+
+		let mut builder = NodeBuilder::new();
+		builder
+			.set_storage_dir_path(storage_dir.to_string_lossy().into_owned())
+			.set_backup_storage_dir_path(backup_dir.to_string_lossy().into_owned());
+		let node = builder
+			.build_with_store(NodeEntropy::from_seed_bytes([42; 64]), primary_store)
+			.unwrap();
+
+		let backup_store = SqliteStore::new(
+			backup_dir,
+			Some(SQLITE_BACKUP_DB_FILE_NAME.to_string()),
+			Some(KV_TABLE_NAME.to_string()),
+		)
+		.unwrap();
+		let runtime = tokio::runtime::Runtime::new().unwrap();
+		assert_eq!(
+			runtime
+				.block_on(async { backup_store.read("namespace", "", "current").await })
+				.unwrap(),
+			vec![1]
+		);
+		assert!(runtime
+			.block_on(async { backup_store.read("namespace", "", "stale").await })
+			.is_err());
+		drop(node);
+	}
+
+	#[cfg(all(feature = "storage-tier", not(feature = "uniffi")))]
+	#[test]
+	fn builder_fails_when_backup_synchronization_cannot_list_primary_keys() {
+		let base_dir = random_storage_path();
+		let _cleanup = CleanupDir(base_dir.clone());
+		let mut builder = NodeBuilder::new();
+		builder
+			.set_storage_dir_path(base_dir.join("node").to_string_lossy().into_owned())
+			.set_backup_storage_dir_path(base_dir.join("backup").to_string_lossy().into_owned());
+
+		let result = builder.build_with_store(
+			NodeEntropy::from_seed_bytes([42; 64]),
+			BuilderTestStore::new(BuilderStoreBehavior::FailListAllKeys),
+		);
+
+		assert!(matches!(result, Err(BuildError::KVStoreSetupFailed)));
+	}
+
 	#[test]
 	fn sanitize_empty_node_alias() {
 		// Empty node alias