(XXX: fuzz) Rm ChannelMonitor merge capabilities in favor of explicit add/update

This removes the ability to merge ChannelMonitors in favor of
explicit ChannelMonitorUpdates. It further removes
ChannelManager::test_restore_channel_monitor in favor of the new
ChannelManager::channel_monitor_updated method, which explicitly
confirms a set of updates instead of providing the latest copy of
each ChannelMonitor to the user.

This removes almost all need for Channels to have the latest
channel_monitor, except for broadcasting the latest local state.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -336,7 +336,7 @@ pub type SimpleRefChannelManager<'a, M> = ChannelManager<InMemoryChannelKeys, &'
 ///
 /// Note that you can be a bit lazier about writing out ChannelManager than you can be with
 /// ChannelMonitors. With ChannelMonitors you MUST write each monitor update out to disk before
-/// returning from ManyChannelMonitor::add_update_monitor, with ChannelManagers, writing updates
+/// returning from ManyChannelMonitor::add_/update_monitor, with ChannelManagers, writing updates
 /// happens out-of-band (and will prevent any other ChannelManager operations from occurring during
 /// the serialization process). If the deserialized version is out-of-date compared to the
 /// ChannelMonitors passed by reference to read(), those channels will be force-closed based on the
@@ -1388,8 +1388,8 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
 			}
 		};
 		// Because we have exclusive ownership of the channel here we can release the channel_state
-		// lock before add_update_monitor
-		if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
+		// lock before add_monitor
+		if let Err(e) = self.monitor.add_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
 			match e {
 				ChannelMonitorUpdateErr::PermanentFailure => {
 					match handle_error!(self, Err(MsgHandleErrInternal::from_finish_shutdown("ChannelMonitor storage failure", *temporary_channel_id, chan.force_shutdown(), None)), chan.get_their_node_id()) {
@@ -2081,117 +2081,6 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
 		}
 	}
 
-	/// Used to restore channels to normal operation after a
-	/// ChannelMonitorUpdateErr::TemporaryFailure was returned from a channel monitor update
-	/// operation.
-	pub fn test_restore_channel_monitor(&self) {
-		let mut close_results = Vec::new();
-		let mut htlc_forwards = Vec::new();
-		let mut htlc_failures = Vec::new();
-		let mut pending_events = Vec::new();
-		let _ = self.total_consistency_lock.read().unwrap();
-
-		{
-			let mut channel_lock = self.channel_state.lock().unwrap();
-			let channel_state = &mut *channel_lock;
-			let short_to_id = &mut channel_state.short_to_id;
-			let pending_msg_events = &mut channel_state.pending_msg_events;
-			channel_state.by_id.retain(|_, channel| {
-				if channel.is_awaiting_monitor_update() {
-					let chan_monitor = channel.channel_monitor().clone();
-					if let Err(e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
-						match e {
-							ChannelMonitorUpdateErr::PermanentFailure => {
-								// TODO: There may be some pending HTLCs that we intended to fail
-								// backwards when a monitor update failed. We should make sure
-								// knowledge of those gets moved into the appropriate in-memory
-								// ChannelMonitor and they get failed backwards once we get
-								// on-chain confirmations.
-								// Note I think #198 addresses this, so once it's merged a test
-								// should be written.
-								if let Some(short_id) = channel.get_short_channel_id() {
-									short_to_id.remove(&short_id);
-								}
-								close_results.push(channel.force_shutdown());
-								if let Ok(update) = self.get_channel_update(&channel) {
-									pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-										msg: update
-									});
-								}
-								false
-							},
-							ChannelMonitorUpdateErr::TemporaryFailure => true,
-						}
-					} else {
-						let (raa, commitment_update, order, pending_forwards, mut pending_failures, needs_broadcast_safe, funding_locked) = channel.monitor_updating_restored();
-						if !pending_forwards.is_empty() {
-							htlc_forwards.push((channel.get_short_channel_id().expect("We can't have pending forwards before funding confirmation"), pending_forwards));
-						}
-						htlc_failures.append(&mut pending_failures);
-
-						macro_rules! handle_cs { () => {
-							if let Some(update) = commitment_update {
-								pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-									node_id: channel.get_their_node_id(),
-									updates: update,
-								});
-							}
-						} }
-						macro_rules! handle_raa { () => {
-							if let Some(revoke_and_ack) = raa {
-								pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK {
-									node_id: channel.get_their_node_id(),
-									msg: revoke_and_ack,
-								});
-							}
-						} }
-						match order {
-							RAACommitmentOrder::CommitmentFirst => {
-								handle_cs!();
-								handle_raa!();
-							},
-							RAACommitmentOrder::RevokeAndACKFirst => {
-								handle_raa!();
-								handle_cs!();
-							},
-						}
-						if needs_broadcast_safe {
-							pending_events.push(events::Event::FundingBroadcastSafe {
-								funding_txo: channel.get_funding_txo().unwrap(),
-								user_channel_id: channel.get_user_id(),
-							});
-						}
-						if let Some(msg) = funding_locked {
-							pending_msg_events.push(events::MessageSendEvent::SendFundingLocked {
-								node_id: channel.get_their_node_id(),
-								msg,
-							});
-							if let Some(announcement_sigs) = self.get_announcement_sigs(channel) {
-								pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures {
-									node_id: channel.get_their_node_id(),
-									msg: announcement_sigs,
-								});
-							}
-							short_to_id.insert(channel.get_short_channel_id().unwrap(), channel.channel_id());
-						}
-						true
-					}
-				} else { true }
-			});
-		}
-
-		self.pending_events.lock().unwrap().append(&mut pending_events);
-
-		for failure in htlc_failures.drain(..) {
-			self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2);
-		}
-		self.forward_htlcs(&mut htlc_forwards[..]);
-
-		for res in close_results.drain(..) {
-			self.finish_force_close_channel(res);
-		}
-	}
-
 	fn internal_open_channel(&self, their_node_id: &PublicKey, their_features: InitFeatures, msg: &msgs::OpenChannel) -> Result<(), MsgHandleErrInternal> {
 		if msg.chain_hash != self.genesis_hash {
 			return Err(MsgHandleErrInternal::send_err_msg_no_close("Unknown genesis block hash", msg.temporary_channel_id.clone()));
@@ -2254,8 +2143,8 @@ impl<ChanSigner: ChannelKeys, M: Deref> ChannelManager<ChanSigner, M> where M::T
 			}
 		};
 		// Because we have exclusive ownership of the channel here we can release the channel_state
-		// lock before add_update_monitor
-		if let Err(e) = self.monitor.add_update_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) {
+		// lock before add_monitor
+		if let Err(e) = self.monitor.add_monitor(monitor_update.get_funding_txo().unwrap(), monitor_update) {
 			match e {
 				ChannelMonitorUpdateErr::PermanentFailure => {
 					// Note that we reply with the new channel_id in error messages if we gave up on the

lightning/src/ln/channelmanager.rs

@@ -130,9 +130,9 @@ pub enum ChannelMonitorUpdateErr {
 }
 
 /// General Err type for ChannelMonitor actions. Generally, this implies that the data provided is
-/// inconsistent with the ChannelMonitor being called. eg for ChannelMonitor::insert_combine this
-/// means you tried to merge two monitors for different channels or for a channel which was
-/// restored from a backup and then generated new commitment updates.
+/// inconsistent with the ChannelMonitor being called. eg for ChannelMonitor::update_monitor this
+/// means you tried to update a monitor for a different channel or the ChannelMonitorUpdate was
+/// corrupted.
 /// Contains a human-readable error message.
 #[derive(Debug)]
 pub struct MonitorUpdateError(pub &'static str);
@@ -149,7 +149,7 @@ impl_writeable!(HTLCUpdate, 0, { payment_hash, payment_preimage, source });
 
 /// Simple trait indicating ability to track a set of ChannelMonitors and multiplex events between
 /// them. Generally should be implemented by keeping a local SimpleManyChannelMonitor and passing
-/// events to it, while also taking any add_update_monitor events and passing them to some remote
+/// events to it, while also taking any add/update_monitor events and passing them to some remote
 /// server(s).
 ///
 /// Note that any updates to a channel's monitor *must* be applied to each instance of the
@@ -163,7 +163,7 @@ impl_writeable!(HTLCUpdate, 0, { payment_hash, payment_preimage, source });
 /// BlockNotifier and call the BlockNotifier's `block_(dis)connected` methods, which will notify
 /// all registered listeners in one go.
 pub trait ManyChannelMonitor<ChanSigner: ChannelKeys>: Send + Sync {
-	/// Adds or updates a monitor for the given `funding_txo`.
+	/// Adds a monitor for the given `funding_txo`.
 	///
 	/// Implementer must also ensure that the funding_txo txid *and* outpoint are registered with
 	/// any relevant ChainWatchInterfaces such that the provided monitor receives block_connected
@@ -175,7 +175,7 @@ pub trait ManyChannelMonitor<ChanSigner: ChannelKeys>: Send + Sync {
 	///
 	/// Any spends of outputs which should have been registered which aren't passed to
 	/// ChannelMonitors via block_connected may result in funds loss.
-	fn add_update_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr>;
+	fn add_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr>;
 
 	/// Updates a monitor for the given `funding_txo`.
 	///
@@ -274,14 +274,11 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys> Simpl
 	}
 
 	/// Adds or updates the monitor which monitors the channel referred to by the given key.
-	pub fn add_update_monitor_by_key(&self, key: Key, monitor: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
+	pub fn add_monitor_by_key(&self, key: Key, monitor: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
 		let mut monitors = self.monitors.lock().unwrap();
-		match monitors.get_mut(&key) {
-			Some(orig_monitor) => {
-				log_trace!(self, "Updating Channel Monitor for channel {}", log_funding_info!(monitor.key_storage));
-				return orig_monitor.insert_combine(monitor);
-			},
-			None => {}
+		let entry = match monitors.entry(key) {
+			hash_map::Entry::Occupied(_) => return Err(MonitorUpdateError("Channel monitor for given key is already present")),
+			hash_map::Entry::Vacant(e) => e,
 		};
 		match monitor.key_storage {
 			Storage::Local { ref funding_info, .. } => {
@@ -305,7 +302,7 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys> Simpl
 				self.chain_monitor.install_watch_outpoint((*txid, idx as u32), script);
 			}
 		}
-		monitors.insert(key, monitor);
+		entry.insert(monitor);
 		Ok(())
 	}
 
@@ -323,8 +320,8 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys> Simpl
 }
 
 impl<ChanSigner: ChannelKeys> ManyChannelMonitor<ChanSigner> for SimpleManyChannelMonitor<OutPoint, ChanSigner> {
-	fn add_update_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr> {
-		match self.add_update_monitor_by_key(funding_txo, monitor) {
+	fn add_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr> {
+		match self.add_monitor_by_key(funding_txo, monitor) {
 			Ok(_) => Ok(()),
 			Err(_) => Err(ChannelMonitorUpdateErr::PermanentFailure),
 		}
@@ -867,7 +864,7 @@ pub struct ChannelMonitor<ChanSigner: ChannelKeys> {
 
 	// We simply modify last_block_hash in Channel's block_connected so that serialization is
 	// consistent but hopefully the users' copy handles block_connected in a consistent way.
-	// (we do *not*, however, update them in insert_combine to ensure any local user copies keep
+	// (we do *not*, however, update them in update_monitor to ensure any local user copies keep
 	// their last_block_hash from its state and not based on updated copies that didn't run through
 	// the full block_connected).
 	pub(crate) last_block_hash: Sha256dHash,
@@ -1497,68 +1494,6 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 		Ok(())
 	}
 
-	/// Combines this ChannelMonitor with the information contained in the other ChannelMonitor.
-	/// After a successful call this ChannelMonitor is up-to-date and is safe to use to monitor the
-	/// chain for new blocks/transactions.
-	pub fn insert_combine(&mut self, mut other: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
-		match self.key_storage {
-			Storage::Local { ref funding_info, .. } => {
-				if funding_info.is_none() { return Err(MonitorUpdateError("Try to combine a Local monitor without funding_info")); }
-				let our_funding_info = funding_info;
-				if let Storage::Local { ref funding_info, .. } = other.key_storage {
-					if funding_info.is_none() { return Err(MonitorUpdateError("Try to combine a Local monitor without funding_info")); }
-					// We should be able to compare the entire funding_txo, but in fuzztarget it's trivially
-					// easy to collide the funding_txo hash and have a different scriptPubKey.
-					if funding_info.as_ref().unwrap().0 != our_funding_info.as_ref().unwrap().0 {
-						return Err(MonitorUpdateError("Funding transaction outputs are not identical!"));
-					}
-				} else {
-					return Err(MonitorUpdateError("Try to combine a Local monitor with a Watchtower one !"));
-				}
-			},
-			Storage::Watchtower { .. } => {
-				if let Storage::Watchtower { .. } = other.key_storage {
-					unimplemented!();
-				} else {
-					return Err(MonitorUpdateError("Try to combine a Watchtower monitor with a Local one !"));
-				}
-			},
-		}
-		let other_min_secret = other.get_min_seen_secret();
-		let our_min_secret = self.get_min_seen_secret();
-		if our_min_secret > other_min_secret {
-			self.provide_secret(other_min_secret, other.get_secret(other_min_secret).unwrap())?;
-		}
-		if let Some(ref local_tx) = self.current_local_signed_commitment_tx {
-			if let Some(ref other_local_tx) = other.current_local_signed_commitment_tx {
-				let our_commitment_number = 0xffffffffffff - ((((local_tx.tx.without_valid_witness().input[0].sequence as u64 & 0xffffff) << 3*8) | (local_tx.tx.without_valid_witness().lock_time as u64 & 0xffffff)) ^ self.commitment_transaction_number_obscure_factor);
-				let other_commitment_number = 0xffffffffffff - ((((other_local_tx.tx.without_valid_witness().input[0].sequence as u64 & 0xffffff) << 3*8) | (other_local_tx.tx.without_valid_witness().lock_time as u64 & 0xffffff)) ^ other.commitment_transaction_number_obscure_factor);
-				if our_commitment_number >= other_commitment_number {
-					self.key_storage = other.key_storage;
-				}
-			}
-		}
-		// TODO: We should use current_remote_commitment_number and the commitment number out of
-		// local transactions to decide how to merge
-		if our_min_secret >= other_min_secret {
-			self.their_cur_revocation_points = other.their_cur_revocation_points;
-			for (txid, htlcs) in other.remote_claimable_outpoints.drain() {
-				self.remote_claimable_outpoints.insert(txid, htlcs);
-			}
-			if let Some(local_tx) = other.prev_local_signed_commitment_tx {
-				self.prev_local_signed_commitment_tx = Some(local_tx);
-			}
-			if let Some(local_tx) = other.current_local_signed_commitment_tx {
-				self.current_local_signed_commitment_tx = Some(local_tx);
-			}
-			self.payment_preimages = other.payment_preimages;
-			self.to_remote_rescue = other.to_remote_rescue;
-		}
-
-		self.current_remote_commitment_number = cmp::min(self.current_remote_commitment_number, other.current_remote_commitment_number);
-		Ok(())
-	}
-
 	/// Gets the update_id from the latest ChannelMonitorUpdate which was applied to this
 	/// ChannelMonitor.
 	pub fn get_latest_update_id(&self) -> u64 {

lightning/src/ln/channelmonitor.rs

@@ -130,7 +130,7 @@ impl<'a, 'b> Drop for Node<'a, 'b> {
 			let chain_watch = Arc::new(chaininterface::ChainWatchInterfaceUtil::new(Network::Testnet, Arc::clone(&self.logger) as Arc<Logger>));
 			let channel_monitor = test_utils::TestChannelMonitor::new(chain_watch.clone(), self.tx_broadcaster.clone(), self.logger.clone(), feeest);
 			for deserialized_monitor in deserialized_monitors.drain(..) {
-				if let Err(_) = channel_monitor.add_update_monitor(deserialized_monitor.get_funding_txo().unwrap(), deserialized_monitor) {
+				if let Err(_) = channel_monitor.add_monitor(deserialized_monitor.get_funding_txo().unwrap(), deserialized_monitor) {
 					panic!();
 				}
 			}

lightning/src/ln/functional_test_utils.rs

@@ -3714,7 +3714,7 @@ fn test_no_txn_manager_serialize_deserialize() {
 	nodes_0_deserialized = nodes_0_deserialized_tmp;
 	assert!(nodes_0_read.is_empty());
 
-	assert!(nodes[0].chan_monitor.add_update_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok());
+	assert!(nodes[0].chan_monitor.add_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok());
 	nodes[0].node = &nodes_0_deserialized;
 	nodes[0].block_notifier.register_listener(nodes[0].node);
 	assert_eq!(nodes[0].node.list_channels().len(), 1);
@@ -3783,7 +3783,7 @@ fn test_simple_manager_serialize_deserialize() {
 	nodes_0_deserialized = nodes_0_deserialized_tmp;
 	assert!(nodes_0_read.is_empty());
 
-	assert!(nodes[0].chan_monitor.add_update_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok());
+	assert!(nodes[0].chan_monitor.add_monitor(chan_0_monitor.get_funding_txo().unwrap(), chan_0_monitor).is_ok());
 	nodes[0].node = &nodes_0_deserialized;
 	check_added_monitors!(nodes[0], 1);
 
@@ -3856,7 +3856,7 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() {
 	}
 
 	for monitor in node_0_monitors.drain(..) {
-		assert!(nodes[0].chan_monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor).is_ok());
+		assert!(nodes[0].chan_monitor.add_monitor(monitor.get_funding_txo().unwrap(), monitor).is_ok());
 		check_added_monitors!(nodes[0], 1);
 	}
 	nodes[0].node = &nodes_0_deserialized;
@@ -6518,7 +6518,7 @@ fn test_data_loss_protect() {
 		}).unwrap().1
 	};
 	nodes[0].node = &node_state_0;
-	assert!(monitor.add_update_monitor(OutPoint { txid: chan.3.txid(), index: 0 }, chan_monitor.clone()).is_ok());
+	assert!(monitor.add_monitor(OutPoint { txid: chan.3.txid(), index: 0 }, chan_monitor.clone()).is_ok());
 	nodes[0].chan_monitor = &monitor;
 	nodes[0].chain_monitor = chain_monitor;
 

lightning/src/ln/functional_tests.rs

@@ -33,7 +33,7 @@ pub enum APIError {
 		/// A human-readable error message
 		err: &'static str
 	},
-	/// An attempt to call add_update_monitor returned an Err (ie you did this!), causing the
+	/// An attempt to call add/update_monitor returned an Err (ie you did this!), causing the
 	/// attempted action to fail.
 	MonitorUpdateFailed,
 }