Process updates before archiving monitors.

Author: arik-so

lightning/src/util/persist.rs

@@ -400,28 +400,34 @@ where
 /// If you have many stale updates stored (such as after a crash with pending lazy deletes), and
 /// would like to get rid of them, consider using the
 /// [`MonitorUpdatingPersister::cleanup_stale_updates`] function.
-pub struct MonitorUpdatingPersister<K: Deref, L: Deref, ES: Deref, SP: Deref>
+pub struct MonitorUpdatingPersister<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
 	ES::Target: EntropySource + Sized,
 	SP::Target: SignerProvider + Sized,
+	BI::Target: BroadcasterInterface,
+	FE::Target: FeeEstimator
 {
 	kv_store: K,
 	logger: L,
 	maximum_pending_updates: u64,
 	entropy_source: ES,
 	signer_provider: SP,
+	broadcaster: BI,
+	fee_estimator: FE
 }
 
 #[allow(dead_code)]
-impl<K: Deref, L: Deref, ES: Deref, SP: Deref>
-	MonitorUpdatingPersister<K, L, ES, SP>
+impl<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
+	MonitorUpdatingPersister<K, L, ES, SP, BI, FE>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
 	ES::Target: EntropySource + Sized,
 	SP::Target: SignerProvider + Sized,
+	BI::Target: BroadcasterInterface,
+	FE::Target: FeeEstimator
 {
 	/// Constructs a new [`MonitorUpdatingPersister`].
 	///
@@ -441,14 +447,16 @@ where
 	/// [`MonitorUpdatingPersister::cleanup_stale_updates`].
 	pub fn new(
 		kv_store: K, logger: L, maximum_pending_updates: u64, entropy_source: ES,
-		signer_provider: SP,
+		signer_provider: SP, broadcaster: BI, fee_estimator: FE
 	) -> Self {
 		MonitorUpdatingPersister {
 			kv_store,
 			logger,
 			maximum_pending_updates,
 			entropy_source,
 			signer_provider,
+			broadcaster,
+			fee_estimator
 		}
 	}
 
@@ -639,13 +647,15 @@ where
 	}
 }
 
-impl<ChannelSigner: EcdsaChannelSigner, K: Deref, L: Deref, ES: Deref, SP: Deref>
-	Persist<ChannelSigner> for MonitorUpdatingPersister<K, L, ES, SP>
+impl<ChannelSigner: EcdsaChannelSigner, K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
+	Persist<ChannelSigner> for MonitorUpdatingPersister<K, L, ES, SP, BI, FE>
 where
 	K::Target: KVStore,
 	L::Target: Logger,
 	ES::Target: EntropySource + Sized,
 	SP::Target: SignerProvider + Sized,
+	BI::Target: BroadcasterInterface,
+	FE::Target: FeeEstimator
 {
 	/// Persists a new channel. This means writing the entire monitor to the
 	/// parametrized [`KVStore`].
@@ -770,13 +780,53 @@ where
 			Ok((_block_hash, monitor)) => monitor,
 			Err(_) => return
 		};
+
+		let mut current_update_id = monitor.get_latest_update_id();
+		loop {
+			current_update_id = match current_update_id.checked_add(1) {
+				Some(next_update_id) => next_update_id,
+				None => break,
+			};
+			let update_name = UpdateName::from(current_update_id);
+			let update = match self.read_monitor_update(&monitor_name, &update_name) {
+				Ok(update) => update,
+				Err(err) if err.kind() == io::ErrorKind::NotFound => {
+					// We can't find any more updates, so we are done.
+					break;
+				}
+				Err(err) => {
+					log_error!(
+						self.logger,
+						"Monitor update read failed. monitor: {} update: {} reason: {:?}",
+						monitor_name.as_str(),
+						update_name.as_str(),
+						err
+					);
+					panic!()
+				},
+			};
+
+			monitor.update_monitor(&update, &self.broadcaster, &self.fee_estimator, &self.logger)
+				.map_err(|e| {
+					log_error!(
+						self.logger,
+						"Monitor update failed. monitor: {} update: {} reason: {:?}",
+						monitor_name.as_str(),
+						update_name.as_str(),
+						e
+					);
+					io::Error::new(io::ErrorKind::Other, "Monitor update failed")
+				})
+				.expect("Could not apply monitor update during archiving");
+		}
+
 		match self.kv_store.write(
 			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
 			ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
 			monitor_name.as_str(),
-			&monitor.encode()
+			&monitor.encode(),
 		) {
-			Ok(()) => {},
+			Ok(()) => {}
 			Err(_e) => return,
 		};
 		let _ = self.kv_store.remove(
@@ -788,12 +838,14 @@ where
 	}
 }
 
-impl<K: Deref, L: Deref, ES: Deref, SP: Deref> MonitorUpdatingPersister<K, L, ES, SP>
+impl<K: Deref, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref> MonitorUpdatingPersister<K, L, ES, SP, BI, FE>
 where
 	ES::Target: EntropySource + Sized,
 	K::Target: KVStore,
 	L::Target: Logger,
-	SP::Target: SignerProvider + Sized
+	SP::Target: SignerProvider + Sized,
+	BI::Target: BroadcasterInterface,
+	FE::Target: FeeEstimator
 {
 	// Cleans up monitor updates for given monitor in range `start..=end`.
 	fn cleanup_in_range(&self, monitor_name: MonitorName, start: u64, end: u64) {
@@ -962,13 +1014,17 @@ mod tests {
 			maximum_pending_updates: persister_0_max_pending_updates,
 			entropy_source: &chanmon_cfgs[0].keys_manager,
 			signer_provider: &chanmon_cfgs[0].keys_manager,
+			broadcaster: &chanmon_cfgs[0].tx_broadcaster,
+			fee_estimator: &chanmon_cfgs[0].fee_estimator,
 		};
 		let persister_1 = MonitorUpdatingPersister {
 			kv_store: &TestStore::new(false),
 			logger: &TestLogger::new(),
 			maximum_pending_updates: persister_1_max_pending_updates,
 			entropy_source: &chanmon_cfgs[1].keys_manager,
 			signer_provider: &chanmon_cfgs[1].keys_manager,
+			broadcaster: &chanmon_cfgs[0].tx_broadcaster,
+			fee_estimator: &chanmon_cfgs[0].fee_estimator,
 		};
 		let mut node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
 		let chain_mon_0 = test_utils::TestChainMonitor::new(
@@ -1129,6 +1185,8 @@ mod tests {
 				maximum_pending_updates: 11,
 				entropy_source: node_cfgs[0].keys_manager,
 				signer_provider: node_cfgs[0].keys_manager,
+				broadcaster: node_cfgs[0].tx_broadcaster,
+				fee_estimator: node_cfgs[0].fee_estimator,
 			};
 			match ro_persister.persist_new_channel(test_txo, &added_monitors[0].1) {
 				ChannelMonitorUpdateStatus::UnrecoverableError => {
@@ -1168,13 +1226,17 @@ mod tests {
 			maximum_pending_updates: test_max_pending_updates,
 			entropy_source: &chanmon_cfgs[0].keys_manager,
 			signer_provider: &chanmon_cfgs[0].keys_manager,
+			broadcaster: &chanmon_cfgs[0].tx_broadcaster,
+			fee_estimator: &chanmon_cfgs[0].fee_estimator,
 		};
 		let persister_1 = MonitorUpdatingPersister {
 			kv_store: &TestStore::new(false),
 			logger: &TestLogger::new(),
 			maximum_pending_updates: test_max_pending_updates,
 			entropy_source: &chanmon_cfgs[1].keys_manager,
 			signer_provider: &chanmon_cfgs[1].keys_manager,
+			broadcaster: &chanmon_cfgs[0].tx_broadcaster,
+			fee_estimator: &chanmon_cfgs[0].fee_estimator,
 		};
 		let mut node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
 		let chain_mon_0 = test_utils::TestChainMonitor::new(