Move ChannelManager::monitor_updated to a MonitorEvent

In the next commit we'll need ChainMonitor to "see" when a monitor
persistence completes, which means `monitor_updated` needs to move
to `ChainMonitor`. The simplest way to then communicate that
information to `ChannelManager` is via `MonitorEvet`s, which seems
to line up ok, even if they're now constructed by multiple
different places.

Author: TheBlueMatt

fuzz/src/chanmon_consistency.rs

@@ -855,22 +855,26 @@ pub fn do_test<Out: test_logger::Output>(data: &[u8], out: Out) {
 
 			0x08 => {
 				if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
-					nodes[0].channel_monitor_updated(&chan_1_funding, *id);
+					monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id);
+					nodes[0].process_monitor_events();
 				}
 			},
 			0x09 => {
 				if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
-					nodes[1].channel_monitor_updated(&chan_1_funding, *id);
+					monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id);
+					nodes[1].process_monitor_events();
 				}
 			},
 			0x0a => {
 				if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
-					nodes[1].channel_monitor_updated(&chan_2_funding, *id);
+					monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id);
+					nodes[1].process_monitor_events();
 				}
 			},
 			0x0b => {
 				if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
-					nodes[2].channel_monitor_updated(&chan_2_funding, *id);
+					monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id);
+					nodes[2].process_monitor_events();
 				}
 			},
 
@@ -1077,16 +1081,20 @@ pub fn do_test<Out: test_logger::Output>(data: &[u8], out: Out) {
 				*monitor_c.persister.update_ret.lock().unwrap() = Ok(());
 
 				if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
-					nodes[0].channel_monitor_updated(&chan_1_funding, *id);
+					monitor_a.chain_monitor.channel_monitor_updated(chan_1_funding, *id);
+					nodes[0].process_monitor_events();
 				}
 				if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_1_funding) {
-					nodes[1].channel_monitor_updated(&chan_1_funding, *id);
+					monitor_b.chain_monitor.channel_monitor_updated(chan_1_funding, *id);
+					nodes[1].process_monitor_events();
 				}
 				if let Some((id, _)) = monitor_b.latest_monitors.lock().unwrap().get(&chan_2_funding) {
-					nodes[1].channel_monitor_updated(&chan_2_funding, *id);
+					monitor_b.chain_monitor.channel_monitor_updated(chan_2_funding, *id);
+					nodes[1].process_monitor_events();
 				}
 				if let Some((id, _)) = monitor_c.latest_monitors.lock().unwrap().get(&chan_2_funding) {
-					nodes[2].channel_monitor_updated(&chan_2_funding, *id);
+					monitor_c.chain_monitor.channel_monitor_updated(chan_2_funding, *id);
+					nodes[2].process_monitor_events();
 				}
 
 				// Next, make sure peers are all connected to each other

lightning/src/chain/chainmonitor.rs

@@ -38,7 +38,7 @@ use util::events::EventHandler;
 use ln::channelmanager::ChannelDetails;
 
 use prelude::*;
-use sync::{RwLock, RwLockReadGuard};
+use sync::{RwLock, RwLockReadGuard, Mutex};
 use core::ops::Deref;
 
 /// `Persist` defines behavior for persisting channel monitors: this could mean
@@ -134,6 +134,7 @@ pub struct ChainMonitor<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: De
 	logger: L,
 	fee_estimator: F,
 	persister: P,
+	pending_monitor_events: Mutex<Vec<MonitorEvent>>,
 }
 
 impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
@@ -207,6 +208,7 @@ where C::Target: chain::Filter,
 			logger,
 			fee_estimator: feeest,
 			persister,
+			pending_monitor_events: Mutex::new(Vec::new()),
 		}
 	}
 
@@ -262,6 +264,29 @@ where C::Target: chain::Filter,
 		self.monitors.write().unwrap().remove(funding_txo).unwrap().monitor
 	}
 
+	/// Indicates the persistence of a [`ChannelMonitor`] has completed after
+	/// [`ChannelMonitorUpdateErr::TemporaryFailure`] was returned from an update operation.
+	///
+	/// All ChannelMonitor updates up to and including highest_applied_update_id must have been
+	/// fully committed in every copy of the given channels' ChannelMonitors.
+	///
+	/// Note that there is no effect to calling with a highest_applied_update_id other than the
+	/// current latest ChannelMonitorUpdate and one call to this function after multiple
+	/// ChannelMonitorUpdateErr::TemporaryFailures is fine. The highest_applied_update_id field
+	/// exists largely only to prevent races between this and concurrent update_monitor calls.
+	///
+	/// Thus, the anticipated use is, at a high level:
+	///  1) This [`ChainMonitor`] calls [`Persist::update_persisted_channel`] which stores the
+	///     update to disk and begins updating any remote (e.g. watchtower/backup) copies,
+	///     returning [`ChannelMonitorUpdateErr::TemporaryFailure`],
+	///  2) once all remote copies are updated, you call this function with the update_id that
+	///     completed, and once it is the latest the Channel will be re-enabled.
+	pub fn channel_monitor_updated(&self, funding_txo: OutPoint, highest_applied_update_id: u64) {
+		self.pending_monitor_events.lock().unwrap().push(MonitorEvent::UpdateCompleted {
+			funding_txo, monitor_update_id: highest_applied_update_id
+		});
+	}
+
 	#[cfg(any(test, feature = "fuzztarget", feature = "_test_utils"))]
 	pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
 		use util::events::EventsProvider;
@@ -431,7 +456,7 @@ where C::Target: chain::Filter,
 	}
 
 	fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
-		let mut pending_monitor_events = Vec::new();
+		let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
 		for monitor_state in self.monitors.read().unwrap().values() {
 			pending_monitor_events.append(&mut monitor_state.monitor.get_and_clear_pending_monitor_events());
 		}

lightning/src/chain/channelmonitor.rs

@@ -131,8 +131,29 @@ pub enum MonitorEvent {
 
 	/// A monitor event that the Channel's commitment transaction was confirmed.
 	CommitmentTxConfirmed(OutPoint),
+
+	/// Indicates a [`ChannelMonitor`] update has completed. See
+	/// [`ChannelMonitorUpdateErr::TemporaryFailure`] for more information on how this is used.
+	///
+	/// [`ChannelMonitorUpdateErr::TemporaryFailure`]: super::ChannelMonitorUpdateErr::TemporaryFailure
+	UpdateCompleted {
+		/// The funding outpoint of the [`ChannelMonitor`] that was updated
+		funding_txo: OutPoint,
+		/// The Update ID from [`ChannelMonitorUpdate::update_id`] which was applied or
+		/// [`ChannelMonitor::get_latest_update_id`].
+		///
+		/// Note that this should only be set to a given update's ID if all previous updates for the
+		/// same [`ChannelMonitor`] have been applied and persisted.
+		monitor_update_id: u64,
+	},
 }
-impl_writeable_tlv_based_enum_upgradable!(MonitorEvent, ;
+impl_writeable_tlv_based_enum_upgradable!(MonitorEvent,
+	// Note that UpdateCompleted is currently never serialized to disk as it is generated only in ChainMonitor
+	(0, UpdateCompleted) => {
+		(0, funding_txo, required),
+		(2, monitor_update_id, required),
+	},
+;
 	(2, HTLCEvent),
 	(4, CommitmentTxConfirmed),
 );
@@ -854,14 +875,19 @@ impl<Signer: Sign> Writeable for ChannelMonitorImpl<Signer> {
 			writer.write_all(&payment_preimage.0[..])?;
 		}
 
-		writer.write_all(&byte_utils::be64_to_array(self.pending_monitor_events.len() as u64))?;
+		writer.write_all(&(self.pending_monitor_events.iter().filter(|ev| match ev {
+			MonitorEvent::HTLCEvent(_) => true,
+			MonitorEvent::CommitmentTxConfirmed(_) => true,
+			_ => false,
+		}).count() as u64).to_be_bytes())?;
 		for event in self.pending_monitor_events.iter() {
 			match event {
 				MonitorEvent::HTLCEvent(upd) => {
 					0u8.write(writer)?;
 					upd.write(writer)?;
 				},
-				MonitorEvent::CommitmentTxConfirmed(_) => 1u8.write(writer)?
+				MonitorEvent::CommitmentTxConfirmed(_) => 1u8.write(writer)?,
+				_ => {}, // Covered in the TLV writes below
 			}
 		}
 

lightning/src/chain/mod.rs

@@ -182,9 +182,10 @@ pub enum ChannelMonitorUpdateErr {
 	/// our state failed, but is expected to succeed at some point in the future).
 	///
 	/// Such a failure will "freeze" a channel, preventing us from revoking old states or
-	/// submitting new commitment transactions to the counterparty. Once the update(s) which failed
-	/// have been successfully applied, ChannelManager::channel_monitor_updated can be used to
-	/// restore the channel to an operational state.
+	/// submitting new commitment transactions to the counterparty. Once the update(s) that failed
+	/// have been successfully applied, a [`MonitorEvent::UpdateCompleted`] event should be returned
+	/// via [`Watch::release_pending_monitor_events`] which will then restore the channel to an
+	/// operational state.
 	///
 	/// Note that a given ChannelManager will *never* re-generate a given ChannelMonitorUpdate. If
 	/// you return a TemporaryFailure you must ensure that it is written to disk safely before
@@ -198,13 +199,14 @@ pub enum ChannelMonitorUpdateErr {
 	/// the channel which would invalidate previous ChannelMonitors are not made when a channel has
 	/// been "frozen".
 	///
-	/// Note that even if updates made after TemporaryFailure succeed you must still call
-	/// channel_monitor_updated to ensure you have the latest monitor and re-enable normal channel
-	/// operation.
+	/// Note that even if updates made after TemporaryFailure succeed you must still provide a
+	/// [`MonitorEvent::UpdateCompleted`] to ensure you have the latest monitor and re-enable
+	/// normal channel operation. Note that this is normally generated through a call to
+	/// [`ChainMonitor::channel_monitor_updated`].
 	///
-	/// Note that the update being processed here will not be replayed for you when you call
-	/// ChannelManager::channel_monitor_updated, so you must store the update itself along
-	/// with the persisted ChannelMonitor on your own local disk prior to returning a
+	/// Note that the update being processed here will not be replayed for you when you return a
+	/// [`MonitorEvent::UpdateCompleted`] event via [`Watch::release_pending_monitor_events`], so
+	/// you must store the update itself on your own local disk prior to returning a
 	/// TemporaryFailure. You may, of course, employ a journaling approach, storing only the
 	/// ChannelMonitorUpdate on disk without updating the monitor itself, replaying the journal at
 	/// reload-time.
@@ -280,6 +282,9 @@ pub trait Watch<ChannelSigner: Sign> {
 
 	/// Returns any monitor events since the last call. Subsequent calls must only return new
 	/// events.
+	///
+	/// For details on asynchronous [`ChannelMonitor`] updating and returning
+	/// [`MonitorEvent::UpdateCompleted`] here, see [`ChannelMonitorUpdateErr::TemporaryFailure`].
 	fn release_pending_monitor_events(&self) -> Vec<MonitorEvent>;
 }