Support async get_per_commitment_point, release_commitment_secret

This PR adds state to the context to allow a `ChannelSigner` implementation to
respond asynchronously to the `get_per_commitment_point`,
`release_commitment_secret`, and `sign_counterparty_commitment` methods, which
are the main signer methods that are called during channel setup and normal
operation.

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::{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,7 @@ use std::sync::atomic;
 use std::io::Cursor;
 use bitcoin::bech32::u5;
 
+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,
@@ -196,10 +197,15 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 	}
 }
 
+struct SignerState {
+	enforcement: Arc<Mutex<EnforcementState>>,
+	unavailable: Arc<Mutex<u32>>,
+}
+
 struct KeyProvider {
 	node_secret: SecretKey,
 	rand_bytes_id: atomic::AtomicU32,
-	enforcement_states: Mutex<HashMap<[u8;32], Arc<Mutex<EnforcementState>>>>,
+	enforcement_states: Mutex<HashMap<[u8;32], SignerState>>,
 }
 
 impl EntropySource for KeyProvider {
@@ -283,8 +289,15 @@ impl SignerProvider for KeyProvider {
 			channel_keys_id,
 			channel_keys_id,
 		);
-		let revoked_commitment = self.make_enforcement_state_cell(keys.commitment_seed);
-		TestChannelSigner::new_with_revoked(keys, revoked_commitment, false)
+		let mut revoked_commitments = self.enforcement_states.lock().unwrap();
+		let new_state = revoked_commitments.entry(keys.commitment_seed)
+			.or_insert(SignerState {
+				enforcement: Arc::new(Mutex::new(EnforcementState::new())),
+				unavailable: Arc::new(Mutex::new(0)),
+			});
+		let mut ret = TestChannelSigner::new_with_revoked(keys, Arc::clone(&new_state.enforcement), false);
+		ret.unavailable = Arc::clone(&new_state.unavailable);
+		ret
 	}
 
 	fn read_chan_signer(&self, buffer: &[u8]) -> Result<Self::EcdsaSigner, DecodeError> {
@@ -316,17 +329,6 @@ impl SignerProvider for KeyProvider {
 	}
 }
 
-impl KeyProvider {
-	fn make_enforcement_state_cell(&self, commitment_seed: [u8; 32]) -> Arc<Mutex<EnforcementState>> {
-		let mut revoked_commitments = self.enforcement_states.lock().unwrap();
-		if !revoked_commitments.contains_key(&commitment_seed) {
-			revoked_commitments.insert(commitment_seed, Arc::new(Mutex::new(EnforcementState::new())));
-		}
-		let cell = revoked_commitments.get(&commitment_seed).unwrap();
-		Arc::clone(cell)
-	}
-}
-
 #[inline]
 fn check_api_err(api_err: APIError, sendable_bounds_violated: bool) {
 	match api_err {
@@ -829,7 +831,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 +846,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 +875,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 +884,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);
 								}
 							}
@@ -1289,15 +1291,108 @@ 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(); },
 
+			0xa0 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xa1 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[0].signer_unblocked(None);
+			}
+			0xa2 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[0].signer_unblocked(None);
+			}
+			0xa3 => {
+				let signer_states = keys_manager_a.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[0].signer_unblocked(None);
+			}
+
+			0xa4 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xa5 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[1].signer_unblocked(None);
+			}
+			0xa6 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[1].signer_unblocked(None);
+			}
+			0xa7 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[1].signer_unblocked(None);
+			}
+
+			0xa8 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xa9 => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[1].signer_unblocked(None);
+			}
+			0xaa => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[1].signer_unblocked(None);
+			}
+			0xab => {
+				let signer_states = keys_manager_b.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 2);
+				*signer_states.values().last().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[1].signer_unblocked(None);
+			}
+
+			0xac => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() = ASYNC_OPS;
+			}
+			0xad => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::GET_PER_COMMITMENT_POINT;
+				nodes[2].signer_unblocked(None);
+			}
+			0xae => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::RELEASE_COMMITMENT_SECRET;
+				nodes[2].signer_unblocked(None);
+			}
+			0xaf => {
+				let signer_states = keys_manager_c.enforcement_states.lock().unwrap();
+				assert_eq!(signer_states.len(), 1);
+				*signer_states.values().next().unwrap().unavailable.lock().unwrap() &= !ops::SIGN_COUNTERPARTY_COMMITMENT;
+				nodes[2].signer_unblocked(None);
+			}
+
 			0xff => {
 				// Test that no channel is in a stuck state where neither party can send funds even
 				// 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(format!("Restoring monitors...\n").as_bytes());
 				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();
@@ -1316,7 +1411,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(format!("Reconnecting peers...\n").as_bytes());
+
 				if chan_a_disconnected {
+					out.locked_write(format!("Reconnecting node 0 and node 1...\n").as_bytes());
 					nodes[0].peer_connected(&nodes[1].get_our_node_id(), &Init {
 						features: nodes[1].init_features(), networks: None, remote_network_address: None
 					}, true).unwrap();
@@ -1326,6 +1424,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(format!("Reconnecting node 1 and node 2...\n").as_bytes());
 					nodes[1].peer_connected(&nodes[2].get_our_node_id(), &Init {
 						features: nodes[2].init_features(), networks: None, remote_network_address: None
 					}, true).unwrap();
@@ -1335,8 +1434,30 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 					chan_b_disconnected = false;
 				}
 
+				out.locked_write(format!("Restoring signers...\n").as_bytes());
+
+				*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;
+
+				for signer_state in keys_manager_a.enforcement_states.lock().unwrap().values() {
+					*signer_state.unavailable.lock().unwrap() = 0;
+				}
+				for signer_state in keys_manager_b.enforcement_states.lock().unwrap().values() {
+					*signer_state.unavailable.lock().unwrap() = 0;
+				}
+				for signer_state in keys_manager_c.enforcement_states.lock().unwrap().values() {
+					*signer_state.unavailable.lock().unwrap() = 0;
+				}
+				nodes[0].signer_unblocked(None);
+				nodes[1].signer_unblocked(None);
+				nodes[2].signer_unblocked(None);
+
+				out.locked_write(format!("Running event queues to quiescence...\n").as_bytes());
+
 				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; }
@@ -1349,6 +1470,8 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 					break;
 				}
 
+				out.locked_write(format!("All channels restored to normal operation.\n").as_bytes());
+
 				// 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) ||

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,