Add ChainMonitor::archive_fully_resolved_monitor_channels

  Archives fully resolved channel monitors by adding them to a backup
  location and removing them from the primary storage & the monitor set.

  This is useful for pruning fully resolved monitors from the monitor
  set and primary storage so they are not reloaded on every new new
  block connection.

  We also add a new function, `archive_persisted_channel` to the
  `Persist` trait that writes the monitor to an archive storage and
  removes it from the primary storage.

Author: jbesraa

fuzz/src/chanmon_consistency.rs

@@ -459,6 +459,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 	let out = SearchingOutput::new(underlying_out);
 	let broadcast = Arc::new(TestBroadcaster{});
 	let router = FuzzRouter {};
+	use std::collections::HashSet;
 
 	macro_rules! make_node {
 		($node_id: expr, $fee_estimator: expr) => { {
@@ -467,7 +468,8 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			let keys_manager = Arc::new(KeyProvider { node_secret, rand_bytes_id: atomic::AtomicU32::new(0), enforcement_states: Mutex::new(new_hash_map()) });
 			let monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), $fee_estimator.clone(),
 				Arc::new(TestPersister {
-					update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed)
+					update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed),
+					archived_channels: Mutex::new(HashSet::new()),
 				}), Arc::clone(&keys_manager)));
 
 			let mut config = UserConfig::default();
@@ -494,7 +496,8 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			let logger: Arc<dyn Logger> = Arc::new(test_logger::TestLogger::new($node_id.to_string(), out.clone()));
 			let chain_monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), $fee_estimator.clone(),
 				Arc::new(TestPersister {
-					update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed)
+					update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed),
+					archived_channels: Mutex::new(HashSet::new()),
 				}), Arc::clone(& $keys_manager)));
 
 			let mut config = UserConfig::default();

fuzz/src/full_stack.rs

@@ -480,7 +480,10 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 
 	let broadcast = Arc::new(TestBroadcaster{ txn_broadcasted: Mutex::new(Vec::new()) });
 	let monitor = Arc::new(chainmonitor::ChainMonitor::new(None, broadcast.clone(), Arc::clone(&logger), fee_est.clone(),
-		Arc::new(TestPersister { update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed) })));
+		Arc::new(TestPersister { 
+			update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed) ,
+			archived_channels: Mutex::new(std::collections::HashSet::new()),
+	})));
 
 	let keys_manager = Arc::new(KeyProvider {
 		node_secret: our_network_key.clone(),

fuzz/src/utils/test_persister.rs

@@ -4,10 +4,12 @@ use lightning::chain::chainmonitor::MonitorUpdateId;
 use lightning::chain::transaction::OutPoint;
 use lightning::util::test_channel_signer::TestChannelSigner;
 
+use std::collections::HashSet;
 use std::sync::Mutex;
 
 pub struct TestPersister {
 	pub update_ret: Mutex<chain::ChannelMonitorUpdateStatus>,
+	pub archived_channels: Mutex<HashSet<OutPoint>>,
 }
 impl chainmonitor::Persist<TestChannelSigner> for TestPersister {
 	fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<TestChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
@@ -17,4 +19,9 @@ impl chainmonitor::Persist<TestChannelSigner> for TestPersister {
 	fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: Option<&channelmonitor::ChannelMonitorUpdate>, _data: &channelmonitor::ChannelMonitor<TestChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
 		self.update_ret.lock().unwrap().clone()
 	}
+
+	fn archive_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
+		self.archived_channels.lock().unwrap().insert(funding_txo);
+		chain::ChannelMonitorUpdateStatus::Completed
+	}
 }

lightning/src/chain/chainmonitor.rs

@@ -194,6 +194,17 @@ pub trait Persist<ChannelSigner: WriteableEcdsaChannelSigner> {
 	///
 	/// [`Writeable::write`]: crate::util::ser::Writeable::write
 	fn update_persisted_channel(&self, channel_funding_outpoint: OutPoint, update: Option<&ChannelMonitorUpdate>, data: &ChannelMonitor<ChannelSigner>, update_id: MonitorUpdateId) -> ChannelMonitorUpdateStatus;
+	/// Archive a channel's data to a backup location.
+	///
+	/// This function can be used to prune a stale channel's monitor.  It is reccommended
+	/// to move the data first to an archive location, and only then remove from the primary
+	/// storage.
+	///
+	/// A stale channel is a channel that has been closed and settled on-chain, and no funds
+	/// can be claimed with its data.
+	///
+	/// Archiving the data is useful for hedging against data loss in case of an unexpected failure/bug.
+	fn archive_persisted_channel(&self, channel_funding_outpoint: OutPoint) -> ChannelMonitorUpdateStatus;
 }
 
 struct MonitorHolder<ChannelSigner: WriteableEcdsaChannelSigner> {
@@ -656,6 +667,28 @@ where C::Target: chain::Filter,
 			}
 		}
 	}
+
+	/// Archives fully resolved channel monitors by adding them to a backup location and
+	/// removing them from the primary storage & the monitor set.
+	///
+	/// This is useful for pruning fully resolved monitors from the monitor set and primary
+	/// storage so they are not reloaded on every new new block connection.
+	///
+	/// The monitor data is archived to an archive namespace so we can still access it in case
+	/// of an unexpected failure/bug.
+	pub fn archive_fully_resolved_channel_monitors(&self, to_archive: Vec<OutPoint>) {
+		let mut monitors = self.monitors.write().unwrap();
+		for funding_txo in to_archive {
+			let channel_monitor = monitors.get(&funding_txo);
+			if let Some(channel_monitor) = channel_monitor {
+				if channel_monitor.monitor.is_fully_resolved() 
+					&& self.persister.archive_persisted_channel(funding_txo) == ChannelMonitorUpdateStatus::Completed {
+						monitors.remove(&funding_txo);
+				};
+			};
+		}
+	}
+
 }
 
 impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>

lightning/src/chain/channelmonitor.rs

@@ -1868,7 +1868,6 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 		let is_all_funds_claimed = self.get_claimable_balances().is_empty();
 		let current_height = self.current_best_block().height;
 		let mut inner = self.inner.lock().unwrap();
-
 		match (inner.balances_empty_height, is_all_funds_claimed) {
 			(Some(h), true) => {
 				// Claimed all funds, Check if reached the threshold.
@@ -1878,7 +1877,7 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitor<Signer> {
 			(Some(_), false) => {
 				// previously assumed we claimed all funds, but we have new funds to claim.
 				// Should not happen in practice.
-				debug_assert!(!is_all_funds_claimed, "Trying to check if monitor is fully resolved before all funds are claimed. Aborting.");
+				debug_assert!(is_all_funds_claimed, "Trying to check if monitor is fully resolved before all funds are claimed.");
 				inner.balances_empty_height = None;
 				return false;
 			},

lightning/src/ln/monitor_tests.rs

@@ -171,7 +171,7 @@ fn do_chanmon_claim_value_coop_close(anchors: bool) {
 		user_config.manually_accept_inbound_channels = true;
 	}
 	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[Some(user_config), Some(user_config)]);
-	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+	let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
 
 	let (_, _, chan_id, funding_tx) =
 		create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 1_000_000, 1_000_000);
@@ -262,6 +262,18 @@ fn do_chanmon_claim_value_coop_close(anchors: bool) {
 
 	assert!(nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());
 	assert!(nodes[1].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());
+
+	// Test we can archived fully resolved channel monitor.
+	assert_eq!(nodes[0].chain_monitor.chain_monitor.list_monitors().len(), 1); // one monitor
+	// first archive should set balances_empty_height to current block height
+	nodes[0].chain_monitor.chain_monitor.archive_fully_resolved_channel_monitors(vec![funding_outpoint]); 
+	connect_blocks(&nodes[0], 2016);
+	// Second call after 2016 blocks, should archive the monitor
+	nodes[0].chain_monitor.chain_monitor.archive_fully_resolved_channel_monitors(vec![funding_outpoint]);
+	assert_eq!(nodes[0].chain_monitor.chain_monitor.list_monitors().len(), 0); // no monitor
+	// cleanup
+	nodes.get_mut(0).unwrap().chain_source.remove_watched_txn_and_outputs(funding_outpoint,
+		funding_tx.txid(), funding_tx.output[0].script_pubkey.clone());
 }
 
 #[test]

lightning/src/util/persist.rs

@@ -58,6 +58,11 @@ pub const CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
 /// The primary namespace under which [`ChannelMonitorUpdate`]s will be persisted.
 pub const CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE: &str = "monitor_updates";
 
+/// The primary namespace under which archived [`ChannelMonitor`]s will be persisted.
+pub const ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE: &str = "archived_monitors";
+/// The secondary namespace under which archived [`ChannelMonitor`]s will be persisted.
+pub const ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+
 /// The primary namespace under which the [`NetworkGraph`] will be persisted.
 pub const NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE: &str = "";
 /// The secondary namespace under which the [`NetworkGraph`] will be persisted.
@@ -214,6 +219,40 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore + ?Sized> Persist<Ch
 			Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
 		}
 	}
+
+	fn archive_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
+		let monitor_name = MonitorName::from(funding_txo);
+		let monitor = match self.read(
+			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+			monitor_name.as_str(),
+		) {
+			Ok(monitor) => monitor,
+			Err(_) => {
+				return chain::ChannelMonitorUpdateStatus::UnrecoverableError;
+			}
+
+		};
+		match self.write(
+			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+			monitor_name.as_str(),
+			&monitor 
+		) {
+			Ok(()) => {}
+			Err(_e) => return chain::ChannelMonitorUpdateStatus::UnrecoverableError 
+		};
+		let key = format!("{}_{}", funding_txo.txid.to_string(), funding_txo.index);
+		match self.remove(
+			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+			&key,
+			false,
+		) {
+			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
+		}
+	}
 }
 
 /// Read previously persisted [`ChannelMonitor`]s from the store.
@@ -720,6 +759,34 @@ where
 			self.persist_new_channel(funding_txo, monitor, monitor_update_call_id)
 		}
 	}
+
+	fn archive_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
+		let monitor_name = MonitorName::from(funding_txo);
+		let monitor = match self.read_monitor(&monitor_name) {
+			Ok((_block_hash, monitor)) => monitor,
+			Err(_) => return chain::ChannelMonitorUpdateStatus::UnrecoverableError
+		};
+		match self.kv_store.write(
+			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+			monitor_name.as_str(),
+			&monitor.encode()
+		) {
+			Ok(()) => {},
+			Err(_e) => {
+				return chain::ChannelMonitorUpdateStatus::UnrecoverableError;
+			} // TODO: Should we return UnrecoverableError here
+		};
+		match self.kv_store.remove(
+			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+			monitor_name.as_str(),
+			false,
+		) {
+			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
+		}
+	}
 }
 
 impl<K: Deref, L: Deref, ES: Deref, SP: Deref> MonitorUpdatingPersister<K, L, ES, SP>

lightning/src/util/test_utils.rs

@@ -10,6 +10,7 @@
 use crate::blinded_path::BlindedPath;
 use crate::blinded_path::payment::ReceiveTlvs;
 use crate::chain;
+use crate::chain::chainmonitor::Persist;
 use crate::chain::WatchedOutput;
 use crate::chain::chaininterface;
 use crate::chain::chaininterface::ConfirmationTarget;
@@ -501,6 +502,12 @@ impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Sig
 		}
 		res
 	}
+
+	fn archive_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
+		let ret = <TestPersister as Persist<TestChannelSigner>>::archive_persisted_channel(&self.persister, funding_txo);
+
+		ret
+	}
 }
 
 pub struct TestPersister {
@@ -513,20 +520,27 @@ pub struct TestPersister {
 	/// When we get an update_persisted_channel call *with* a ChannelMonitorUpdate, we insert the
 	/// MonitorUpdateId here.
 	pub offchain_monitor_updates: Mutex<HashMap<OutPoint, HashSet<MonitorUpdateId>>>,
+	/// When we get an archive_persisted_channel call, we insert the OutPoint here.
+	pub archived_channels: Mutex<HashSet<OutPoint>>,
 }
 impl TestPersister {
 	pub fn new() -> Self {
 		Self {
 			update_rets: Mutex::new(VecDeque::new()),
 			chain_sync_monitor_persistences: Mutex::new(new_hash_map()),
 			offchain_monitor_updates: Mutex::new(new_hash_map()),
+			archived_channels: Mutex::new(new_hash_set()),
 		}
 	}
 
 	/// Queue an update status to return.
 	pub fn set_update_ret(&self, next_ret: chain::ChannelMonitorUpdateStatus) {
 		self.update_rets.lock().unwrap().push_back(next_ret);
 	}
+	// Check if the given OutPoint has been archived.
+	pub fn is_archived(&self, funding_txo: OutPoint) -> bool {
+		self.archived_channels.lock().unwrap().contains(&funding_txo)
+	}
 }
 impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Signer> for TestPersister {
 	fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<Signer>, _id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
@@ -549,6 +563,20 @@ impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Sig
 		}
 		ret
 	}
+
+	fn archive_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus { 
+		self.archived_channels.lock().unwrap().insert(funding_txo);
+		// remove the channel from the offchain_monitor_updates map
+		match self.offchain_monitor_updates.lock().unwrap().remove(&funding_txo) {
+			Some(_) => {},
+			None => {
+				// If the channel was not in the offchain_monitor_updates map, it should be in the
+				// chain_sync_monitor_persistences map.
+				assert!(self.chain_sync_monitor_persistences.lock().unwrap().remove(&funding_txo).is_some());
+			}
+		};
+		chain::ChannelMonitorUpdateStatus::Completed
+	}
 }
 
 pub struct TestStore {
@@ -1360,6 +1388,10 @@ impl TestChainSource {
 			watched_outputs: Mutex::new(new_hash_set()),
 		}
 	}
+	pub fn remove_watched_txn_and_outputs(&self, outpoint: OutPoint, txid: Txid, script_pubkey: ScriptBuf) {
+		self.watched_outputs.lock().unwrap().remove(&(outpoint, script_pubkey.clone())); 
+		self.watched_txn.lock().unwrap().remove(&(txid, script_pubkey));
+	}
 }
 
 impl UtxoLookup for TestChainSource {