Track manual BOLT11 invoices

Manual BOLT11 invoices need a pending entry before HTLC arrival.
That lets duplicate registrations be rejected without scanning
finalized payments.

Keep the legacy hash-derived payment ID in this step; the next
commit switches BOLT11 payments to randomized IDs.

Co-Authored-By: HAL 9000

Author: tnull

src/builder.rs

@@ -2245,6 +2245,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;
@@ -50,10 +51,11 @@ use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore;
 use crate::payment::store::{
 	PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus,
 };
-use crate::payment::PaymentMetadata;
+use crate::payment::{PaymentMetadata, PendingPaymentDetails};
 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,47 @@ 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(())
+	}
+
+	async fn find_pending_inbound_payment(
+		&self, payment_hash: &PaymentHash,
+	) -> Result<Option<PendingPaymentDetails>, ReplayEvent> {
+		self.prune_expired_pending_payments().await?;
+		Ok(self.pending_payment_store.list_by_payment_hash(payment_hash).into_iter().find(
+			|payment| {
+				payment.details.direction == PaymentDirection::Inbound
+					&& payment.details.status == PaymentStatus::Pending
+					&& matches!(&payment.details.kind, PaymentKind::Bolt11 { hash, .. } if hash == payment_hash)
+			},
+		))
+	}
+
 	pub async fn handle_event(&self, event: LdkEvent) -> Result<(), ReplayEvent> {
 		match event {
 			LdkEvent::FundingGenerationReady {
@@ -718,6 +763,11 @@ where
 			} => {
 				let payment_id = PaymentId(payment_hash.0);
 				let payment_info = self.payment_store.get(&payment_id);
+				let pending_payment = if payment_info.is_none() {
+					self.find_pending_inbound_payment(&payment_hash).await?
+				} else {
+					None
+				};
 				if let Some(info) = payment_info.as_ref() {
 					if info.direction == PaymentDirection::Outbound {
 						log_info!(
@@ -803,6 +853,17 @@ where
 							counterparty_skimmed_fee_msat,
 						);
 						self.fail_claimable_payment(payment_id, &payment_hash).await?;
+						if pending_payment.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());
+							}
+						}
 						return Ok(());
 					};
 
@@ -815,6 +876,17 @@ where
 							max_total_opening_fee_msat,
 						);
 						self.fail_claimable_payment(payment_id, &payment_hash).await?;
+						if pending_payment.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());
+							}
+						}
 						return Ok(());
 					}
 
@@ -838,6 +910,57 @@ where
 					}
 				}
 
+				let payment_info = if let Some(pending_payment) = pending_payment.as_ref() {
+					let mut payment = pending_payment.details.clone();
+					if let PaymentKind::Bolt11 {
+						counterparty_skimmed_fee_msat: stored_fee, ..
+					} = &mut payment.kind
+					{
+						if counterparty_skimmed_fee_msat > 0 {
+							*stored_fee = Some(counterparty_skimmed_fee_msat);
+						}
+					}
+
+					match self.payment_store.insert(payment.clone()).await {
+						Ok(false) => (),
+						Ok(true) => {
+							log_error!(
+								self.logger,
+								"Bolt11InvoicePayment with ID {} was previously known",
+								payment_id,
+							);
+							debug_assert!(false);
+						},
+						Err(e) => {
+							log_error!(
+								self.logger,
+								"Failed to insert payment with ID {}: {}",
+								payment_id,
+								e
+							);
+							return Err(ReplayEvent());
+						},
+					}
+
+					let mut pending_payment = pending_payment.clone();
+					pending_payment.expires_at = None;
+					if let Err(e) =
+						self.pending_payment_store.insert_or_update(pending_payment).await
+					{
+						log_error!(
+							self.logger,
+							"Failed to update pending payment with ID {}: {}",
+							payment_id,
+							e
+						);
+						return Err(ReplayEvent());
+					}
+
+					Some(payment)
+				} else {
+					payment_info
+				};
+
 				if let Some(info) = payment_info {
 					// If this is known by the store but ChannelManager doesn't know the preimage,
 					// the payment has been registered via `_for_hash` variants and needs to be manually claimed via
@@ -1088,6 +1211,16 @@ where
 					},
 				}
 
+				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());
+				}
+
 				let event = Event::PaymentReceived {
 					payment_id,
 					payment_hash,

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,
@@ -948,6 +950,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),
@@ -966,6 +969,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),