Add ChannelReady event

This adds a `ChannelReady` event that is emitted as soon as a new
channel becomes usable, i.e., after both sides have sent
`channel_ready`.

Author: tnull

lightning/src/ln/channel.rs

@@ -736,6 +736,9 @@ pub(super) struct Channel<Signer: Sign> {
 	// don't currently support node id aliases and eventually privacy should be provided with
 	// blinded paths instead of simple scid+node_id aliases.
 	outbound_scid_alias: u64,
+
+	// We track whether we already emitted a `ChannelReady` event.
+	channel_ready_event_emitted: bool,
 }
 
 #[cfg(any(test, fuzzing))]
@@ -1063,6 +1066,8 @@ impl<Signer: Sign> Channel<Signer> {
 			latest_inbound_scid_alias: None,
 			outbound_scid_alias,
 
+			channel_ready_event_emitted: false,
+
 			#[cfg(any(test, fuzzing))]
 			historical_inbound_htlc_fulfills: HashSet::new(),
 
@@ -1397,6 +1402,8 @@ impl<Signer: Sign> Channel<Signer> {
 			latest_inbound_scid_alias: None,
 			outbound_scid_alias,
 
+			channel_ready_event_emitted: false,
+
 			#[cfg(any(test, fuzzing))]
 			historical_inbound_htlc_fulfills: HashSet::new(),
 
@@ -2174,7 +2181,14 @@ impl<Signer: Sign> Channel<Signer> {
 		&self.get_counterparty_pubkeys().funding_pubkey
 	}
 
-	pub fn funding_created<L: Deref>(&mut self, msg: &msgs::FundingCreated, best_block: BestBlock, logger: &L) -> Result<(msgs::FundingSigned, ChannelMonitor<Signer>, Option<msgs::ChannelReady>), ChannelError> where L::Target: Logger {
+	/// Handles a received [`FundingCreated`] message.
+	///
+	/// May return the resulting [`ChannelReady`] message, as well as
+	/// a `bool` signalling whether a corresponding event should be emitted.
+	///
+	/// [`FundingCreated`]: msgs::FundingCreated
+	/// [`ChannelReady`]: msgs::ChannelReady
+	pub fn funding_created<L: Deref>(&mut self, msg: &msgs::FundingCreated, best_block: BestBlock, logger: &L) -> Result<(msgs::FundingSigned, ChannelMonitor<Signer>, (Option<msgs::ChannelReady>, bool)), ChannelError> where L::Target: Logger {
 		if self.is_outbound() {
 			return Err(ChannelError::Close("Received funding_created for an outbound channel?".to_owned()));
 		}
@@ -2252,9 +2266,16 @@ impl<Signer: Sign> Channel<Signer> {
 		}, channel_monitor, self.check_get_channel_ready(0)))
 	}
 
-	/// Handles a funding_signed message from the remote end.
+	/// Handles a received [`FundingSigned`] message.
+	///
 	/// If this call is successful, broadcast the funding transaction (and not before!)
-	pub fn funding_signed<L: Deref>(&mut self, msg: &msgs::FundingSigned, best_block: BestBlock, logger: &L) -> Result<(ChannelMonitor<Signer>, Transaction, Option<msgs::ChannelReady>), ChannelError> where L::Target: Logger {
+	///
+	/// May return the resulting [`ChannelReady`] message, as well as
+	/// a `bool` signalling whether a corresponding event should be emitted.
+	///
+	/// [`FundingSigned`]: msgs::FundingSigned
+	/// [`ChannelReady`]: msgs::ChannelReady
+	pub fn funding_signed<L: Deref>(&mut self, msg: &msgs::FundingSigned, best_block: BestBlock, logger: &L) -> Result<(ChannelMonitor<Signer>, Transaction, (Option<msgs::ChannelReady>, bool)), ChannelError> where L::Target: Logger {
 		if !self.is_outbound() {
 			return Err(ChannelError::Close("Received funding_signed for an inbound channel?".to_owned()));
 		}
@@ -4598,6 +4619,17 @@ impl<Signer: Sign> Channel<Signer> {
 		self.prev_config.map(|prev_config| prev_config.0)
 	}
 
+	// Checks whether we previously had emitted a `ChannelReady` event and sets the value if not.
+	pub(crate) fn should_emit_channel_ready_event(&mut self) -> bool {
+		if self.channel_ready_event_emitted {
+			// Do nothing, already emitted.
+			false
+		} else {
+			self.channel_ready_event_emitted = true;
+			true
+		}
+	}
+
 	/// Tracks the number of ticks elapsed since the previous [`ChannelConfig`] was updated. Once
 	/// [`EXPIRE_PREV_CONFIG_TICKS`] is reached, the previous config is considered expired and will
 	/// no longer be considered when forwarding HTLCs.
@@ -4858,12 +4890,12 @@ impl<Signer: Sign> Channel<Signer> {
 		self.channel_update_status = status;
 	}
 
-	fn check_get_channel_ready(&mut self, height: u32) -> Option<msgs::ChannelReady> {
+	fn check_get_channel_ready(&mut self, height: u32) -> (Option<msgs::ChannelReady>, bool) {
 		// Called:
 		//  * always when a new block/transactions are confirmed with the new height
 		//  * when funding is signed with a height of 0
 		if self.funding_tx_confirmation_height == 0 && self.minimum_depth != Some(0) {
-			return None;
+			return (None, false);
 		}
 
 		let funding_tx_confirmations = height as i64 - self.funding_tx_confirmation_height as i64 + 1;
@@ -4872,7 +4904,7 @@ impl<Signer: Sign> Channel<Signer> {
 		}
 
 		if funding_tx_confirmations < self.minimum_depth.unwrap_or(0) as i64 {
-			return None;
+			return (None, false);
 		}
 
 		let non_shutdown_state = self.channel_state & (!MULTI_STATE_FLAGS);
@@ -4904,27 +4936,34 @@ impl<Signer: Sign> Channel<Signer> {
 		if need_commitment_update {
 			if self.channel_state & (ChannelState::MonitorUpdateInProgress as u32) == 0 {
 				if self.channel_state & (ChannelState::PeerDisconnected as u32) == 0 {
+					let emit_channel_ready_event = self.should_emit_channel_ready_event();
 					let next_per_commitment_point =
 						self.holder_signer.get_per_commitment_point(INITIAL_COMMITMENT_NUMBER - 1, &self.secp_ctx);
-					return Some(msgs::ChannelReady {
+					return (Some(msgs::ChannelReady {
 						channel_id: self.channel_id,
 						next_per_commitment_point,
 						short_channel_id_alias: Some(self.outbound_scid_alias),
-					});
+					}), emit_channel_ready_event);
 				}
 			} else {
 				self.monitor_pending_channel_ready = true;
 			}
 		}
-		None
+		(None, false)
 	}
 
 	/// When a transaction is confirmed, we check whether it is or spends the funding transaction
 	/// In the first case, we store the confirmation height and calculating the short channel id.
 	/// In the second, we simply return an Err indicating we need to be force-closed now.
+	///
+	/// May return the resulting [`ChannelReady`] and [`AnnouncementSignatures`] messages, as well as
+	/// a `bool` signalling whether a corresponding `ChannelReady` event should be emitted.
+	///
+	/// [`ChannelReady`]: msgs::ChannelReady
+	/// [`AnnouncementSignatures`]: msgs::AnnouncementSignatures
 	pub fn transactions_confirmed<L: Deref>(&mut self, block_hash: &BlockHash, height: u32,
 		txdata: &TransactionData, genesis_block_hash: BlockHash, node_pk: PublicKey, logger: &L)
-	-> Result<(Option<msgs::ChannelReady>, Option<msgs::AnnouncementSignatures>), ClosureReason> where L::Target: Logger {
+	-> Result<((Option<msgs::ChannelReady>, bool), Option<msgs::AnnouncementSignatures>), ClosureReason> where L::Target: Logger {
 		if let Some(funding_txo) = self.get_funding_txo() {
 			for &(index_in_block, tx) in txdata.iter() {
 				// Check if the transaction is the expected funding transaction, and if it is,
@@ -4968,10 +5007,10 @@ impl<Signer: Sign> Channel<Signer> {
 					// If we allow 1-conf funding, we may need to check for channel_ready here and
 					// send it immediately instead of waiting for a best_block_updated call (which
 					// may have already happened for this block).
-					if let Some(channel_ready) = self.check_get_channel_ready(height) {
+					if let (Some(channel_ready), emit_channel_ready_event) = self.check_get_channel_ready(height) {
 						log_info!(logger, "Sending a channel_ready to our peer for channel {}", log_bytes!(self.channel_id));
 						let announcement_sigs = self.get_announcement_sigs(node_pk, genesis_block_hash, height, logger);
-						return Ok((Some(channel_ready), announcement_sigs));
+						return Ok(((Some(channel_ready), emit_channel_ready_event), announcement_sigs));
 					}
 				}
 				for inp in tx.input.iter() {
@@ -4982,7 +5021,7 @@ impl<Signer: Sign> Channel<Signer> {
 				}
 			}
 		}
-		Ok((None, None))
+		Ok(((None, false), None))
 	}
 
 	/// When a new block is connected, we check the height of the block against outbound holding
@@ -4994,15 +5033,19 @@ impl<Signer: Sign> Channel<Signer> {
 	/// requirements apply - no calls may be made except those explicitly stated to be allowed
 	/// post-shutdown.
 	///
-	/// May return some HTLCs (and their payment_hash) which have timed out and should be failed
-	/// back.
+	/// May return the resulting [`ChannelReady`] and [`AnnouncementSignatures`] messages, as well as
+	/// a `bool` signalling whether a corresponding `ChannelReady` event should be emitted and some HTLCs (and
+	/// their payment_hash) which have timed out and should be failed back.
+	///
+	/// [`ChannelReady`]: msgs::ChannelReady
+	/// [`AnnouncementSignatures`]: msgs::AnnouncementSignatures
 	pub fn best_block_updated<L: Deref>(&mut self, height: u32, highest_header_time: u32, genesis_block_hash: BlockHash, node_pk: PublicKey, logger: &L)
-	-> Result<(Option<msgs::ChannelReady>, Vec<(HTLCSource, PaymentHash)>, Option<msgs::AnnouncementSignatures>), ClosureReason> where L::Target: Logger {
+	-> Result<((Option<msgs::ChannelReady>, bool), Vec<(HTLCSource, PaymentHash)>, Option<msgs::AnnouncementSignatures>), ClosureReason> where L::Target: Logger {
 		self.do_best_block_updated(height, highest_header_time, Some((genesis_block_hash, node_pk)), logger)
 	}
 
 	fn do_best_block_updated<L: Deref>(&mut self, height: u32, highest_header_time: u32, genesis_node_pk: Option<(BlockHash, PublicKey)>, logger: &L)
-	-> Result<(Option<msgs::ChannelReady>, Vec<(HTLCSource, PaymentHash)>, Option<msgs::AnnouncementSignatures>), ClosureReason> where L::Target: Logger {
+	-> Result<((Option<msgs::ChannelReady>, bool), Vec<(HTLCSource, PaymentHash)>, Option<msgs::AnnouncementSignatures>), ClosureReason> where L::Target: Logger {
 		let mut timed_out_htlcs = Vec::new();
 		// This mirrors the check in ChannelManager::decode_update_add_htlc_onion, refusing to
 		// forward an HTLC when our counterparty should almost certainly just fail it for expiring
@@ -5022,12 +5065,12 @@ impl<Signer: Sign> Channel<Signer> {
 
 		self.update_time_counter = cmp::max(self.update_time_counter, highest_header_time);
 
-		if let Some(channel_ready) = self.check_get_channel_ready(height) {
+		if let (Some(channel_ready), emit_channel_ready_event) = self.check_get_channel_ready(height) {
 			let announcement_sigs = if let Some((genesis_block_hash, node_pk)) = genesis_node_pk {
 				self.get_announcement_sigs(node_pk, genesis_block_hash, height, logger)
 			} else { None };
 			log_info!(logger, "Sending a channel_ready to our peer for channel {}", log_bytes!(self.channel_id));
-			return Ok((Some(channel_ready), timed_out_htlcs, announcement_sigs));
+			return Ok(((Some(channel_ready), emit_channel_ready_event), timed_out_htlcs, announcement_sigs));
 		}
 
 		let non_shutdown_state = self.channel_state & (!MULTI_STATE_FLAGS);
@@ -5067,7 +5110,7 @@ impl<Signer: Sign> Channel<Signer> {
 		let announcement_sigs = if let Some((genesis_block_hash, node_pk)) = genesis_node_pk {
 			self.get_announcement_sigs(node_pk, genesis_block_hash, height, logger)
 		} else { None };
-		Ok((None, timed_out_htlcs, announcement_sigs))
+		Ok(((None, false), timed_out_htlcs, announcement_sigs))
 	}
 
 	/// Indicates the funding transaction is no longer confirmed in the main chain. This may
@@ -5083,8 +5126,9 @@ impl<Signer: Sign> Channel<Signer> {
 			// time we saw and it will be ignored.
 			let best_time = self.update_time_counter;
 			match self.do_best_block_updated(reorg_height, best_time, None, logger) {
-				Ok((channel_ready, timed_out_htlcs, announcement_sigs)) => {
+				Ok(((channel_ready, emit_channel_ready_event), timed_out_htlcs, announcement_sigs)) => {
 					assert!(channel_ready.is_none(), "We can't generate a funding with 0 confirmations?");
+					assert!(!emit_channel_ready_event, "We shouldn't emit a ChannelReady event for a funding with 0 confirmations?");
 					assert!(timed_out_htlcs.is_empty(), "We can't have accepted HTLCs with a timeout before our funding confirmation?");
 					assert!(announcement_sigs.is_none(), "We can't generate an announcement_sigs with 0 confirmations?");
 					Ok(())
@@ -6666,6 +6710,8 @@ impl<'a, Signer: Sign, K: Deref> ReadableArgs<(&'a K, u32)> for Channel<Signer>
 			// Later in the ChannelManager deserialization phase we scan for channels and assign scid aliases if its missing
 			outbound_scid_alias: outbound_scid_alias.unwrap_or(0),
 
+			channel_ready_event_emitted: false,
+
 			#[cfg(any(test, fuzzing))]
 			historical_inbound_htlc_fulfills,
 

lightning/src/ln/channelmanager.rs

@@ -1466,6 +1466,17 @@ macro_rules! send_channel_ready {
 	}
 }
 
+macro_rules! emit_channel_ready_event {
+	($pending_events: expr, $channel: expr) => {
+		$pending_events.push(events::Event::ChannelReady {
+			channel_id: $channel.channel_id(),
+			user_channel_id: $channel.get_user_id(),
+			counterparty_node_id: $channel.get_counterparty_node_id(),
+			channel_type: $channel.get_channel_type().clone(),
+		});
+	}
+}
+
 macro_rules! handle_chan_restoration_locked {
 	($self: ident, $channel_lock: expr, $channel_state: expr, $channel_entry: expr,
 	 $raa: expr, $commitment_update: expr, $order: expr, $chanmon_update: expr,
@@ -1509,6 +1520,11 @@ macro_rules! handle_chan_restoration_locked {
 				});
 			}
 
+			if $channel_entry.get().is_usable() && $channel_entry.get_mut().should_emit_channel_ready_event() {
+				let mut pending_events = $self.pending_events.lock().unwrap();
+				emit_channel_ready_event!(pending_events, $channel_entry.get());
+			}
+
 			let funding_broadcastable: Option<Transaction> = $funding_broadcastable; // Force type-checking to resolve
 			if let Some(monitor_update) = chanmon_update {
 				// We only ever broadcast a funding transaction in response to a funding_signed
@@ -4530,7 +4546,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 	}
 
 	fn internal_funding_created(&self, counterparty_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result<(), MsgHandleErrInternal> {
-		let ((funding_msg, monitor, mut channel_ready), mut chan) = {
+		let ((funding_msg, monitor, (mut channel_ready, emit_channel_ready_event)), mut chan) = {
 			let best_block = *self.best_block.read().unwrap();
 			let mut channel_lock = self.channel_state.lock().unwrap();
 			let channel_state = &mut *channel_lock;
@@ -4594,6 +4610,10 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 				if let Some(msg) = channel_ready {
 					send_channel_ready!(channel_state.short_to_chan_info, channel_state.pending_msg_events, chan, msg);
 				}
+				if emit_channel_ready_event {
+					let mut pending_events = self.pending_events.lock().unwrap();
+					emit_channel_ready_event!(pending_events, chan);
+				}
 				e.insert(chan);
 			}
 		}
@@ -4610,7 +4630,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 					if chan.get().get_counterparty_node_id() != *counterparty_node_id {
 						return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id));
 					}
-					let (monitor, funding_tx, channel_ready) = match chan.get_mut().funding_signed(&msg, best_block, &self.logger) {
+					let (monitor, funding_tx, (channel_ready, emit_channel_ready_event)) = match chan.get_mut().funding_signed(&msg, best_block, &self.logger) {
 						Ok(update) => update,
 						Err(e) => try_chan_entry!(self, Err(e), channel_state, chan),
 					};
@@ -4632,6 +4652,11 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 					if let Some(msg) = channel_ready {
 						send_channel_ready!(channel_state.short_to_chan_info, channel_state.pending_msg_events, chan.get(), msg);
 					}
+
+					if emit_channel_ready_event {
+						let mut pending_events = self.pending_events.lock().unwrap();
+						emit_channel_ready_event!(pending_events, chan.get());
+					}
 					funding_tx
 				},
 				hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
@@ -4671,6 +4696,11 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 							msg,
 						});
 					}
+
+				}
+				if chan.get().is_usable() && chan.get_mut().should_emit_channel_ready_event() {
+					let mut pending_events = self.pending_events.lock().unwrap();
+					emit_channel_ready_event!(pending_events, chan.get());
 				}
 				Ok(())
 			},
@@ -5775,9 +5805,9 @@ where
 		self.do_chain_event(None, |channel| {
 			if let Some(funding_txo) = channel.get_funding_txo() {
 				if funding_txo.txid == *txid {
-					channel.funding_transaction_unconfirmed(&self.logger).map(|()| (None, Vec::new(), None))
-				} else { Ok((None, Vec::new(), None)) }
-			} else { Ok((None, Vec::new(), None)) }
+					channel.funding_transaction_unconfirmed(&self.logger).map(|()| ((None, false), Vec::new(), None))
+				} else { Ok(((None, false), Vec::new(), None)) }
+			} else { Ok(((None, false), Vec::new(), None)) }
 		});
 	}
 }
@@ -5793,7 +5823,7 @@ where
 	/// Calls a function which handles an on-chain event (blocks dis/connected, transactions
 	/// un/confirmed, etc) on each channel, handling any resulting errors or messages generated by
 	/// the function.
-	fn do_chain_event<FN: Fn(&mut Channel<Signer>) -> Result<(Option<msgs::ChannelReady>, Vec<(HTLCSource, PaymentHash)>, Option<msgs::AnnouncementSignatures>), ClosureReason>>
+	fn do_chain_event<FN: Fn(&mut Channel<Signer>) -> Result<((Option<msgs::ChannelReady>, bool), Vec<(HTLCSource, PaymentHash)>, Option<msgs::AnnouncementSignatures>), ClosureReason>>
 			(&self, height_opt: Option<u32>, f: FN) {
 		// Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called
 		// during initialization prior to the chain_monitor being fully configured in some cases.
@@ -5808,7 +5838,7 @@ where
 			let pending_msg_events = &mut channel_state.pending_msg_events;
 			channel_state.by_id.retain(|_, channel| {
 				let res = f(channel);
-				if let Ok((channel_ready_opt, mut timed_out_pending_htlcs, announcement_sigs)) = res {
+				if let Ok(((channel_ready_opt, emit_channel_ready_event), mut timed_out_pending_htlcs, announcement_sigs)) = res {
 					for (source, payment_hash) in timed_out_pending_htlcs.drain(..) {
 						let (failure_code, data) = self.get_htlc_inbound_temp_fail_err_and_data(0x1000|14 /* expiry_too_soon */, &channel);
 						timed_out_htlcs.push((source, payment_hash, HTLCFailReason::Reason {
@@ -5829,6 +5859,10 @@ where
 							log_trace!(self.logger, "Sending channel_ready WITHOUT channel_update for {}", log_bytes!(channel.channel_id()));
 						}
 					}
+					if emit_channel_ready_event {
+						let mut pending_events = self.pending_events.lock().unwrap();
+						emit_channel_ready_event!(pending_events, channel);
+					}
 					if let Some(announcement_sigs) = announcement_sigs {
 						log_trace!(self.logger, "Sending announcement_signatures for channel {}", log_bytes!(channel.channel_id()));
 						pending_msg_events.push(events::MessageSendEvent::SendAnnouncementSignatures {