Track manual BOLT11 invoices

Manual BOLT11 invoices need a pending entry before HTLC arrival.
That lets duplicate registrations be rejected while keeping randomized
payment IDs.

Store the pending entries with their invoice expiry and resolve BOLT11
claimable events by scanning those entries.

Co-Authored-By: HAL 9000

Author: tnull

src/builder.rs

@@ -2149,6 +2149,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		pending_payment_store,
 		lnurl_auth,
 		is_running,
 		node_metrics,

src/event.rs

@@ -10,6 +10,7 @@ use core::task::{Poll, Waker};
 use std::collections::VecDeque;
 use std::ops::Deref;
 use std::sync::{Arc, Mutex};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use bitcoin::blockdata::locktime::absolute::LockTime;
 use bitcoin::secp256k1::PublicKey;
@@ -53,7 +54,8 @@ use crate::payment::store::{
 use crate::payment::PaymentMetadata;
 use crate::runtime::Runtime;
 use crate::types::{
-	CustomTlvRecord, DynStore, KeysManager, OnionMessenger, PaymentStore, Sweeper, Wallet,
+	CustomTlvRecord, DynStore, KeysManager, OnionMessenger, PaymentStore, PendingPaymentStore,
+	Sweeper, Wallet,
 };
 use crate::{
 	hex_utils, BumpTransactionEventHandler, ChannelManager, Error, Graph, PeerInfo, PeerStore,
@@ -529,6 +531,7 @@ where
 	network_graph: Arc<Graph>,
 	liquidity_source: Arc<LiquiditySource<Arc<Logger>>>,
 	payment_store: Arc<PaymentStore>,
+	pending_payment_store: Arc<PendingPaymentStore>,
 	peer_store: Arc<PeerStore<L>>,
 	keys_manager: Arc<KeysManager>,
 	runtime: Arc<Runtime>,
@@ -549,10 +552,10 @@ where
 		channel_manager: Arc<ChannelManager>, connection_manager: Arc<ConnectionManager<L>>,
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Arc<LiquiditySource<Arc<Logger>>>, payment_store: Arc<PaymentStore>,
-		peer_store: Arc<PeerStore<L>>, keys_manager: Arc<KeysManager>,
-		static_invoice_store: Option<StaticInvoiceStore>, onion_messenger: Arc<OnionMessenger>,
-		om_mailbox: Option<Arc<OnionMessageMailbox>>, runtime: Arc<Runtime>, logger: L,
-		config: Arc<Config>,
+		pending_payment_store: Arc<PendingPaymentStore>, peer_store: Arc<PeerStore<L>>,
+		keys_manager: Arc<KeysManager>, static_invoice_store: Option<StaticInvoiceStore>,
+		onion_messenger: Arc<OnionMessenger>, om_mailbox: Option<Arc<OnionMessageMailbox>>,
+		runtime: Arc<Runtime>, logger: L, config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -564,6 +567,7 @@ where
 			network_graph,
 			liquidity_source,
 			payment_store,
+			pending_payment_store,
 			peer_store,
 			keys_manager,
 			logger,
@@ -605,6 +609,59 @@ where
 		})
 	}
 
+	fn current_time_secs() -> u64 {
+		SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs()
+	}
+
+	async fn prune_expired_pending_payments(&self) -> Result<(), ReplayEvent> {
+		let now = Self::current_time_secs();
+		let expired_payment_ids = self
+			.pending_payment_store
+			.list_filter(|payment| payment.has_expired(now))
+			.into_iter()
+			.map(|payment| payment.details.id)
+			.collect::<Vec<_>>();
+
+		for payment_id in expired_payment_ids {
+			if let Err(e) = self.pending_payment_store.remove(&payment_id).await {
+				log_error!(
+					self.logger,
+					"Failed to remove expired pending payment with ID {}: {}",
+					payment_id,
+					e
+				);
+				return Err(ReplayEvent());
+			}
+		}
+
+		Ok(())
+	}
+
+	fn find_inbound_payment_by_hash(&self, payment_hash: &PaymentHash) -> Option<PaymentDetails> {
+		self.payment_store
+			.list_filter(|payment| {
+				payment.direction == PaymentDirection::Inbound
+					&& matches!(&payment.kind, PaymentKind::Bolt11 { hash, .. } if hash == payment_hash)
+			})
+			.first()
+			.cloned()
+	}
+
+	async fn find_pending_inbound_payment_id(
+		&self, payment_hash: &PaymentHash,
+	) -> Result<Option<PaymentId>, ReplayEvent> {
+		self.prune_expired_pending_payments().await?;
+		Ok(self
+			.pending_payment_store
+			.list_filter(|payment| {
+				payment.details.direction == PaymentDirection::Inbound
+					&& payment.details.status == PaymentStatus::Pending
+					&& matches!(&payment.details.kind, PaymentKind::Bolt11 { hash, .. } if hash == payment_hash)
+			})
+			.first()
+			.map(|payment| payment.details.id))
+	}
+
 	fn resolve_inbound_payment_id(
 		&self, event_payment_id: PaymentId, payment_hash: &PaymentHash,
 	) -> (PaymentId, Option<PaymentDetails>) {
@@ -619,6 +676,10 @@ where
 			}
 		}
 
+		if let Some(info) = self.find_inbound_payment_by_hash(payment_hash) {
+			return (info.id, Some(info));
+		}
+
 		(event_payment_id, None)
 	}
 
@@ -735,8 +796,16 @@ where
 				..
 			} => {
 				let event_payment_id = payment_id.unwrap_or(PaymentId(payment_hash.0));
-				let (payment_id, payment_info) =
+				let (mut payment_id, payment_info) =
 					self.resolve_inbound_payment_id(event_payment_id, &payment_hash);
+				let pending_payment_id = if payment_info.is_none() {
+					self.find_pending_inbound_payment_id(&payment_hash).await?
+				} else {
+					None
+				};
+				if let Some(pending_payment_id) = pending_payment_id {
+					payment_id = pending_payment_id;
+				}
 				if let Some(info) = payment_info.as_ref() {
 					if info.direction == PaymentDirection::Outbound {
 						log_info!(
@@ -911,6 +980,19 @@ where
 									return Err(ReplayEvent());
 								},
 							}
+
+							if pending_payment_id.is_some() {
+								if let Err(e) = self.pending_payment_store.remove(&payment_id).await
+								{
+									log_error!(
+										self.logger,
+										"Failed to remove pending payment with ID {}: {}",
+										payment_id,
+										e
+									);
+									return Err(ReplayEvent());
+								}
+							}
 						}
 
 						if payment_preimage.is_none() {

src/lib.rs

@@ -176,8 +176,8 @@ use runtime::Runtime;
 pub use tokio;
 use types::{
 	Broadcaster, BumpTransactionEventHandler, ChainMonitor, ChannelManager, DynStore, Graph,
-	HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, Router, Scorer, Sweeper,
-	Wallet,
+	HRNResolver, KeysManager, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore,
+	Router, Scorer, Sweeper, Wallet,
 };
 pub use types::{
 	ChannelCounterparty, ChannelDetails, CustomTlvRecord, PeerDetails, ReserveType, UserChannelId,
@@ -244,6 +244,7 @@ pub struct Node {
 	scorer: Arc<Mutex<Scorer>>,
 	peer_store: Arc<PeerStore<Arc<Logger>>>,
 	payment_store: Arc<PaymentStore>,
+	pending_payment_store: Arc<PendingPaymentStore>,
 	lnurl_auth: Arc<LnurlAuth>,
 	is_running: Arc<RwLock<bool>>,
 	node_metrics: Arc<PersistedNodeMetrics>,
@@ -605,6 +606,7 @@ impl Node {
 			Arc::clone(&self.network_graph),
 			Arc::clone(&self.liquidity_source),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.pending_payment_store),
 			Arc::clone(&self.peer_store),
 			Arc::clone(&self.keys_manager),
 			static_invoice_store,
@@ -949,6 +951,7 @@ impl Node {
 			Arc::clone(&self.connection_manager),
 			Arc::clone(&self.liquidity_source),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.pending_payment_store),
 			Arc::clone(&self.peer_store),
 			Arc::clone(&self.config),
 			Arc::clone(&self.is_running),
@@ -968,6 +971,7 @@ impl Node {
 			Arc::clone(&self.connection_manager),
 			Arc::clone(&self.liquidity_source),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.pending_payment_store),
 			Arc::clone(&self.peer_store),
 			Arc::clone(&self.config),
 			Arc::clone(&self.is_running),

src/payment/bolt11.rs

@@ -10,6 +10,7 @@
 //! [BOLT 11]: https://github.com/lightning/bolts/blob/master/11-payment-encoding.md
 
 use std::sync::{Arc, RwLock};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use bitcoin::hashes::sha256::Hash as Sha256;
 use bitcoin::hashes::Hash;
@@ -23,7 +24,7 @@ use lightning::sign::EntropySource;
 use lightning_invoice::{
 	Bolt11Invoice as LdkBolt11Invoice, Bolt11InvoiceDescription as LdkBolt11InvoiceDescription,
 };
-use lightning_types::payment::{PaymentHash, PaymentPreimage};
+use lightning_types::payment::{PaymentHash, PaymentPreimage, PaymentSecret};
 
 use crate::config::{Config, LDK_PAYMENT_RETRY_TIMEOUT};
 use crate::connection::ConnectionManager;
@@ -36,9 +37,10 @@ use crate::payment::store::{
 	LSPS2Parameters, PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind,
 	PaymentStatus,
 };
+use crate::payment::PendingPaymentDetails;
 use crate::peer_store::{PeerInfo, PeerStore};
 use crate::runtime::Runtime;
-use crate::types::{ChannelManager, KeysManager, PaymentStore};
+use crate::types::{ChannelManager, KeysManager, PaymentStore, PendingPaymentStore};
 
 #[cfg(not(feature = "uniffi"))]
 type Bolt11Invoice = LdkBolt11Invoice;
@@ -74,6 +76,7 @@ pub struct Bolt11Payment {
 	connection_manager: Arc<ConnectionManager<Arc<Logger>>>,
 	liquidity_source: Arc<LiquiditySource<Arc<Logger>>>,
 	payment_store: Arc<PaymentStore>,
+	pending_payment_store: Arc<PendingPaymentStore>,
 	peer_store: Arc<PeerStore<Arc<Logger>>>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
@@ -85,8 +88,8 @@ impl Bolt11Payment {
 		runtime: Arc<Runtime>, channel_manager: Arc<ChannelManager>,
 		keys_manager: Arc<KeysManager>, connection_manager: Arc<ConnectionManager<Arc<Logger>>>,
 		liquidity_source: Arc<LiquiditySource<Arc<Logger>>>, payment_store: Arc<PaymentStore>,
-		peer_store: Arc<PeerStore<Arc<Logger>>>, config: Arc<Config>,
-		is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
+		pending_payment_store: Arc<PendingPaymentStore>, peer_store: Arc<PeerStore<Arc<Logger>>>,
+		config: Arc<Config>, is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
 	) -> Self {
 		Self {
 			runtime,
@@ -95,17 +98,94 @@ impl Bolt11Payment {
 			connection_manager,
 			liquidity_source,
 			payment_store,
+			pending_payment_store,
 			peer_store,
 			config,
 			is_running,
 			logger,
 		}
 	}
 
+	fn current_time_secs() -> u64 {
+		SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs()
+	}
+
+	fn prune_expired_pending_payments(&self) -> Result<(), Error> {
+		let now = Self::current_time_secs();
+		let expired_payment_ids = self
+			.pending_payment_store
+			.list_filter(|payment| payment.has_expired(now))
+			.into_iter()
+			.map(|payment| payment.details.id)
+			.collect::<Vec<_>>();
+
+		for payment_id in expired_payment_ids {
+			self.runtime.block_on(self.pending_payment_store.remove(&payment_id))?;
+		}
+
+		Ok(())
+	}
+
+	fn has_pending_or_succeeded_inbound_payment(&self, payment_hash: &PaymentHash) -> bool {
+		!self
+			.payment_store
+			.list_filter(|payment| {
+				payment.direction == PaymentDirection::Inbound
+					&& matches!(&payment.kind, PaymentKind::Bolt11 { hash, .. } if hash == payment_hash)
+					&& matches!(payment.status, PaymentStatus::Pending | PaymentStatus::Succeeded)
+			})
+			.is_empty()
+			|| !self
+				.pending_payment_store
+				.list_filter(|payment| {
+					payment.details.direction == PaymentDirection::Inbound
+						&& matches!(&payment.details.kind, PaymentKind::Bolt11 { hash, .. } if hash == payment_hash)
+						&& matches!(
+							payment.details.status,
+							PaymentStatus::Pending | PaymentStatus::Succeeded
+						)
+				})
+				.is_empty()
+	}
+
+	fn register_manual_claim_invoice(
+		&self, payment_hash: PaymentHash, amount_msat: Option<u64>, payment_secret: PaymentSecret,
+		expiry_secs: u32,
+	) -> Result<(), Error> {
+		let payment_id = PaymentId(self.keys_manager.get_secure_random_bytes());
+		let kind = PaymentKind::Bolt11 {
+			hash: payment_hash,
+			preimage: None,
+			secret: Some(payment_secret),
+			counterparty_skimmed_fee_msat: None,
+		};
+		let payment = PaymentDetails::new(
+			payment_id,
+			kind,
+			amount_msat,
+			None,
+			PaymentDirection::Inbound,
+			PaymentStatus::Pending,
+		);
+		let expires_at = Some(Self::current_time_secs().saturating_add(expiry_secs as u64));
+		let pending_payment =
+			PendingPaymentDetails::new_with_expiry(payment, Vec::new(), expires_at);
+		self.runtime.block_on(self.pending_payment_store.insert_or_update(pending_payment))?;
+		Ok(())
+	}
+
 	pub(crate) fn receive_inner(
 		&self, amount_msat: Option<u64>, invoice_description: &LdkBolt11InvoiceDescription,
 		expiry_secs: u32, manual_claim_payment_hash: Option<PaymentHash>,
 	) -> Result<LdkBolt11Invoice, Error> {
+		if let Some(payment_hash) = manual_claim_payment_hash {
+			self.prune_expired_pending_payments()?;
+			if self.has_pending_or_succeeded_inbound_payment(&payment_hash) {
+				log_error!(self.logger, "Payment error: an invoice must not be paid twice.");
+				return Err(Error::DuplicatePayment);
+			}
+		}
+
 		let invoice = {
 			let invoice_params = Bolt11InvoiceParameters {
 				amount_msats: amount_msat,
@@ -127,6 +207,15 @@ impl Bolt11Payment {
 			}
 		};
 
+		if let Some(payment_hash) = manual_claim_payment_hash {
+			self.register_manual_claim_invoice(
+				payment_hash,
+				amount_msat,
+				*invoice.payment_secret(),
+				expiry_secs,
+			)?;
+		}
+
 		Ok(invoice)
 	}
 
@@ -135,6 +224,14 @@ impl Bolt11Payment {
 		expiry_secs: u32, max_total_lsp_fee_limit_msat: Option<u64>,
 		max_proportional_lsp_fee_limit_ppm_msat: Option<u64>, payment_hash: Option<PaymentHash>,
 	) -> Result<LdkBolt11Invoice, Error> {
+		if let Some(payment_hash) = payment_hash {
+			self.prune_expired_pending_payments()?;
+			if self.has_pending_or_succeeded_inbound_payment(&payment_hash) {
+				log_error!(self.logger, "Payment error: an invoice must not be paid twice.");
+				return Err(Error::DuplicatePayment);
+			}
+		}
+
 		let connection_manager = Arc::clone(&self.connection_manager);
 		let (invoice, chosen_lsp) = self.runtime.block_on(async move {
 			if let Some(amount_msat) = amount_msat {
@@ -167,6 +264,15 @@ impl Bolt11Payment {
 		let peer_info = PeerInfo { node_id: chosen_lsp.node_id, address: chosen_lsp.address };
 		self.runtime.block_on(self.peer_store.add_peer(peer_info))?;
 
+		if let Some(payment_hash) = payment_hash {
+			self.register_manual_claim_invoice(
+				payment_hash,
+				amount_msat,
+				*invoice.payment_secret(),
+				expiry_secs,
+			)?;
+		}
+
 		Ok(invoice)
 	}
 }