Support normal channel operation with async signing

This is a do-over of lightningdevkit#2653, wherein we support asynchronous signing for
'normal' channel operation. This involves allowing the following
`ChannelSigner` methods to return an `Err` result, indicating that the
requested value is not available:

- get_per_commitment_point
- release_commitment_secret
- sign_counterparty_commitment

When the value does become available, channel operation can be resumed by
invoking `signer_unblocked`.

Note that this adds the current and next per-commitment point to the state
that is persisted by the channel monitor.

Author: waterson

fuzz/src/chanmon_consistency.rs

@@ -49,7 +49,7 @@ use lightning::ln::functional_test_utils::*;
 use lightning::offers::invoice::{BlindedPayInfo, UnsignedBolt12Invoice};
 use lightning::offers::invoice_request::UnsignedInvoiceRequest;
 use lightning::onion_message::messenger::{Destination, MessageRouter, OnionMessagePath};
-use lightning::util::test_channel_signer::{TestChannelSigner, EnforcementState};
+use lightning::util::test_channel_signer::{TestChannelSigner, EnforcementState, ops};
 use lightning::util::errors::APIError;
 use lightning::util::logger::Logger;
 use lightning::util::config::UserConfig;
@@ -72,6 +72,8 @@ use std::sync::atomic;
 use std::io::Cursor;
 use bitcoin::bech32::u5;
 
+#[allow(unused)]
+const ASYNC_OPS: u32 = ops::GET_PER_COMMITMENT_POINT | ops::RELEASE_COMMITMENT_SECRET | ops::SIGN_COUNTERPARTY_COMMITMENT;
 const MAX_FEE: u32 = 10_000;
 struct FuzzEstimator {
 	ret_val: atomic::AtomicU32,
@@ -297,7 +299,6 @@ impl SignerProvider for KeyProvider {
 			inner,
 			state,
 			disable_revocation_policy_check: false,
-			available: Arc::new(Mutex::new(true)),
 		})
 	}
 
@@ -829,7 +830,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == node_id {
 									for update_add in update_add_htlcs.iter() {
-										out.locked_write(format!("Delivering update_add_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_add_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										if !$corrupt_forward {
 											dest.handle_update_add_htlc(&nodes[$node].get_our_node_id(), update_add);
 										} else {
@@ -844,19 +845,19 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 										}
 									}
 									for update_fulfill in update_fulfill_htlcs.iter() {
-										out.locked_write(format!("Delivering update_fulfill_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fulfill_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fulfill_htlc(&nodes[$node].get_our_node_id(), update_fulfill);
 									}
 									for update_fail in update_fail_htlcs.iter() {
-										out.locked_write(format!("Delivering update_fail_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fail_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fail_htlc(&nodes[$node].get_our_node_id(), update_fail);
 									}
 									for update_fail_malformed in update_fail_malformed_htlcs.iter() {
-										out.locked_write(format!("Delivering update_fail_malformed_htlc to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fail_malformed_htlc to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fail_malformed_htlc(&nodes[$node].get_our_node_id(), update_fail_malformed);
 									}
 									if let Some(msg) = update_fee {
-										out.locked_write(format!("Delivering update_fee to node {}.\n", idx).as_bytes());
+										out.locked_write(format!("Delivering update_fee to node {} from node {}.\n", idx, $node).as_bytes());
 										dest.handle_update_fee(&nodes[$node].get_our_node_id(), &msg);
 									}
 									let processed_change = !update_add_htlcs.is_empty() || !update_fulfill_htlcs.is_empty() ||
@@ -873,7 +874,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 										} });
 										break;
 									}
-									out.locked_write(format!("Delivering commitment_signed to node {}.\n", idx).as_bytes());
+									out.locked_write(format!("Delivering commitment_signed to node {} from node {}.\n", idx, $node).as_bytes());
 									dest.handle_commitment_signed(&nodes[$node].get_our_node_id(), &commitment_signed);
 									break;
 								}
@@ -882,15 +883,15 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 						events::MessageSendEvent::SendRevokeAndACK { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
-									out.locked_write(format!("Delivering revoke_and_ack to node {}.\n", idx).as_bytes());
+									out.locked_write(format!("Delivering revoke_and_ack to node {} from node {}.\n", idx, $node).as_bytes());
 									dest.handle_revoke_and_ack(&nodes[$node].get_our_node_id(), msg);
 								}
 							}
 						},
 						events::MessageSendEvent::SendChannelReestablish { ref node_id, ref msg } => {
 							for (idx, dest) in nodes.iter().enumerate() {
 								if dest.get_our_node_id() == *node_id {
-									out.locked_write(format!("Delivering channel_reestablish to node {}.\n", idx).as_bytes());
+									out.locked_write(format!("Delivering channel_reestablish to node {} from node {}.\n", idx, $node).as_bytes());
 									dest.handle_channel_reestablish(&nodes[$node].get_our_node_id(), msg);
 								}
 							}
@@ -913,7 +914,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 						_ => if out.may_fail.load(atomic::Ordering::Acquire) {
 							return;
 						} else {
-							panic!("Unhandled message event {:?}", event)
+							panic!("Unhandled message event on node {}, {:?}", $node, event)
 						},
 					}
 					if $limit_events != ProcessMessages::AllMessages {
@@ -1289,6 +1290,118 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			},
 			0x89 => { fee_est_c.ret_val.store(253, atomic::Ordering::Release); nodes[2].maybe_update_chan_fees(); },
 
+			#[cfg(async_signing)]
+			0xa0 => {
+				let states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_unavailable(ASYNC_OPS);
+			}
+			#[cfg(async_signing)]
+			0xa1 => {
+				let states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::GET_PER_COMMITMENT_POINT);
+				nodes[0].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xa2 => {
+				let states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::RELEASE_COMMITMENT_SECRET);
+				nodes[0].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xa3 => {
+				let states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::SIGN_COUNTERPARTY_COMMITMENT);
+				nodes[0].signer_unblocked(None);
+			}
+
+			#[cfg(async_signing)]
+			0xa4 => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().next().unwrap().lock().unwrap().set_signer_unavailable(ASYNC_OPS);
+			}
+			#[cfg(async_signing)]
+			0xa5 => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::GET_PER_COMMITMENT_POINT);
+				nodes[1].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xa6 => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::RELEASE_COMMITMENT_SECRET);
+				nodes[1].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xa7 => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::SIGN_COUNTERPARTY_COMMITMENT);
+				nodes[1].signer_unblocked(None);
+			}
+
+			#[cfg(async_signing)]
+			0xa8 => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().last().unwrap().lock().unwrap().set_signer_unavailable(ASYNC_OPS);
+			}
+			#[cfg(async_signing)]
+			0xa9 => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().last().unwrap().lock().unwrap().set_signer_available(ops::GET_PER_COMMITMENT_POINT);
+				nodes[1].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xaa => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().last().unwrap().lock().unwrap().set_signer_available(ops::RELEASE_COMMITMENT_SECRET);
+				nodes[1].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xab => {
+				let states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 2);
+				states.values().last().unwrap().lock().unwrap().set_signer_available(ops::SIGN_COUNTERPARTY_COMMITMENT);
+				nodes[1].signer_unblocked(None);
+			}
+
+			#[cfg(async_signing)]
+			0xac => {
+				let states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_unavailable(ASYNC_OPS);
+			}
+			#[cfg(async_signing)]
+			0xad => {
+				let states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::GET_PER_COMMITMENT_POINT);
+				nodes[2].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xae => {
+				let states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::RELEASE_COMMITMENT_SECRET);
+				nodes[2].signer_unblocked(None);
+			}
+			#[cfg(async_signing)]
+			0xaf => {
+				let states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(states.len(), 1);
+				states.values().next().unwrap().lock().unwrap().set_signer_available(ops::SIGN_COUNTERPARTY_COMMITMENT);
+				nodes[2].signer_unblocked(None);
+			}
+
 			0xf0 => {
 				let pending_updates = monitor_a.chain_monitor.list_pending_monitor_updates().remove(&chan_1_funding).unwrap();
 				if let Some(id) = pending_updates.get(0) {
@@ -1382,10 +1495,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				// after we resolve all pending events.
 				// First make sure there are no pending monitor updates, resetting the error state
 				// and calling force_channel_monitor_updated for each monitor.
-				*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
-				*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
-				*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
-
+				out.locked_write(b"Restoring monitors...\n");
 				if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
 					monitor_a.chain_monitor.force_channel_monitor_updated(chan_1_funding, *id);
 					nodes[0].process_monitor_events();
@@ -1404,7 +1514,10 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				}
 
 				// Next, make sure peers are all connected to each other
+				out.locked_write(b"Reconnecting peers...\n");
+
 				if chan_a_disconnected {
+					out.locked_write(b"Reconnecting node 0 and node 1...\n");
 					nodes[0].peer_connected(&nodes[1].get_our_node_id(), &Init {
 						features: nodes[1].init_features(), networks: None, remote_network_address: None
 					}, true).unwrap();
@@ -1414,6 +1527,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 					chan_a_disconnected = false;
 				}
 				if chan_b_disconnected {
+					out.locked_write(b"Reconnecting node 1 and node 2...\n");
 					nodes[1].peer_connected(&nodes[2].get_our_node_id(), &Init {
 						features: nodes[2].init_features(), networks: None, remote_network_address: None
 					}, true).unwrap();
@@ -1423,8 +1537,33 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 					chan_b_disconnected = false;
 				}
 
+				out.locked_write(b"Restoring signers...\n");
+
+				*monitor_a.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
+				*monitor_b.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
+				*monitor_c.persister.update_ret.lock().unwrap() = ChannelMonitorUpdateStatus::Completed;
+
+				#[cfg(async_signing)]
+				{
+					for state in keys_manager_a.enforcement_states.lock().unwrap().values() {
+						state.lock().unwrap().set_signer_available(!0);
+					}
+					for state in keys_manager_b.enforcement_states.lock().unwrap().values() {
+						state.lock().unwrap().set_signer_available(!0);
+					}
+					for state in keys_manager_c.enforcement_states.lock().unwrap().values() {
+						state.lock().unwrap().set_signer_available(!0);
+					}
+					nodes[0].signer_unblocked(None);
+					nodes[1].signer_unblocked(None);
+					nodes[2].signer_unblocked(None);
+				}
+
+				out.locked_write(b"Running event queues to quiescence...\n");
+
 				for i in 0..std::usize::MAX {
 					if i == 100 { panic!("It may take may iterations to settle the state, but it should not take forever"); }
+
 					// Then, make sure any current forwards make their way to their destination
 					if process_msg_events!(0, false, ProcessMessages::AllMessages) { continue; }
 					if process_msg_events!(1, false, ProcessMessages::AllMessages) { continue; }
@@ -1437,13 +1576,34 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 					break;
 				}
 
+				out.locked_write(b"All channels restored to normal operation.\n");
+
 				// Finally, make sure that at least one end of each channel can make a substantial payment
 				assert!(
 					send_payment(&nodes[0], &nodes[1], chan_a, 10_000_000, &mut payment_id, &mut payment_idx) ||
 					send_payment(&nodes[1], &nodes[0], chan_a, 10_000_000, &mut payment_id, &mut payment_idx));
+				out.locked_write(b"Successfully sent a payment between node 0 and node 1.\n");
+
 				assert!(
 					send_payment(&nodes[1], &nodes[2], chan_b, 10_000_000, &mut payment_id, &mut payment_idx) ||
 					send_payment(&nodes[2], &nodes[1], chan_b, 10_000_000, &mut payment_id, &mut payment_idx));
+				out.locked_write(b"Successfully sent a payment between node 1 and node 2.\n");
+
+				out.locked_write(b"Flushing pending messages.\n");
+				for i in 0..std::usize::MAX {
+					if i == 100 { panic!("It may take may iterations to settle the state, but it should not take forever"); }
+
+					// Then, make sure any current forwards make their way to their destination
+					if process_msg_events!(0, false, ProcessMessages::AllMessages) { continue; }
+					if process_msg_events!(1, false, ProcessMessages::AllMessages) { continue; }
+					if process_msg_events!(2, false, ProcessMessages::AllMessages) { continue; }
+					// ...making sure any pending PendingHTLCsForwardable events are handled and
+					// payments claimed.
+					if process_events!(0, false) { continue; }
+					if process_events!(1, false) { continue; }
+					if process_events!(2, false) { continue; }
+					break;
+				}
 
 				last_htlc_clear_fee_a = fee_est_a.ret_val.load(atomic::Ordering::Acquire);
 				last_htlc_clear_fee_b = fee_est_b.ret_val.load(atomic::Ordering::Acquire);

lightning/src/chain/channelmonitor.rs

@@ -2944,9 +2944,7 @@ impl<Signer: WriteableEcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 							},
 							commitment_txid: htlc.commitment_txid,
 							per_commitment_number: htlc.per_commitment_number,
-							per_commitment_point: self.onchain_tx_handler.signer.get_per_commitment_point(
-								htlc.per_commitment_number, &self.onchain_tx_handler.secp_ctx,
-							),
+							per_commitment_point: htlc.per_commitment_point,
 							feerate_per_kw: 0,
 							htlc: htlc.htlc,
 							preimage: htlc.preimage,

lightning/src/chain/onchaintx.rs

@@ -178,6 +178,7 @@ pub(crate) struct ExternalHTLCClaim {
 	pub(crate) htlc: HTLCOutputInCommitment,
 	pub(crate) preimage: Option<PaymentPreimage>,
 	pub(crate) counterparty_sig: Signature,
+	pub(crate) per_commitment_point: bitcoin::secp256k1::PublicKey,
 }
 
 // Represents the different types of claims for which events are yielded externally to satisfy said
@@ -1177,9 +1178,11 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner> OnchainTxHandler<ChannelSigner>
 				})
 				.map(|(htlc_idx, htlc)| {
 					let counterparty_htlc_sig = holder_commitment.counterparty_htlc_sigs[htlc_idx];
+
 					ExternalHTLCClaim {
 						commitment_txid: trusted_tx.txid(),
 						per_commitment_number: trusted_tx.commitment_number(),
+						per_commitment_point: trusted_tx.per_commitment_point(),
 						htlc: htlc.clone(),
 						preimage: *preimage,
 						counterparty_sig: counterparty_htlc_sig,