Allow sending closing tx signatures asynchronously

alecchendev · alecchendev · commit b38ae325f5be · 2024-07-03T13:22:43.000-07:00
diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
@@ -909,6 +909,8 @@ pub(super) struct SignerResumeUpdates {
 	pub commitment_update: Option<msgs::CommitmentUpdate>,
 	pub funding_signed: Option<msgs::FundingSigned>,
 	pub channel_ready: Option<msgs::ChannelReady>,
+	pub closing_signed: Option<msgs::ClosingSigned>,
+	pub signed_closing_tx: Option<Transaction>,
 }
 
 /// The return value of `channel_reestablish`
@@ -1227,6 +1229,9 @@ pub(super) struct ChannelContext<SP: Deref> where SP::Target: SignerProvider {
 	/// [`msgs::FundingCreated`] or [`msgs::FundingSigned`] depending on if this channel is
 	/// outbound or inbound.
 	signer_pending_funding: bool,
+	/// If we attempted to sign a cooperative close transaction but the signer wasn't ready, then this
+	/// will be set to `true`.
+	signer_pending_closing: bool,
 
 	// pending_update_fee is filled when sending and receiving update_fee.
 	//
@@ -1258,7 +1263,9 @@ pub(super) struct ChannelContext<SP: Deref> where SP::Target: SignerProvider {
 	/// Max to_local and to_remote outputs in a remote-generated commitment transaction
 	counterparty_max_commitment_tx_output: Mutex<(u64, u64)>,
 
-	last_sent_closing_fee: Option<(u64, Signature)>, // (fee, holder_sig)
+	// (fee, skip_remote_output, fee_range, holder_sig)
+	last_sent_closing_fee: Option<(u64, bool, ClosingSignedFeeRange, Option<Signature>)>,
+	last_received_closing_sig: Option<Signature>,
 	target_closing_feerate_sats_per_kw: Option<u32>,
 
 	/// If our counterparty sent us a closing_signed while we were waiting for a `ChannelMonitor`
@@ -1686,6 +1693,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider  {
 
 			signer_pending_commitment_update: false,
 			signer_pending_funding: false,
+			signer_pending_closing: false,
 
 
 			#[cfg(debug_assertions)]
@@ -1694,6 +1702,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider  {
 			counterparty_max_commitment_tx_output: Mutex::new((value_to_self_msat, (channel_value_satoshis * 1000 - msg_push_msat).saturating_sub(value_to_self_msat))),
 
 			last_sent_closing_fee: None,
+			last_received_closing_sig: None,
 			pending_counterparty_closing_signed: None,
 			expecting_peer_commitment_signed: false,
 			closing_fee_limits: None,
@@ -1911,6 +1920,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider  {
 
 			signer_pending_commitment_update: false,
 			signer_pending_funding: false,
+			signer_pending_closing: false,
 
 			// We'll add our counterparty's `funding_satoshis` to these max commitment output assertions
 			// when we receive `accept_channel2`.
@@ -1920,6 +1930,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider  {
 			counterparty_max_commitment_tx_output: Mutex::new((channel_value_satoshis * 1000 - push_msat, push_msat)),
 
 			last_sent_closing_fee: None,
+			last_received_closing_sig: None,
 			pending_counterparty_closing_signed: None,
 			expecting_peer_commitment_signed: false,
 			closing_fee_limits: None,
@@ -5387,15 +5398,35 @@ impl<SP: Deref> Channel<SP> where
 			self.check_get_channel_ready(0, logger)
 		} else { None };
 
-		log_trace!(logger, "Signer unblocked with {} commitment_update, {} funding_signed and {} channel_ready",
+		let (closing_signed, signed_closing_tx) = if self.context.signer_pending_closing {
+			debug_assert!(self.context.last_sent_closing_fee.is_some());
+			if let Some((fee, skip_remote_output, fee_range, holder_sig)) = self.context.last_sent_closing_fee.clone() {
+				debug_assert!(holder_sig.is_none());
+				log_trace!(logger, "Attempting to generate pending closing_signed...");
+				let (closing_tx, fee) = self.build_closing_transaction(fee, skip_remote_output);
+				let closing_signed = self.get_closing_signed_msg(&closing_tx, skip_remote_output,
+					fee, fee_range.min_fee_satoshis, fee_range.max_fee_satoshis, logger);
+				let signed_tx = if let (Some(ClosingSigned { signature, .. }), Some(counterparty_sig)) =
+					(closing_signed.as_ref(), self.context.last_received_closing_sig) {
+					Some(self.build_signed_closing_transaction(&closing_tx, &counterparty_sig, signature))
+				} else { None };
+				(closing_signed, signed_tx)
+			} else { (None, None) }
+		} else { (None, None) };
+
+		log_trace!(logger, "Signer unblocked with {} commitment_update, {} funding_signed, {} channel_ready, {} closing_signed, and {} signed_closing_tx",
 			if commitment_update.is_some() { "a" } else { "no" },
 			if funding_signed.is_some() { "a" } else { "no" },
-			if channel_ready.is_some() { "a" } else { "no" });
+			if channel_ready.is_some() { "a" } else { "no" },
+			if closing_signed.is_some() { "a" } else { "no" },
+			if signed_closing_tx.is_some() { "a" } else { "no" });
 
 		SignerResumeUpdates {
 			commitment_update,
 			funding_signed,
 			channel_ready,
+			closing_signed,
+			signed_closing_tx,
 		}
 	}
 
@@ -5801,9 +5832,6 @@ impl<SP: Deref> Channel<SP> where
 			our_min_fee, our_max_fee, total_fee_satoshis);
 
 		let closing_signed = self.get_closing_signed_msg(&closing_tx, false, total_fee_satoshis, our_min_fee, our_max_fee, logger);
-		if closing_signed.is_none() {
-			return Err(ChannelError::close("Failed to get signature for closing transaction.".to_owned()));
-		}
 		Ok((closing_signed, None, None))
 	}
 
@@ -5957,26 +5985,26 @@ impl<SP: Deref> Channel<SP> where
 	) -> Option<msgs::ClosingSigned>
 		where L::Target: Logger
 	{
-		match &self.context.holder_signer {
-			ChannelSignerType::Ecdsa(ecdsa) => {
-				let fee_range = msgs::ClosingSignedFeeRange {
-					min_fee_satoshis,
-					max_fee_satoshis,
-				};
-				let sig = ecdsa.sign_closing_transaction(closing_tx, &self.context.secp_ctx).ok()?;
-
-				self.context.last_sent_closing_fee = Some((fee_satoshis, sig.clone()));
-				Some(msgs::ClosingSigned {
-					channel_id: self.context.channel_id,
-					fee_satoshis,
-					signature: sig,
-					fee_range: Some(fee_range),
-				})
-			},
+		let sig = match &self.context.holder_signer {
+			ChannelSignerType::Ecdsa(ecdsa) => ecdsa.sign_closing_transaction(closing_tx, &self.context.secp_ctx).ok(),
 			// TODO (taproot|arik)
 			#[cfg(taproot)]
 			_ => todo!()
+		};
+		if sig.is_none() {
+			log_trace!(logger, "Closing transaction signature unavailable, waiting on signer");
+			self.context.signer_pending_closing = true;
+		} else {
+			self.context.signer_pending_closing = false;
 		}
+		let fee_range = msgs::ClosingSignedFeeRange { min_fee_satoshis, max_fee_satoshis };
+		self.context.last_sent_closing_fee = Some((fee_satoshis, skip_remote_output, fee_range.clone(), sig.clone()));
+		sig.map(|signature| msgs::ClosingSigned {
+			channel_id: self.context.channel_id,
+			fee_satoshis,
+			signature,
+			fee_range: Some(fee_range),
+		})
 	}
 
 	pub fn closing_signed<F: Deref, L: Deref>(
@@ -6039,7 +6067,7 @@ impl<SP: Deref> Channel<SP> where
 		};
 
 		assert!(self.context.shutdown_scriptpubkey.is_some());
-		if let Some((last_fee, sig)) = self.context.last_sent_closing_fee {
+		if let Some((last_fee, _, _, Some(sig))) = self.context.last_sent_closing_fee {
 			if last_fee == msg.fee_satoshis {
 				let shutdown_result = ShutdownResult {
 					closure_reason,
@@ -6072,9 +6100,6 @@ impl<SP: Deref> Channel<SP> where
 				};
 
 				let closing_signed = self.get_closing_signed_msg(&closing_tx, skip_remote_output, used_fee, our_min_fee, our_max_fee, logger);
-				if closing_signed.is_none() {
-					return Err(ChannelError::close("Failed to get signature for closing transaction.".to_owned()));
-				}
 				let (signed_tx, shutdown_result) = if $new_fee == msg.fee_satoshis {
 					let shutdown_result = ShutdownResult {
 						closure_reason,
@@ -6090,6 +6115,7 @@ impl<SP: Deref> Channel<SP> where
 					};
 					self.context.channel_state = ChannelState::ShutdownComplete;
 					self.context.update_time_counter += 1;
+					self.context.last_received_closing_sig = Some(msg.signature.clone());
 					let tx = closing_signed.as_ref().map(|ClosingSigned { signature, .. }|
 						self.build_signed_closing_transaction(&closing_tx, &msg.signature, signature));
 					(tx, Some(shutdown_result))
@@ -6127,7 +6153,7 @@ impl<SP: Deref> Channel<SP> where
 		} else {
 			// Old fee style negotiation. We don't bother to enforce whether they are complying
 			// with the "making progress" requirements, we just comply and hope for the best.
-			if let Some((last_fee, _)) = self.context.last_sent_closing_fee {
+			if let Some((last_fee, _, _, _)) = self.context.last_sent_closing_fee {
 				if msg.fee_satoshis > last_fee {
 					if msg.fee_satoshis < our_max_fee {
 						propose_fee!(msg.fee_satoshis);
@@ -9282,6 +9308,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 
 				signer_pending_commitment_update: false,
 				signer_pending_funding: false,
+				signer_pending_closing: false,
 
 				pending_update_fee,
 				holding_cell_update_fee,
@@ -9296,6 +9323,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 				counterparty_max_commitment_tx_output: Mutex::new((0, 0)),
 
 				last_sent_closing_fee: None,
+				last_received_closing_sig: None,
 				pending_counterparty_closing_signed: None,
 				expecting_peer_commitment_signed: false,
 				closing_fee_limits: None,
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
@@ -8192,7 +8192,8 @@ where
 	pub fn signer_unblocked(&self, channel_opt: Option<(PublicKey, ChannelId)>) {
 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self);
 
-		let unblock_chan = |phase: &mut ChannelPhase<SP>, pending_msg_events: &mut Vec<MessageSendEvent>| {
+		// Returns whether we should remove this channel as it's just been closed.
+		let unblock_chan = |phase: &mut ChannelPhase<SP>, pending_msg_events: &mut Vec<MessageSendEvent>| -> bool {
 			let node_id = phase.context().get_counterparty_node_id();
 			match phase {
 				ChannelPhase::Funded(chan) => {
@@ -8212,6 +8213,29 @@ where
 					if let Some(msg) = msgs.channel_ready {
 						send_channel_ready!(self, pending_msg_events, chan, msg);
 					}
+					if let Some(msg) = msgs.closing_signed {
+						pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
+							node_id,
+							msg,
+						});
+					}
+					if let Some(broadcast_tx) = msgs.signed_closing_tx {
+						let channel_id = chan.context.channel_id();
+						let counterparty_node_id = chan.context.get_counterparty_node_id();
+						let logger = WithContext::from(&self.logger, Some(counterparty_node_id), Some(channel_id), None);
+						log_info!(logger, "Broadcasting closing tx {}", log_tx!(broadcast_tx));
+						self.tx_broadcaster.broadcast_transactions(&[&broadcast_tx]);
+
+						if let Ok(update) = self.get_channel_update_for_broadcast(&chan) {
+							pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+								msg: update
+							});
+						}
+
+						// We should return true to remove the channel if we just
+						// broadcasted the closing transaction.
+						true
+					} else { false }
 				}
 				ChannelPhase::UnfundedOutboundV1(chan) => {
 					if let Some(msg) = chan.signer_maybe_unblocked(&self.logger) {
@@ -8220,8 +8244,9 @@ where
 							msg,
 						});
 					}
+					false
 				}
-				ChannelPhase::UnfundedInboundV1(_) => {},
+				ChannelPhase::UnfundedInboundV1(_) => false,
 			}
 		};
 
@@ -8230,17 +8255,25 @@ where
 			if let Some(peer_state_mutex) = per_peer_state.get(&counterparty_node_id) {
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
-				if let Some(chan) = peer_state.channel_by_id.get_mut(&channel_id) {
-					unblock_chan(chan, &mut peer_state.pending_msg_events);
+				let should_remove = if let Some(chan) = peer_state.channel_by_id.get_mut(&channel_id) {
+					unblock_chan(chan, &mut peer_state.pending_msg_events)
+				} else { false };
+				if should_remove {
+					log_trace!(self.logger, "Removing channel after unblocking signer");
+					peer_state.channel_by_id.remove(&channel_id);
 				}
 			}
 		} else {
 			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
 				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
-				for (_, chan) in peer_state.channel_by_id.iter_mut() {
-					unblock_chan(chan, &mut peer_state.pending_msg_events);
-				}
+				peer_state.channel_by_id.retain(|_, chan| {
+					let should_remove = unblock_chan(chan, &mut peer_state.pending_msg_events);
+					if should_remove {
+						log_trace!(self.logger, "Removing channel after unblocking signer");
+					}
+					!should_remove
+				});
 			}
 		}
 	}
