channelmanager: Create FundRecoverer to take our node in offline mode so that we can just send a BogusChannelReestablish and close all the StubChannelMonitors and sweep the funds from the events.

Author: Aditya Sharma

lightning/src/chain/chainmonitor.rs

@@ -165,8 +165,8 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> {
 	fn archive_persisted_channel(&self, channel_funding_outpoint: OutPoint);
 }
 
-struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
-	monitor: ChannelMonitor<ChannelSigner>,
+pub(crate) struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
+	pub(crate) monitor: ChannelMonitor<ChannelSigner>,
 	/// The full set of pending monitor updates for this Channel.
 	///
 	/// Note that this lock must be held from [`ChannelMonitor::update_monitor`] through to
@@ -181,7 +181,7 @@ struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> {
 	/// could cause users to have a full [`ChannelMonitor`] on disk as well as a
 	/// [`ChannelMonitorUpdate`] which was already applied. While this isn't an issue for the
 	/// LDK-provided update-based [`Persist`], it is somewhat surprising for users so we avoid it.
-	pending_monitor_updates: Mutex<Vec<u64>>,
+	pub(crate) pending_monitor_updates: Mutex<Vec<u64>>,
 }
 
 impl<ChannelSigner: EcdsaChannelSigner> MonitorHolder<ChannelSigner> {
@@ -195,8 +195,8 @@ impl<ChannelSigner: EcdsaChannelSigner> MonitorHolder<ChannelSigner> {
 /// Note that this holds a mutex in [`ChainMonitor`] and may block other events until it is
 /// released.
 pub struct LockedChannelMonitor<'a, ChannelSigner: EcdsaChannelSigner> {
-	lock: RwLockReadGuard<'a, HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
-	funding_txo: OutPoint,
+	pub(crate) lock: RwLockReadGuard<'a, HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
+	pub(crate) funding_txo: OutPoint,
 }
 
 impl<ChannelSigner: EcdsaChannelSigner> Deref for LockedChannelMonitor<'_, ChannelSigner> {
@@ -247,73 +247,19 @@ pub struct ChainMonitor<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F
 	event_notifier: Notifier,
 }
 
-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);
 
@@ -338,7 +284,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)
@@ -354,7 +300,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(..) {
@@ -365,13 +311,85 @@ 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(())
 	}
 
+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> 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.
 	///
 	/// When an optional chain source implementing [`chain::Filter`] is provided, the chain monitor
@@ -750,17 +768,6 @@ where C::Target: chain::Filter,
 	    L::Target: Logger,
 	    P::Target: Persist<ChannelSigner>,
 {
-	fn get_stub_cids_with_counterparty(&self, counterparty_node_id: PublicKey) -> Vec<ChannelId> {
-		let stub_monitors = self.stub_monitors.read().unwrap();
-		let mut stubs = vec![];
-		for (_, mon) in stub_monitors.iter() {
-			if mon.get_counterparty_node_id() == Some(counterparty_node_id) {
-				stubs.push(mon.channel_id());
-			}
-		}
-		stubs
-	}
-
 	fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor<ChannelSigner>) -> Result<ChannelMonitorUpdateStatus, ()> {
 		let logger = WithChannelMonitor::from(&self.logger, &monitor, None);
 		let mut monitors = self.monitors.write().unwrap();

lightning/src/chain/channelmonitor.rs

@@ -1440,6 +1440,13 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		})
 	}
 
+	pub(crate) fn merge_commitment_secret(&mut self, monitor: ChannelMonitor<Signer>) {
+		if self.get_min_seen_secret() > monitor.get_min_seen_secret() {
+			let inner = monitor.inner.lock().unwrap();
+			self.inner.lock().unwrap().commitment_secrets = inner.commitment_secrets.clone();
+		}
+	}
+
 	/// Returns a [`ChannelMonitor`] using [`StubChannelMonitor`] and other
 	/// important information to sweep funds and create penalty transactions.
 	pub(crate) fn new_stub(secp_ctx: Secp256k1<secp256k1::All>, stub_channel: &StubChannelMonitor, keys: Signer, channel_parameters: ChannelTransactionParameters ,funding_info_scriptbuf: ScriptBuf, destination_script: ScriptBuf) -> ChannelMonitor<Signer> {
@@ -1689,6 +1696,10 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitor<Signer> {
 		}
 	}
 
+	pub fn update_latest_state_from_new_stubmonitor(&self, stub: &StubChannelMonitor) {
+		self.inner.lock().unwrap().update_latest_state_from_new_stubmonitor(stub);
+	}
+
 	/// Get the list of HTLCs who's status has been updated on chain. This should be called by
 	/// ChannelManager via [`chain::Watch::release_pending_monitor_events`].
 	pub fn get_and_clear_pending_monitor_events(&self) -> Vec<MonitorEvent> {
@@ -3532,13 +3543,14 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		self.current_holder_commitment_number
 	}
 
-	/// Updates the [`StubChannelMonitor`] when we receive a new more recent
-	/// peer storage from our peer. This souldn't be called through [`ChannelMonitor`].
-	fn update_latest_state_from_new_stubchannelmonitor(&mut self, stub: &StubChannelMonitor<Signer>) {
-		let inner = stub.inner.lock().unwrap();
-		self.commitment_secrets = inner.commitment_secrets.clone();
-		self.counterparty_claimable_outpoints = inner.counterparty_claimable_outpoints.clone();
-		self.their_cur_per_commitment_points = inner.their_cur_per_commitment_points.clone();
+	/// Updates the [`ChannelMonitor`] when we receive a new more recent
+	/// peer storage from our peer.
+	fn update_latest_state_from_new_stubmonitor(&mut self, stub: &StubChannelMonitor) {
+		let mut latest_state = new_hash_map();
+		latest_state.insert(stub.latest_state.unwrap(), Vec::new());
+		self.commitment_secrets = stub.commitment_secrets.clone();
+		self.counterparty_claimable_outpoints = latest_state;
+		self.their_cur_per_commitment_points = stub.their_cur_per_commitment_points.clone();
 	}
 
 	/// Attempts to claim a counterparty commitment transaction's outputs using the revocation key and

lightning/src/chain/mod.rs

@@ -305,15 +305,6 @@ pub trait Watch<ChannelSigner: EcdsaChannelSigner> {
 	/// For details on asynchronous [`ChannelMonitor`] updating and returning
 	/// [`MonitorEvent::Completed`] here, see [`ChannelMonitorUpdateStatus::InProgress`].
 	fn release_pending_monitor_events(&self) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, Option<PublicKey>)>;
-
-	/// Retrieves a list of channel IDs for [`StubChannelMonitor`] associated with a specific counterparty node ID.
-	///
-	/// This function searches through the collection of [`StubChannelMonitor`] and collects the channel IDs
-	/// of those monitors that have the specified counterparty node ID.
-	///
-	/// This is used by [`FundRecoverer`] to fetch all the [`ChannelId`] with a peer that needs recovery so that we can send them
-	/// `BogusChannelReestablish`.
-	fn get_stub_cids_with_counterparty(&self, counterparty_node_id: PublicKey) -> Vec<ChannelId>;
 }
 
 /// The `Filter` trait defines behavior for indicating chain activity of interest pertaining to

lightning/src/events/mod.rs

@@ -20,7 +20,7 @@ pub use bump_transaction::BumpTransactionEvent;
 
 use crate::blinded_path::message::OffersContext;
 use crate::blinded_path::payment::{Bolt12OfferContext, Bolt12RefundContext, PaymentContext, PaymentContextRef};
-use crate::chain::transaction;
+use crate::chain::{transaction, BestBlock};
 use crate::ln::channelmanager::{InterceptId, PaymentId, RecipientOnionFields};
 use crate::ln::channel::FUNDING_CONF_DEADLINE_BLOCKS;
 use crate::ln::features::ChannelTypeFeatures;
@@ -2281,6 +2281,13 @@ impl MaybeReadable for Event {
 	}
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum RecoveryEvent {
+	RescanBlock {
+		rescan_from: BestBlock
+	},
+}
+
 /// An event generated by ChannelManager which indicates a message should be sent to a peer (or
 /// broadcast to most peers).
 /// These events are handled by PeerManager::process_events if you are using a PeerManager.
@@ -2630,6 +2637,10 @@ pub trait EventsProvider {
 	fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler;
 }
 
+pub trait RecoverEventsProvider: EventsProvider {
+	fn process_pending_recovery_events<RH: Deref>(&self, handler:RH) where RH::Target: RecoveryHandler;
+}
+
 /// An error type that may be returned to LDK in order to safely abort event handling if it can't
 /// currently succeed (e.g., due to a persistence failure).
 ///
@@ -2661,3 +2672,19 @@ impl<T: EventHandler> EventHandler for Arc<T> {
 		self.deref().handle_event(event)
 	}
 }
+
+pub trait RecoveryHandler {
+	fn handle_recovery_event(&self, event: RecoveryEvent) -> Result<(), ReplayEvent>;
+}
+
+impl<F> RecoveryHandler for F where F: Fn(RecoveryEvent) -> Result<(), ReplayEvent> {
+	fn handle_recovery_event(&self, event: RecoveryEvent) -> Result<(), ReplayEvent> {
+		self(event)
+	}
+}
+
+impl<T: RecoveryHandler> RecoveryHandler for Arc<T> {
+	fn handle_recovery_event(&self, event: RecoveryEvent) -> Result<(), ReplayEvent> {
+		self.deref().handle_recovery_event(event)
+	}
+}