Wire PaymentMetadataStore into Builder, Node, Bolt11Payment, and EventHandler

Add `read_payment_metadata` to the startup `tokio::join!` call in the
builder to load persisted metadata entries. Construct an
`Arc<PaymentMetadataStore>` and thread it through to `Node`,
`Bolt11Payment` (along with `KeysManager`), and `EventHandler`.

Update `Node::remove_payment` to also call
`payment_metadata_store.remove_payment_id` so the reverse index stays
consistent when payment store entries are removed.

Generated with the assistance of AI tools.

Co-Authored-By: HAL 9000

Author: tnull

src/builder.rs

@@ -56,8 +56,8 @@ use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
 	read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
-	read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_pending_payments,
-	read_scorer, write_node_metrics,
+	read_node_metrics, read_output_sweeper, read_payment_metadata, read_payments, read_peer_info,
+	read_pending_payments, read_scorer, write_node_metrics,
 };
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
@@ -72,6 +72,7 @@ use crate::lnurl_auth::LnurlAuth;
 use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger};
 use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
+use crate::payment::metadata_store::PaymentMetadataStore;
 use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
@@ -1267,12 +1268,13 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (payment_store_res, node_metris_res, pending_payment_store_res) =
+	let (payment_store_res, node_metris_res, pending_payment_store_res, payment_metadata_res) =
 		runtime.block_on(async move {
 			tokio::join!(
 				read_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
 				read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
-				read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref))
+				read_pending_payments(&*kv_store_ref, Arc::clone(&logger_ref)),
+				read_payment_metadata(&*kv_store_ref, Arc::clone(&logger_ref))
 			)
 		});
 
@@ -1303,6 +1305,18 @@ fn build_with_store_internal(
 		},
 	};
 
+	let payment_metadata_store = match payment_metadata_res {
+		Ok(metadata_entries) => Arc::new(PaymentMetadataStore::new(
+			metadata_entries,
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read payment metadata from store: {}", e);
+			return Err(BuildError::ReadFailed);
+		},
+	};
+
 	let (chain_source, chain_tip_opt) = match chain_data_source_config {
 		Some(ChainDataSourceConfig::Esplora { server_url, headers, sync_config }) => {
 			let sync_config = sync_config.unwrap_or(EsploraSyncConfig::default());
@@ -1996,6 +2010,7 @@ fn build_with_store_internal(
 		scorer,
 		peer_store,
 		payment_store,
+		payment_metadata_store,
 		lnurl_auth,
 		is_running,
 		node_metrics,

src/event.rs

@@ -49,6 +49,7 @@ use crate::liquidity::LiquiditySource;
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore;
+use crate::payment::metadata_store::PaymentMetadataStore;
 use crate::payment::store::{
 	PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus,
 };
@@ -507,6 +508,7 @@ where
 	network_graph: Arc<Graph>,
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	payment_store: Arc<PaymentStore>,
+	payment_metadata_store: Arc<PaymentMetadataStore>,
 	peer_store: Arc<PeerStore<L>>,
 	keys_manager: Arc<KeysManager>,
 	runtime: Arc<Runtime>,
@@ -527,10 +529,11 @@ where
 		channel_manager: Arc<ChannelManager>, connection_manager: Arc<ConnectionManager<L>>,
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<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>,
+		payment_store: Arc<PaymentStore>, payment_metadata_store: Arc<PaymentMetadataStore>,
+		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,
@@ -542,6 +545,7 @@ where
 			network_graph,
 			liquidity_source,
 			payment_store,
+			payment_metadata_store,
 			peer_store,
 			keys_manager,
 			logger,

src/io/utils.rs

@@ -44,6 +44,7 @@ use crate::io::{
 	NODE_METRICS_KEY, NODE_METRICS_PRIMARY_NAMESPACE, NODE_METRICS_SECONDARY_NAMESPACE,
 };
 use crate::logger::{log_error, LdkLogger, Logger};
+use crate::payment::metadata_store::PaymentMetadataEntry;
 use crate::payment::PendingPaymentDetails;
 use crate::peer_store::PeerStore;
 use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
@@ -298,6 +299,83 @@ where
 	Ok(res)
 }
 
+/// Read previously persisted payment metadata from the store.
+pub(crate) async fn read_payment_metadata<L: Deref>(
+	kv_store: &DynStore, logger: L,
+) -> Result<Vec<PaymentMetadataEntry>, std::io::Error>
+where
+	L::Target: LdkLogger,
+{
+	let mut res = Vec::new();
+
+	let mut stored_keys = KVStore::list(
+		&*kv_store,
+		PAYMENT_METADATA_PERSISTENCE_PRIMARY_NAMESPACE,
+		PAYMENT_METADATA_PERSISTENCE_SECONDARY_NAMESPACE,
+	)
+	.await?;
+
+	const BATCH_SIZE: usize = 50;
+
+	let mut set = tokio::task::JoinSet::new();
+
+	// Fill JoinSet with tasks if possible
+	while set.len() < BATCH_SIZE && !stored_keys.is_empty() {
+		if let Some(next_key) = stored_keys.pop() {
+			let fut = KVStore::read(
+				&*kv_store,
+				PAYMENT_METADATA_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_METADATA_PERSISTENCE_SECONDARY_NAMESPACE,
+				&next_key,
+			);
+			set.spawn(fut);
+			debug_assert!(set.len() <= BATCH_SIZE);
+		}
+	}
+
+	while let Some(read_res) = set.join_next().await {
+		// Exit early if we get an IO error.
+		let reader = read_res
+			.map_err(|e| {
+				log_error!(logger, "Failed to read PaymentMetadataEntry: {}", e);
+				set.abort_all();
+				e
+			})?
+			.map_err(|e| {
+				log_error!(logger, "Failed to read PaymentMetadataEntry: {}", e);
+				set.abort_all();
+				e
+			})?;
+
+		// Refill set for every finished future, if we still have something to do.
+		if let Some(next_key) = stored_keys.pop() {
+			let fut = KVStore::read(
+				&*kv_store,
+				PAYMENT_METADATA_PERSISTENCE_PRIMARY_NAMESPACE,
+				PAYMENT_METADATA_PERSISTENCE_SECONDARY_NAMESPACE,
+				&next_key,
+			);
+			set.spawn(fut);
+			debug_assert!(set.len() <= BATCH_SIZE);
+		}
+
+		// Handle result.
+		let entry = PaymentMetadataEntry::read(&mut &*reader).map_err(|e| {
+			log_error!(logger, "Failed to deserialize PaymentMetadataEntry: {}", e);
+			std::io::Error::new(
+				std::io::ErrorKind::InvalidData,
+				"Failed to deserialize PaymentMetadataEntry",
+			)
+		})?;
+		res.push(entry);
+	}
+
+	debug_assert!(set.is_empty());
+	debug_assert!(stored_keys.is_empty());
+
+	Ok(res)
+}
+
 /// Read `OutputSweeper` state from the store.
 pub(crate) async fn read_output_sweeper(
 	broadcaster: Arc<Broadcaster>, fee_estimator: Arc<OnchainFeeEstimator>,

src/lib.rs

@@ -165,6 +165,7 @@ use lnurl_auth::LnurlAuth;
 use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use payment::asynchronous::static_invoice_store::StaticInvoiceStore;
+use payment::metadata_store::PaymentMetadataStore;
 use payment::{
 	Bolt11Payment, Bolt12Payment, OnchainPayment, PaymentDetails, SpontaneousPayment,
 	UnifiedPayment,
@@ -233,6 +234,7 @@ pub struct Node {
 	scorer: Arc<Mutex<Scorer>>,
 	peer_store: Arc<PeerStore<Arc<Logger>>>,
 	payment_store: Arc<PaymentStore>,
+	payment_metadata_store: Arc<PaymentMetadataStore>,
 	lnurl_auth: Arc<LnurlAuth>,
 	is_running: Arc<RwLock<bool>>,
 	node_metrics: Arc<RwLock<NodeMetrics>>,
@@ -600,6 +602,7 @@ impl Node {
 			Arc::clone(&self.network_graph),
 			self.liquidity_source.clone(),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.payment_metadata_store),
 			Arc::clone(&self.peer_store),
 			Arc::clone(&self.keys_manager),
 			static_invoice_store,
@@ -891,6 +894,8 @@ impl Node {
 			Arc::clone(&self.connection_manager),
 			self.liquidity_source.clone(),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.payment_metadata_store),
+			Arc::clone(&self.keys_manager),
 			Arc::clone(&self.peer_store),
 			Arc::clone(&self.config),
 			Arc::clone(&self.is_running),
@@ -909,6 +914,8 @@ impl Node {
 			Arc::clone(&self.connection_manager),
 			self.liquidity_source.clone(),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.payment_metadata_store),
+			Arc::clone(&self.keys_manager),
 			Arc::clone(&self.peer_store),
 			Arc::clone(&self.config),
 			Arc::clone(&self.is_running),
@@ -1884,7 +1891,9 @@ impl Node {
 
 	/// Remove the payment with the given id from the store.
 	pub fn remove_payment(&self, payment_id: &PaymentId) -> Result<(), Error> {
-		self.payment_store.remove(&payment_id)
+		self.payment_store.remove(&payment_id)?;
+		self.payment_metadata_store.remove_payment_id(payment_id)?;
+		Ok(())
 	}
 
 	/// Retrieves an overview of all known balances.

src/payment/bolt11.rs

@@ -30,13 +30,14 @@ use crate::error::Error;
 use crate::ffi::{maybe_deref, maybe_try_convert_enum, maybe_wrap};
 use crate::liquidity::LiquiditySource;
 use crate::logger::{log_error, log_info, LdkLogger, Logger};
+use crate::payment::metadata_store::PaymentMetadataStore;
 use crate::payment::store::{
 	LSPFeeLimits, PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind,
 	PaymentStatus,
 };
 use crate::peer_store::{PeerInfo, PeerStore};
 use crate::runtime::Runtime;
-use crate::types::{ChannelManager, PaymentStore};
+use crate::types::{ChannelManager, KeysManager, PaymentStore};
 
 #[cfg(not(feature = "uniffi"))]
 type Bolt11Invoice = LdkBolt11Invoice;
@@ -61,6 +62,8 @@ pub struct Bolt11Payment {
 	connection_manager: Arc<ConnectionManager<Arc<Logger>>>,
 	liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
 	payment_store: Arc<PaymentStore>,
+	payment_metadata_store: Arc<PaymentMetadataStore>,
+	keys_manager: Arc<KeysManager>,
 	peer_store: Arc<PeerStore<Arc<Logger>>>,
 	config: Arc<Config>,
 	is_running: Arc<RwLock<bool>>,
@@ -72,7 +75,8 @@ impl Bolt11Payment {
 		runtime: Arc<Runtime>, channel_manager: Arc<ChannelManager>,
 		connection_manager: Arc<ConnectionManager<Arc<Logger>>>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
-		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<Arc<Logger>>>,
+		payment_store: Arc<PaymentStore>, payment_metadata_store: Arc<PaymentMetadataStore>,
+		keys_manager: Arc<KeysManager>, peer_store: Arc<PeerStore<Arc<Logger>>>,
 		config: Arc<Config>, is_running: Arc<RwLock<bool>>, logger: Arc<Logger>,
 	) -> Self {
 		Self {
@@ -81,6 +85,8 @@ impl Bolt11Payment {
 			connection_manager,
 			liquidity_source,
 			payment_store,
+			payment_metadata_store,
+			keys_manager,
 			peer_store,
 			config,
 			is_running,