Wire semisync backup retry into builder and node startup

Thread BackupMode through TierStoreConfig and update backup-store
configuration to distinguish best-effort backup writes from semisync
behavior.

Build a local retry queue store for SemiSync during tier-store
construction, retain the concrete TierStore on Node, and spawn the
background backup retry task during Node::start() with shutdown
integration via stop_sender.

Also update TierStore backup result handling to enqueue concrete
PendingBackupOp values for durable retry, and refresh the related
backup and retry-task documentation.

Author: enigbe

src/builder.rs

@@ -58,7 +58,7 @@ use crate::event::EventQueue;
 use crate::fee_estimator::OnchainFeeEstimator;
 use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
-use crate::io::tier_store::TierStore;
+use crate::io::tier_store::{BackupMode, BackupRetryQueue, TierStore};
 use crate::io::utils::{
 	read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
 	read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_pending_payments,
@@ -157,16 +157,18 @@ impl std::fmt::Debug for LogWriterConfig {
 }
 
 #[derive(Default)]
-struct TierStoreConfig {
-	ephemeral: Option<Arc<DynStore>>,
-	backup: Option<Arc<DynStore>>,
+pub(crate) struct TierStoreConfig {
+	pub(crate) ephemeral: Option<Arc<DynStore>>,
+	pub(crate) backup: Option<Arc<DynStore>>,
+	pub(crate) backup_mode: BackupMode,
 }
 
 impl std::fmt::Debug for TierStoreConfig {
 	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 		f.debug_struct("TierStoreConfig")
 			.field("ephemeral", &self.ephemeral.as_ref().map(|_| "Arc<DynStore>"))
 			.field("backup", &self.backup.as_ref().map(|_| "Arc<DynStore>"))
+			.field("backup_mode", &self.backup_mode)
 			.finish()
 	}
 }
@@ -650,13 +652,26 @@ impl NodeBuilder {
 	/// When building with tiered storage, this store receives a second durable
 	/// copy of data written to the primary store.
 	///
-	/// Writes and removals for primary-backed data only succeed once both the
-	/// primary and backup stores complete successfully.
-	///
-	/// If not set, durable data will be stored only in the primary store.
-	pub fn set_backup_store(&mut self, backup_store: Arc<DynStore>) -> &mut Self {
-		let tier_store_config = self.tier_store_config.get_or_insert(TierStoreConfig::default());
-		tier_store_config.backup = Some(backup_store);
+	/// Backup failure handling depends on `backup_mode`:
+	/// - [`BackupMode::BestEffortBackup`] logs backup failures and still returns
+	///   success as long as the primary store succeeds.
+	/// - [`BackupMode::SemiSync`] returns success only if a failed backup
+	///   operation can be durably persisted locally and enqueued for
+	///   asynchronous retry.
+	///
+	/// Note: in [`BackupMode::SemiSync`], writes and removals are still not atomic
+	/// across the primary and backup stores. A call may return an error after the
+	/// primary store has already been updated if immediate backup replication
+	/// fails and the retry intent cannot be durably persisted locally.
+	///
+	/// If no backup store is configured, durable data will be stored only in the
+	/// primary store.
+	pub fn set_backup_store(
+		&mut self, backup_store: Arc<DynStore>, backup_mode: BackupMode,
+	) -> &mut Self {
+		let cfg = self.tier_store_config.get_or_insert(TierStoreConfig::default());
+		cfg.backup = Some(backup_store);
+		cfg.backup_mode = backup_mode;
 		self
 	}
 
@@ -852,14 +867,41 @@ impl NodeBuilder {
 
 		let ts_config = self.tier_store_config.as_ref();
 		let mut tier_store = TierStore::new(primary_store, Arc::clone(&logger));
+
 		if let Some(config) = ts_config {
 			config.ephemeral.as_ref().map(|s| tier_store.set_ephemeral_store(Arc::clone(s)));
-			config.backup.as_ref().map(|s| tier_store.set_backup_store(Arc::clone(s), None));
+
+			if let Some(backup_store) = config.backup.as_ref() {
+				match config.backup_mode {
+					BackupMode::BestEffortBackup => {
+						tier_store.set_backup_store(Arc::clone(backup_store), None);
+					},
+					BackupMode::SemiSync => {
+						let retry_store: Arc<DynStore> =
+							Arc::new(DynStoreWrapper(FilesystemStore::new(
+								PathBuf::from(&self.config.storage_dir_path)
+									.join("backup_retry_queue"),
+							)));
+
+						let retry_queue = Arc::new(BackupRetryQueue::new(
+							Arc::clone(&retry_store),
+							Arc::clone(&logger),
+						));
+						tier_store.set_backup_store(
+							Arc::clone(backup_store),
+							Some(Arc::clone(&retry_queue)),
+						);
+					},
+				}
+			}
 		}
 
 		let seed_bytes = node_entropy.to_seed_bytes();
 		let config = Arc::new(self.config.clone());
 
+		let kv_store: Arc<DynStore> = Arc::new(DynStoreWrapper(tier_store.clone()));
+		let tier_store = Arc::new(tier_store);
+
 		build_with_store_internal(
 			config,
 			self.chain_data_source_config.as_ref(),
@@ -871,7 +913,8 @@ impl NodeBuilder {
 			seed_bytes,
 			runtime,
 			logger,
-			Arc::new(DynStoreWrapper(tier_store)),
+			kv_store,
+			tier_store,
 		)
 	}
 }
@@ -1311,6 +1354,7 @@ fn build_with_store_internal(
 	pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
 	async_payments_role: Option<AsyncPaymentsRole>, recovery_mode: bool, seed_bytes: [u8; 64],
 	runtime: Arc<Runtime>, logger: Arc<Logger>, kv_store: Arc<DynStore>,
+	tier_store: Arc<TierStore>,
 ) -> Result<Node, BuildError> {
 	optionally_install_rustls_cryptoprovider();
 
@@ -2131,6 +2175,7 @@ fn build_with_store_internal(
 		om_mailbox,
 		async_payments_role,
 		hrn_resolver: Arc::new(hrn_resolver),
+		tier_store,
 		#[cfg(cycle_tests)]
 		_leak_checker,
 	})

src/io/tier_store.rs

@@ -44,16 +44,16 @@ use std::time::Duration;
 /// Note that dual-store writes and removals are not atomic across the primary
 /// and backup stores. One store may already reflect the change even if the
 /// overall operation returns an error.
+#[derive(Clone)]
 pub(crate) struct TierStore {
 	inner: Arc<TierStoreInner>,
-	logger: Arc<Logger>,
 }
 
 impl TierStore {
 	pub fn new(primary_store: Arc<DynStore>, logger: Arc<Logger>) -> Self {
 		let inner = Arc::new(TierStoreInner::new(primary_store, Arc::clone(&logger)));
 
-		Self { inner, logger }
+		Self { inner }
 	}
 
 	/// Configures a backup store for primary-backed data.
@@ -98,6 +98,10 @@ impl TierStore {
 
 		inner.ephemeral_store = Some(ephemeral);
 	}
+
+	pub(crate) fn backup_store(&self) -> Option<&BackupStore> {
+		self.inner.backup_store.as_ref()
+	}
 }
 
 impl KVStore for TierStore {
@@ -188,11 +192,11 @@ impl KVStoreSync for TierStore {
 	}
 }
 
-struct BackupStore {
+pub(crate) struct BackupStore {
 	/// Store may be remote.
-	store: Arc<DynStore>,
+	pub(crate) store: Arc<DynStore>,
 	/// Present only when backup failures should be durably queued for retry.
-	retry_queue: Option<Arc<BackupRetryQueue<Arc<Logger>>>>,
+	pub(crate) retry_queue: Option<Arc<BackupRetryQueue<Arc<Logger>>>>,
 }
 
 struct TierStoreInner {
@@ -763,6 +767,7 @@ const RETRY_MAX_BACKOFF_MS: u64 = 60_000;
 const RETRY_BACKOFF_MULTIPLIER: u64 = 2;
 
 /// Controls how TierStore treats backup-store failures for primary-backed writes and removals.
+#[derive(Debug)]
 pub enum BackupMode {
 	/// Writes must succeed on the primary store.
 	///
@@ -780,6 +785,12 @@ pub enum BackupMode {
 	SemiSync,
 }
 
+impl Default for BackupMode {
+	fn default() -> Self {
+		Self::BestEffortBackup
+	}
+}
+
 /// A pending operation against the backup store that failed on the write path
 /// and needs to be retried asynchronously.
 ///
@@ -1070,49 +1081,61 @@ impl Future for RetryQueueFuture {
 /// Runs the background retry loop for the given `BackupRetryQueue` against
 /// `backup_store`.
 ///
-/// Wakes when the queue has entries, iterates all pending ops, applies each
-/// against the backup store, and removes successfully retried entries.
-/// Uses exponential backoff with jitter on persistent failures.
-/// Throttles error logging to avoid log spam.
+/// The task waits for pending queue entries, a periodic fallback wake, or a
+/// shutdown signal. When woken, it snapshots the current queue and retries
+/// each pending operation against the backup store.
 ///
-/// This function runs for node lifetime and should be spawned via
-/// `tokio::spawn` during `build()` when `BackupMode::SemiSync` is configured.
-pub(crate) async fn run_backup_retry_task<L: Deref + Send + Sync + 'static>(
-	queue: Arc<BackupRetryQueue<L>>, backup_store: Arc<DynStore>, logger: L,
-) where
-	L::Target: LdkLogger,
-{
+/// `Write` retries write the originally queued buffer directly to the backup
+/// store and never re-read from primary. `Remove` retries issue a remove
+/// against the backup store and treat `io::ErrorKind::NotFound` as success.
+///
+/// Successfully retried entries are removed from the queue and the updated
+/// queue state is persisted locally. Failed entries remain queued for later
+/// retry. If any retries fail in a round, the task waits using exponential
+/// backoff with jitter before retrying again.
+///
+/// This function runs for node lifetime and should be spawned as a background
+/// task during [`Node::start`] when semisynchronous backup retry is configured.
+///
+/// [`Node::start`]: crate::Node::start
+pub(crate) async fn run_backup_retry_task(
+	queue: Arc<BackupRetryQueue<Arc<Logger>>>, backup_store: Arc<DynStore>, logger: Arc<Logger>,
+	mut stop_receiver: tokio::sync::watch::Receiver<()>,
+) {
 	let mut backoff_ms = RETRY_INITIAL_BACKOFF_MS;
-	let mut consecutive_failures: usize = 0;
 
 	loop {
-		// Wait until there is something to do.
-		queue.wait_for_entries().await;
+		let fallback_sleep = tokio::time::sleep(Duration::from_secs(60));
+		tokio::pin!(fallback_sleep);
+
+		tokio::select! {
+			_ = stop_receiver.changed() => break,
+			_ = queue.wait_for_entries() => {},
+			_ = &mut fallback_sleep => {},
+		}
+
+		if queue.is_empty() {
+			continue;
+		}
 
 		let entries = queue.snapshot();
-		let mut all_succeeded = true;
+		let mut any_failed = false;
 
 		for (key, op) in &entries {
+			if stop_receiver.has_changed().unwrap_or(false) {
+				return;
+			}
+
 			let (pn, sn, k) = key;
 
-			// Before applying, check whether this key is still in the queue
-			// with the same op. A newer op may have replaced it since we
-			// took the snapshot.
+			// Skip stale snapshot entries that were removed or replaced
+			// after we took the snapshot.
 			{
 				let locked = queue.pending.lock().expect("lock");
 				match locked.get(key) {
-					None => {
-						// Entry was removed (completed by a concurrent retry).
-						continue;
-					},
-					Some(current_op) => {
-						// If the op in the queue differs from our snapshot,
-						// a newer op replaced it. Skip — next iteration will
-						// pick up the newer op.
-						if !ops_match(current_op, op) {
-							continue;
-						}
-					},
+					None => continue,
+					Some(current_op) if !ops_match(current_op, op) => continue,
+					Some(_) => {},
 				}
 			}
 
@@ -1121,43 +1144,39 @@ pub(crate) async fn run_backup_retry_task<L: Deref + Send + Sync + 'static>(
 					KVStoreSync::write(backup_store.as_ref(), pn, sn, k, buf.clone())
 				},
 				PendingBackupOp::Remove { lazy } => {
-					KVStoreSync::remove(backup_store.as_ref(), pn, sn, k, *lazy)
+					match KVStoreSync::remove(backup_store.as_ref(), pn, sn, k, *lazy) {
+						Ok(()) => Ok(()),
+						Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
+						Err(e) => Err(e),
+					}
 				},
 			};
 
 			match result {
 				Ok(()) => {
 					queue.remove_completed(key);
-					consecutive_failures = 0;
-					backoff_ms = RETRY_INITIAL_BACKOFF_MS;
 				},
 				Err(e) => {
-					all_succeeded = false;
-					consecutive_failures += 1;
-					// Throttle logging — only log every 8 failures to avoid
-					// spam on persistent remote store outages.
-					if consecutive_failures == 1 || consecutive_failures.is_power_of_two() {
-						log_error!(
-							logger,
-							"Backup retry failed for key {}/{}/{} \
-                            (attempt {}): {}",
-							pn,
-							sn,
-							k,
-							consecutive_failures,
-							e
-						);
-					}
+					any_failed = true;
+					log_error!(logger, "Backup retry failed for key {}/{}/{}: {}", pn, sn, k, e);
 				},
 			}
 		}
 
-		if !all_succeeded {
-			// Exponential backoff with a small jitter.
+		if any_failed {
 			let jitter_ms = (backoff_ms / 10).max(1);
-			let sleep_ms = backoff_ms + (jitter_ms % 17); // deterministic but offset
-			tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
+			let sleep_ms = backoff_ms + (jitter_ms % 17);
+			let sleep = tokio::time::sleep(Duration::from_millis(sleep_ms));
+			tokio::pin!(sleep);
+
+			tokio::select! {
+				_ = stop_receiver.changed() => break,
+				_ = &mut sleep => {},
+			}
+
 			backoff_ms = (backoff_ms * RETRY_BACKOFF_MULTIPLIER).min(RETRY_MAX_BACKOFF_MS);
+		} else {
+			backoff_ms = RETRY_INITIAL_BACKOFF_MS;
 		}
 	}
 }

src/lib.rs

@@ -180,6 +180,7 @@ use types::{
 pub use types::{ChannelDetails, CustomTlvRecord, PeerDetails, SyncAndAsyncKVStore, UserChannelId};
 pub use vss_client;
 
+use crate::io::tier_store::{run_backup_retry_task, TierStore};
 use crate::scoring::setup_background_pathfinding_scores_sync;
 use crate::wallet::FundingAmount;
 
@@ -239,6 +240,7 @@ pub struct Node {
 	om_mailbox: Option<Arc<OnionMessageMailbox>>,
 	async_payments_role: Option<AsyncPaymentsRole>,
 	hrn_resolver: Arc<HRNResolver>,
+	tier_store: Arc<TierStore>,
 	#[cfg(cycle_tests)]
 	_leak_checker: LeakChecker,
 }
@@ -346,6 +348,26 @@ impl Node {
 			);
 		}
 
+		// Spawn background task to asynchronously retry failed backup operations.
+		if let Some(backup_store) = self.tier_store.backup_store() {
+			if let Some(retry_queue) = backup_store.retry_queue.as_ref() {
+				let retry_queue = Arc::clone(retry_queue);
+				let retry_backup_store = Arc::clone(&backup_store.store);
+				let retry_logger = Arc::clone(&self.logger);
+				let stop_retry = self.stop_sender.subscribe();
+
+				self.runtime.spawn_background_task(async move {
+					run_backup_retry_task(
+						retry_queue,
+						retry_backup_store,
+						retry_logger,
+						stop_retry,
+					)
+					.await;
+				});
+			}
+		}
+
 		if let Some(listening_addresses) = &self.config.listening_addresses {
 			// Setup networking
 			let peer_manager_connection_handler = Arc::clone(&self.peer_manager);