f Split prun_stale_monitors to archive_.. and remove_..

Author: jbesraa

lightning/src/chain/chainmonitor.rs

@@ -194,12 +194,35 @@ 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;
-
-	/// Prune a channel's data.
+	/// Archive a channel's data to a backup location.
+	///
+	/// Invoked by [`ChainMonitor::archive_stale_channel_monitors`].
+	///
+	/// This function can be used as a first step in pruning a channel's monitor data.
+	/// The data should be moved to an archive location, and then removed
+	/// using [`Persist::remove_persisted_channel`].
+	///
+	/// This is usually called when a channel monitor is in a stale state, ie no
+	/// fund can be claimed with its data. This is a good time to
+	/// archive the channel data to a backup location, before removing
+	/// it from the primary storage.
 	///
-	/// This is called when a channel is stale, ie we dont have balance to claim and its
-	/// closed.
-	fn prune_persisted_channel(&self, channel_funding_outpoint: OutPoint) -> bool;
+	/// 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;
+	/// *Dangerous* - removes a channel's data.
+	///
+	/// Only call this when the data have been archived using [`Persist::archive_persisted_channel`].
+	///
+	/// This function will remove the persisted channel data for the
+	/// given `channel_funding_outpoint` and should only be used when
+	/// the data is stale and no longer needed.` A stale channel
+	/// monitor is one where no funds can be claimed with its data.
+	///
+	/// Invoked by [`ChainMonitor::archive_stale_channel_monitors`]
+	/// after verifying the channel's data is no longer needed and has been archived using
+	/// [`Persist::archive_persisted_channel`].
+	fn remove_persisted_channel(&self, channel_funding_outpoint: OutPoint) -> ChannelMonitorUpdateStatus;
 }
 
 struct MonitorHolder<ChannelSigner: WriteableEcdsaChannelSigner> {
@@ -663,21 +686,23 @@ where C::Target: chain::Filter,
 		}
 	}
 
-	pub fn prune_stale_channel_monitors(&self, to_prune: Vec<OutPoint>) {
+	/// Archives channel monitors which are stale, meaning they have
+	/// no funds to claim and are eligible for removal. 
+	///
+	/// This is useful for pruning stale channels from the monitor set.
+	///
+	/// The monitor data is archived to a backup location and then removed from the primary storage.
+	pub fn archive_stale_channel_monitors(&self, to_archive: Vec<OutPoint>) {
 		let mut monitors = self.monitors.write().unwrap();
-		for funding_txo in to_prune {
+		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_stale() {
-					log_info!(self.logger, "Pruning stale ChannelMonitor for
-						channel {}", log_funding_info!(channel_monitor.monitor));
-					//TODO: save the channel monitor to disk in an archived namespace before removing it
-
-					self.persister.prune_persisted_channel(funding_txo);
-					monitors.remove(&funding_txo);
-				}
-			}
-
+				if channel_monitor.monitor.is_stale() 
+					&& self.persister.archive_persisted_channel(funding_txo) == ChannelMonitorUpdateStatus::Completed 
+					&& self.persister.remove_persisted_channel(funding_txo) == ChannelMonitorUpdateStatus::Completed {
+						monitors.remove(&funding_txo);
+				};
+			};
 		}
 	}
 }
@@ -1133,26 +1158,8 @@ mod tests {
 		}).is_err());
 	}
 
-	#[test]
-	fn prune_stale_channel_monitor() {
-		// Test that we can prune a ChannelMonitor that has no active channel.
-		let chanmon_cfgs = create_chanmon_cfgs(2);
-		let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
-		let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
-		let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
-		create_announced_chan_between_nodes(&nodes, 0, 1);
-		// Get a route for later and rebalance the channel somewhat
-		send_payment(&nodes[0], &[&nodes[1]], 10_000_000);
-		// First route a payment that we will claim on chain and give the recipient the preimage.
-		let (payment_preimage, payment_hash, ..) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000);
-		nodes[1].node.claim_funds(payment_preimage);
-		expect_payment_claimed!(nodes[1], payment_hash, 1_000_000);
-		nodes[1].node.get_and_clear_pending_msg_events();
-		let binding = node_cfgs[1].chain_monitor.added_monitors.lock().unwrap();
-		let monitors_b = binding.first().unwrap();
-		let outpoint = monitors_b.0.clone();
-		dbg!(&outpoint);
-		// nodes[1].chain_monitor().unwrap().chain_monitor.prune_stale_channel_monitors(vec![outpoint]); // lock order problem
-		assert!(false);
-	}
+	// #[test]
+	// fn archive_stale_channel_monitors() {
+	// 	unimplemented!()
+	// }
 }

lightning/src/util/persist.rs

@@ -58,10 +58,10 @@ 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 [`ChannelMonitor`]s will be persisted.
-pub const PRUNED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE: &str = "pruned_monitors";
-/// The secondary namespace under which [`ChannelMonitor`]s will be persisted.
-pub const PRUNED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+/// 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 = "";
@@ -247,15 +247,38 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore> Persist<ChannelSign
 		}
 	}
 
-	fn prune_persisted_channel(&self, funding_txo: OutPoint) -> bool {
+	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(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_e) => chain::ChannelMonitorUpdateStatus::UnrecoverableError // TODO: Should we return UnrecoverableError here?
+		}
+	}
+
+	fn remove_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
 		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(()) => true,
-			Err(_) => false
+			&key,
+			false,
+		) {
+			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
 		}
 	}
 }
@@ -290,15 +313,38 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner> Persist<ChannelSigner> for dyn
 		}
 	}
 
-	fn prune_persisted_channel(&self, funding_txo: OutPoint) -> bool {
+	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(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_e) => chain::ChannelMonitorUpdateStatus::UnrecoverableError // TODO: Should we return UnrecoverableError here?
+		}
+	}
+
+	fn remove_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
 		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(()) => true,
-			Err(_) => false
+			&key,
+			false,
+		) {
+			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
 		}
 	}
 }
@@ -623,6 +669,46 @@ where
 			}
 		}
 	}
+	/// Read an archived channel monitor.
+	fn read_archived_channel_monitor(&self, archived_monitor_name: &MonitorName) -> Result<(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>), io::Error>  { 
+		let outpoint: OutPoint = archived_monitor_name.try_into()?;
+		let mut monitor_cursor = io::Cursor::new(self.kv_store.read(
+				ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+				ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+				archived_monitor_name.as_str(),
+		)?);
+		match <(BlockHash, ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>)>::read(
+			&mut monitor_cursor,
+			(&*self.entropy_source, &*self.signer_provider),
+		) {
+			Ok((blockhash, channel_monitor)) => {
+				if channel_monitor.get_funding_txo().0.txid != outpoint.txid
+					|| channel_monitor.get_funding_txo().0.index != outpoint.index
+				{
+					log_error!(
+						self.logger,
+						"Archived ChannelMonitor {} was stored under the wrong key!",
+						archived_monitor_name.as_str()
+					);
+					Err(io::Error::new(
+						io::ErrorKind::InvalidData,
+						"Archived ChannelMonitor was stored under the wrong key",
+					))
+				} else {
+					Ok((blockhash, channel_monitor))
+				}
+			}
+			Err(e) => {
+				log_error!(
+					self.logger,
+					"Failed to read archived ChannelMonitor {}, reason: {}",
+					archived_monitor_name.as_str(),
+					e,
+				);
+				Err(io::Error::new(io::ErrorKind::InvalidData, "Failed to read archived ChannelMonitor"))
+			}
+		}
+	}
 
 	/// Read a channel monitor update.
 	fn read_monitor_update(
@@ -808,26 +894,33 @@ where
 		}
 	}
 
-	fn prune_persisted_channel(&self, funding_txo: OutPoint) -> bool {
+	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(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_e) => chain::ChannelMonitorUpdateStatus::UnrecoverableError // TODO: Should we return UnrecoverableError here?
+		}
+	}
+
+	fn remove_persisted_channel(&self, funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
 		let monitor_name = MonitorName::from(funding_txo);
 		match self.kv_store.remove(
 			CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
 			CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
 			monitor_name.as_str(),
 			false,
 		) {
-			Ok(()) => true,
-			Err(e) => {
-				log_error!(
-					self.logger,
-					"Failed to remove ChannelMonitor {}/{}/{} reason: {}",
-					CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
-					CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
-					monitor_name.as_str(),
-					e
-				);
-				false
-			}
+			Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
+			Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
 		}
 	}
 }

lightning/src/util/test_utils.rs

@@ -502,8 +502,12 @@ impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Sig
 		res
 	}
 
-	fn prune_persisted_channel(&self, _funding_txo: OutPoint) -> bool {
-		false
+	fn archive_persisted_channel(&self, _funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus {
+		unimplemented!()
+	}
+
+	fn remove_persisted_channel(&self, _funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus { 
+		unimplemented!()
 	}
 }
 
@@ -554,8 +558,12 @@ impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Sig
 		ret
 	}
 
-	fn prune_persisted_channel(&self, _funding_txo: OutPoint) -> bool { 
-		false
+	fn archive_persisted_channel(&self, _funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus { 
+		unimplemented!()
+	}
+
+	fn remove_persisted_channel(&self, _funding_txo: OutPoint) -> chain::ChannelMonitorUpdateStatus { 
+		unimplemented!()
 	}
 }