Review comments

Author: wvanlint

lightning/src/ln/channel.rs

@@ -3652,17 +3652,17 @@ impl<SP: Deref> Channel<SP> where
 	/// resent.
 	/// No further message handling calls may be made until a channel_reestablish dance has
 	/// completed.
-	pub fn remove_uncommitted_htlcs_and_mark_paused<L: Deref>(&mut self, logger: &L)  where L::Target: Logger {
+	/// May return `Err(())`, which implies [`Self::force_shutdown`] should be called immediately.
+	pub fn remove_uncommitted_htlcs_and_mark_paused<L: Deref>(&mut self, logger: &L) -> Result<(), ()> where L::Target: Logger {
 		assert_eq!(self.context.channel_state & ChannelState::ShutdownComplete as u32, 0);
 		if self.context.channel_state & !STATE_FLAGS < ChannelState::FundingSent as u32 {
-			self.context.channel_state = ChannelState::ShutdownComplete as u32;
-			return;
+			return Err(());
 		}
 
 		if self.context.channel_state & (ChannelState::PeerDisconnected as u32) == (ChannelState::PeerDisconnected as u32) {
 			// While the below code should be idempotent, it's simpler to just return early, as
 			// redundant disconnect events can fire, though they should be rare.
-			return;
+			return Ok(());
 		}
 
 		if self.context.announcement_sigs_state == AnnouncementSigsState::MessageSent || self.context.announcement_sigs_state == AnnouncementSigsState::Committed {
@@ -3723,6 +3723,7 @@ impl<SP: Deref> Channel<SP> where
 
 		self.context.channel_state |= ChannelState::PeerDisconnected as u32;
 		log_trace!(logger, "Peer disconnection resulted in {} remote-announced HTLC drops on channel {}", inbound_drop_count, &self.context.channel_id());
+		Ok(())
 	}
 
 	/// Indicates that a ChannelMonitor update is in progress and has not yet been fully persisted.
@@ -7940,15 +7941,15 @@ mod tests {
 
 		// Now disconnect the two nodes and check that the commitment point in
 		// Node B's channel_reestablish message is sane.
-		node_b_chan.remove_uncommitted_htlcs_and_mark_paused(&&logger);
+		assert!(node_b_chan.remove_uncommitted_htlcs_and_mark_paused(&&logger).is_ok());
 		let msg = node_b_chan.get_channel_reestablish(&&logger);
 		assert_eq!(msg.next_local_commitment_number, 1); // now called next_commitment_number
 		assert_eq!(msg.next_remote_commitment_number, 0); // now called next_revocation_number
 		assert_eq!(msg.your_last_per_commitment_secret, [0; 32]);
 
 		// Check that the commitment point in Node A's channel_reestablish message
 		// is sane.
-		node_a_chan.remove_uncommitted_htlcs_and_mark_paused(&&logger);
+		assert!(node_a_chan.remove_uncommitted_htlcs_and_mark_paused(&&logger).is_ok());
 		let msg = node_a_chan.get_channel_reestablish(&&logger);
 		assert_eq!(msg.next_local_commitment_number, 1); // now called next_commitment_number
 		assert_eq!(msg.next_remote_commitment_number, 0); // now called next_revocation_number

lightning/src/ln/channelmanager.rs

@@ -2047,8 +2047,12 @@ macro_rules! handle_monitor_update_completion {
 				));
 				if let Some(channel_state) = channel_state {
 					channel_state.2 = true;
+				} else {
+					debug_assert!(false, "Missing channel batch state for channel which completed initial monitor update");
 				}
 				batch_completed = batch_state.iter().all(|(_, _, completed)| *completed);
+			} else {
+				debug_assert!(false, "Missing batch state for channel which completed initial monitor update");
 			}
 
 			// When all channels in a batched funding transaction have become ready, it is not necessary
@@ -2707,7 +2711,8 @@ where
 			let mut funding_batch_states = self.funding_batch_states.lock().unwrap();
 			let affected_channels = funding_batch_states.remove(&txid).into_iter().flatten();
 			let per_peer_state = self.per_peer_state.read().unwrap();
-			for (channel_id, counterparty_node_id, _state) in affected_channels {
+			let mut has_uncompleted_channel = None;
+			for (channel_id, counterparty_node_id, state) in affected_channels {
 				if let Some(peer_state_mutex) = per_peer_state.get(&counterparty_node_id) {
 					let mut peer_state = peer_state_mutex.lock().unwrap();
 					if let Some(mut chan) = peer_state.channel_by_id.remove(&channel_id) {
@@ -2716,7 +2721,12 @@ where
 						shutdown_results.push(chan.context_mut().force_shutdown(false));
 					}
 				}
+				has_uncompleted_channel = Some(has_uncompleted_channel.map_or(!state, |v| v || !state));
 			}
+			debug_assert!(
+				has_uncompleted_channel.unwrap_or(true),
+				"Closing a batch where all channels have completed initial monitor update",
+			);
 		}
 		for shutdown_result in shutdown_results.drain(..) {
 			self.finish_close_channel(shutdown_result);
@@ -3828,7 +3838,7 @@ where
 
 	/// Call this upon creation of a batch funding transaction for the given channels.
 	///
-	/// Return values are identical to `funding_transaction_generated`, respective to
+	/// Return values are identical to [`Self::funding_transaction_generated`], respective to
 	/// each individual channel and transaction output.
 	///
 	/// Do NOT broadcast the funding transaction yourself. This batch funding transcaction
@@ -3867,7 +3877,6 @@ where
 			}
 		}
 
-
 		let txid = funding_transaction.txid();
 		let is_batch_funding = temporary_channels.len() > 1;
 		let mut funding_batch_states = if is_batch_funding {
@@ -3876,8 +3885,15 @@ where
 			None
 		};
 		let mut funding_batch_state = funding_batch_states.as_mut().and_then(|states| {
-			states.insert(txid, Vec::new());
-			states.get_mut(&txid)
+			if states.contains_key(&txid) {
+				result = result.clone().and(Err(APIError::APIMisuseError {
+					err: "Batch funding transaction with the same txid already exists".to_owned()
+				}));
+				None
+			} else {
+				states.insert(txid, Vec::new());
+				states.get_mut(&txid)
+			}
 		});
 		for (channel_idx, &(temporary_channel_id, counterparty_node_id)) in temporary_channels.iter().enumerate() {
 			result = result.and_then(|_| self.funding_transaction_generated_intern(
@@ -7995,35 +8011,24 @@ where
 				peer_state.channel_by_id.retain(|_, phase| {
 					let context = match phase {
 						ChannelPhase::Funded(chan) => {
-							if !chan.context.is_funding_broadcast() {
-								update_maps_on_chan_removal!(self, &chan.context);
-								self.issue_channel_close_events(&chan.context, ClosureReason::DisconnectedPeer);
-								// It is possible to have persisted the monitor upon funding_signed
-								// but not have broadcast the transaction, especially for batch funding.
-								// The monitor should be moved to the correct state.
-								failed_channels.push(chan.context.force_shutdown(false));
-								return false;
-							} else {
-								chan.remove_uncommitted_htlcs_and_mark_paused(&self.logger);
+							if chan.remove_uncommitted_htlcs_and_mark_paused(&self.logger).is_ok() {
 								// We only retain funded channels that are not shutdown.
-								if !chan.is_shutdown() {
-									return true;
-								}
+								return true;
 							}
-							&chan.context
+							&mut chan.context
 						},
 						// Unfunded channels will always be removed.
 						ChannelPhase::UnfundedOutboundV1(chan) => {
-							&chan.context
+							&mut chan.context
 						},
 						ChannelPhase::UnfundedInboundV1(chan) => {
-							&chan.context
+							&mut chan.context
 						},
 					};
 					// Clean up for removal.
 					update_maps_on_chan_removal!(self, &context);
 					self.issue_channel_close_events(&context, ClosureReason::DisconnectedPeer);
-					failed_channels.push((None, Vec::new(), None));
+					failed_channels.push(context.force_shutdown(false));
 					false
 				});
 				// Note that we don't bother generating any events for pre-accept channels -
@@ -9262,7 +9267,10 @@ where
 						log_error!(args.logger, " The ChannelMonitor for channel {} is at counterparty commitment transaction number {} but the ChannelManager is at counterparty commitment transaction number {}.",
 							&channel.context.channel_id(), monitor.get_cur_counterparty_commitment_number(), channel.get_cur_counterparty_commitment_transaction_number());
 					}
-					let (monitor_update, mut new_failed_htlcs, _batch_funding_txid) = channel.context.force_shutdown(true);
+					let (monitor_update, mut new_failed_htlcs, batch_funding_txid) = channel.context.force_shutdown(true);
+					if batch_funding_txid.is_some() {
+						return Err(DecodeError::InvalidValue);
+					}
 					if let Some((counterparty_node_id, funding_txo, update)) = monitor_update {
 						close_background_events.push(BackgroundEvent::MonitorUpdateRegeneratedOnStartup {
 							counterparty_node_id, funding_txo, update

lightning/src/ln/functional_tests.rs

@@ -10514,10 +10514,10 @@ fn test_disconnect_in_funding_batch() {
 			..
 		} if channel_id == &channel_id_2
 	)));
-	assert!(events.iter().any(|e| matches!(
+	assert_eq!(events.iter().filter(|e| matches!(
 		e,
 		Event::DiscardFunding { .. },
-	)));
+	)).count(), 2);
 
 	// The monitor should become closed.
 	check_added_monitors(&nodes[0], 1);
@@ -10558,19 +10558,32 @@ fn test_batch_funding_close_after_funding_signed() {
 	nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &funding_signed_msg);
 	check_added_monitors(&nodes[0], 1);
 
+	// Go through the funding_created and funding_signed flow with node 2.
+	nodes[2].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created_msgs[1]);
+	check_added_monitors(&nodes[2], 1);
+	expect_channel_pending_event(&nodes[2], &nodes[0].node.get_our_node_id());
+
+	let funding_signed_msg = get_event_msg!(nodes[2], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id());
+	chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+	nodes[0].node.handle_funding_signed(&nodes[2].node.get_our_node_id(), &funding_signed_msg);
+	check_added_monitors(&nodes[0], 1);
+
 	// The transaction should not have been broadcast before all channels are ready.
 	assert_eq!(nodes[0].tx_broadcaster.txn_broadcast().len(), 0);
 
-	// Force-close the channel for which we've received funding_signed.
+	// Force-close the channel for which we've completed the initial monitor.
 	let channel_id_1 = OutPoint { txid: tx.txid(), index: 0 }.to_channel_id();
 	let channel_id_2 = OutPoint { txid: tx.txid(), index: 1 }.to_channel_id();
 	nodes[0].node.force_close_broadcasting_latest_txn(&channel_id_1, &nodes[1].node.get_our_node_id()).unwrap();
-	check_added_monitors(&nodes[0], 1);
+	check_added_monitors(&nodes[0], 2);
 	{
 		let mut monitor_updates = nodes[0].chain_monitor.monitor_updates.lock().unwrap();
-		let monitor_updates = monitor_updates.get(&channel_id_1).unwrap();
-		assert_eq!(monitor_updates.len(), 1);
-		assert_eq!(monitor_updates[0].update_id, CLOSED_CHANNEL_UPDATE_ID);
+		let monitor_updates_1 = monitor_updates.get(&channel_id_1).unwrap();
+		assert_eq!(monitor_updates_1.len(), 1);
+		assert_eq!(monitor_updates_1[0].update_id, CLOSED_CHANNEL_UPDATE_ID);
+		let monitor_updates_2 = monitor_updates.get(&channel_id_2).unwrap();
+		assert_eq!(monitor_updates_2.len(), 1);
+		assert_eq!(monitor_updates_2[0].update_id, CLOSED_CHANNEL_UPDATE_ID);
 	}
 	let msg_events = nodes[0].node.get_and_clear_pending_msg_events();
 	match msg_events[0] {
@@ -10604,10 +10617,10 @@ fn test_batch_funding_close_after_funding_signed() {
 			..
 		} if channel_id == &channel_id_2
 	)));
-	assert!(events.iter().any(|e| matches!(
+	assert_eq!(events.iter().filter(|e| matches!(
 		e,
 		Event::DiscardFunding { .. },
-	)));
+	)).count(), 2);
 
 	// Ensure the channels don't exist anymore.
 	assert!(nodes[0].node.list_channels().is_empty());