Merge branch 'lightningdevkit:main' into issue1462

Author: mattfaltyn

lightning-invoice/src/payment.rs

@@ -1558,7 +1558,7 @@ mod tests {
 
 		assert!(invoice_payer.pay_invoice(&create_invoice_from_channelmanager_and_duration_since_epoch(
 			&nodes[1].node, nodes[1].keys_manager, Currency::Bitcoin, Some(100_010_000), "Invoice".to_string(),
-			duration_since_epoch()).unwrap())
+			duration_since_epoch(), 3600).unwrap())
 			.is_ok());
 		let htlc_msgs = nodes[0].node.get_and_clear_pending_msg_events();
 		assert_eq!(htlc_msgs.len(), 2);
@@ -1604,7 +1604,7 @@ mod tests {
 
 		assert!(invoice_payer.pay_invoice(&create_invoice_from_channelmanager_and_duration_since_epoch(
 			&nodes[1].node, nodes[1].keys_manager, Currency::Bitcoin, Some(100_010_000), "Invoice".to_string(),
-			duration_since_epoch()).unwrap())
+			duration_since_epoch(), 3600).unwrap())
 			.is_ok());
 		let htlc_msgs = nodes[0].node.get_and_clear_pending_msg_events();
 		assert_eq!(htlc_msgs.len(), 2);
@@ -1686,7 +1686,7 @@ mod tests {
 
 		assert!(invoice_payer.pay_invoice(&create_invoice_from_channelmanager_and_duration_since_epoch(
 			&nodes[1].node, nodes[1].keys_manager, Currency::Bitcoin, Some(100_010_000), "Invoice".to_string(),
-			duration_since_epoch()).unwrap())
+			duration_since_epoch(), 3600).unwrap())
 			.is_ok());
 		let htlc_updates = SendEvent::from_node(&nodes[0]);
 		check_added_monitors!(nodes[0], 1);

lightning-invoice/src/utils.rs

@@ -1,6 +1,6 @@
 //! Convenient utilities to create an invoice.
 
-use {CreationError, Currency, DEFAULT_EXPIRY_TIME, Invoice, InvoiceBuilder, SignOrCreationError};
+use {CreationError, Currency, Invoice, InvoiceBuilder, SignOrCreationError};
 use payment::{Payer, Router};
 
 use crate::{prelude::*, Description, InvoiceDescription, Sha256};
@@ -20,7 +20,6 @@ use lightning::routing::network_graph::{NetworkGraph, RoutingFees};
 use lightning::routing::router::{Route, RouteHint, RouteHintHop, RouteParameters, find_route};
 use lightning::util::logger::Logger;
 use secp256k1::PublicKey;
-use core::convert::TryInto;
 use core::ops::Deref;
 use core::time::Duration;
 use sync::Mutex;
@@ -162,7 +161,8 @@ fn _create_phantom_invoice<Signer: Sign, K: Deref>(
 		.current_timestamp()
 		.payment_hash(Hash::from_slice(&payment_hash.0).unwrap())
 		.payment_secret(payment_secret)
-		.min_final_cltv_expiry(MIN_FINAL_CLTV_EXPIRY.into());
+		.min_final_cltv_expiry(MIN_FINAL_CLTV_EXPIRY.into())
+		.expiry_time(Duration::from_secs(invoice_expiry_delta_secs.into()));
 	if let Some(amt) = amt_msat {
 		invoice = invoice.amount_milli_satoshis(amt);
 	}
@@ -212,9 +212,12 @@ fn _create_phantom_invoice<Signer: Sign, K: Deref>(
 /// method stores the invoice's payment secret and preimage in `ChannelManager`, so (a) the user
 /// doesn't have to store preimage/payment secret information and (b) `ChannelManager` can verify
 /// that the payment secret is valid when the invoice is paid.
+///
+/// `invoice_expiry_delta_secs` describes the number of seconds that the invoice is valid for
+/// in excess of the current time.
 pub fn create_invoice_from_channelmanager<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>(
 	channelmanager: &ChannelManager<Signer, M, T, K, F, L>, keys_manager: K, network: Currency,
-	amt_msat: Option<u64>, description: String
+	amt_msat: Option<u64>, description: String, invoice_expiry_delta_secs: u32
 ) -> Result<Invoice, SignOrCreationError<()>>
 where
 	M::Target: chain::Watch<Signer>,
@@ -227,7 +230,8 @@ where
 	let duration = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH)
 		.expect("for the foreseeable future this shouldn't happen");
 	create_invoice_from_channelmanager_and_duration_since_epoch(
-		channelmanager, keys_manager, network, amt_msat, description, duration
+		channelmanager, keys_manager, network, amt_msat, description, duration,
+		invoice_expiry_delta_secs
 	)
 }
 
@@ -238,9 +242,12 @@ where
 /// doesn't have to store preimage/payment secret information and (b) `ChannelManager` can verify
 /// that the payment secret is valid when the invoice is paid.
 /// Use this variant if you want to pass the `description_hash` to the invoice.
+///
+/// `invoice_expiry_delta_secs` describes the number of seconds that the invoice is valid for
+/// in excess of the current time.
 pub fn create_invoice_from_channelmanager_with_description_hash<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>(
 	channelmanager: &ChannelManager<Signer, M, T, K, F, L>, keys_manager: K, network: Currency,
-	amt_msat: Option<u64>, description_hash: Sha256,
+	amt_msat: Option<u64>, description_hash: Sha256, invoice_expiry_delta_secs: u32
 ) -> Result<Invoice, SignOrCreationError<()>>
 where
 	M::Target: chain::Watch<Signer>,
@@ -256,7 +263,8 @@ where
 		.expect("for the foreseeable future this shouldn't happen");
 
 	create_invoice_from_channelmanager_with_description_hash_and_duration_since_epoch(
-		channelmanager, keys_manager, network, amt_msat, description_hash, duration,
+		channelmanager, keys_manager, network, amt_msat,
+		description_hash, duration, invoice_expiry_delta_secs
 	)
 }
 
@@ -266,6 +274,7 @@ where
 pub fn create_invoice_from_channelmanager_with_description_hash_and_duration_since_epoch<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>(
 	channelmanager: &ChannelManager<Signer, M, T, K, F, L>, keys_manager: K, network: Currency,
 	amt_msat: Option<u64>, description_hash: Sha256, duration_since_epoch: Duration,
+	invoice_expiry_delta_secs: u32
 ) -> Result<Invoice, SignOrCreationError<()>>
 where
 	M::Target: chain::Watch<Signer>,
@@ -277,7 +286,7 @@ where
 	_create_invoice_from_channelmanager_and_duration_since_epoch(
 		channelmanager, keys_manager, network, amt_msat,
 		InvoiceDescription::Hash(&description_hash),
-		duration_since_epoch,
+		duration_since_epoch, invoice_expiry_delta_secs
 	)
 }
 
@@ -287,6 +296,7 @@ where
 pub fn create_invoice_from_channelmanager_and_duration_since_epoch<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>(
 	channelmanager: &ChannelManager<Signer, M, T, K, F, L>, keys_manager: K, network: Currency,
 	amt_msat: Option<u64>, description: String, duration_since_epoch: Duration,
+	invoice_expiry_delta_secs: u32
 ) -> Result<Invoice, SignOrCreationError<()>>
 where
 	M::Target: chain::Watch<Signer>,
@@ -300,13 +310,14 @@ where
 		InvoiceDescription::Direct(
 			&Description::new(description).map_err(SignOrCreationError::CreationError)?,
 		),
-		duration_since_epoch,
+		duration_since_epoch, invoice_expiry_delta_secs
 	)
 }
 
 fn _create_invoice_from_channelmanager_and_duration_since_epoch<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>(
 	channelmanager: &ChannelManager<Signer, M, T, K, F, L>, keys_manager: K, network: Currency,
 	amt_msat: Option<u64>, description: InvoiceDescription, duration_since_epoch: Duration,
+	invoice_expiry_delta_secs: u32
 ) -> Result<Invoice, SignOrCreationError<()>>
 where
 	M::Target: chain::Watch<Signer>,
@@ -320,7 +331,7 @@ where
 	// `create_inbound_payment` only returns an error if the amount is greater than the total bitcoin
 	// supply.
 	let (payment_hash, payment_secret) = channelmanager
-		.create_inbound_payment(amt_msat, DEFAULT_EXPIRY_TIME.try_into().unwrap())
+		.create_inbound_payment(amt_msat, invoice_expiry_delta_secs)
 		.map_err(|()| SignOrCreationError::CreationError(CreationError::InvalidAmount))?;
 	let our_node_pubkey = channelmanager.get_our_node_id();
 
@@ -337,7 +348,8 @@ where
 		.payment_hash(Hash::from_slice(&payment_hash.0).unwrap())
 		.payment_secret(payment_secret)
 		.basic_mpp()
-		.min_final_cltv_expiry(MIN_FINAL_CLTV_EXPIRY.into());
+		.min_final_cltv_expiry(MIN_FINAL_CLTV_EXPIRY.into())
+		.expiry_time(Duration::from_secs(invoice_expiry_delta_secs.into()));
 	if let Some(amt) = amt_msat {
 		invoice = invoice.amount_milli_satoshis(amt);
 	}
@@ -526,12 +538,14 @@ mod test {
 		let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
 		let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
 		create_unannounced_chan_between_nodes_with_value(&nodes, 0, 1, 100000, 10001, InitFeatures::known(), InitFeatures::known());
+		let non_default_invoice_expiry_secs = 4200;
 		let invoice = create_invoice_from_channelmanager_and_duration_since_epoch(
 			&nodes[1].node, nodes[1].keys_manager, Currency::BitcoinTestnet, Some(10_000), "test".to_string(),
-			Duration::from_secs(1234567)).unwrap();
+			Duration::from_secs(1234567), non_default_invoice_expiry_secs).unwrap();
 		assert_eq!(invoice.amount_pico_btc(), Some(100_000));
 		assert_eq!(invoice.min_final_cltv_expiry(), MIN_FINAL_CLTV_EXPIRY as u64);
 		assert_eq!(invoice.description(), InvoiceDescription::Direct(&Description("test".to_string())));
+		assert_eq!(invoice.expiry_time(), Duration::from_secs(non_default_invoice_expiry_secs.into()));
 
 		// Invoice SCIDs should always use inbound SCID aliases over the real channel ID, if one is
 		// available.
@@ -592,7 +606,7 @@ mod test {
 		let description_hash = crate::Sha256(Hash::hash("Testing description_hash".as_bytes()));
 		let invoice = ::utils::create_invoice_from_channelmanager_with_description_hash_and_duration_since_epoch(
 			&nodes[1].node, nodes[1].keys_manager, Currency::BitcoinTestnet, Some(10_000),
-			description_hash, Duration::from_secs(1234567),
+			description_hash, Duration::from_secs(1234567), 3600
 		).unwrap();
 		assert_eq!(invoice.amount_pico_btc(), Some(100_000));
 		assert_eq!(invoice.min_final_cltv_expiry(), MIN_FINAL_CLTV_EXPIRY as u64);
@@ -752,7 +766,7 @@ mod test {
 	) {
 		let invoice = create_invoice_from_channelmanager_and_duration_since_epoch(
 			&invoice_node.node, invoice_node.keys_manager, Currency::BitcoinTestnet, invoice_amt, "test".to_string(),
-			Duration::from_secs(1234567)).unwrap();
+			Duration::from_secs(1234567), 3600).unwrap();
 		let hints = invoice.private_routes();
 
 		for hint in hints {
@@ -799,8 +813,13 @@ mod test {
 		} else {
 			None
 		};
+		let non_default_invoice_expiry_secs = 4200;
 
-		let invoice = ::utils::create_phantom_invoice::<EnforcingSigner, &test_utils::TestKeysInterface>(Some(payment_amt), payment_hash, "test".to_string(), 3600, route_hints, &nodes[1].keys_manager, Currency::BitcoinTestnet).unwrap();
+		let invoice =
+			::utils::create_phantom_invoice::<EnforcingSigner, &test_utils::TestKeysInterface>(
+				Some(payment_amt), payment_hash, "test".to_string(), non_default_invoice_expiry_secs,
+				route_hints, &nodes[1].keys_manager, Currency::BitcoinTestnet
+			).unwrap();
 		let (payment_hash, payment_secret) = (PaymentHash(invoice.payment_hash().into_inner()), *invoice.payment_secret());
 		let payment_preimage = if user_generated_pmt_hash {
 			user_payment_preimage
@@ -811,6 +830,7 @@ mod test {
 		assert_eq!(invoice.min_final_cltv_expiry(), MIN_FINAL_CLTV_EXPIRY as u64);
 		assert_eq!(invoice.description(), InvoiceDescription::Direct(&Description("test".to_string())));
 		assert_eq!(invoice.route_hints().len(), 2);
+		assert_eq!(invoice.expiry_time(), Duration::from_secs(non_default_invoice_expiry_secs.into()));
 		assert!(!invoice.features().unwrap().supports_basic_mpp());
 
 		let payment_params = PaymentParameters::from_node_id(invoice.recover_payee_pub_key())
@@ -931,10 +951,17 @@ mod test {
 		];
 
 		let description_hash = crate::Sha256(Hash::hash("Description hash phantom invoice".as_bytes()));
-		let invoice = ::utils::create_phantom_invoice_with_description_hash::<EnforcingSigner,&test_utils::TestKeysInterface>(Some(payment_amt), None, 3600, description_hash, route_hints, &nodes[1].keys_manager, Currency::BitcoinTestnet).unwrap();
-
+		let non_default_invoice_expiry_secs = 4200;
+		let invoice = ::utils::create_phantom_invoice_with_description_hash::<
+			EnforcingSigner, &test_utils::TestKeysInterface,
+		>(
+			Some(payment_amt), None, non_default_invoice_expiry_secs, description_hash,
+			route_hints, &nodes[1].keys_manager, Currency::BitcoinTestnet
+		)
+		.unwrap();
 		assert_eq!(invoice.amount_pico_btc(), Some(200_000));
 		assert_eq!(invoice.min_final_cltv_expiry(), MIN_FINAL_CLTV_EXPIRY as u64);
+		assert_eq!(invoice.expiry_time(), Duration::from_secs(non_default_invoice_expiry_secs.into()));
 		assert_eq!(invoice.description(), InvoiceDescription::Hash(&crate::Sha256(Sha256::hash("Description hash phantom invoice".as_bytes()))));
 	}
 

lightning-net-tokio/src/lib.rs

@@ -120,11 +120,28 @@ struct Connection {
 	id: u64,
 }
 impl Connection {
+	async fn poll_event_process<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, mut event_receiver: mpsc::Receiver<()>) where
+			CMH: ChannelMessageHandler + 'static + Send + Sync,
+			RMH: RoutingMessageHandler + 'static + Send + Sync,
+			L: Logger + 'static + ?Sized + Send + Sync,
+			UMH: CustomMessageHandler + 'static + Send + Sync {
+		loop {
+			if event_receiver.recv().await.is_none() {
+				return;
+			}
+			peer_manager.process_events();
+		}
+	}
+
 	async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
-			CMH: ChannelMessageHandler + 'static,
-			RMH: RoutingMessageHandler + 'static,
-			L: Logger + 'static + ?Sized,
-			UMH: CustomMessageHandler + 'static {
+			CMH: ChannelMessageHandler + 'static + Send + Sync,
+			RMH: RoutingMessageHandler + 'static + Send + Sync,
+			L: Logger + 'static + ?Sized + Send + Sync,
+			UMH: CustomMessageHandler + 'static + Send + Sync {
+		// Create a waker to wake up poll_event_process, above
+		let (event_waker, event_receiver) = mpsc::channel(1);
+		tokio::spawn(Self::poll_event_process(Arc::clone(&peer_manager), event_receiver));
+
 		// 8KB is nice and big but also should never cause any issues with stack overflowing.
 		let mut buf = [0; 8192];
 
@@ -175,7 +192,14 @@ impl Connection {
 					Err(_) => break Disconnect::PeerDisconnected,
 				},
 			}
-			peer_manager.process_events();
+			let _ = event_waker.try_send(());
+
+			// At this point we've processed a message or two, and reset the ping timer for this
+			// peer, at least in the "are we still receiving messages" context, if we don't give up
+			// our timeslice to another task we may just spin on this peer, starving other peers
+			// and eventually disconnecting them for ping timeouts. Instead, we explicitly yield
+			// here.
+			tokio::task::yield_now().await;
 		};
 		let writer_option = us.lock().unwrap().writer.take();
 		if let Some(mut writer) = writer_option {
@@ -443,6 +467,9 @@ impl peer_handler::SocketDescriptor for SocketDescriptor {
 					// pause read given we're now waiting on the remote end to ACK (and in
 					// accordance with the send_data() docs).
 					us.read_paused = true;
+					// Further, to avoid any current pending read causing a `read_event` call, wake
+					// up the read_waker and restart its loop.
+					let _ = us.read_waker.try_send(());
 					return written_len;
 				},
 			}

lightning/src/debug_sync.rs

@@ -362,3 +362,5 @@ fn read_write_lockorder_fail() {
 		let _a = a.write().unwrap();
 	}
 }
+
+pub type FairRwLock<T> = RwLock<T>;

lightning/src/lib.rs

@@ -193,6 +193,8 @@ mod sync {
 	pub use debug_sync::*;
 	#[cfg(not(test))]
 	pub use ::std::sync::{Arc, Mutex, Condvar, MutexGuard, RwLock, RwLockReadGuard};
+	#[cfg(not(test))]
+	pub use crate::util::fairrwlock::FairRwLock;
 }
 
 #[cfg(not(feature = "std"))]