feat(cbf): auto-restart bip157 node with exponential backoff

When node.run() exits (e.g. NoReachablePeers from kyoto lightningdevkit#558), the
background task rebuilds the node, swaps the requester, and respawns
channel processing tasks, up to MAX_RESTART_RETRIES (5) attempts with
doubling backoff starting at 500ms.

- Change cbf_runtime_status from Mutex<> to Arc<Mutex<>> so it can be
  shared with the async restart loop
- Extract build_cbf_node_static() that takes explicit params instead of
  &self, enabling calls from 'static async blocks
- Move all task spawning (info/warn/event + node.run) into a single
  restart loop inside spawn_background_task

AI: claude

Author: febyeji

src/chain/cbf.rs

@@ -79,7 +79,7 @@ pub(super) struct CbfChainSource {
 	/// Fee estimation back-end.
 	fee_source: FeeSource,
 	/// Tracks whether the bip157 node is running and holds the command handle.
-	cbf_runtime_status: Mutex<CbfRuntimeStatus>,
+	cbf_runtime_status: Arc<Mutex<CbfRuntimeStatus>>,
 	/// Latest chain tip hash, updated by the background event processing task.
 	latest_tip: Arc<Mutex<Option<BlockHash>>>,
 	/// Scripts to match against compact block filters during a scan.
@@ -147,7 +147,7 @@ impl CbfChainSource {
 			None => FeeSource::Cbf,
 		};
 
-		let cbf_runtime_status = Mutex::new(CbfRuntimeStatus::Stopped);
+		let cbf_runtime_status = Arc::new(Mutex::new(CbfRuntimeStatus::Stopped));
 		let latest_tip = Arc::new(Mutex::new(None));
 		let watched_scripts = Arc::new(RwLock::new(Vec::new()));
 		let matched_block_hashes = Arc::new(Mutex::new(Vec::new()));
@@ -183,46 +183,57 @@ impl CbfChainSource {
 
 	/// Build a new bip157 node and client from the current configuration.
 	///
-	/// If an on-chain wallet reference is available, a `ChainState::Checkpoint`
-	/// is derived from the wallet's persisted checkpoint (walked back by
-	/// `REORG_SAFETY_BLOCKS`) so the node resumes near its last known height
-	/// instead of re-syncing from genesis.
+	/// Delegates to [`Self::build_cbf_node_static`], passing all needed fields.
 	fn build_cbf_node(&self) -> (CbfNode, Client) {
-		let network = self.config.network;
+		let wallet = self.onchain_wallet.lock().unwrap().clone();
+		Self::build_cbf_node_static(
+			&self.peers,
+			&self.sync_config,
+			&self.config,
+			wallet.as_ref(),
+			&self.logger,
+		)
+	}
+
+	/// Static version of the builder: takes all required parameters explicitly
+	/// so it can be called from an `async move` block without borrowing `self`.
+	fn build_cbf_node_static(
+		peers: &[String], sync_config: &CbfSyncConfig, config: &Config,
+		wallet: Option<&Arc<Wallet>>, logger: &Logger,
+	) -> (CbfNode, Client) {
+		let network = config.network;
 
 		let mut builder = Builder::new(network);
 
 		// Configure data directory under the node's storage path.
-		let data_dir = std::path::PathBuf::from(&self.config.storage_dir_path).join("bip157_data");
+		let data_dir = std::path::PathBuf::from(&config.storage_dir_path).join("bip157_data");
 		builder = builder.data_dir(data_dir);
 
 		// Add configured peers.
-		let peers: Vec<TrustedPeer> = self
-			.peers
+		let trusted_peers: Vec<TrustedPeer> = peers
 			.iter()
 			.filter_map(|peer_str| {
 				peer_str.parse::<SocketAddr>().ok().map(TrustedPeer::from_socket_addr)
 			})
 			.collect();
-		if !peers.is_empty() {
-			builder = builder.add_peers(peers);
+		if !trusted_peers.is_empty() {
+			builder = builder.add_peers(trusted_peers);
 		}
 
 		// Require multiple peers to agree on filter headers before accepting them,
 		// as recommended by BIP 157 to mitigate malicious peer attacks.
-		builder = builder.required_peers(self.sync_config.required_peers);
+		builder = builder.required_peers(sync_config.required_peers);
 
 		// Request witness data so segwit transactions include full witnesses,
 		// required for Lightning channel operations.
 		builder = builder.fetch_witness_data();
 
 		// Set peer response timeout from user configuration (default: 30s).
-		builder =
-			builder.response_timeout(Duration::from_secs(self.sync_config.response_timeout_secs));
+		builder = builder.response_timeout(Duration::from_secs(sync_config.response_timeout_secs));
 
 		// If we have a wallet reference, derive a chain_state checkpoint so the
 		// bip157 node can skip already-synced headers on restart.
-		if let Some(wallet) = self.onchain_wallet.lock().unwrap().as_ref() {
+		if let Some(wallet) = wallet {
 			let cp = wallet.latest_checkpoint();
 			let target_height = cp.height().saturating_sub(REORG_SAFETY_BLOCKS);
 			// Walk the checkpoint chain back to the target height.
@@ -237,7 +248,7 @@ impl CbfChainSource {
 				let header_cp = HeaderCheckpoint::new(cursor.height(), cursor.hash());
 				builder = builder.chain_state(ChainState::Checkpoint(header_cp));
 				log_debug!(
-					self.logger,
+					logger,
 					"CBF builder: resuming from checkpoint height={}, hash={}",
 					cursor.height(),
 					cursor.hash(),
@@ -249,6 +260,11 @@ impl CbfChainSource {
 	}
 
 	/// Start the bip157 node and spawn background tasks for event processing.
+	///
+	/// The node runs inside a restart loop: if `node.run()` returns an error,
+	/// the loop rebuilds the node, swaps the requester, and respawns channel
+	/// processing tasks — up to [`MAX_RESTART_RETRIES`] consecutive failures
+	/// with exponential backoff starting at [`INITIAL_BACKOFF_MS`].
 	pub(crate) fn start(&self, runtime: Arc<Runtime>, onchain_wallet: Arc<Wallet>) {
 		let mut status = self.cbf_runtime_status.lock().unwrap();
 		if matches!(*status, CbfRuntimeStatus::Started { .. }) {
@@ -263,42 +279,118 @@ impl CbfChainSource {
 
 		let Client { requester, info_rx, warn_rx, event_rx } = client;
 
-		// Spawn the bip157 node in the background.
-		let node_logger = Arc::clone(&self.logger);
-		runtime.spawn_background_task(async move {
-			if let Err(e) = node.run().await {
-				log_error!(node_logger, "CBF node exited with error: {:?}", e);
-			}
-		});
-
-		// Spawn a task to log info messages.
-		let info_logger = Arc::clone(&self.logger);
-		runtime
-			.spawn_cancellable_background_task(Self::process_info_messages(info_rx, info_logger));
-
-		// Spawn a task to log warning messages.
-		let warn_logger = Arc::clone(&self.logger);
-		runtime
-			.spawn_cancellable_background_task(Self::process_warn_messages(warn_rx, warn_logger));
-
-		// Spawn a task to process events.
-		let event_state = CbfEventState {
-			latest_tip: Arc::clone(&self.latest_tip),
-			watched_scripts: Arc::clone(&self.watched_scripts),
-			matched_block_hashes: Arc::clone(&self.matched_block_hashes),
-			sync_completion_tx: Arc::clone(&self.sync_completion_tx),
-			filter_skip_height: Arc::clone(&self.filter_skip_height),
-		};
-		let event_logger = Arc::clone(&self.logger);
-		runtime.spawn_cancellable_background_task(Self::process_events(
-			event_rx,
-			event_state,
-			event_logger,
-		));
+		*status = CbfRuntimeStatus::Started { requester };
+		drop(status);
 
 		log_info!(self.logger, "CBF chain source started.");
 
-		*status = CbfRuntimeStatus::Started { requester };
+		// Clone all Arc references needed by the restart loop so the async
+		// block is 'static (no borrows of `self`).
+		let restart_status = Arc::clone(&self.cbf_runtime_status);
+		let restart_logger = Arc::clone(&self.logger);
+		let restart_latest_tip = Arc::clone(&self.latest_tip);
+		let restart_watched_scripts = Arc::clone(&self.watched_scripts);
+		let restart_matched_block_hashes = Arc::clone(&self.matched_block_hashes);
+		let restart_sync_completion_tx = Arc::clone(&self.sync_completion_tx);
+		let restart_filter_skip_height = Arc::clone(&self.filter_skip_height);
+		let restart_peers = self.peers.clone();
+		let restart_sync_config = self.sync_config.clone();
+		let restart_config = Arc::clone(&self.config);
+		let restart_wallet = Arc::clone(&onchain_wallet);
+
+		runtime.spawn_background_task(async move {
+			let mut current_node = node;
+			let mut current_info_rx = info_rx;
+			let mut current_warn_rx = warn_rx;
+			let mut current_event_rx = event_rx;
+			let mut retries = 0u32;
+			let mut backoff_ms = INITIAL_BACKOFF_MS;
+
+			loop {
+				// Spawn channel processing tasks for this iteration.
+				let info_handle = tokio::spawn(Self::process_info_messages(
+					current_info_rx,
+					Arc::clone(&restart_logger),
+				));
+				let warn_handle = tokio::spawn(Self::process_warn_messages(
+					current_warn_rx,
+					Arc::clone(&restart_logger),
+				));
+				let event_state = CbfEventState {
+					latest_tip: Arc::clone(&restart_latest_tip),
+					watched_scripts: Arc::clone(&restart_watched_scripts),
+					matched_block_hashes: Arc::clone(&restart_matched_block_hashes),
+					sync_completion_tx: Arc::clone(&restart_sync_completion_tx),
+					filter_skip_height: Arc::clone(&restart_filter_skip_height),
+				};
+				let event_handle = tokio::spawn(Self::process_events(
+					current_event_rx,
+					event_state,
+					Arc::clone(&restart_logger),
+				));
+
+				// Run the node until it exits.
+				match current_node.run().await {
+					Ok(()) => {
+						log_info!(restart_logger, "CBF node shut down cleanly.");
+						break;
+					},
+					Err(e) => {
+						retries += 1;
+						if retries > MAX_RESTART_RETRIES {
+							log_error!(
+								restart_logger,
+								"CBF node failed {} times, giving up: {:?}",
+								retries,
+								e,
+							);
+							break;
+						}
+						log_error!(
+							restart_logger,
+							"CBF node exited with error (attempt {}/{}): {:?}. \
+							 Restarting in {}ms.",
+							retries,
+							MAX_RESTART_RETRIES,
+							e,
+							backoff_ms,
+						);
+
+						tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
+						backoff_ms = backoff_ms.saturating_mul(2);
+
+						// Abort old channel processing tasks.
+						info_handle.abort();
+						warn_handle.abort();
+						event_handle.abort();
+
+						// Rebuild the node from scratch.
+						let (new_node, new_client) = Self::build_cbf_node_static(
+							&restart_peers,
+							&restart_sync_config,
+							&restart_config,
+							Some(&restart_wallet),
+							&restart_logger,
+						);
+						let Client {
+							requester: new_requester,
+							info_rx: new_info_rx,
+							warn_rx: new_warn_rx,
+							event_rx: new_event_rx,
+						} = new_client;
+
+						// Swap the requester so callers pick up the new handle.
+						*restart_status.lock().unwrap() =
+							CbfRuntimeStatus::Started { requester: new_requester };
+
+						current_node = new_node;
+						current_info_rx = new_info_rx;
+						current_warn_rx = new_warn_rx;
+						current_event_rx = new_event_rx;
+					},
+				}
+			}
+		});
 	}
 
 	/// Shut down the bip157 node and stop all background tasks.