Support async get_per_commitment_point, release_commitment_secret

Apologies in advance for the hair-ball. Mostly just wanted to get a 50,000 foot
overview to see if things were headed in the right direction. If things look
okay, I can take a step back and chop this into more modestly-sized PRs.

In general, 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.

These changes seem to work as advertised during normal channel operation
(creation and payment exchange), and do not obviously fail during channel
re-establishment or across node restart. That said, there are a lot more test
scenarios to evaluate here.

The details are below.

Adds the RevokeAndAck and RAACommitemnt ordering to the `SignerResumeUpdates`
struct that is returned from `signer_maybe_unblocked`.

Adds `signer_maybe_unblocked` for both inbound and outbound unfunded channel
states. We need these now because `get_per_commitment_point` is asynchronous
-- and necessary for us to proceed out of the unfunded state into normal
operation. Adds appropriate `SignerResumeUpdates` classes for both inbound and
outbound unfunded states.

Maintains `cur_holder_commitment_point` and `prev_holder_commitment_secret` on
the channel context. By making these part of the context state, we can access
them at any point we need them without requiring a signer call to
regenerate. These are updated appropriately throughout the channel state
machine by calling a new context method `update_holder_per_commitment`.

Add several flags to indicate messages that may now be pending the remote
signer unblocking:

- `signer_pending_revoke_and_ack`, set when we're waiting to send the
  revoke-and-ack to the counterparty. This might _not_ just be us waiting on
  the signer; for example, if the commitment order requires sending the RAA
  after the commitment update, then even though we _might_ be able to generate
  the RAA (e.g., because we have the secret), we will not do so while the
  commitment update is pending (e.g., because we're missing the latest point
  or do not have a signed counterparty commitment).

- `signer_pending_channel_ready`, set when we're waiting to send the
  channel-ready to the counterparty.

- `signer_pending_commitment_point`, set when we're waiting for the signer to
  return us the commitment point for the current state.

- `signer_pending_released_secret`, set when we're waiting for the signer to
  release the commitment secret for the previous state.

This state (current commitment point, previous secret, flags) is persisted in
the channel monitor, and restored when the channel is unpickled.

When a monitor update completes, we may still be pending results from the
remote signer. If that is the case, we ensure that we correctly maintain the
above state flags before the channel state is resumed. For example, if the
_monitor_ is pending a commitment signed, but we were not able to retrieve the
commitment update from the signer, then we ensure that
`signer_pending_commitment_update` is set.

When unblocking the signer, we need to ensure that we honor message ordering
constraints. For the commitment update and revoke-and-ack, ensure that we can
honor the context's current `resend_order`. For example, assume we must send
the RAA before the CU, and could potentially send a CU because we have the
commitment point and a signed counterparty commitment transaction. _But_, we
have not yet received the previous state's commitment secret. In this case, we
ensure that no messages are emitted until the commitment secret is released,
at which point the signer-unblocked call will emit both messages.

A similar situation exists with the `channel_ready` message and the
`funding_signed` / `funding_created` messages at channel startup: we make sure
that we don't emit `channel_ready` before the funding message.

There is at least one unsolved problem here: during channel re-establishment,
we need to request an _arbitrary_ commitment point from the signer in order to
verify that the counterparty's giving us a legitimate secret. For the time
being, I've simply commented out this check; however, this is not a viable
solution. There are a few options to consider here:

1. We could require that an asynchronous signer _cache_ the previous
   commitment points, so that any such request must necessarily succeed
   synchronously.

2. We could attempt to factor this method as to admit an asynchronous response
   that would restart the channel re-establish once the commitment point has
   been provided by the signer and the counterparty's secret can be verified.

The former places some additional burden on a remote signer (albeit minimal)
and seems reasonable to me.

As for testing...

For testing asynchronous channel signing, replaces the simple boolean
("everything is on, or everything is off") with flags that let us toggle
different signing methods on and off. This lets us (sort of) simulate the
signer returning responses in an arbitrary order.

Adds a fully-exploded "send a one-hop payment" test to the async test
suite. At each step, simulate the async signer being unavailable, and then
unblocking. Vary the order to check all possible orderings of
`get_per_commitment_point`, `release_commitment_secret`, and
`sign_counterparty_commitment` being provided by the signer.

But there is a lot more to be done here. Many of the odd-ball cases in the PR
_aren't_ covered by unit tests and were instead uncovered by running the code
in situ with an LND counterparty. So there are a lot more tests to write here.

Author: waterson

fuzz/src/chanmon_consistency.rs

@@ -266,7 +266,7 @@ impl SignerProvider for KeyProvider {
 			inner,
 			state,
 			disable_revocation_policy_check: false,
-			available: Arc::new(Mutex::new(true)),
+			unavailable: Arc::new(Mutex::new(0)),
 		})
 	}
 

lightning/src/chain/channelmonitor.rs

@@ -2832,9 +2832,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,
 							htlc: htlc.htlc,
 							preimage: htlc.preimage,
 							counterparty_sig: htlc.counterparty_sig,

lightning/src/chain/onchaintx.rs

@@ -179,6 +179,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
@@ -1192,9 +1193,16 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner> OnchainTxHandler<ChannelSigner>
 				})
 				.map(|(htlc_idx, htlc)| {
 					let counterparty_htlc_sig = holder_commitment.counterparty_htlc_sigs[htlc_idx];
+
+					// TODO(waterson) fallible: move this somewhere!
+					let per_commitment_point = self.signer.get_per_commitment_point(
+						trusted_tx.commitment_number(), &self.secp_ctx,
+					).unwrap();
+					
 					ExternalHTLCClaim {
 						commitment_txid: trusted_tx.txid(),
 						per_commitment_number: trusted_tx.commitment_number(),
+						per_commitment_point: per_commitment_point,
 						htlc: htlc.clone(),
 						preimage: *preimage,
 						counterparty_sig: counterparty_htlc_sig,

lightning/src/ln/async_signer_tests.rs

@@ -2372,7 +2372,7 @@ where
 			.ok_or_else(|| APIError::APIMisuseError{ err: format!("Not connected to node: {}", their_network_key) })?;
 
 		let mut peer_state = peer_state_mutex.lock().unwrap();
-		let channel = {
+		let mut channel = {
 			let outbound_scid_alias = self.create_and_insert_outbound_scid_alias();
 			let their_features = &peer_state.latest_features;
 			let config = if override_config.is_some() { override_config.as_ref().unwrap() } else { &self.default_configuration };
@@ -2387,8 +2387,11 @@ where
 				},
 			}
 		};
-		let res = channel.get_open_channel(self.genesis_hash.clone());
-
+		let opt_msg = channel.get_open_channel(self.genesis_hash.clone());
+		if opt_msg.is_none() {
+			channel.signer_pending_open_channel = true;
+		}
+			
 		let temporary_channel_id = channel.context.channel_id();
 		match peer_state.channel_by_id.entry(temporary_channel_id) {
 			hash_map::Entry::Occupied(_) => {
@@ -2401,10 +2404,13 @@ where
 			hash_map::Entry::Vacant(entry) => { entry.insert(ChannelPhase::UnfundedOutboundV1(channel)); }
 		}
 
-		peer_state.pending_msg_events.push(events::MessageSendEvent::SendOpenChannel {
-			node_id: their_network_key,
-			msg: res,
-		});
+		if let Some(msg) = opt_msg {
+			peer_state.pending_msg_events.push(events::MessageSendEvent::SendOpenChannel {
+				node_id: their_network_key,
+				msg,
+			});
+		};
+		
 		Ok(temporary_channel_id)
 	}
 
@@ -5659,6 +5665,12 @@ where
 			emit_channel_ready_event!(pending_events, channel);
 		}
 
+
+		log_debug!(self.logger, "Outgoing message queue is:");
+		for msg in pending_msg_events.iter() {
+			log_debug!(self.logger, "  {:?}", msg);
+		}
+		
 		htlc_forwards
 	}
 
@@ -5809,10 +5821,14 @@ where
 		let outbound_scid_alias = self.create_and_insert_outbound_scid_alias();
 		channel.context.set_outbound_scid_alias(outbound_scid_alias);
 
-		peer_state.pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel {
-			node_id: channel.context.get_counterparty_node_id(),
-			msg: channel.accept_inbound_channel(),
-		});
+		match channel.accept_inbound_channel() {
+			Some(msg) =>
+				peer_state.pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel {
+					node_id: channel.context.get_counterparty_node_id(),
+					msg
+				}),
+			None => channel.signer_pending_accept_channel = true,
+		};
 
 		peer_state.channel_by_id.insert(temporary_channel_id.clone(), ChannelPhase::UnfundedInboundV1(channel));
 
@@ -5964,10 +5980,15 @@ where
 		let outbound_scid_alias = self.create_and_insert_outbound_scid_alias();
 		channel.context.set_outbound_scid_alias(outbound_scid_alias);
 
-		peer_state.pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel {
-			node_id: counterparty_node_id.clone(),
-			msg: channel.accept_inbound_channel(),
-		});
+		match channel.accept_inbound_channel() {
+			Some(msg) =>
+				peer_state.pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel {
+					node_id: channel.context.get_counterparty_node_id(),
+					msg
+				}),
+			None => channel.signer_pending_accept_channel = true,
+		};
+
 		peer_state.channel_by_id.insert(channel_id, ChannelPhase::UnfundedInboundV1(channel));
 		Ok(())
 	}
@@ -6027,6 +6048,12 @@ where
 				Some(ChannelPhase::UnfundedInboundV1(inbound_chan)) => {
 					match inbound_chan.funding_created(msg, best_block, &self.signer_provider, &self.logger) {
 						Ok(res) => res,
+						Err((inbound_chan, ChannelError::Ignore(_))) => {
+							// If we get an `Ignore` error then something transient went wrong. Put the channel
+							// back into the table and bail.
+							peer_state.channel_by_id.insert(msg.temporary_channel_id, ChannelPhase::UnfundedInboundV1(inbound_chan));
+							return Ok(());
+						},
 						Err((mut inbound_chan, err)) => {
 							// We've already removed this inbound channel from the map in `PeerState`
 							// above so at this point we just need to clean up any lingering entries
@@ -6147,6 +6174,7 @@ where
 		match peer_state.channel_by_id.entry(msg.channel_id) {
 			hash_map::Entry::Occupied(mut chan_phase_entry) => {
 				if let ChannelPhase::Funded(chan) = chan_phase_entry.get_mut() {
+					log_debug!(self.logger, "<== channel_ready");
 					let announcement_sigs_opt = try_chan_phase_entry!(self, chan.channel_ready(&msg, &self.node_signer,
 						self.genesis_hash.clone(), &self.default_configuration, &self.best_block.read().unwrap(), &self.logger), chan_phase_entry);
 					if let Some(announcement_sigs) = announcement_sigs_opt {
@@ -6367,6 +6395,7 @@ where
 							_ => pending_forward_info
 						}
 					};
+					log_debug!(self.logger, "<== update_add_htlc: htlc_id={} amount_msat={}", msg.htlc_id, msg.amount_msat);
 					try_chan_phase_entry!(self, chan.update_add_htlc(&msg, pending_forward_info, create_pending_htlc_status, &self.fee_estimator, &self.logger), chan_phase_entry);
 				} else {
 					return try_chan_phase_entry!(self, Err(ChannelError::Close(
@@ -6392,6 +6421,7 @@ where
 			match peer_state.channel_by_id.entry(msg.channel_id) {
 				hash_map::Entry::Occupied(mut chan_phase_entry) => {
 					if let ChannelPhase::Funded(chan) = chan_phase_entry.get_mut() {
+						log_debug!(self.logger, "<== update_fulfill_htlc: htlc_id={}", msg.htlc_id);
 						let res = try_chan_phase_entry!(self, chan.update_fulfill_htlc(&msg), chan_phase_entry);
 						if let HTLCSource::PreviousHopData(prev_hop) = &res.0 {
 							peer_state.actions_blocking_raa_monitor_updates.entry(msg.channel_id)
@@ -6485,6 +6515,7 @@ where
 			hash_map::Entry::Occupied(mut chan_phase_entry) => {
 				if let ChannelPhase::Funded(chan) = chan_phase_entry.get_mut() {
 					let funding_txo = chan.context.get_funding_txo();
+					log_debug!(self.logger, "<== commitment_signed: {} htlcs", msg.htlc_signatures.len());
 					let monitor_update_opt = try_chan_phase_entry!(self, chan.commitment_signed(&msg, &self.logger), chan_phase_entry);
 					if let Some(monitor_update) = monitor_update_opt {
 						handle_new_monitor_update!(self, funding_txo.unwrap(), monitor_update, peer_state_lock,
@@ -6656,6 +6687,7 @@ where
 								&peer_state.actions_blocking_raa_monitor_updates, funding_txo,
 								*counterparty_node_id)
 						} else { false };
+						log_debug!(self.logger, "<== revoke_and_ack");
 						let (htlcs_to_fail, monitor_update_opt) = try_chan_phase_entry!(self,
 							chan.revoke_and_ack(&msg, &self.fee_estimator, &self.logger, mon_update_blocked), chan_phase_entry);
 						if let Some(monitor_update) = monitor_update_opt {
@@ -6997,28 +7029,51 @@ where
 
 		let unblock_chan = |phase: &mut ChannelPhase<SP>, pending_msg_events: &mut Vec<MessageSendEvent>| {
 			let node_id = phase.context().get_counterparty_node_id();
-			if let ChannelPhase::Funded(chan) = phase {
-				let msgs = chan.signer_maybe_unblocked(&self.logger);
-				if let Some(updates) = msgs.commitment_update {
-					pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
-						node_id,
-						updates,
-					});
-				}
-				if let Some(msg) = msgs.funding_signed {
-					pending_msg_events.push(events::MessageSendEvent::SendFundingSigned {
-						node_id,
-						msg,
-					});
+			match phase {
+				ChannelPhase::Funded(chan) => {
+					let msgs = chan.signer_maybe_unblocked(&self.logger);
+					match (msgs.commitment_update, msgs.raa) {
+						(Some(cu), Some(raa)) if msgs.order == RAACommitmentOrder::CommitmentFirst => {
+							pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { node_id, updates: cu });
+							pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { node_id, msg: raa });
+						},
+						(Some(cu), Some(raa)) if msgs.order == RAACommitmentOrder::RevokeAndACKFirst => {
+							pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { node_id, msg: raa });
+							pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { node_id, updates: cu });
+						},
+						(Some(cu), _) => pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { node_id, updates: cu }),
+						(_, Some(raa)) => pending_msg_events.push(events::MessageSendEvent::SendRevokeAndACK { node_id, msg: raa }),
+						(_, _) => (),
+					};
+					if let Some(msg) = msgs.funding_signed {
+						pending_msg_events.push(events::MessageSendEvent::SendFundingSigned {
+							node_id,
+							msg,
+						});
+					}
+					if let Some(msg) = msgs.funding_created {
+						pending_msg_events.push(events::MessageSendEvent::SendFundingCreated {
+							node_id,
+							msg,
+						});
+					}
+					if let Some(msg) = msgs.channel_ready {
+						send_channel_ready!(self, pending_msg_events, chan, msg);
+					}
 				}
-				if let Some(msg) = msgs.funding_created {
-					pending_msg_events.push(events::MessageSendEvent::SendFundingCreated {
-						node_id,
-						msg,
-					});
+				ChannelPhase::UnfundedInboundV1(chan) => {
+					let msgs = chan.signer_maybe_unblocked(&self.logger);
+					let node_id = phase.context().get_counterparty_node_id();
+					if let Some(msg) = msgs.accept_channel {
+						pending_msg_events.push(events::MessageSendEvent::SendAcceptChannel { node_id, msg });
+					}
 				}
-				if let Some(msg) = msgs.channel_ready {
-					send_channel_ready!(self, pending_msg_events, chan, msg);
+				ChannelPhase::UnfundedOutboundV1(chan) => {
+					let msgs = chan.signer_maybe_unblocked(&self.genesis_hash, &self.logger);
+					let node_id = phase.context().get_counterparty_node_id();
+					if let Some(msg) = msgs.open_channel {
+						pending_msg_events.push(events::MessageSendEvent::SendOpenChannel { node_id, msg });
+					}
 				}
 			}
 		};
@@ -8318,11 +8373,13 @@ where
 				let mut peer_state_lock = peer_state_mutex_opt.unwrap().lock().unwrap();
 				let peer_state = &mut *peer_state_lock;
 				if let Some(ChannelPhase::UnfundedOutboundV1(chan)) = peer_state.channel_by_id.get_mut(&msg.channel_id) {
-					if let Ok(msg) = chan.maybe_handle_error_without_close(self.genesis_hash, &self.fee_estimator) {
-						peer_state.pending_msg_events.push(events::MessageSendEvent::SendOpenChannel {
-							node_id: *counterparty_node_id,
-							msg,
-						});
+					if let Ok(opt_msg) = chan.maybe_handle_error_without_close(self.genesis_hash, &self.fee_estimator) {
+						if let Some(msg) = opt_msg {
+							peer_state.pending_msg_events.push(events::MessageSendEvent::SendOpenChannel {
+								node_id: *counterparty_node_id,
+								msg,
+							});
+						}
 						return;
 					}
 				}

lightning/src/ln/channel.rs

@@ -32,6 +32,8 @@ use crate::util::config::{UserConfig, MaxDustHTLCExposure};
 use crate::util::ser::{ReadableArgs, Writeable};
 #[cfg(test)]
 use crate::util::logger::Logger;
+#[cfg(test)]
+use crate::util::test_channel_signer::ops;
 
 use bitcoin::blockdata::block::{Block, BlockHeader};
 use bitcoin::blockdata::transaction::{Transaction, TxOut};
@@ -438,14 +440,14 @@ impl<'a, 'b, 'c> Node<'a, 'b, 'c> {
 	pub fn get_block_header(&self, height: u32) -> BlockHeader {
 		self.blocks.lock().unwrap()[height as usize].0.header
 	}
+
 	/// Changes the channel signer's availability for the specified peer and channel.
 	///
 	/// When `available` is set to `true`, the channel signer will behave normally. When set to
 	/// `false`, the channel signer will act like an off-line remote signer and will return `Err` for
-	/// several of the signing methods. Currently, only `get_per_commitment_point` and
-	/// `release_commitment_secret` are affected by this setting.
+	/// several of the signing methods.
 	#[cfg(test)]
-	pub fn set_channel_signer_available(&self, peer_id: &PublicKey, chan_id: &ChannelId, available: bool) {
+	pub fn set_channel_signer_ops_available(&self, peer_id: &PublicKey, chan_id: &ChannelId, mask: u32, available: bool) {
 		let per_peer_state = self.node.per_peer_state.read().unwrap();
 		let chan_lock = per_peer_state.get(peer_id).unwrap().lock().unwrap();
 		let signer = (|| {
@@ -454,8 +456,9 @@ impl<'a, 'b, 'c> Node<'a, 'b, 'c> {
 				None => panic!("Couldn't find a channel with id {}", chan_id),
 			}
 		})();
-		log_debug!(self.logger, "Setting channel signer for {} as available={}", chan_id, available);
-		signer.as_ecdsa().unwrap().set_available(available);
+		log_debug!(self.logger, "Setting channel signer for {} as {}available for {} (mask={})",
+			chan_id, if available { "" } else { "un" }, ops::string_from(mask), mask);
+		signer.as_ecdsa().unwrap().set_ops_available(mask, available);
 	}
 }
 
@@ -3206,15 +3209,15 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 				} else { panic!("Unexpected event! {:?}", announcement_event[0]); }
 			}
 		} else {
-			assert!(chan_msgs.0.is_none());
+			assert!(chan_msgs.0.is_none(), "did not expect to have a ChannelReady for node 1");
 		}
 		if pending_raa.0 {
 			assert!(chan_msgs.3 == RAACommitmentOrder::RevokeAndACKFirst);
 			node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &chan_msgs.1.unwrap());
 			assert!(node_a.node.get_and_clear_pending_msg_events().is_empty());
 			check_added_monitors!(node_a, 1);
 		} else {
-			assert!(chan_msgs.1.is_none());
+			assert!(chan_msgs.1.is_none(), "did not expect to have a RevokeAndACK for node 1");
 		}
 		if pending_htlc_adds.0 != 0 || pending_htlc_claims.0 != 0 || pending_htlc_fails.0 != 0 ||
 			pending_cell_htlc_claims.0 != 0 || pending_cell_htlc_fails.0 != 0 ||
@@ -3247,7 +3250,7 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 				check_added_monitors!(node_b, if pending_responding_commitment_signed_dup_monitor.0 { 0 } else { 1 });
 			}
 		} else {
-			assert!(chan_msgs.2.is_none());
+			assert!(chan_msgs.2.is_none(), "did not expect to have commitment updates for node 1");
 		}
 	}
 
@@ -3264,15 +3267,15 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 				}
 			}
 		} else {
-			assert!(chan_msgs.0.is_none());
+			assert!(chan_msgs.0.is_none(), "did not expect to have a ChannelReady for node 2");
 		}
 		if pending_raa.1 {
 			assert!(chan_msgs.3 == RAACommitmentOrder::RevokeAndACKFirst);
 			node_b.node.handle_revoke_and_ack(&node_a.node.get_our_node_id(), &chan_msgs.1.unwrap());
 			assert!(node_b.node.get_and_clear_pending_msg_events().is_empty());
 			check_added_monitors!(node_b, 1);
 		} else {
-			assert!(chan_msgs.1.is_none());
+			assert!(chan_msgs.1.is_none(), "did not expect to have a RevokeAndACK for node 2");
 		}
 		if pending_htlc_adds.1 != 0 || pending_htlc_claims.1 != 0 || pending_htlc_fails.1 != 0 ||
 			pending_cell_htlc_claims.1 != 0 || pending_cell_htlc_fails.1 != 0 ||
@@ -3305,7 +3308,7 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
 				check_added_monitors!(node_a, if pending_responding_commitment_signed_dup_monitor.1 { 0 } else { 1 });
 			}
 		} else {
-			assert!(chan_msgs.2.is_none());
+			assert!(chan_msgs.2.is_none(), "did not expect to have commitment updates for node 2");
 		}
 	}
 }