fixup! Add commitment transaction broadcast as a ChannelMonitor event

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -2970,11 +2970,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 	/// Process pending events from the ManyChannelMonitor.
 	fn process_pending_monitor_events(&self) {
 		let mut failed_channels = Vec::new();
-		// TODO: Event release to users and serialization is currently race-y: it's very easy for a
-		// user to serialize a ChannelManager with pending events in it and lose those events on
-		// restart. This is doubly true for the fail/fulfill-backs from monitor events!
 		{
-			//TODO: This behavior should be documented.
 			for monitor_event in self.monitor.get_and_clear_pending_monitor_events() {
 				match monitor_event {
 					MonitorEvent::HTLCEvent(htlc_update) => {
@@ -3022,6 +3018,8 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 				L::Target: Logger,
 {
 	fn get_and_clear_pending_msg_events(&self) -> Vec<events::MessageSendEvent> {
+		//TODO: This behavior should be documented. It's non-intuitive that we query
+		// ChannelMonitors when clearing other events.
 		self.process_pending_monitor_events();
 
 		let mut ret = Vec::new();
@@ -3039,6 +3037,8 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 				L::Target: Logger,
 {
 	fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
+		//TODO: This behavior should be documented. It's non-intuitive that we query
+		// ChannelMonitors when clearing other events.
 		self.process_pending_monitor_events();
 
 		let mut ret = Vec::new();

lightning/src/ln/channelmonitor.rs

@@ -1934,7 +1934,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			claimable_outpoints.push(ClaimRequest { absolute_timelock: height, aggregable: false, outpoint: BitcoinOutPoint { txid: self.funding_info.0.txid.clone(), vout: self.funding_info.0.index as u32 }, witness_data: InputMaterial::Funding { funding_redeemscript: self.funding_redeemscript.clone() }});
 		}
 		if should_broadcast {
+			self.pending_monitor_events.push(MonitorEvent::CommitmentTxBroadcasted(self.funding_info.0));
 			if let Some(commitment_tx) = self.onchain_tx_handler.get_fully_signed_local_tx(&self.funding_redeemscript) {
+				self.local_tx_signed = true;
 				let (mut new_outpoints, new_outputs, _) = self.broadcast_by_local_state(&commitment_tx, &self.current_local_commitment_tx);
 				if !new_outputs.is_empty() {
 					watch_outputs.push((self.current_local_commitment_tx.txid.clone(), new_outputs));
@@ -1963,10 +1965,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			}
 		}
 
-		let broadcasted_commit_tx = self.onchain_tx_handler.block_connected(txn_matched, claimable_outpoints, height, &*broadcaster, &*fee_estimator, &*logger);
-		if broadcasted_commit_tx {
-			self.pending_monitor_events.push(MonitorEvent::CommitmentTxBroadcasted(self.funding_info.0));
-		}
+		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,10 +1987,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			//- maturing spendable output has transaction paying us has been disconnected
 		}
 
-		let broadcasted_commit_tx = self.onchain_tx_handler.block_disconnected(height, broadcaster, fee_estimator, logger);
-		if broadcasted_commit_tx {
-			self.pending_monitor_events.push(MonitorEvent::CommitmentTxBroadcasted(self.funding_info.0));
-		}
+		self.onchain_tx_handler.block_disconnected(height, broadcaster, fee_estimator, logger);
 
 		self.last_block_hash = block_hash.clone();
 	}
@@ -2165,7 +2161,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 					}
 				} else if offered_preimage_claim {
 					if !self.pending_monitor_events.iter().any(
-						|update| if let &MonitorEvent::HTLCEvent(ref upd) = update { upd.source == source } else { false }) {
+						|update| if let &MonitorEvent::HTLCEvent(ref upd) = update {
+							upd.source == source
+						} else { false }) {
 						payment_preimage.0.copy_from_slice(&input.witness[1]);
 						self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
 							source,

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) -> bool
+	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)
 		where B::Target: BroadcasterInterface,
 		      F::Target: FeeEstimator,
 					L::Target: Logger,
@@ -666,7 +666,6 @@ 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 = 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).
@@ -703,7 +702,6 @@ 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 = true;
 			}
 		}
 
@@ -823,17 +821,15 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 			if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &claim_material, &*fee_estimator, &*logger) {
 				log_trace!(logger, "Broadcast onchain {}", log_tx!(bump_tx));
 				broadcaster.broadcast_transaction(&bump_tx);
-				broadcasted = true;
 				if let Some(claim_material) = self.pending_claim_requests.get_mut(first_claim_txid) {
 					claim_material.height_timer = new_timer;
 					claim_material.feerate_previous = new_feerate;
 				}
 			}
 		}
-		broadcasted
 	}
 
-	pub(super) fn block_disconnected<B: Deref, F: Deref, L: Deref>(&mut self, height: u32, broadcaster: B, fee_estimator: F, logger: L) -> bool
+	pub(super) fn block_disconnected<B: Deref, F: Deref, L: Deref>(&mut self, height: u32, broadcaster: B, fee_estimator: F, logger: L)
 		where B::Target: BroadcasterInterface,
 		      F::Target: FeeEstimator,
 					L::Target: Logger,
@@ -858,13 +854,11 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 				}
 			}
 		}
-		let mut broadcasted = false;
 		for (_, claim_material) in bump_candidates.iter_mut() {
 			if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &claim_material, &*fee_estimator, &*logger) {
 				claim_material.height_timer = new_timer;
 				claim_material.feerate_previous = new_feerate;
 				broadcaster.broadcast_transaction(&bump_tx);
-				broadcasted = true;
 			}
 		}
 		for (ancestor_claim_txid, claim_material) in bump_candidates.drain() {
@@ -881,7 +875,6 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		for req in remove_request {
 			self.pending_claim_requests.remove(&req);
 		}
-		broadcasted
 	}
 
 	pub(super) fn provide_latest_local_tx(&mut self, tx: LocalCommitmentTransaction) -> Result<(), ()> {