Persist ChannelMonitors after new blocks are connected

This resolves several user complaints (and issues in the sample
node) where startup is substantially delayed as we're always
waiting for the chain data to sync.

Further, in an upcoming PR, we'll be reloading pending payments
from ChannelMonitors on restart, at which point we'll need the
change here which avoids handling events until after the user
has confirmed the `ChannelMonitor` has been persisted to disk.
It will avoid a race where we
 * send a payment/HTLC (persisting the monitor to disk with the
   HTLC pending),
 * force-close the channel, removing the channel entry from the
   ChannelManager entirely,
 * persist the ChannelManager,
 * connect a block which contains a fulfill of the HTLC, generating
   a claim event,
 * handle the claim event while the `ChannelMonitor` is being
   persisted,
 * persist the ChannelManager (before the CHannelMonitor is
   persisted fully),
 * restart, reloading the HTLC as a pending payment in the
   ChannelManager, which now has no references to it except from
   the ChannelMonitor which still has the pending HTLC,
 * replay the block connection, generating a duplicate PaymentSent
   event.

Author: TheBlueMatt

fuzz/src/utils/test_persister.rs

@@ -14,7 +14,7 @@ impl chainmonitor::Persist<EnforcingSigner> for TestPersister {
 		self.update_ret.lock().unwrap().clone()
 	}
 
-	fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>, _update_id: MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
+	fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &Option<channelmonitor::ChannelMonitorUpdate>, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>, _update_id: MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
 		self.update_ret.lock().unwrap().clone()
 	}
 }

lightning-persister/src/lib.rs

@@ -159,13 +159,18 @@ impl FilesystemPersister {
 }
 
 impl<ChannelSigner: Sign> chainmonitor::Persist<ChannelSigner> for FilesystemPersister {
+	// TODO: We really need a way for the persister to inform the user that its time to crash/shut
+	// down once these start returning failure.
+	// A PermanentFailure implies we need to shut down since we're force-closing channels without
+	// even broadcasting!
+
 	fn persist_new_channel(&self, funding_txo: OutPoint, monitor: &ChannelMonitor<ChannelSigner>, _update_id: chainmonitor::MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
 		let filename = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
 		util::write_to_file(self.path_to_monitor_data(), filename, monitor)
 			.map_err(|_| chain::ChannelMonitorUpdateErr::PermanentFailure)
 	}
 
-	fn update_persisted_channel(&self, funding_txo: OutPoint, _update: &ChannelMonitorUpdate, monitor: &ChannelMonitor<ChannelSigner>, _update_id: chainmonitor::MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
+	fn update_persisted_channel(&self, funding_txo: OutPoint, _update: &Option<ChannelMonitorUpdate>, monitor: &ChannelMonitor<ChannelSigner>, _update_id: chainmonitor::MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
 		let filename = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
 		util::write_to_file(self.path_to_monitor_data(), filename, monitor)
 			.map_err(|_| chain::ChannelMonitorUpdateErr::PermanentFailure)

lightning/src/chain/chainmonitor.rs

@@ -32,6 +32,7 @@ use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, MonitorUpdated, TransactionOutputs};
 use chain::transaction::{OutPoint, TransactionData};
 use chain::keysinterface::Sign;
+use util::atomic_counter::AtomicCounter;
 use util::logger::Logger;
 use util::events;
 use util::events::EventHandler;
@@ -40,10 +41,12 @@ use ln::channelmanager::ChannelDetails;
 use prelude::*;
 use sync::{RwLock, RwLockReadGuard, Mutex};
 use core::ops::Deref;
+use core::sync::atomic::{AtomicBool, Ordering};
 
 #[derive(Clone, Copy, Hash, PartialEq, Eq)]
 pub(crate) enum MonitorUpdate {
 	MonitorUpdateId(u64),
+	SyncPersistId(u64),
 }
 
 /// An opaque identifier describing a specific [`Persist`] method call.
@@ -90,6 +93,9 @@ pub trait Persist<ChannelSigner: Sign> {
 	/// updated monitor itself to disk/backups. See the `Persist` trait documentation for more
 	/// details.
 	///
+	/// During blockchain synchronization operations, this may be called with no
+	/// [`ChannelMonitorUpdate`], in which case the full [`ChannelMonitor`] needs to be persisted.
+	///
 	/// If an implementer chooses to persist the updates only, they need to make
 	/// sure that all the updates are applied to the `ChannelMonitors` *before*
 	/// the set of channel monitors is given to the `ChannelManager`
@@ -107,7 +113,7 @@ pub trait Persist<ChannelSigner: Sign> {
 	/// [`ChannelMonitorUpdateErr`] for requirements when returning errors.
 	///
 	/// [`Writeable::write`]: crate::util::ser::Writeable::write
-	fn update_persisted_channel(&self, id: OutPoint, update: &ChannelMonitorUpdate, data: &ChannelMonitor<ChannelSigner>, update_id: MonitorUpdateId) -> Result<(), ChannelMonitorUpdateErr>;
+	fn update_persisted_channel(&self, id: OutPoint, update: &Option<ChannelMonitorUpdate>, data: &ChannelMonitor<ChannelSigner>, update_id: MonitorUpdateId) -> Result<(), ChannelMonitorUpdateErr>;
 }
 
 struct MonitorHolder<ChannelSigner: Sign> {
@@ -118,7 +124,24 @@ struct MonitorHolder<ChannelSigner: Sign> {
 	/// update_persisted_channel, the user returns a TemporaryFailure, and then calls
 	/// channel_monitor_updated immediately, racing our insertion of the pending update into the
 	/// contained Vec.
+	///
+	/// Beyond the synchronization of updates themselves, we cannot handle user events until after
+	/// any chain updates have been stored on disk. Thus, we scan this list when returning updates
+	/// to the ChannelManager, refusing to return any updates for a ChannelMonitor which is still
+	/// being persisted fully ro disk after a chain update.
+	///
+	/// This avoids the possibility of handling, e.g. an on-chain claim, generating a claim monitor
+	/// event, resulting in the relevant ChannelManager generating a PaymentSent event and dropping
+	/// the pending payment entry, and then reloading before the monitor is persisted, resulting in
+	/// the ChannelManager re-adding the same payment entry, before the same block is replayed,
+	/// resulting in a duplicate PaymentSent event.
 	pending_monitor_updates: Mutex<Vec<MonitorUpdateId>>,
+	/// When the user returns a PermanentFailure error from an update_persisted_channel call during
+	/// block processing, we inform the ChannelManager that the channel should be closed
+	/// asynchronously. In order to ensure no further changes happen before the ChannelManager has
+	/// processed the closure event, we set this to true and return PermanentFailure for any other
+	/// chain::Watch events.
+	channel_closed: AtomicBool,
 }
 
 /// A read-only reference to a current ChannelMonitor.
@@ -154,6 +177,7 @@ pub struct ChainMonitor<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: De
         P::Target: Persist<ChannelSigner>,
 {
 	monitors: RwLock<HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
+	sync_persistence_id: AtomicCounter,
 	chain_source: Option<C>,
 	broadcaster: T,
 	logger: L,
@@ -183,26 +207,50 @@ where C::Target: chain::Filter,
 		FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>
 	{
 		let mut dependent_txdata = Vec::new();
-		let monitors = self.monitors.read().unwrap();
-		for monitor_state in monitors.values() {
-			let mut txn_outputs = process(&monitor_state.monitor, txdata);
+		{
+			let monitors = self.monitors.write().unwrap();
+			for (funding_outpoint, monitor_state) in monitors.iter() {
+				let monitor = &monitor_state.monitor;
+				let mut txn_outputs;
+				{
+					txn_outputs = process(monitor, txdata);
+					let update_id = MonitorUpdateId {
+						contents: MonitorUpdate::SyncPersistId(self.sync_persistence_id.get_increment()),
+					};
+					let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
+
+					log_trace!(self.logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor));
+					match self.persister.update_persisted_channel(*funding_outpoint, &None, monitor, update_id) {
+						Ok(()) =>
+							log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
+						Err(ChannelMonitorUpdateErr::PermanentFailure) => {
+							monitor_state.channel_closed.store(true, Ordering::Release);
+							self.user_provided_events.lock().unwrap().push(MonitorEvent::UpdateFailed(*funding_outpoint));
+						},
+						Err(ChannelMonitorUpdateErr::TemporaryFailure) => {
+							log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
+							pending_monitor_updates.push(update_id);
+						},
+					}
+				}
 
-			// 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.
-			if let Some(ref chain_source) = self.chain_source {
-				let block_hash = header.block_hash();
-				for (txid, mut outputs) in txn_outputs.drain(..) {
-					for (idx, output) in outputs.drain(..) {
-						// Register any new outputs with the chain source for filtering and recurse
-						// if it indicates that there are dependent transactions within the block
-						// that had not been previously included in txdata.
-						let output = WatchedOutput {
-							block_hash: Some(block_hash),
-							outpoint: OutPoint { txid, index: idx as u16 },
-							script_pubkey: output.script_pubkey,
-						};
-						if let Some(tx) = chain_source.register_output(output) {
-							dependent_txdata.push(tx);
+				// 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.
+				if let Some(ref chain_source) = self.chain_source {
+					let block_hash = header.block_hash();
+					for (txid, mut outputs) in txn_outputs.drain(..) {
+						for (idx, output) in outputs.drain(..) {
+							// Register any new outputs with the chain source for filtering and recurse
+							// if it indicates that there are dependent transactions within the block
+							// that had not been previously included in txdata.
+							let output = WatchedOutput {
+								block_hash: Some(block_hash),
+								outpoint: OutPoint { txid, index: idx as u16 },
+								script_pubkey: output.script_pubkey,
+							};
+							if let Some(tx) = chain_source.register_output(output) {
+								dependent_txdata.push(tx);
+							}
 						}
 					}
 				}
@@ -228,6 +276,7 @@ where C::Target: chain::Filter,
 	pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P) -> Self {
 		Self {
 			monitors: RwLock::new(HashMap::new()),
+			sync_persistence_id: AtomicCounter::new(),
 			chain_source,
 			broadcaster,
 			logger,
@@ -300,7 +349,7 @@ where C::Target: chain::Filter,
 		pending_monitor_updates.retain(|update_id| *update_id != completed_update_id);
 
 		match completed_update_id {
-			MonitorUpdateId { .. } => {
+			MonitorUpdateId { contents: MonitorUpdate::MonitorUpdateId(_) } => {
 				let monitor_update_pending_updates =  pending_monitor_updates.iter().filter(|update_id|
 					if let MonitorUpdate::MonitorUpdateId(_) = update_id.contents { true } else { false }).count();
 				if monitor_update_pending_updates != 0 {
@@ -312,7 +361,12 @@ where C::Target: chain::Filter,
 					funding_txo,
 					monitor_update_id: monitor_data.monitor.get_latest_update_id(),
 				}));
-			}
+			},
+			MonitorUpdateId { contents: MonitorUpdate::SyncPersistId(_) } => {
+				// We've already done everything we need to, the next time release_monitor_events
+				// is called, any events for this ChannelMonitor will be returned if there's no
+				// more SyncPersistId events left.
+			},
 		}
 	}
 
@@ -458,7 +512,11 @@ where C::Target: chain::Filter,
 				monitor.load_outputs_to_watch(chain_source);
 			}
 		}
-		entry.insert(MonitorHolder { monitor, pending_monitor_updates: Mutex::new(pending_monitor_updates) });
+		entry.insert(MonitorHolder {
+			monitor,
+			pending_monitor_updates: Mutex::new(pending_monitor_updates),
+			channel_closed: AtomicBool::new(false),
+		});
 		update_res
 	}
 
@@ -492,7 +550,7 @@ where C::Target: chain::Filter,
 					contents: MonitorUpdate::MonitorUpdateId(update.update_id),
 				};
 				let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
-				let persist_res = self.persister.update_persisted_channel(funding_txo, &update, monitor, update_id);
+				let persist_res = self.persister.update_persisted_channel(funding_txo, &Some(update), monitor, update_id);
 				if let Err(e) = persist_res {
 					if e == ChannelMonitorUpdateErr::TemporaryFailure {
 						pending_monitor_updates.push(update_id);
@@ -501,6 +559,8 @@ where C::Target: chain::Filter,
 				}
 				if update_res.is_err() {
 					Err(ChannelMonitorUpdateErr::PermanentFailure)
+				} else if monitor_state.channel_closed.load(Ordering::Acquire) {
+					Err(ChannelMonitorUpdateErr::PermanentFailure)
 				} else {
 					persist_res
 				}
@@ -511,7 +571,17 @@ where C::Target: chain::Filter,
 	fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
 		let mut pending_monitor_events = self.user_provided_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());
+			let pending_monitor_update_count = monitor_state.pending_monitor_updates.lock().unwrap()
+				.iter().filter(|update_id|
+					if let MonitorUpdate::SyncPersistId(_) = update_id.contents { true } else { false })
+				.count();
+			if pending_monitor_update_count > 0 {
+				log_info!(self.logger, "A Channel Monitor sync is still in progress, refusing to provide monitor events!");
+			} else if monitor_state.channel_closed.load(Ordering::Acquire) {
+				log_info!(self.logger, "A Channel Monitor sync failed, refusing to provide monitor events!");
+			} else {
+				pending_monitor_events.append(&mut monitor_state.monitor.get_and_clear_pending_monitor_events());
+			}
 		}
 		pending_monitor_events
 	}

lightning/src/chain/channelmonitor.rs

@@ -157,12 +157,20 @@ pub enum MonitorEvent {
 	///
 	/// [`ChannelMonitorUpdateErr::TemporaryFailure`]: super::ChannelMonitorUpdateErr::TemporaryFailure
 	UpdateCompleted(MonitorUpdated),
+
+	/// Indicates a [`ChannelMonitor`] update has failed. See
+	/// [`ChannelMonitorUpdateErr::PermanentFailure`] for more information on how this is used.
+	///
+	/// [`ChannelMonitorUpdateErr::PermanentFailure`]: super::ChannelMonitorUpdateErr::PermanentFailure
+	UpdateFailed(OutPoint),
 }
 impl_writeable_tlv_based_enum_upgradable!(MonitorEvent, ;
 	(0, HTLCEvent),
-	// Note that UpdateCompleted is currently never serialized to disk as it is generated only in ChainMonitor
+	// Note that UpdateCompleted and UpdateFailed is currently never serialized to disk as they are
+	// generated only in ChainMonitor
 	(1, UpdateCompleted),
 	(2, CommitmentTxConfirmed),
+	(3, UpdateFailed),
 );
 
 /// Simple structure sent back by `chain::Watch` when an HTLC from a forward channel is detected on
@@ -656,7 +664,17 @@ pub(crate) struct ChannelMonitorImpl<Signer: Sign> {
 
 	payment_preimages: HashMap<PaymentHash, PaymentPreimage>,
 
+	// Note that MonitorEvents MUST NOT be generated during update processing, only generated
+	// during chain data processing. This prevents a race in ChainMonitor::update_channel (and
+	// presumably user implementations thereof as well) where we update the in-memory channel
+	// object, then before the persistence finishes (as its all under a read-lock), we return
+	// pending events to the user or to the relevant ChannelManager. Then, on reload, we'll have
+	// the pre-event state here, but have processed the event in the ChannelManager.
+	// Note that because the `event_lock` in `ChainMonitor` is only taken in
+	// block/transaction-connected events and *not* during block/transaction-disconnected events,
+	// we further MUST NOT generate events during block/transaction-disconnection.
 	pending_monitor_events: Vec<MonitorEvent>,
+
 	pending_events: Vec<Event>,
 
 	// Used to track on-chain events (i.e., transactions part of channels confirmed on chain) on

lightning/src/chain/mod.rs

@@ -281,6 +281,10 @@ pub trait Watch<ChannelSigner: Sign> {
 
 	/// Returns any monitor events since the last call. Subsequent calls must only return new
 	/// events.
+	///
+	/// Note that after any block- or transaction-connection calls to a [`ChannelMonitor`], no
+	/// further events may be returned here until the [`ChannelMonitor`] has been fully persisted
+	/// to disk.
 	fn release_pending_monitor_events(&self) -> Vec<MonitorEvent>;
 }
 

lightning/src/ln/channelmanager.rs

@@ -4087,7 +4087,8 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 						self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), htlc_update.source, &htlc_update.payment_hash, HTLCFailReason::Reason { failure_code: 0x4000 | 8, data: Vec::new() });
 					}
 				},
-				MonitorEvent::CommitmentTxConfirmed(funding_outpoint) => {
+				MonitorEvent::CommitmentTxConfirmed(funding_outpoint) |
+				MonitorEvent::UpdateFailed(funding_outpoint) => {
 					let mut channel_lock = self.channel_state.lock().unwrap();
 					let channel_state = &mut *channel_lock;
 					let by_id = &mut channel_state.by_id;
@@ -4103,7 +4104,12 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 								msg: update
 							});
 						}
-						self.issue_channel_close_events(&chan, ClosureReason::CommitmentTxConfirmed);
+						let reason = if let MonitorEvent::UpdateFailed(_) = monitor_event {
+							ClosureReason::ProcessingError { err: "Failed to persist ChannelMonitor update during chain sync".to_string() }
+						} else {
+							ClosureReason::CommitmentTxConfirmed
+						};
+						self.issue_channel_close_events(&chan, reason);
 						pending_msg_events.push(events::MessageSendEvent::HandleError {
 							node_id: chan.get_counterparty_node_id(),
 							action: msgs::ErrorAction::SendErrorMessage {
@@ -5437,20 +5443,21 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable f
 ///
 /// At a high-level, the process for deserializing a ChannelManager and resuming normal operation
 /// is:
-/// 1) Deserialize all stored ChannelMonitors.
-/// 2) Deserialize the ChannelManager by filling in this struct and calling:
-///    <(BlockHash, ChannelManager)>::read(reader, args)
-///    This may result in closing some Channels if the ChannelMonitor is newer than the stored
-///    ChannelManager state to ensure no loss of funds. Thus, transactions may be broadcasted.
-/// 3) If you are not fetching full blocks, register all relevant ChannelMonitor outpoints the same
-///    way you would handle a `chain::Filter` call using ChannelMonitor::get_outputs_to_watch() and
-///    ChannelMonitor::get_funding_txo().
-/// 4) Reconnect blocks on your ChannelMonitors.
-/// 5) Disconnect/connect blocks on the ChannelManager.
-/// 6) Move the ChannelMonitors into your local chain::Watch.
+/// 1) Deserialize all stored [`ChannelMonitor`]s.
+/// 2) Deserialize the [`ChannelManager`] by filling in this struct and calling:
+///    `<(BlockHash, ChannelManager)>::read(reader, args)`
+///    This may result in closing some channels if the [`ChannelMonitor`] is newer than the stored
+///    [`ChannelManager`] state to ensure no loss of funds. Thus, transactions may be broadcasted.
+/// 3) If you are not fetching full blocks, register all relevant [`ChannelMonitor`] outpoints the
+///    same way you would handle a [`chain::Filter`] call using
+///    [`ChannelMonitor::get_outputs_to_watch`] and [`ChannelMonitor::get_funding_txo`].
+/// 4) Reconnect blocks on your [`ChannelMonitor`]s.
+/// 5) Re-persist the [`ChannelMonitor`]s to ensure the latest state is on disk.
+/// 6) Disconnect/connect blocks on the [`ChannelManager`].
+/// 7) Move the [`ChannelMonitor`]s into your local [`chain::Watch`].
 ///
-/// Note that the ordering of #4-6 is not of importance, however all three must occur before you
-/// call any other methods on the newly-deserialized ChannelManager.
+/// Note that the ordering of #4-7 is not of importance, however all four must occur before you
+/// call any other methods on the newly-deserialized [`ChannelManager`].
 ///
 /// Note that because some channels may be closed during deserialization, it is critical that you
 /// always deserialize only the latest version of a ChannelManager and ChannelMonitors available to

lightning/src/util/test_utils.rs

@@ -188,7 +188,7 @@ impl<Signer: keysinterface::Sign> chainmonitor::Persist<Signer> for TestPersiste
 		ret
 	}
 
-	fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<Signer>, _id: MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
+	fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &Option<channelmonitor::ChannelMonitorUpdate>, _data: &channelmonitor::ChannelMonitor<Signer>, _id: MonitorUpdateId) -> Result<(), chain::ChannelMonitorUpdateErr> {
 		let ret = self.update_ret.lock().unwrap().clone();
 		if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
 			*self.update_ret.lock().unwrap() = next_ret;