Make ChainMonitor::monitors private and expose monitor via getter

Exposing a `RwLock<HashMap<>>` directly was always a bit strange,
and in upcoming changes we'd like to change the internal
datastructure in `ChainMonitor`.

Further, the use of `RwLock` and `HashMap` meant we weren't able
to expose the ChannelMonitors themselves to users in bindings,
leaving a bindings/rust API gap.

Thus, we take this opportunity go expose ChannelMonitors directly
via a wrapper, hiding the internals of `ChainMonitor` behind
getters. We also update tests to use the new API.

Author: TheBlueMatt

lightning/src/chain/chainmonitor.rs

@@ -38,7 +38,7 @@ use util::events::EventHandler;
 use ln::channelmanager::ChannelDetails;
 
 use prelude::*;
-use sync::RwLock;
+use sync::{RwLock, RwLockReadGuard};
 use core::ops::Deref;
 
 /// `Persist` defines behavior for persisting channel monitors: this could mean
@@ -92,6 +92,26 @@ pub trait Persist<ChannelSigner: Sign> {
 	fn update_persisted_channel(&self, id: OutPoint, update: &ChannelMonitorUpdate, data: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr>;
 }
 
+struct MonitorHolder<ChannelSigner: Sign> {
+	monitor: ChannelMonitor<ChannelSigner>,
+}
+
+/// A read-only reference to a current ChannelMonitor.
+///
+/// Note that this holds a mutex in [`ChainMonitor`] and may block other events until it is
+/// released.
+pub struct LockedChannelMonitor<'a, ChannelSigner: Sign> {
+	lock: RwLockReadGuard<'a, HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
+	funding_txo: OutPoint,
+}
+
+impl<ChannelSigner: Sign> Deref for LockedChannelMonitor<'_, ChannelSigner> {
+	type Target = ChannelMonitor<ChannelSigner>;
+	fn deref(&self) -> &ChannelMonitor<ChannelSigner> {
+		&self.lock.get(&self.funding_txo).expect("Checked at construction").monitor
+	}
+}
+
 /// An implementation of [`chain::Watch`] for monitoring channels.
 ///
 /// Connected and disconnected blocks must be provided to `ChainMonitor` as documented by
@@ -108,8 +128,7 @@ pub struct ChainMonitor<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: De
         L::Target: Logger,
         P::Target: Persist<ChannelSigner>,
 {
-	/// The monitors
-	pub monitors: RwLock<HashMap<OutPoint, ChannelMonitor<ChannelSigner>>>,
+	monitors: RwLock<HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
 	chain_source: Option<C>,
 	broadcaster: T,
 	logger: L,
@@ -139,8 +158,8 @@ where C::Target: chain::Filter,
 	{
 		let mut dependent_txdata = Vec::new();
 		let monitors = self.monitors.read().unwrap();
-		for monitor in monitors.values() {
-			let mut txn_outputs = process(monitor, txdata);
+		for monitor_state in monitors.values() {
+			let mut txn_outputs = process(&monitor_state.monitor, txdata);
 
 			// 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.
@@ -203,19 +222,41 @@ where C::Target: chain::Filter,
 	pub fn get_claimable_balances(&self, ignored_channels: &[&ChannelDetails]) -> Vec<Balance> {
 		let mut ret = Vec::new();
 		let monitors = self.monitors.read().unwrap();
-		for (_, monitor) in monitors.iter().filter(|(funding_outpoint, _)| {
+		for (_, monitor_state) in monitors.iter().filter(|(funding_outpoint, _)| {
 			for chan in ignored_channels {
 				if chan.funding_txo.as_ref() == Some(funding_outpoint) {
 					return false;
 				}
 			}
 			true
 		}) {
-			ret.append(&mut monitor.get_claimable_balances());
+			ret.append(&mut monitor_state.monitor.get_claimable_balances());
 		}
 		ret
 	}
 
+	/// Gets the [`LockedChannelMonitor`] for a given funding outpoint, returning an `Err` if no
+	/// such [`ChannelMonitor`] is currently being monitored for.
+	pub fn get_current_monitor(&self, funding_txo: OutPoint) -> Result<LockedChannelMonitor<'_, ChannelSigner>, ()> {
+		let lock = self.monitors.read().unwrap();
+		if lock.get(&funding_txo).is_some() {
+			Ok(LockedChannelMonitor { lock, funding_txo })
+		} else {
+			Err(())
+		}
+	}
+
+	/// Gets the current list of [`ChannelMonitor`]s being monitored for, by their funding
+	/// outpoint.
+	pub fn list_current_monitors(&self) -> Vec<OutPoint> {
+		self.monitors.read().unwrap().keys().map(|outpoint| *outpoint).collect()
+	}
+
+	#[cfg(test)]
+	pub fn remove_monitor(&self, funding_txo: &OutPoint) -> ChannelMonitor<ChannelSigner> {
+		self.monitors.write().unwrap().remove(funding_txo).unwrap().monitor
+	}
+
 	#[cfg(any(test, feature = "fuzztarget", feature = "_test_utils"))]
 	pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
 		use util::events::EventsProvider;
@@ -248,8 +289,8 @@ where
 	fn block_disconnected(&self, header: &BlockHeader, height: u32) {
 		let monitors = self.monitors.read().unwrap();
 		log_debug!(self.logger, "Latest block {} at height {} removed via block_disconnected", header.block_hash(), height);
-		for monitor in monitors.values() {
-			monitor.block_disconnected(
+		for monitor_state in monitors.values() {
+			monitor_state.monitor.block_disconnected(
 				header, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
 		}
 	}
@@ -275,8 +316,8 @@ where
 	fn transaction_unconfirmed(&self, txid: &Txid) {
 		log_debug!(self.logger, "Transaction {} reorganized out of chain", txid);
 		let monitors = self.monitors.read().unwrap();
-		for monitor in monitors.values() {
-			monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+		for monitor_state in monitors.values() {
+			monitor_state.monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
 		}
 	}
 
@@ -294,8 +335,8 @@ where
 	fn get_relevant_txids(&self) -> Vec<Txid> {
 		let mut txids = Vec::new();
 		let monitors = self.monitors.read().unwrap();
-		for monitor in monitors.values() {
-			txids.append(&mut monitor.get_relevant_txids());
+		for monitor_state in monitors.values() {
+			txids.append(&mut monitor_state.monitor.get_relevant_txids());
 		}
 
 		txids.sort_unstable();
@@ -338,7 +379,7 @@ where C::Target: chain::Filter,
 				monitor.load_outputs_to_watch(chain_source);
 			}
 		}
-		entry.insert(monitor);
+		entry.insert(MonitorHolder { monitor });
 		Ok(())
 	}
 
@@ -359,7 +400,8 @@ where C::Target: chain::Filter,
 				#[cfg(not(any(test, feature = "fuzztarget")))]
 				Err(ChannelMonitorUpdateErr::PermanentFailure)
 			},
-			Some(monitor) => {
+			Some(monitor_state) => {
+				let monitor = &monitor_state.monitor;
 				log_trace!(self.logger, "Updating Channel Monitor for channel {}", log_funding_info!(monitor));
 				let update_res = monitor.update_monitor(&update, &self.broadcaster, &self.fee_estimator, &self.logger);
 				if let Err(e) = &update_res {
@@ -382,8 +424,8 @@ where C::Target: chain::Filter,
 
 	fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
 		let mut pending_monitor_events = Vec::new();
-		for monitor in self.monitors.read().unwrap().values() {
-			pending_monitor_events.append(&mut monitor.get_and_clear_pending_monitor_events());
+		for monitor_state in self.monitors.read().unwrap().values() {
+			pending_monitor_events.append(&mut monitor_state.monitor.get_and_clear_pending_monitor_events());
 		}
 		pending_monitor_events
 	}
@@ -404,8 +446,8 @@ impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> even
 	/// [`SpendableOutputs`]: events::Event::SpendableOutputs
 	fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler {
 		let mut pending_events = Vec::new();
-		for monitor in self.monitors.read().unwrap().values() {
-			pending_events.append(&mut monitor.get_and_clear_pending_events());
+		for monitor_state in self.monitors.read().unwrap().values() {
+			pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events());
 		}
 		for event in pending_events.drain(..) {
 			handler.handle_event(&event);

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -118,8 +118,7 @@ fn test_monitor_and_persister_update_fail() {
 		blocks: Arc::new(Mutex::new(vec![(genesis_block(Network::Testnet).header, 200); 200])),
 	};
 	let chain_mon = {
-		let monitors = nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap();
-		let monitor = monitors.get(&outpoint).unwrap();
+		let monitor = nodes[0].chain_monitor.chain_monitor.get_current_monitor(outpoint).unwrap();
 		let mut w = test_utils::TestVecWriter(Vec::new());
 		monitor.write(&mut w).unwrap();
 		let new_monitor = <(BlockHash, ChannelMonitor<EnforcingSigner>)>::read(
@@ -2317,7 +2316,7 @@ fn do_channel_holding_cell_serialize(disconnect: bool, reload_a: bool) {
 		if reload_a {
 			let nodes_0_serialized = nodes[0].node.encode();
 			let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
-			nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+			get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap();
 
 			persister = test_utils::TestPersister::new();
 			let keys_manager = &chanmon_cfgs[0].keys_manager;

lightning/src/ln/functional_test_utils.rs

@@ -274,10 +274,9 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
 			let feeest = test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) };
 			let mut deserialized_monitors = Vec::new();
 			{
-				let old_monitors = self.chain_monitor.chain_monitor.monitors.read().unwrap();
-				for (_, old_monitor) in old_monitors.iter() {
+				for outpoint in self.chain_monitor.chain_monitor.list_current_monitors() {
 					let mut w = test_utils::TestVecWriter(Vec::new());
-					old_monitor.write(&mut w).unwrap();
+					self.chain_monitor.chain_monitor.get_current_monitor(outpoint).unwrap().write(&mut w).unwrap();
 					let (_, deserialized_monitor) = <(BlockHash, ChannelMonitor<EnforcingSigner>)>::read(
 						&mut io::Cursor::new(&w.0), self.keys_manager).unwrap();
 					deserialized_monitors.push(deserialized_monitor);
@@ -437,20 +436,35 @@ macro_rules! get_feerate {
 	}
 }
 
-/// Returns any local commitment transactions for the channel.
+/// Returns a channel monitor given a channel id, making some naive assumptions
 #[macro_export]
-macro_rules! get_local_commitment_txn {
+macro_rules! get_monitor {
 	($node: expr, $channel_id: expr) => {
 		{
-			let monitors = $node.chain_monitor.chain_monitor.monitors.read().unwrap();
-			let mut commitment_txn = None;
-			for (funding_txo, monitor) in monitors.iter() {
-				if funding_txo.to_channel_id() == $channel_id {
-					commitment_txn = Some(monitor.unsafe_get_latest_holder_commitment_txn(&$node.logger));
+			use bitcoin::hashes::Hash;
+			let mut monitor = None;
+			// Assume funding vout is either 0 or 1 blindly
+			for index in 0..2 {
+				if let Ok(mon) = $node.chain_monitor.chain_monitor.get_current_monitor(
+					$crate::chain::transaction::OutPoint {
+						txid: bitcoin::Txid::from_slice(&$channel_id[..]).unwrap(), index
+					})
+				{
+					monitor = Some(mon);
 					break;
 				}
 			}
-			commitment_txn.unwrap()
+			monitor.unwrap()
+		}
+	}
+}
+
+/// Returns any local commitment transactions for the channel.
+#[macro_export]
+macro_rules! get_local_commitment_txn {
+	($node: expr, $channel_id: expr) => {
+		{
+			$crate::get_monitor!($node, $channel_id).unsafe_get_latest_holder_commitment_txn(&$node.logger)
 		}
 	}
 }