Allow sending closing tx signatures asynchronously

alecchendev · alecchendev · commit 2fd77d0cf390 · 2024-08-23T11:02:01.000-07:00
diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
@@ -906,6 +906,8 @@ pub(super) struct SignerResumeUpdates {
 	pub funding_signed: Option<msgs::FundingSigned>,
 	pub channel_ready: Option<msgs::ChannelReady>,
 	pub order: RAACommitmentOrder,
+	pub closing_signed: Option<msgs::ClosingSigned>,
+	pub signed_closing_tx: Option<Transaction>,
 }
 
 /// The return value of `channel_reestablish`
@@ -1268,6 +1270,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.
 	//
@@ -1299,7 +1304,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_sats, 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`
@@ -1736,6 +1743,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider  {
 			signer_pending_revoke_and_ack: false,
 			signer_pending_commitment_update: false,
 			signer_pending_funding: false,
+			signer_pending_closing: false,
 
 
 			#[cfg(debug_assertions)]
@@ -1744,6 +1752,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,
@@ -1968,6 +1977,7 @@ impl<SP: Deref> ChannelContext<SP> where SP::Target: SignerProvider  {
 			signer_pending_revoke_and_ack: false,
 			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`.
@@ -1977,6 +1987,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,
@@ -5490,19 +5501,44 @@ impl<SP: Deref> Channel<SP> where
 			commitment_update = None;
 		}
 
-		log_trace!(logger, "Signer unblocked with {} commitment_update, {} revoke_and_ack, {} funding_signed and {} channel_ready, with resend order {:?}",
+		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) {
+					let funding_redeemscript = self.context.get_funding_redeemscript();
+					let sighash = closing_tx.trust().get_sighash_all(&funding_redeemscript, self.context.channel_value_satoshis);
+					debug_assert!(self.context.secp_ctx.verify_ecdsa(&sighash, &counterparty_sig,
+						&self.context.get_counterparty_pubkeys().funding_pubkey).is_ok());
+					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, {} revoke_and_ack, with resend order {:?}, {} funding_signed, {} channel_ready,
+					{} closing_signed, and {} signed_closing_tx",
 			if commitment_update.is_some() { "a" } else { "no" },
 			if revoke_and_ack.is_some() { "a" } else { "no" },
+			self.context.resend_order,
 			if funding_signed.is_some() { "a" } else { "no" },
 			if channel_ready.is_some() { "a" } else { "no" },
-			self.context.resend_order);
+			if closing_signed.is_some() { "a" } else { "no" },
+			if signed_closing_tx.is_some() { "a" } else { "no" });
 
 		SignerResumeUpdates {
 			commitment_update,
 			revoke_and_ack,
 			funding_signed,
 			channel_ready,
 			order: self.context.resend_order.clone(),
+			closing_signed,
+			signed_closing_tx,
 		}
 	}
 
@@ -5952,9 +5988,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))
 	}
 
@@ -6108,33 +6141,36 @@ 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>(
 		&mut self, fee_estimator: &LowerBoundedFeeEstimator<F>, msg: &msgs::ClosingSigned, logger: &L)
 		-> Result<(Option<msgs::ClosingSigned>, Option<Transaction>, Option<ShutdownResult>), ChannelError>
 		where F::Target: FeeEstimator, L::Target: Logger
 	{
+		if self.is_shutdown_pending_signature() {
+			return Err(ChannelError::Warn(String::from("Remote end sent us a closing_signed while fully shutdown and just waiting on the final closing signature")));
+		}
 		if !self.context.channel_state.is_both_sides_shutdown() {
 			return Err(ChannelError::close("Remote end sent us a closing_signed before both sides provided a shutdown".to_owned()));
 		}
@@ -6190,7 +6226,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,
@@ -6223,9 +6259,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,
@@ -6239,8 +6272,11 @@ impl<SP: Deref> Channel<SP> where
 						unbroadcasted_funding_tx: self.context.unbroadcasted_funding(),
 						channel_funding_txo: self.context.get_funding_txo(),
 					};
-					self.context.channel_state = ChannelState::ShutdownComplete;
+					if closing_signed.is_some() {
+						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))
@@ -6278,7 +6314,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);
@@ -6601,6 +6637,12 @@ impl<SP: Deref> Channel<SP> where
 		matches!(self.context.channel_state, ChannelState::ShutdownComplete)
 	}
 
+	pub fn is_shutdown_pending_signature(&self) -> bool {
+		matches!(self.context.channel_state, ChannelState::ChannelReady(_))
+			&& self.context.signer_pending_closing
+			&& self.context.last_received_closing_sig.is_some()
+	}
+
 	pub fn channel_update_status(&self) -> ChannelUpdateStatus {
 		self.context.channel_update_status
 	}
@@ -9466,6 +9508,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
 				signer_pending_revoke_and_ack: false,
 				signer_pending_commitment_update: false,
 				signer_pending_funding: false,
+				signer_pending_closing: false,
 
 				pending_update_fee,
 				holding_cell_update_fee,
@@ -9480,6 +9523,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
@@ -7795,7 +7795,7 @@ where
 					if let ChannelPhase::Funded(chan) = chan_phase_entry.get_mut() {
 						let logger = WithChannelContext::from(&self.logger, &chan.context, None);
 						let (closing_signed, tx, shutdown_result) = try_chan_phase_entry!(self, chan.closing_signed(&self.fee_estimator, &msg, &&logger), chan_phase_entry);
-						debug_assert_eq!(shutdown_result.is_some(), chan.is_shutdown());
+						debug_assert_eq!(shutdown_result.is_some(), chan.is_shutdown() || chan.is_shutdown_pending_signature());
 						if let Some(msg) = closing_signed {
 							peer_state.pending_msg_events.push(events::MessageSendEvent::SendClosingSigned {
 								node_id: counterparty_node_id.clone(),
@@ -8617,7 +8617,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) => {
@@ -8652,6 +8653,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) {
@@ -8660,28 +8684,31 @@ where
 							msg,
 						});
 					}
+					false
 				}
-				ChannelPhase::UnfundedInboundV1(_) => {},
+				ChannelPhase::UnfundedInboundV1(_) => false,
 			}
 		};
 
 		let per_peer_state = self.per_peer_state.read().unwrap();
-		if let Some((counterparty_node_id, channel_id)) = channel_opt {
-			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);
-				}
-			}
-		} 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);
+		let per_peer_state_iter = per_peer_state.iter().filter(|(cp_id, _)| {
+			if let Some((counterparty_node_id, _)) = channel_opt {
+				**cp_id == counterparty_node_id
+			} else { true }
+		});
+		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;
+			peer_state.channel_by_id.retain(|_, chan| {
+				let should_remove = match channel_opt {
+					Some((_, channel_id)) if chan.context().channel_id() != channel_id => false,
+					_ => unblock_chan(chan, &mut peer_state.pending_msg_events),
+				};
+				if should_remove {
+					log_trace!(self.logger, "Removing channel after unblocking signer");
 				}
-			}
+				!should_remove
+			});
 		}
 	}
 
@@ -8711,8 +8738,8 @@ where
 											node_id: chan.context.get_counterparty_node_id(), msg,
 										});
 									}
-									debug_assert_eq!(shutdown_result_opt.is_some(), chan.is_shutdown());
 									if let Some(shutdown_result) = shutdown_result_opt {
+										debug_assert!(chan.is_shutdown() || chan.is_shutdown_pending_signature());
 										shutdown_results.push(shutdown_result);
 									}
 									if let Some(tx) = tx_opt {
