From: Matt Corallo <649246+TheBlueMatt@users.noreply.github.com>
Date: Sat, 15 May 2021 00:44:40 +0000 (+0000)
Subject: Merge pull request #916 from TheBlueMatt/2021-05-fix-disabled-announcements
X-Git-Tag: v0.0.98~25
X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=commitdiff_plain;h=e0986de47796848efa1619624dde0fe6e9908d81;hp=-c;p=rust-lightning

Merge pull request #916 from TheBlueMatt/2021-05-fix-disabled-announcements

Avoid persisting a ChannelManager after each timer tick and send update_channel re-enable messages
---

e0986de47796848efa1619624dde0fe6e9908d81
diff --combined lightning/src/ln/channelmanager.rs
index a0168bdce,922912932..54bd3c89f
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@@ -45,7 -45,7 +45,7 @@@ use chain::transaction::{OutPoint, Tran
  // construct one themselves.
  use ln::{PaymentHash, PaymentPreimage, PaymentSecret};
  pub use ln::channel::CounterpartyForwardingInfo;
- use ln::channel::{Channel, ChannelError};
+ use ln::channel::{Channel, ChannelError, ChannelUpdateStatus};
  use ln::features::{InitFeatures, NodeFeatures};
  use routing::router::{Route, RouteHop};
  use ln::msgs;
@@@ -468,8 -468,8 +468,8 @@@ pub struct ChannelManager<Signer: Sign
  	/// Essentially just when we're serializing ourselves out.
  	/// Taken first everywhere where we are making changes before any other locks.
  	/// When acquiring this lock in read mode, rather than acquiring it directly, call
- 	/// `PersistenceNotifierGuard::new(..)` and pass the lock to it, to ensure the PersistenceNotifier
- 	/// the lock contains sends out a notification when the lock is released.
+ 	/// `PersistenceNotifierGuard::notify_on_drop(..)` and pass the lock to it, to ensure the
+ 	/// PersistenceNotifier the lock contains sends out a notification when the lock is released.
  	total_consistency_lock: RwLock<()>,
  
  	persistence_notifier: PersistenceNotifier,
@@@ -522,32 -522,50 +522,50 @@@ impl BestBlock 
  	pub fn height(&self) -> u32 { self.height }
  }
  
+ #[derive(Copy, Clone, PartialEq)]
+ enum NotifyOption {
+ 	DoPersist,
+ 	SkipPersist,
+ }
+ 
  /// Whenever we release the `ChannelManager`'s `total_consistency_lock`, from read mode, it is
  /// desirable to notify any listeners on `await_persistable_update_timeout`/
- /// `await_persistable_update` that new updates are available for persistence. Therefore, this
+ /// `await_persistable_update` when new updates are available for persistence. Therefore, this
  /// struct is responsible for locking the total consistency lock and, upon going out of scope,
  /// sending the aforementioned notification (since the lock being released indicates that the
  /// updates are ready for persistence).
- struct PersistenceNotifierGuard<'a> {
+ ///
+ /// We allow callers to either always notify by constructing with `notify_on_drop` or choose to
+ /// notify or not based on whether relevant changes have been made, providing a closure to
+ /// `optionally_notify` which returns a `NotifyOption`.
+ struct PersistenceNotifierGuard<'a, F: Fn() -> NotifyOption> {
  	persistence_notifier: &'a PersistenceNotifier,
+ 	should_persist: F,
  	// We hold onto this result so the lock doesn't get released immediately.
  	_read_guard: RwLockReadGuard<'a, ()>,
  }
  
- impl<'a> PersistenceNotifierGuard<'a> {
- 	fn new(lock: &'a RwLock<()>, notifier: &'a PersistenceNotifier) -> Self {
+ impl<'a> PersistenceNotifierGuard<'a, fn() -> NotifyOption> { // We don't care what the concrete F is here, it's unused
+ 	fn notify_on_drop(lock: &'a RwLock<()>, notifier: &'a PersistenceNotifier) -> PersistenceNotifierGuard<'a, impl Fn() -> NotifyOption> {
+ 		PersistenceNotifierGuard::optionally_notify(lock, notifier, || -> NotifyOption { NotifyOption::DoPersist })
+ 	}
+ 
+ 	fn optionally_notify<F: Fn() -> NotifyOption>(lock: &'a RwLock<()>, notifier: &'a PersistenceNotifier, persist_check: F) -> PersistenceNotifierGuard<'a, F> {
  		let read_guard = lock.read().unwrap();
  
- 		Self {
+ 		PersistenceNotifierGuard {
  			persistence_notifier: notifier,
+ 			should_persist: persist_check,
  			_read_guard: read_guard,
  		}
  	}
  }
  
- impl<'a> Drop for PersistenceNotifierGuard<'a> {
+ impl<'a, F: Fn() -> NotifyOption> Drop for PersistenceNotifierGuard<'a, F> {
  	fn drop(&mut self) {
- 		self.persistence_notifier.notify();
+ 		if (self.should_persist)() == NotifyOption::DoPersist {
+ 			self.persistence_notifier.notify();
+ 		}
  	}
  }
  
@@@ -943,7 -961,7 +961,7 @@@ impl<Signer: Sign, M: Deref, T: Deref, 
  		let channel = Channel::new_outbound(&self.fee_estimator, &self.keys_manager, their_network_key, channel_value_satoshis, push_msat, user_id, config)?;
  		let res = channel.get_open_channel(self.genesis_hash.clone());
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		// We want to make sure the lock is actually acquired by PersistenceNotifierGuard.
  		debug_assert!(&self.total_consistency_lock.try_write().is_err());
  
@@@ -1024,7 -1042,7 +1042,7 @@@
  	///
  	/// May generate a SendShutdown message event on success, which should be relayed.
  	pub fn close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		let (mut failed_htlcs, chan_option) = {
  			let mut channel_state_lock = self.channel_state.lock().unwrap();
@@@ -1114,7 -1132,7 +1132,7 @@@
  	/// Force closes a channel, immediately broadcasting the latest local commitment transaction to
  	/// the chain and rejecting new HTLCs on the given channel. Fails if channel_id is unknown to the manager.
  	pub fn force_close_channel(&self, channel_id: &[u8; 32]) -> Result<(), APIError> {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		match self.force_close_channel_with_peer(channel_id, None) {
  			Ok(counterparty_node_id) => {
  				self.channel_state.lock().unwrap().pending_msg_events.push(
@@@ -1459,7 -1477,7 +1477,7 @@@
  		}
  		let onion_packet = onion_utils::construct_onion_packet(onion_payloads, onion_keys, prng_seed, payment_hash);
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		let err: Result<(), _> = loop {
  			let mut channel_lock = self.channel_state.lock().unwrap();
@@@ -1686,7 -1704,7 +1704,7 @@@
  	/// not currently support replacing a funding transaction on an existing channel. Instead,
  	/// create a new channel with a conflicting funding transaction.
  	pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_transaction: Transaction) -> Result<(), APIError> {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		for inp in funding_transaction.input.iter() {
  			if inp.witness.is_empty() {
@@@ -1769,7 -1787,7 +1787,7 @@@
  	///
  	/// Panics if addresses is absurdly large (more than 500).
  	pub fn broadcast_node_announcement(&self, rgb: [u8; 3], alias: [u8; 32], mut addresses: Vec<NetAddress>) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		if addresses.len() > 500 {
  			panic!("More than half the message size was taken up by public addresses!");
@@@ -1803,7 -1821,7 +1821,7 @@@
  	/// Should only really ever be called in response to a PendingHTLCsForwardable event.
  	/// Will likely generate further events.
  	pub fn process_pending_htlc_forwards(&self) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		let mut new_events = Vec::new();
  		let mut failed_forwards = Vec::new();
@@@ -2094,9 -2112,13 +2112,13 @@@
  	/// BroadcastChannelUpdate events in timer_tick_occurred.
  	///
  	/// Expects the caller to have a total_consistency_lock read lock.
- 	fn process_background_events(&self) {
+ 	fn process_background_events(&self) -> bool {
  		let mut background_events = Vec::new();
  		mem::swap(&mut *self.pending_background_events.lock().unwrap(), &mut background_events);
+ 		if background_events.is_empty() {
+ 			return false;
+ 		}
+ 
  		for event in background_events.drain(..) {
  			match event {
  				BackgroundEvent::ClosingMonitorUpdate((funding_txo, update)) => {
@@@ -2106,6 -2128,7 +2128,7 @@@
  				},
  			}
  		}
+ 		true
  	}
  
  	#[cfg(any(test, feature = "_test_utils"))]
@@@ -2121,25 -2144,42 +2144,42 @@@
  	///
  	/// Note that in some rare cases this may generate a `chain::Watch::update_channel` call.
  	pub fn timer_tick_occurred(&self) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
- 		self.process_background_events();
+ 		PersistenceNotifierGuard::optionally_notify(&self.total_consistency_lock, &self.persistence_notifier, || {
+ 			let mut should_persist = NotifyOption::SkipPersist;
+ 			if self.process_background_events() { should_persist = NotifyOption::DoPersist; }
  
- 		let mut channel_state_lock = self.channel_state.lock().unwrap();
- 		let channel_state = &mut *channel_state_lock;
- 		for (_, chan) in channel_state.by_id.iter_mut() {
- 			if chan.is_disabled_staged() && !chan.is_live() {
- 				if let Ok(update) = self.get_channel_update(&chan) {
- 					channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
- 						msg: update
- 					});
+ 			let mut channel_state_lock = self.channel_state.lock().unwrap();
+ 			let channel_state = &mut *channel_state_lock;
+ 			for (_, chan) in channel_state.by_id.iter_mut() {
+ 				match chan.channel_update_status() {
+ 					ChannelUpdateStatus::Enabled if !chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::DisabledStaged),
+ 					ChannelUpdateStatus::Disabled if chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::EnabledStaged),
+ 					ChannelUpdateStatus::DisabledStaged if chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::Enabled),
+ 					ChannelUpdateStatus::EnabledStaged if !chan.is_live() => chan.set_channel_update_status(ChannelUpdateStatus::Disabled),
+ 					ChannelUpdateStatus::DisabledStaged if !chan.is_live() => {
+ 						if let Ok(update) = self.get_channel_update(&chan) {
+ 							channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+ 								msg: update
+ 							});
+ 						}
+ 						should_persist = NotifyOption::DoPersist;
+ 						chan.set_channel_update_status(ChannelUpdateStatus::Disabled);
+ 					},
+ 					ChannelUpdateStatus::EnabledStaged if chan.is_live() => {
+ 						if let Ok(update) = self.get_channel_update(&chan) {
+ 							channel_state.pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+ 								msg: update
+ 							});
+ 						}
+ 						should_persist = NotifyOption::DoPersist;
+ 						chan.set_channel_update_status(ChannelUpdateStatus::Enabled);
+ 					},
+ 					_ => {},
  				}
- 				chan.to_fresh();
- 			} else if chan.is_disabled_staged() && chan.is_live() {
- 				chan.to_fresh();
- 			} else if chan.is_disabled_marked() {
- 				chan.to_disabled_staged();
  			}
- 		}
+ 
+ 			should_persist
+ 		});
  	}
  
  	/// Indicates that the preimage for payment_hash is unknown or the received amount is incorrect
@@@ -2148,7 -2188,7 +2188,7 @@@
  	/// Returns false if no payment was found to fail backwards, true if the process of failing the
  	/// HTLC backwards has been started.
  	pub fn fail_htlc_backwards(&self, payment_hash: &PaymentHash) -> bool {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		let mut channel_state = Some(self.channel_state.lock().unwrap());
  		let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(payment_hash);
@@@ -2328,7 -2368,7 +2368,7 @@@
  	pub fn claim_funds(&self, payment_preimage: PaymentPreimage) -> bool {
  		let payment_hash = PaymentHash(Sha256::hash(&payment_preimage.0).into_inner());
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		let mut channel_state = Some(self.channel_state.lock().unwrap());
  		let removed_source = channel_state.as_mut().unwrap().claimable_htlcs.remove(&payment_hash);
@@@ -2512,7 -2552,7 +2552,7 @@@
  	///  4) 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) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		let mut close_results = Vec::new();
  		let mut htlc_forwards = Vec::new();
@@@ -3283,7 -3323,7 +3323,7 @@@
  	/// (C-not exported) Cause its doc(hidden) anyway
  	#[doc(hidden)]
  	pub fn update_fee(&self, channel_id: [u8;32], feerate_per_kw: u32) -> Result<(), APIError> {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let counterparty_node_id;
  		let err: Result<(), _> = loop {
  			let mut channel_state_lock = self.channel_state.lock().unwrap();
@@@ -3407,7 -3447,7 +3447,7 @@@
  
  		let payment_secret = PaymentSecret(self.keys_manager.get_secure_random_bytes());
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let mut payment_secrets = self.pending_inbound_payments.lock().unwrap();
  		match payment_secrets.entry(payment_hash) {
  			hash_map::Entry::Vacant(e) => {
@@@ -3477,7 -3517,7 +3517,7 @@@
  	/// `invoice_expiry_delta_secs` describes the number of seconds that the invoice is valid for
  	/// in excess of the current time. This should roughly match the expiry time set in the invoice.
  	/// After this many seconds, we will remove the inbound payment, resulting in any attempts to
 -	/// pay the invoice failing. The BOLT spec suggests 7,200 secs as a default validity time for
 +	/// pay the invoice failing. The BOLT spec suggests 3,600 secs as a default validity time for
  	/// invoices when no timeout is set.
  	///
  	/// Note that we use block header time to time-out pending inbound payments (with some margin
@@@ -3564,7 -3604,7 +3604,7 @@@ wher
  	}
  
  	fn block_disconnected(&self, header: &BlockHeader, height: u32) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let new_height = height - 1;
  		{
  			let mut best_block = self.best_block.write().unwrap();
@@@ -3595,7 -3635,7 +3635,7 @@@ wher
  		let block_hash = header.block_hash();
  		log_trace!(self.logger, "{} transactions included in block {} at height {} provided", txdata.len(), block_hash, height);
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		self.do_chain_event(Some(height), |channel| channel.transactions_confirmed(&block_hash, height, txdata, &self.logger).map(|a| (a, Vec::new())));
  	}
  
@@@ -3607,7 -3647,7 +3647,7 @@@
  		let block_hash = header.block_hash();
  		log_trace!(self.logger, "New best block: {} at height {}", block_hash, height);
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		*self.best_block.write().unwrap() = BestBlock::new(block_hash, height);
  
@@@ -3649,7 -3689,7 +3689,7 @@@
  	}
  
  	fn transaction_unconfirmed(&self, txid: &Txid) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		self.do_chain_event(None, |channel| {
  			if let Some(funding_txo) = channel.get_funding_txo() {
  				if funding_txo.txid == *txid {
@@@ -3795,92 -3835,92 +3835,92 @@@ impl<Signer: Sign, M: Deref , T: Deref 
          L::Target: Logger,
  {
  	fn handle_open_channel(&self, counterparty_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_open_channel(counterparty_node_id, their_features, msg), *counterparty_node_id);
  	}
  
  	fn handle_accept_channel(&self, counterparty_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::AcceptChannel) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_accept_channel(counterparty_node_id, their_features, msg), *counterparty_node_id);
  	}
  
  	fn handle_funding_created(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingCreated) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_funding_created(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_funding_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingSigned) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_funding_signed(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_funding_locked(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingLocked) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_funding_locked(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_shutdown(&self, counterparty_node_id: &PublicKey, their_features: &InitFeatures, msg: &msgs::Shutdown) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_shutdown(counterparty_node_id, their_features, msg), *counterparty_node_id);
  	}
  
  	fn handle_closing_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::ClosingSigned) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_closing_signed(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_update_add_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_update_add_htlc(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_update_fulfill_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_update_fulfill_htlc(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_update_fail_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_update_fail_htlc(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_update_fail_malformed_htlc(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_update_fail_malformed_htlc(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_commitment_signed(&self, counterparty_node_id: &PublicKey, msg: &msgs::CommitmentSigned) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_commitment_signed(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_revoke_and_ack(&self, counterparty_node_id: &PublicKey, msg: &msgs::RevokeAndACK) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_revoke_and_ack(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_update_fee(&self, counterparty_node_id: &PublicKey, msg: &msgs::UpdateFee) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_update_fee(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_announcement_signatures(&self, counterparty_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_announcement_signatures(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_channel_update(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelUpdate) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_channel_update(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn handle_channel_reestablish(&self, counterparty_node_id: &PublicKey, msg: &msgs::ChannelReestablish) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let _ = handle_error!(self, self.internal_channel_reestablish(counterparty_node_id, msg), *counterparty_node_id);
  	}
  
  	fn peer_disconnected(&self, counterparty_node_id: &PublicKey, no_connection_possible: bool) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  		let mut failed_channels = Vec::new();
  		let mut failed_payments = Vec::new();
  		let mut no_channels_remain = true;
@@@ -3916,7 -3956,6 +3956,6 @@@
  						// on peer disconnect here, there will need to be corresponding changes in
  						// reestablish logic.
  						let failed_adds = chan.remove_uncommitted_htlcs_and_mark_paused(&self.logger);
- 						chan.to_disabled_marked();
  						if !failed_adds.is_empty() {
  							let chan_update = self.get_channel_update(&chan).map(|u| u.encode_with_len()).unwrap(); // Cannot add/recv HTLCs before we have a short_id so unwrap is safe
  							failed_payments.push((chan_update, failed_adds));
@@@ -3974,7 -4013,7 +4013,7 @@@
  	fn peer_connected(&self, counterparty_node_id: &PublicKey, init_msg: &msgs::Init) {
  		log_debug!(self.logger, "Generating channel_reestablish events for {}", log_pubkey!(counterparty_node_id));
  
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		{
  			let mut peer_state_lock = self.per_peer_state.write().unwrap();
@@@ -4014,7 -4053,7 +4053,7 @@@
  	}
  
  	fn handle_error(&self, counterparty_node_id: &PublicKey, msg: &msgs::ErrorMessage) {
- 		let _persistence_guard = PersistenceNotifierGuard::new(&self.total_consistency_lock, &self.persistence_notifier);
+ 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
  
  		if msg.channel_id == [0; 32] {
  			for chan in self.list_channels() {
@@@ -4049,6 -4088,10 +4088,10 @@@ impl PersistenceNotifier 
  		loop {
  			let &(ref mtx, ref cvar) = &self.persistence_lock;
  			let mut guard = mtx.lock().unwrap();
+ 			if *guard {
+ 				*guard = false;
+ 				return;
+ 			}
  			guard = cvar.wait(guard).unwrap();
  			let result = *guard;
  			if result {
@@@ -4064,6 -4107,10 +4107,10 @@@
  		loop {
  			let &(ref mtx, ref cvar) = &self.persistence_lock;
  			let mut guard = mtx.lock().unwrap();
+ 			if *guard {
+ 				*guard = false;
+ 				return true;
+ 			}
  			guard = cvar.wait_timeout(guard, max_wait).unwrap().0;
  			// Due to spurious wakeups that can happen on `wait_timeout`, here we need to check if the
  			// desired wait time has actually passed, and if not then restart the loop with a reduced wait