Add FundRecoverer to recover funds from PeerStorage

Introduce FundRecoverer, a utility to recover funds from PeerStorage while running the node
in offline mode.

 - Define Fundrecoverer and Implement MessageSendEventsProvider, chain::Confirm, ChannelMessageHandler, events::EventsProvider.
 - Write Handler for your_peer_storage, peer_connected, channel_reestablish.
 - Implement functions to process new blocks defined in chain::Confirm.
 - Define RecoveryEvent which would be used to notify upstream to rescan the blockchain on the go.
 - Write watch_dummy() to persist the StubChannelMonitor and load outputs to watch.
 - Define ReadUtilOpt for defining parameters to be passed to ChannelMonitor::read_util in case of stub ChannelMonitors and otherwise.
 - Write utility for ChannelMonitor::read() to read ChannelMonitors from PeerStorage.

Author: Aditya Sharma

lightning/src/chain/chainmonitor.rs

@@ -254,73 +254,19 @@ pub struct ChainMonitor<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F
 	our_peerstorage_encryption_key: [u8;32],
 }
 
-impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
-where C::Target: chain::Filter,
-	    T::Target: BroadcasterInterface,
-	    F::Target: FeeEstimator,
-	    L::Target: Logger,
-	    P::Target: Persist<ChannelSigner>,
-{
-	/// Dispatches to per-channel monitors, which are responsible for updating their on-chain view
-	/// of a channel and reacting accordingly based on transactions in the given chain data. See
-	/// [`ChannelMonitor::block_connected`] for details. Any HTLCs that were resolved on chain will
-	/// be returned by [`chain::Watch::release_pending_monitor_events`].
-	///
-	/// Calls back to [`chain::Filter`] if any monitor indicated new outputs to watch. Subsequent
-	/// calls must not exclude any transactions matching the new outputs nor any in-block
-	/// descendants of such transactions. It is not necessary to re-fetch the block to obtain
-	/// updated `txdata`.
-	///
-	/// Calls which represent a new blockchain tip height should set `best_height`.
-	fn process_chain_data<FN>(&self, header: &Header, best_height: Option<u32>, txdata: &TransactionData, process: FN)
+pub(crate) fn update_monitor_with_chain_data_util <FN, P: Deref, ChannelSigner, C:Deref, L:Deref>(
+	persister: &P, chain_source: &Option<C>, logger: &L, header: &Header, best_height: Option<u32>, txdata: &TransactionData, process: FN, funding_outpoint: &OutPoint,
+	monitor_state: &MonitorHolder<ChannelSigner>, channel_count: usize,
+	) -> Result<(), ()>
 	where
-		FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>
+	C::Target: chain::Filter,
+	FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>,
+	P::Target: Persist<ChannelSigner>,
+	L::Target: Logger,
+	ChannelSigner: EcdsaChannelSigner,
 	{
-		let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down.";
-		let funding_outpoints = hash_set_from_iter(self.monitors.read().unwrap().keys().cloned());
-		let channel_count = funding_outpoints.len();
-		for funding_outpoint in funding_outpoints.iter() {
-			let monitor_lock = self.monitors.read().unwrap();
-			if let Some(monitor_state) = monitor_lock.get(funding_outpoint) {
-				if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state, channel_count).is_err() {
-					// Take the monitors lock for writing so that we poison it and any future
-					// operations going forward fail immediately.
-					core::mem::drop(monitor_lock);
-					let _poison = self.monitors.write().unwrap();
-					log_error!(self.logger, "{}", err_str);
-					panic!("{}", err_str);
-				}
-			}
-		}
-
-		// do some followup cleanup if any funding outpoints were added in between iterations
-		let monitor_states = self.monitors.write().unwrap();
-		for (funding_outpoint, monitor_state) in monitor_states.iter() {
-			if !funding_outpoints.contains(funding_outpoint) {
-				if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state, channel_count).is_err() {
-					log_error!(self.logger, "{}", err_str);
-					panic!("{}", err_str);
-				}
-			}
-		}
-
-		if let Some(height) = best_height {
-			// If the best block height is being updated, update highest_chain_height under the
-			// monitors write lock.
-			let old_height = self.highest_chain_height.load(Ordering::Acquire);
-			let new_height = height as usize;
-			if new_height > old_height {
-				self.highest_chain_height.store(new_height, Ordering::Release);
-			}
-		}
-	}
-
-	fn update_monitor_with_chain_data<FN>(
-		&self, header: &Header, best_height: Option<u32>, txdata: &TransactionData, process: FN, funding_outpoint: &OutPoint,
-		monitor_state: &MonitorHolder<ChannelSigner>, channel_count: usize,
-	) -> Result<(), ()> where FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs> {
 		let monitor = &monitor_state.monitor;
-		let logger = WithChannelMonitor::from(&self.logger, &monitor, None);
+		let logger = WithChannelMonitor::from(logger, &monitor, None);
 
 		let mut txn_outputs = process(monitor, txdata);
 
@@ -345,7 +291,7 @@ where C::Target: chain::Filter,
 			// `ChannelMonitorUpdate` after a channel persist for a channel with the same
 			// `latest_update_id`.
 			let _pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
-			match self.persister.update_persisted_channel(*funding_outpoint, None, monitor) {
+			match persister.update_persisted_channel(*funding_outpoint, None, monitor) {
 				ChannelMonitorUpdateStatus::Completed =>
 					log_trace!(logger, "Finished syncing Channel Monitor for channel {} for block-data",
 						log_funding_info!(monitor)
@@ -361,7 +307,7 @@ where C::Target: chain::Filter,
 
 		// Register any new outputs with the chain source for filtering, storing any dependent
 		// transactions from within the block that previously had not been included in txdata.
-		if let Some(ref chain_source) = self.chain_source {
+		if let Some(ref chain_source_ref) = chain_source {
 			let block_hash = header.block_hash();
 			for (txid, mut outputs) in txn_outputs.drain(..) {
 				for (idx, output) in outputs.drain(..) {
@@ -372,13 +318,62 @@ where C::Target: chain::Filter,
 						script_pubkey: output.script_pubkey,
 					};
 					log_trace!(logger, "Adding monitoring for spends of outpoint {} to the filter", output.outpoint);
-					chain_source.register_output(output);
+					chain_source_ref.register_output(output);
 				}
 			}
 		}
 		Ok(())
 	}
 
+// Utility function for process_chain_data to prevent code duplication in [`FundRecoverer`]
+pub(crate) fn process_chain_data_util<FN, ChannelSigner: EcdsaChannelSigner, L: Deref, P: Deref, C: Deref>(persister: &P, chain_source: &Option<C>,
+	logger: &L, monitors: &RwLock<HashMap<OutPoint, MonitorHolder<ChannelSigner>>>, highest_chain_height: &AtomicUsize,
+	header: &Header, best_height: Option<u32>, txdata: &TransactionData, process: FN)
+where
+	FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>,
+	L::Target: Logger,
+	P::Target: Persist<ChannelSigner>,
+	C::Target: chain::Filter,
+{
+	let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down.";
+		let funding_outpoints = hash_set_from_iter(monitors.read().unwrap().keys().cloned());
+		let channel_count = funding_outpoints.len();
+		for funding_outpoint in funding_outpoints.iter() {
+			let monitor_lock = monitors.read().unwrap();
+			if let Some(monitor_state) = monitor_lock.get(funding_outpoint) {
+				if update_monitor_with_chain_data_util(persister, chain_source, logger, header, best_height, txdata, &process, funding_outpoint, &monitor_state, channel_count).is_err() {
+					// Take the monitors lock for writing so that we poison it and any future
+					// operations going forward fail immediately.
+					core::mem::drop(monitor_lock);
+					let _poison = monitors.write().unwrap();
+					log_error!(logger, "{}", err_str);
+					panic!("{}", err_str);
+				}
+			}
+		}
+
+		// do some followup cleanup if any funding outpoints were added in between iterations
+		let monitor_states = monitors.write().unwrap();
+		for (funding_outpoint, monitor_state) in monitor_states.iter() {
+			if !funding_outpoints.contains(funding_outpoint) {
+				if update_monitor_with_chain_data_util(persister, chain_source, logger, header, best_height, txdata, &process, funding_outpoint, &monitor_state, channel_count).is_err() {
+					log_error!(logger, "{}", err_str);
+					panic!("{}", err_str);
+				}
+			}
+		}
+
+		if let Some(height) = best_height {
+			// If the best block height is being updated, update highest_chain_height under the
+			// monitors write lock.
+			let old_height = highest_chain_height.load(Ordering::Acquire);
+			let new_height = height as usize;
+			if new_height > old_height {
+				highest_chain_height.store(new_height, Ordering::Release);
+			}
+		}
+}
+
 impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> MessageSendEventsProvider for ChainMonitor<ChannelSigner, C, T, F, L, P>
 where C::Target: chain::Filter,
 		T::Target: BroadcasterInterface,
@@ -425,6 +420,30 @@ where C::Target: chain::Filter,
 	}
 }
 
+impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
+where C::Target: chain::Filter,
+	    T::Target: BroadcasterInterface,
+	    F::Target: FeeEstimator,
+	    L::Target: Logger,
+	    P::Target: Persist<ChannelSigner>,
+{
+	/// Dispatches to per-channel monitors, which are responsible for updating their on-chain view
+	/// of a channel and reacting accordingly based on transactions in the given chain data. See
+	/// [`ChannelMonitor::block_connected`] for details. Any HTLCs that were resolved on chain will
+	/// be returned by [`chain::Watch::release_pending_monitor_events`].
+	///
+	/// Calls back to [`chain::Filter`] if any monitor indicated new outputs to watch. Subsequent
+	/// calls must not exclude any transactions matching the new outputs nor any in-block
+	/// descendants of such transactions. It is not necessary to re-fetch the block to obtain
+	/// updated `txdata`.
+	///
+	/// Calls which represent a new blockchain tip height should set `best_height`.
+	fn process_chain_data<FN>(&self, header: &Header, best_height: Option<u32>, txdata: &TransactionData, process: FN)
+	where
+		FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>,
+	{
+		process_chain_data_util(&self.persister, &self.chain_source, &self.logger, &self.monitors, &self.highest_chain_height, header, best_height, txdata, process);
+	}
 
 	/// Creates a new `ChainMonitor` used to watch on-chain activity pertaining to channels.
 	///