Send initial closing_signed message asynchronously and handle errs

When we added the support for external signing, many of the
signing functions were allowed to return an error, closing the
channel in such a case. `sign_closing_transaction` is one such
function which can now return an error, except instead of handling
it properly we'd simply never send a `closing_signed` message,
hanging the channel until users intervene and force-close it.

Piping the channel-closing error back through the various callsites
(several of which already have pending results by the time they
call `maybe_propose_first_closing_signed`) may be rather
complicated, so instead we simply attempt to propose the initial
`closing_signed` in `get_and_clear_pending_msg_events` like we do
for holding-cell freeing.

This simplifies a few function interfaces and has no impact on
behavior, aside from a few message-ordering edge-cases, as seen in
the two small test changes required.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -141,7 +141,7 @@ fn test_monitor_and_persister_update_fail() {
 	assert_eq!(updates.update_fulfill_htlcs.len(), 1);
 	nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fulfill_htlcs[0]);
 	if let Some(ref mut channel) = nodes[0].node.channel_state.lock().unwrap().by_id.get_mut(&chan.2) {
-		if let Ok((_, _, _, update)) = channel.commitment_signed(&updates.commitment_signed, &node_cfgs[0].fee_estimator, &node_cfgs[0].logger) {
+		if let Ok((_, _, update)) = channel.commitment_signed(&updates.commitment_signed, &node_cfgs[0].logger) {
 			// Check that even though the persister is returning a TemporaryFailure,
 			// because the update is bogus, ultimately the error that's returned
 			// should be a PermanentFailure.

lightning/src/ln/channel.rs

@@ -2169,9 +2169,8 @@ impl<Signer: Sign> Channel<Signer> {
 		Ok(())
 	}
 
-	pub fn commitment_signed<F: Deref, L: Deref>(&mut self, msg: &msgs::CommitmentSigned, fee_estimator: &F, logger: &L) -> Result<(msgs::RevokeAndACK, Option<msgs::CommitmentSigned>, Option<msgs::ClosingSigned>, ChannelMonitorUpdate), (Option<ChannelMonitorUpdate>, ChannelError)>
-	where F::Target: FeeEstimator,
-				L::Target: Logger
+	pub fn commitment_signed<L: Deref>(&mut self, msg: &msgs::CommitmentSigned, logger: &L) -> Result<(msgs::RevokeAndACK, Option<msgs::CommitmentSigned>, ChannelMonitorUpdate), (Option<ChannelMonitorUpdate>, ChannelError)>
+		where L::Target: Logger
 	{
 		if (self.channel_state & (ChannelState::ChannelFunded as u32)) != (ChannelState::ChannelFunded as u32) {
 			return Err((None, ChannelError::Close("Got commitment signed message when channel was not in an operational state".to_owned())));
@@ -2346,12 +2345,10 @@ impl<Signer: Sign> Channel<Signer> {
 			}
 			log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updated HTLC state but awaiting a monitor update resolution to reply.",
 				log_bytes!(self.channel_id));
-			// TODO: Call maybe_propose_first_closing_signed on restoration (or call it here and
-			// re-send the message on restoration)
 			return Err((Some(monitor_update), ChannelError::Ignore("Previous monitor update failure prevented generation of RAA".to_owned())));
 		}
 
-		let (commitment_signed, closing_signed) = if need_commitment && (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32)) == 0 {
+		let commitment_signed = if need_commitment && (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32)) == 0 {
 			// If we're AwaitingRemoteRevoke we can't send a new commitment here, but that's ok -
 			// we'll send one right away when we get the revoke_and_ack when we
 			// free_holding_cell_htlcs().
@@ -2360,10 +2357,8 @@ impl<Signer: Sign> Channel<Signer> {
 			// strictly increasing by one, so decrement it here.
 			self.latest_monitor_update_id = monitor_update.update_id;
 			monitor_update.updates.append(&mut additional_update.updates);
-			(Some(msg), None)
-		} else if !need_commitment {
-			(None, self.maybe_propose_first_closing_signed(fee_estimator))
-		} else { (None, None) };
+			Some(msg)
+		} else { None };
 
 		log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updating HTLC state and responding with{} a revoke_and_ack.",
 			log_bytes!(self.channel_id()), if commitment_signed.is_some() { " our own commitment_signed and" } else { "" });
@@ -2372,7 +2367,7 @@ impl<Signer: Sign> Channel<Signer> {
 			channel_id: self.channel_id,
 			per_commitment_secret,
 			next_per_commitment_point,
-		}, commitment_signed, closing_signed, monitor_update))
+		}, commitment_signed, monitor_update))
 	}
 
 	/// Public version of the below, checking relevant preconditions first.
@@ -2504,9 +2499,8 @@ impl<Signer: Sign> Channel<Signer> {
 	/// waiting on this revoke_and_ack. The generation of this new commitment_signed may also fail,
 	/// generating an appropriate error *after* the channel state has been updated based on the
 	/// revoke_and_ack message.
-	pub fn revoke_and_ack<F: Deref, L: Deref>(&mut self, msg: &msgs::RevokeAndACK, fee_estimator: &F, logger: &L) -> Result<(Option<msgs::CommitmentUpdate>, Vec<(PendingHTLCInfo, u64)>, Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, Option<msgs::ClosingSigned>, ChannelMonitorUpdate, Vec<(HTLCSource, PaymentHash)>), ChannelError>
-		where F::Target: FeeEstimator,
-					L::Target: Logger,
+	pub fn revoke_and_ack<L: Deref>(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result<(Option<msgs::CommitmentUpdate>, Vec<(PendingHTLCInfo, u64)>, Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, ChannelMonitorUpdate, Vec<(HTLCSource, PaymentHash)>), ChannelError>
+		where L::Target: Logger,
 	{
 		if (self.channel_state & (ChannelState::ChannelFunded as u32)) != (ChannelState::ChannelFunded as u32) {
 			return Err(ChannelError::Close("Got revoke/ACK message when channel was not in an operational state".to_owned()));
@@ -2686,7 +2680,7 @@ impl<Signer: Sign> Channel<Signer> {
 			self.monitor_pending_forwards.append(&mut to_forward_infos);
 			self.monitor_pending_failures.append(&mut revoked_htlcs);
 			log_debug!(logger, "Received a valid revoke_and_ack for channel {} but awaiting a monitor update resolution to reply.", log_bytes!(self.channel_id()));
-			return Ok((None, Vec::new(), Vec::new(), None, monitor_update, Vec::new()))
+			return Ok((None, Vec::new(), Vec::new(), monitor_update, Vec::new()))
 		}
 
 		match self.free_holding_cell_htlcs(logger)? {
@@ -2705,7 +2699,7 @@ impl<Signer: Sign> Channel<Signer> {
 				self.latest_monitor_update_id = monitor_update.update_id;
 				monitor_update.updates.append(&mut additional_update.updates);
 
-				Ok((Some(commitment_update), to_forward_infos, revoked_htlcs, None, monitor_update, htlcs_to_fail))
+				Ok((Some(commitment_update), to_forward_infos, revoked_htlcs, monitor_update, htlcs_to_fail))
 			},
 			(None, htlcs_to_fail) => {
 				if require_commitment {
@@ -2725,14 +2719,13 @@ impl<Signer: Sign> Channel<Signer> {
 						update_fail_malformed_htlcs,
 						update_fee: None,
 						commitment_signed
-					}), to_forward_infos, revoked_htlcs, None, monitor_update, htlcs_to_fail))
+					}), to_forward_infos, revoked_htlcs, monitor_update, htlcs_to_fail))
 				} else {
 					log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.channel_id()));
-					Ok((None, to_forward_infos, revoked_htlcs, self.maybe_propose_first_closing_signed(fee_estimator), monitor_update, htlcs_to_fail))
+					Ok((None, to_forward_infos, revoked_htlcs, monitor_update, htlcs_to_fail))
 				}
 			}
 		}
-
 	}
 
 	/// Adds a pending update to this channel. See the doc for send_htlc for
@@ -3119,13 +3112,17 @@ impl<Signer: Sign> Channel<Signer> {
 		}
 	}
 
-	fn maybe_propose_first_closing_signed<F: Deref>(&mut self, fee_estimator: &F) -> Option<msgs::ClosingSigned>
-		where F::Target: FeeEstimator
+	pub fn maybe_propose_first_closing_signed<F: Deref, L: Deref>(&mut self, fee_estimator: &F, logger: &L)
+		-> Result<Option<msgs::ClosingSigned>, ChannelError>
+		where F::Target: FeeEstimator, L::Target: Logger
 	{
 		if !self.is_outbound() || !self.pending_inbound_htlcs.is_empty() || !self.pending_outbound_htlcs.is_empty() ||
-				self.channel_state & (BOTH_SIDES_SHUTDOWN_MASK | ChannelState::AwaitingRemoteRevoke as u32) != BOTH_SIDES_SHUTDOWN_MASK ||
+				self.channel_state &
+					(BOTH_SIDES_SHUTDOWN_MASK | ChannelState::AwaitingRemoteRevoke as u32 |
+					 ChannelState::PeerDisconnected as u32 | ChannelState::MonitorUpdateFailed as u32)
+				!= BOTH_SIDES_SHUTDOWN_MASK ||
 				self.last_sent_closing_fee.is_some() || self.pending_update_fee.is_some() {
-			return None;
+			return Ok(None);
 		}
 
 		let mut proposed_feerate = fee_estimator.get_est_sat_per_1000_weight(ConfirmationTarget::Background);
@@ -3134,26 +3131,24 @@ impl<Signer: Sign> Channel<Signer> {
 		}
 		let tx_weight = self.get_closing_transaction_weight(Some(&self.get_closing_scriptpubkey()), Some(self.counterparty_shutdown_scriptpubkey.as_ref().unwrap()));
 		let proposed_total_fee_satoshis = proposed_feerate as u64 * tx_weight / 1000;
+		log_trace!(logger, "Proposing initial closing signed for our counterparty with a feerate of {} sat/kWeight (= {} sats)",
+			proposed_feerate, proposed_total_fee_satoshis);
 
 		let (closing_tx, total_fee_satoshis) = self.build_closing_transaction(proposed_total_fee_satoshis, false);
 		let sig = self.holder_signer
 			.sign_closing_transaction(&closing_tx, &self.secp_ctx)
-			.ok();
-		assert!(closing_tx.get_weight() as u64 <= tx_weight);
-		if sig.is_none() { return None; }
+			.map_err(|()| ChannelError::Close("Failed to get signature for closing transaction.".to_owned()))?;
 
-		self.last_sent_closing_fee = Some((proposed_feerate, total_fee_satoshis, sig.clone().unwrap()));
-		Some(msgs::ClosingSigned {
+		self.last_sent_closing_fee = Some((proposed_feerate, total_fee_satoshis, sig.clone()));
+		Ok(Some(msgs::ClosingSigned {
 			channel_id: self.channel_id,
 			fee_satoshis: total_fee_satoshis,
-			signature: sig.unwrap(),
+			signature: sig,
 			fee_range: None,
-		})
+		}))
 	}
 
-	pub fn shutdown<F: Deref>(&mut self, fee_estimator: &F, their_features: &InitFeatures, msg: &msgs::Shutdown) -> Result<(Option<msgs::Shutdown>, Option<msgs::ClosingSigned>, Vec<(HTLCSource, PaymentHash)>), ChannelError>
-		where F::Target: FeeEstimator
-	{
+	pub fn shutdown(&mut self, their_features: &InitFeatures, msg: &msgs::Shutdown) -> Result<(Option<msgs::Shutdown>, Vec<(HTLCSource, PaymentHash)>), ChannelError> {
 		if self.channel_state & (ChannelState::PeerDisconnected as u32) == ChannelState::PeerDisconnected as u32 {
 			return Err(ChannelError::Close("Peer sent shutdown when we needed a channel_reestablish".to_owned()));
 		}
@@ -3217,7 +3212,7 @@ impl<Signer: Sign> Channel<Signer> {
 		self.channel_state |= ChannelState::LocalShutdownSent as u32;
 		self.update_time_counter += 1;
 
-		Ok((shutdown, self.maybe_propose_first_closing_signed(fee_estimator), dropped_outbound_htlcs))
+		Ok((shutdown, dropped_outbound_htlcs))
 	}
 
 	fn build_signed_closing_transaction(&self, tx: &mut Transaction, counterparty_sig: &Signature, sig: &Signature) {

lightning/src/ln/channelmanager.rs

@@ -3058,19 +3058,13 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 							log_bytes!(msg.channel_id),
 							if chan_entry.get().we_want_shutdown() { " after we initiated shutdown" } else { "" });
 					}
-					let (shutdown, closing_signed, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&self.fee_estimator, &their_features, &msg), channel_state, chan_entry);
+					let (shutdown, dropped_htlcs) = try_chan_entry!(self, chan_entry.get_mut().shutdown(&their_features, &msg), channel_state, chan_entry);
 					if let Some(msg) = shutdown {
 						channel_state.pending_msg_events.push(events::MessageSendEvent::SendShutdown {
 							node_id: counterparty_node_id.clone(),
 							msg,
 						});
 					}
-					if let Some(msg) = closing_signed {
-						channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
-							node_id: counterparty_node_id.clone(),
-							msg,
-						});
-					}
 					if chan_entry.get().is_shutdown() {
 						if let Some(short_id) = chan_entry.get().get_short_channel_id() {
 							channel_state.short_to_id.remove(&short_id);
@@ -3266,8 +3260,8 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 				if chan.get().get_counterparty_node_id() != *counterparty_node_id {
 					return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id));
 				}
-				let (revoke_and_ack, commitment_signed, closing_signed, monitor_update) =
-					match chan.get_mut().commitment_signed(&msg, &self.fee_estimator, &self.logger) {
+				let (revoke_and_ack, commitment_signed, monitor_update) =
+					match chan.get_mut().commitment_signed(&msg, &self.logger) {
 						Err((None, e)) => try_chan_entry!(self, Err(e), channel_state, chan),
 						Err((Some(update), e)) => {
 							assert!(chan.get().is_awaiting_monitor_update());
@@ -3279,7 +3273,6 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 					};
 				if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
 					return_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::RevokeAndACKFirst, true, commitment_signed.is_some());
-					//TODO: Rebroadcast closing_signed if present on monitor update restoration
 				}
 				channel_state.pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK {
 					node_id: counterparty_node_id.clone(),
@@ -3298,12 +3291,6 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 						},
 					});
 				}
-				if let Some(msg) = closing_signed {
-					channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
-						node_id: counterparty_node_id.clone(),
-						msg,
-					});
-				}
 				Ok(())
 			},
 			hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
@@ -3358,12 +3345,12 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 						break Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id));
 					}
 					let was_frozen_for_monitor = chan.get().is_awaiting_monitor_update();
-					let (commitment_update, pending_forwards, pending_failures, closing_signed, monitor_update, htlcs_to_fail_in) =
-						break_chan_entry!(self, chan.get_mut().revoke_and_ack(&msg, &self.fee_estimator, &self.logger), channel_state, chan);
+					let (commitment_update, pending_forwards, pending_failures, monitor_update, htlcs_to_fail_in) =
+						break_chan_entry!(self, chan.get_mut().revoke_and_ack(&msg, &self.logger), channel_state, chan);
 					htlcs_to_fail = htlcs_to_fail_in;
 					if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
 						if was_frozen_for_monitor {
-							assert!(commitment_update.is_none() && closing_signed.is_none() && pending_forwards.is_empty() && pending_failures.is_empty());
+							assert!(commitment_update.is_none() && pending_forwards.is_empty() && pending_failures.is_empty());
 							break Err(MsgHandleErrInternal::ignore_no_close("Previous monitor update failure prevented responses to RAA".to_owned()));
 						} else {
 							if let Err(e) = handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, commitment_update.is_some(), pending_forwards, pending_failures) {
@@ -3377,12 +3364,6 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 							updates,
 						});
 					}
-					if let Some(msg) = closing_signed {
-						channel_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
-							node_id: counterparty_node_id.clone(),
-							msg,
-						});
-					}
 					break Ok((pending_forwards, pending_failures, chan.get().get_short_channel_id().expect("RAA should only work on a short-id-available channel"), chan.get().get_funding_txo().unwrap()))
 				},
 				hash_map::Entry::Vacant(_) => break Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
@@ -3682,7 +3663,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 			});
 		}
 
-		let has_update = has_monitor_update || !failed_htlcs.is_empty();
+		let has_update = has_monitor_update || !failed_htlcs.is_empty() || !handle_errors.is_empty();
 		for (failures, channel_id) in failed_htlcs.drain(..) {
 			self.fail_holding_cell_htlcs(failures, channel_id);
 		}
@@ -3694,6 +3675,46 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 		has_update
 	}
 
+	/// Check whether any channels have finished removing all pending updates after a shutdown
+	/// exchange and can now send a closing_signed.
+	/// Returns whether any closing_signed messages were generated.
+	fn maybe_generate_initial_closing_signed(&self) -> bool {
+		let mut handle_errors: Vec<(PublicKey, Result<(), _>)> = Vec::new();
+		let mut has_update = false;
+		{
+			let mut channel_state_lock = self.channel_state.lock().unwrap();
+			let channel_state = &mut *channel_state_lock;
+			let by_id = &mut channel_state.by_id;
+			let short_to_id = &mut channel_state.short_to_id;
+			let pending_msg_events = &mut channel_state.pending_msg_events;
+
+			by_id.retain(|channel_id, chan| {
+				match chan.maybe_propose_first_closing_signed(&self.fee_estimator, &self.logger) {
+					Ok(Some(msg)) => {
+						pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
+							node_id: chan.get_counterparty_node_id(), msg,
+						});
+						has_update = true;
+						true
+					},
+					Ok(None) => true,
+					Err(e) => {
+						has_update = true;
+						let (close_channel, res) = convert_chan_err!(self, e, short_to_id, chan, channel_id);
+						handle_errors.push((chan.get_counterparty_node_id(), Err(res)));
+						!close_channel
+					}
+				}
+			});
+		}
+
+		for (counterparty_node_id, err) in handle_errors.drain(..) {
+			let _ = handle_error!(self, err, counterparty_node_id);
+		}
+
+		has_update
+	}
+
 	/// Handle a list of channel failures during a block_connected or block_disconnected call,
 	/// pushing the channel monitor update (if any) to the background events queue and removing the
 	/// Channel object.
@@ -3847,6 +3868,9 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> MessageSend
 			if self.check_free_holding_cells() {
 				result = NotifyOption::DoPersist;
 			}
+			if self.maybe_generate_initial_closing_signed() {
+				result = NotifyOption::DoPersist;
+			}
 
 			let mut pending_events = Vec::new();
 			let mut channel_state = self.channel_state.lock().unwrap();