Add monitor_did_broadcast method to ManyChannelMonitor

This allows us to replace the call to channel.channel_monitor()
.would_broadcast_at_height() in ChannelManager. This allows
us to next get rid of the Channel's copy of the ChannelMonitor.

Author: valentinewallace

fuzz/src/chanmon_consistency.rs

@@ -138,6 +138,10 @@ impl channelmonitor::ManyChannelMonitor for TestChannelMonitor {
 	fn get_and_clear_pending_htlcs_updated(&self) -> Vec<HTLCUpdate> {
 		return self.simple_monitor.get_and_clear_pending_htlcs_updated();
 	}
+
+	fn get_monitor_did_broadcast(&self, funding_txo: OutPoint) -> Result<bool, channelmonitor::ChannelMonitorUpdateErr> {
+		self.simple_monitor.get_monitor_did_broadcast(funding_txo)
+	}
 }
 
 struct KeyProvider {

lightning/src/ln/channelmanager.rs

@@ -2992,9 +2992,39 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 			}
 		}
 
+		let mut failed_channels = Vec::new();
+		let mut channel_lock = self.channel_state.lock().unwrap();
+		let channel_state = &mut *channel_lock;
+		let pending_msg_events = &mut channel_state.pending_msg_events;
+		let short_to_id = &mut channel_state.short_to_id;
+		channel_state.by_id.retain(|_, chan| {
+			if let Some(funding_tx) = chan.get_funding_txo() {
+				match self.monitor.get_monitor_did_broadcast(funding_tx) {
+					Ok(broadcasted) => {
+						if broadcasted {
+							if let Some(short_id) = chan.get_short_channel_id() {
+								short_to_id.remove(&short_id);
+							}
+							failed_channels.push(chan.force_shutdown(false));
+							if let Ok(update) = self.get_channel_update(&chan) {
+								pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+									msg: update
+								});
+							}
+							return false
+						}
+					},
+					_ => {},
+				}
+			}
+			true
+		});
+		for failure in failed_channels.drain(..) {
+			self.finish_force_close_channel(failure);
+		}
+
 		let mut ret = Vec::new();
-		let mut channel_state = self.channel_state.lock().unwrap();
-		mem::swap(&mut ret, &mut channel_state.pending_msg_events);
+		mem::swap(&mut ret, pending_msg_events);
 		ret
 	}
 }
@@ -3104,21 +3134,6 @@ impl<ChanSigner: ChannelKeys, M: Deref + Sync + Send, T: Deref + Sync + Send, K:
 						}
 					}
 				}
-				if channel.is_funding_initiated() && channel.channel_monitor().would_broadcast_at_height(height, &self.logger) {
-					if let Some(short_id) = channel.get_short_channel_id() {
-						short_to_id.remove(&short_id);
-					}
-					// If would_broadcast_at_height() is true, the channel_monitor will broadcast
-					// the latest local tx for us, so we should skip that here (it doesn't really
-					// hurt anything, but does make tests a bit simpler).
-					failed_channels.push(channel.force_shutdown(false));
-					if let Ok(update) = self.get_channel_update(&channel) {
-						pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-							msg: update
-						});
-					}
-					return false;
-				}
 				true
 			});
 

lightning/src/ln/channelmonitor.rs

@@ -299,6 +299,13 @@ impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send, L:
 		}
 		pending_htlcs_updated
 	}
+
+	fn get_monitor_did_broadcast(&self, funding_txo: OutPoint) -> Result<bool, ChannelMonitorUpdateErr> {
+		match self.monitors.lock().unwrap().get(&funding_txo) {
+			Some(monitor) => Ok(monitor.broadcasted()),
+			None => Err(ChannelMonitorUpdateErr::PermanentFailure)
+		}
+	}
 }
 
 impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref, C: Deref> events::EventsProvider for SimpleManyChannelMonitor<Key, ChanSigner, T, F, L, C>
@@ -813,6 +820,9 @@ pub struct ChannelMonitor<ChanSigner: ChannelKeys> {
 	// may occur, and we fail any such monitor updates.
 	local_tx_signed: bool,
 
+	// Whether the commitment transaction associated with this channel has been broadcasted.
+	broadcasted: bool,
+
 	// 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 update_monitor to ensure any local user copies keep
@@ -884,6 +894,9 @@ pub trait ManyChannelMonitor: Send + Sync {
 	/// ChannelMonitor::get_and_clear_pending_htlcs_updated() for each ChannelMonitor and return
 	/// the full list.
 	fn get_and_clear_pending_htlcs_updated(&self) -> Vec<HTLCUpdate>;
+
+	/// Gets whether the monitor broadcasted the channel's commitment transaction.
+	fn get_monitor_did_broadcast(&self, funding_txo: OutPoint) -> Result<bool, ChannelMonitorUpdateErr>;
 }
 
 #[cfg(any(test, feature = "fuzztarget"))]
@@ -1113,6 +1126,7 @@ impl<ChanSigner: ChannelKeys + Writeable> ChannelMonitor<ChanSigner> {
 
 		self.lockdown_from_offchain.write(writer)?;
 		self.local_tx_signed.write(writer)?;
+		self.broadcasted.write(writer)?;
 
 		Ok(())
 	}
@@ -1197,6 +1211,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 
 			lockdown_from_offchain: false,
 			local_tx_signed: false,
+			broadcasted: false,
 
 			last_block_hash: Default::default(),
 			secp_ctx: Secp256k1::new(),
@@ -1350,6 +1365,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 	{
 		for tx in self.get_latest_local_commitment_txn(logger).iter() {
 			broadcaster.broadcast_transaction(tx);
+			self.broadcasted = true;
 		}
 	}
 
@@ -1966,7 +1982,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 				}
 			}
 		}
-		self.onchain_tx_handler.block_connected(txn_matched, claimable_outpoints, height, &*broadcaster, &*fee_estimator, &*logger);
+		self.broadcasted = self.onchain_tx_handler.block_connected(txn_matched, claimable_outpoints, height, &*broadcaster, &*fee_estimator, &*logger);
 
 		self.last_block_hash = block_hash.clone();
 		for &(ref txid, ref output_scripts) in watch_outputs.iter() {
@@ -1988,11 +2004,16 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			//- maturing spendable output has transaction paying us has been disconnected
 		}
 
-		self.onchain_tx_handler.block_disconnected(height, broadcaster, fee_estimator, logger);
+		self.broadcasted = self.onchain_tx_handler.block_disconnected(height, broadcaster, fee_estimator, logger);
 
 		self.last_block_hash = block_hash.clone();
 	}
 
+	/// Whether our local commitment transaction has been broadcasted.
+	fn broadcasted(&self) -> bool {
+		self.broadcasted
+	}
+
 	pub(super) fn would_broadcast_at_height<L: Deref>(&self, height: u32, logger: &L) -> bool where L::Target: Logger {
 		// We need to consider all HTLCs which are:
 		//  * in any unrevoked remote commitment transaction, as they could broadcast said
@@ -2483,6 +2504,7 @@ impl<ChanSigner: ChannelKeys + Readable> Readable for (BlockHash, ChannelMonitor
 
 		let lockdown_from_offchain = Readable::read(reader)?;
 		let local_tx_signed = Readable::read(reader)?;
+		let broadcasted = Readable::read(reader)?;
 
 		Ok((last_block_hash.clone(), ChannelMonitor {
 			latest_update_id,
@@ -2527,6 +2549,8 @@ impl<ChanSigner: ChannelKeys + Readable> Readable for (BlockHash, ChannelMonitor
 			lockdown_from_offchain,
 			local_tx_signed,
 
+			broadcasted,
+
 			last_block_hash,
 			secp_ctx: Secp256k1::new(),
 		}))

lightning/src/ln/functional_tests.rs

@@ -2420,13 +2420,21 @@ fn channel_monitor_network_test() {
 	// CLTV expires at TEST_FINAL_CLTV + 1 (current height) + 1 (added in send_payment for
 	// buffer space).
 
-	{
+	let (close_chan_update_1, close_chan_update_2) = {
 		let mut header = BlockHeader { version: 0x20000000, prev_blockhash: Default::default(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 		nodes[3].block_notifier.block_connected_checked(&header, 2, &Vec::new()[..], &[0; 0]);
 		for i in 3..TEST_FINAL_CLTV + 2 + LATENCY_GRACE_PERIOD_BLOCKS + 1 {
 			header = BlockHeader { version: 0x20000000, prev_blockhash: header.bitcoin_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 			nodes[3].block_notifier.block_connected_checked(&header, i, &Vec::new()[..], &[0; 0]);
 		}
+		let events = nodes[3].node.get_and_clear_pending_msg_events();
+		assert_eq!(events.len(), 1);
+		let close_chan_update_1 = match events[0] {
+			MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
+				msg.clone()
+			},
+			_ => panic!("Unexpected event"),
+		};
 		check_added_monitors!(nodes[3], 1);
 
 		// Clear bumped claiming txn spending node 2 commitment tx. Bumped txn are generated after reaching some height timer.
@@ -2451,16 +2459,25 @@ fn channel_monitor_network_test() {
 			header = BlockHeader { version: 0x20000000, prev_blockhash: header.bitcoin_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 			nodes[4].block_notifier.block_connected_checked(&header, i, &Vec::new()[..], &[0; 0]);
 		}
-
+		let events = nodes[4].node.get_and_clear_pending_msg_events();
+		assert_eq!(events.len(), 1);
+		let close_chan_update_2 = match events[0] {
+			MessageSendEvent::BroadcastChannelUpdate { ref msg } => {
+				msg.clone()
+			},
+			_ => panic!("Unexpected event"),
+		};
 		check_added_monitors!(nodes[4], 1);
 		test_txn_broadcast(&nodes[4], &chan_4, None, HTLCType::SUCCESS);
 
 		header = BlockHeader { version: 0x20000000, prev_blockhash: header.bitcoin_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
 		nodes[4].block_notifier.block_connected(&Block { header, txdata: vec![node_txn[0].clone()] }, TEST_FINAL_CLTV - 5);
 
 		check_preimage_claim(&nodes[4], &node_txn);
-	}
-	get_announce_close_broadcast_events(&nodes, 3, 4);
+		(close_chan_update_1, close_chan_update_2)
+	};
+	nodes[3].net_graph_msg_handler.handle_channel_update(&close_chan_update_2).unwrap();
+	nodes[4].net_graph_msg_handler.handle_channel_update(&close_chan_update_1).unwrap();
 	assert_eq!(nodes[3].node.list_channels().len(), 0);
 	assert_eq!(nodes[4].node.list_channels().len(), 0);
 }

lightning/src/ln/onchaintx.rs

@@ -657,7 +657,7 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		None
 	}
 
-	pub(super) fn block_connected<B: Deref, F: Deref, L: Deref>(&mut self, txn_matched: &[&Transaction], claimable_outpoints: Vec<ClaimRequest>, height: u32, broadcaster: B, fee_estimator: F, logger: L)
+	pub(super) fn block_connected<B: Deref, F: Deref, L: Deref>(&mut self, txn_matched: &[&Transaction], claimable_outpoints: Vec<ClaimRequest>, height: u32, broadcaster: B, fee_estimator: F, logger: L) -> bool
 		where B::Target: BroadcasterInterface,
 		      F::Target: FeeEstimator,
 					L::Target: Logger,
@@ -666,6 +666,7 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		let mut new_claims = Vec::new();
 		let mut aggregated_claim = HashMap::new();
 		let mut aggregated_soonest = ::std::u32::MAX;
+		let mut broadcasted_tx = false;
 
 		// Try to aggregate outputs if their timelock expiration isn't imminent (absolute_timelock
 		// <= CLTV_SHARED_CLAIM_BUFFER) and they don't require an immediate nLockTime (aggregable).
@@ -702,6 +703,7 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 				self.pending_claim_requests.insert(txid, claim_material);
 				log_trace!(logger, "Broadcast onchain {}", log_tx!(tx));
 				broadcaster.broadcast_transaction(&tx);
+				broadcasted_tx = true;
 			}
 		}
 
@@ -827,13 +829,16 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 				}
 			}
 		}
+
+		broadcasted_tx
 	}
 
-	pub(super) fn block_disconnected<B: Deref, F: Deref, L: Deref>(&mut self, height: u32, broadcaster: B, fee_estimator: F, logger: L)
+	pub(super) fn block_disconnected<B: Deref, F: Deref, L: Deref>(&mut self, height: u32, broadcaster: B, fee_estimator: F, logger: L) -> bool
 		where B::Target: BroadcasterInterface,
 		      F::Target: FeeEstimator,
 					L::Target: Logger,
 	{
+		let mut broadcasted_tx = false;
 		let mut bump_candidates = HashMap::new();
 		if let Some(events) = self.onchain_events_waiting_threshold_conf.remove(&(height + ANTI_REORG_DELAY - 1)) {
 			//- our claim tx on a commitment tx output
@@ -859,6 +864,7 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 				claim_material.height_timer = new_timer;
 				claim_material.feerate_previous = new_feerate;
 				broadcaster.broadcast_transaction(&bump_tx);
+				broadcasted_tx = true;
 			}
 		}
 		for (ancestor_claim_txid, claim_material) in bump_candidates.drain() {
@@ -875,6 +881,8 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		for req in remove_request {
 			self.pending_claim_requests.remove(&req);
 		}
+
+		broadcasted_tx
 	}
 
 	pub(super) fn provide_latest_local_tx(&mut self, tx: LocalCommitmentTransaction) -> Result<(), ()> {

lightning/src/util/test_utils.rs

@@ -132,6 +132,10 @@ impl<'a> channelmonitor::ManyChannelMonitor for TestChannelMonitor<'a> {
 	fn get_and_clear_pending_htlcs_updated(&self) -> Vec<HTLCUpdate> {
 		return self.simple_monitor.get_and_clear_pending_htlcs_updated();
 	}
+
+	fn get_monitor_did_broadcast(&self, funding_txo: OutPoint) -> Result<bool, channelmonitor::ChannelMonitorUpdateErr> {
+		self.simple_monitor.get_monitor_did_broadcast(funding_txo)
+	}
 }
 
 pub struct TestBroadcaster {