X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fchannelmanager.rs;h=ae8625d89f4f60eaba2fe8127a5b957a75b5d301;hb=593d8c4610f082441563d4906d64175a354b1cfc;hp=91a853132064049abca699a211de956e6e5cdc22;hpb=384c4dc7753e4b7ac53ea380e52809babd8f0f9b;p=rust-lightning

diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 91a85313..ae8625d8 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -53,7 +53,7 @@ use crate::ln::msgs::{ChannelMessageHandler, DecodeError, LightningError, MAX_VA
 use crate::ln::wire::Encode;
 use crate::chain::keysinterface::{Sign, KeysInterface, KeysManager, Recipient};
 use crate::util::config::{UserConfig, ChannelConfig};
-use crate::util::events::{EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
+use crate::util::events::{Event, EventHandler, EventsProvider, MessageSendEvent, MessageSendEventsProvider, ClosureReason, HTLCDestination};
 use crate::util::{byte_utils, events};
 use crate::util::wakers::{Future, Notifier};
 use crate::util::scid_utils::fake_scid;
@@ -112,8 +112,8 @@ pub(super) struct PendingHTLCInfo {
 	pub(super) routing: PendingHTLCRouting,
 	pub(super) incoming_shared_secret: [u8; 32],
 	payment_hash: PaymentHash,
-	pub(super) amt_to_forward: u64,
-	pub(super) amt_incoming: Option<u64>, // Added in 0.0.113
+	pub(super) incoming_amt_msat: Option<u64>, // Added in 0.0.113
+	pub(super) outgoing_amt_msat: u64,
 	pub(super) outgoing_cltv_value: u32,
 }
 
@@ -2138,13 +2138,6 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 		}
 
 		let routing = match hop_data.format {
-			msgs::OnionHopDataFormat::Legacy { .. } => {
-				return Err(ReceiveError {
-					err_code: 0x4000|0x2000|3,
-					err_data: Vec::new(),
-					msg: "We require payment_secrets",
-				});
-			},
 			msgs::OnionHopDataFormat::NonFinalNode { .. } => {
 				return Err(ReceiveError {
 					err_code: 0x4000|22,
@@ -2197,8 +2190,8 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 			routing,
 			payment_hash,
 			incoming_shared_secret: shared_secret,
-			amt_incoming: Some(amt_msat),
-			amt_to_forward: amt_msat,
+			incoming_amt_msat: Some(amt_msat),
+			outgoing_amt_msat: amt_msat,
 			outgoing_cltv_value: hop_data.outgoing_cltv_value,
 		})
 	}
@@ -2280,7 +2273,6 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 				};
 
 				let short_channel_id = match next_hop_data.format {
-					msgs::OnionHopDataFormat::Legacy { short_channel_id } => short_channel_id,
 					msgs::OnionHopDataFormat::NonFinalNode { short_channel_id } => short_channel_id,
 					msgs::OnionHopDataFormat::FinalNode { .. } => {
 						return_err!("Final Node OnionHopData provided for us as an intermediary node", 0x4000 | 22, &[0;0]);
@@ -2294,14 +2286,14 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 					},
 					payment_hash: msg.payment_hash.clone(),
 					incoming_shared_secret: shared_secret,
-					amt_incoming: Some(msg.amount_msat),
-					amt_to_forward: next_hop_data.amt_to_forward,
+					incoming_amt_msat: Some(msg.amount_msat),
+					outgoing_amt_msat: next_hop_data.amt_to_forward,
 					outgoing_cltv_value: next_hop_data.outgoing_cltv_value,
 				})
 			}
 		};
 
-		if let &PendingHTLCStatus::Forward(PendingHTLCInfo { ref routing, ref amt_to_forward, ref outgoing_cltv_value, .. }) = &pending_forward_info {
+		if let &PendingHTLCStatus::Forward(PendingHTLCInfo { ref routing, ref outgoing_amt_msat, ref outgoing_cltv_value, .. }) = &pending_forward_info {
 			// If short_channel_id is 0 here, we'll reject the HTLC as there cannot be a channel
 			// with a short_channel_id of 0. This is important as various things later assume
 			// short_channel_id is non-0 in any ::Forward.
@@ -2352,10 +2344,10 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 						if !chan.is_live() { // channel_disabled
 							break Some(("Forwarding channel is not in a ready state.", 0x1000 | 20, chan_update_opt));
 						}
-						if *amt_to_forward < chan.get_counterparty_htlc_minimum_msat() { // amount_below_minimum
+						if *outgoing_amt_msat < chan.get_counterparty_htlc_minimum_msat() { // amount_below_minimum
 							break Some(("HTLC amount was below the htlc_minimum_msat", 0x1000 | 11, chan_update_opt));
 						}
-						if let Err((err, code)) = chan.htlc_satisfies_config(&msg, *amt_to_forward, *outgoing_cltv_value) {
+						if let Err((err, code)) = chan.htlc_satisfies_config(&msg, *outgoing_amt_msat, *outgoing_cltv_value) {
 							break Some((err, code, chan_update_opt));
 						}
 						chan_update_opt
@@ -3157,8 +3149,8 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 									HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 										prev_short_channel_id, prev_htlc_id, prev_funding_outpoint,
 										forward_info: PendingHTLCInfo {
-											routing, incoming_shared_secret, payment_hash, amt_to_forward,
-											outgoing_cltv_value, amt_incoming: _
+											routing, incoming_shared_secret, payment_hash, outgoing_amt_msat,
+											outgoing_cltv_value, incoming_amt_msat: _
 										}
 									}) => {
 										macro_rules! failure_handler {
@@ -3220,7 +3212,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 												};
 												match next_hop {
 													onion_utils::Hop::Receive(hop_data) => {
-														match self.construct_recv_pending_htlc_info(hop_data, incoming_shared_secret, payment_hash, amt_to_forward, outgoing_cltv_value, Some(phantom_shared_secret)) {
+														match self.construct_recv_pending_htlc_info(hop_data, incoming_shared_secret, payment_hash, outgoing_amt_msat, outgoing_cltv_value, Some(phantom_shared_secret)) {
 															Ok(info) => phantom_receives.push((prev_short_channel_id, prev_funding_outpoint, vec![(info, prev_htlc_id)])),
 															Err(ReceiveError { err_code, err_data, msg }) => failed_payment!(msg, err_code, err_data, Some(phantom_shared_secret))
 														}
@@ -3264,8 +3256,8 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 									HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 										prev_short_channel_id, prev_htlc_id, prev_funding_outpoint ,
 										forward_info: PendingHTLCInfo {
-											incoming_shared_secret, payment_hash, amt_to_forward, outgoing_cltv_value,
-											routing: PendingHTLCRouting::Forward { onion_packet, .. }, amt_incoming: _,
+											incoming_shared_secret, payment_hash, outgoing_amt_msat, outgoing_cltv_value,
+											routing: PendingHTLCRouting::Forward { onion_packet, .. }, incoming_amt_msat: _,
 										},
 									}) => {
 										log_trace!(self.logger, "Adding HTLC from short id {} with payment_hash {} to channel with short id {} after delay", prev_short_channel_id, log_bytes!(payment_hash.0), short_chan_id);
@@ -3277,7 +3269,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 											// Phantom payments are only PendingHTLCRouting::Receive.
 											phantom_shared_secret: None,
 										});
-										match chan.get_mut().send_htlc(amt_to_forward, payment_hash, outgoing_cltv_value, htlc_source.clone(), onion_packet, &self.logger) {
+										match chan.get_mut().send_htlc(outgoing_amt_msat, payment_hash, outgoing_cltv_value, htlc_source.clone(), onion_packet, &self.logger) {
 											Err(e) => {
 												if let ChannelError::Ignore(msg) = e {
 													log_trace!(self.logger, "Failed to forward HTLC with payment_hash {}: {}", log_bytes!(payment_hash.0), msg);
@@ -3391,7 +3383,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 							HTLCForwardInfo::AddHTLC(PendingAddHTLCInfo {
 								prev_short_channel_id, prev_htlc_id, prev_funding_outpoint,
 								forward_info: PendingHTLCInfo {
-									routing, incoming_shared_secret, payment_hash, amt_to_forward, ..
+									routing, incoming_shared_secret, payment_hash, outgoing_amt_msat, ..
 								}
 							}) => {
 								let (cltv_expiry, onion_payload, payment_data, phantom_shared_secret) = match routing {
@@ -3413,9 +3405,9 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 										incoming_packet_shared_secret: incoming_shared_secret,
 										phantom_shared_secret,
 									},
-									value: amt_to_forward,
+									value: outgoing_amt_msat,
 									timer_ticks: 0,
-									total_msat: if let Some(data) = &payment_data { data.total_msat } else { amt_to_forward },
+									total_msat: if let Some(data) = &payment_data { data.total_msat } else { outgoing_amt_msat },
 									cltv_expiry,
 									onion_payload,
 								};
@@ -3522,7 +3514,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 														e.insert((purpose.clone(), vec![claimable_htlc]));
 														new_events.push(events::Event::PaymentReceived {
 															payment_hash,
-															amount_msat: amt_to_forward,
+															amount_msat: outgoing_amt_msat,
 															purpose,
 														});
 													},
@@ -5714,7 +5706,7 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 	#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 	pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
 		let events = core::cell::RefCell::new(Vec::new());
-		let event_handler = |event: &events::Event| events.borrow_mut().push(event.clone());
+		let event_handler = |event: events::Event| events.borrow_mut().push(event);
 		self.process_pending_events(&event_handler);
 		events.into_inner()
 	}
@@ -5728,6 +5720,39 @@ impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelManager<M, T, K, F
 	pub fn clear_pending_payments(&self) {
 		self.pending_outbound_payments.lock().unwrap().clear()
 	}
+
+	/// Processes any events asynchronously in the order they were generated since the last call
+	/// using the given event handler.
+	///
+	/// See the trait-level documentation of [`EventsProvider`] for requirements.
+	pub async fn process_pending_events_async<Future: core::future::Future, H: Fn(Event) -> Future>(
+		&self, handler: H
+	) {
+		// We'll acquire our total consistency lock until the returned future completes so that
+		// we can be sure no other persists happen while processing events.
+		let _read_guard = self.total_consistency_lock.read().unwrap();
+
+		let mut result = NotifyOption::SkipPersist;
+
+		// TODO: This behavior should be documented. It's unintuitive that we query
+		// ChannelMonitors when clearing other events.
+		if self.process_pending_monitor_events() {
+			result = NotifyOption::DoPersist;
+		}
+
+		let pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
+		if !pending_events.is_empty() {
+			result = NotifyOption::DoPersist;
+		}
+
+		for event in pending_events {
+			handler(event).await;
+		}
+
+		if result == NotifyOption::DoPersist {
+			self.persistence_notifier.notify();
+		}
+	}
 }
 
 impl<M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> MessageSendEventsProvider for ChannelManager<M, T, K, F, L>
@@ -5791,13 +5816,13 @@ where
 				result = NotifyOption::DoPersist;
 			}
 
-			let mut pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
+			let pending_events = mem::replace(&mut *self.pending_events.lock().unwrap(), vec![]);
 			if !pending_events.is_empty() {
 				result = NotifyOption::DoPersist;
 			}
 
-			for event in pending_events.drain(..) {
-				handler.handle_event(&event);
+			for event in pending_events {
+				handler.handle_event(event);
 			}
 
 			result
@@ -6472,9 +6497,9 @@ impl_writeable_tlv_based!(PendingHTLCInfo, {
 	(0, routing, required),
 	(2, incoming_shared_secret, required),
 	(4, payment_hash, required),
-	(6, amt_to_forward, required),
+	(6, outgoing_amt_msat, required),
 	(8, outgoing_cltv_value, required),
-	(9, amt_incoming, option),
+	(9, incoming_amt_msat, option),
 });