io: Read only as many objects as a store needs

Seeding a store meant reading its entire namespace, which for a
bounded cache means fetching a node's whole payment history at
startup only to drop all but the newest entries. The previous commit
sidestepped that by not seeding the payment store at all, leaving it
cold and no longer catching unreadable payment data at build time.

Give the reader a bound instead, and seed the payment store with the
newest 50 payments. That matches the storage backends' page size, so
warming the cache costs a single page listing and one batch of reads,
and the first page of `Node::list_payments` is answered without going
to the store.

Take the keys from the paginated listing rather than `KVStore::list`,
which is documented to return them in arbitrary order and would
therefore make "the newest 50" meaningless. Objects now come back in
the store's own creation order, newest first, where before they came
back in whatever order the reads happened to finish.

Note the cache treats the objects it is seeded with as increasingly
recently used, so a newest-first read has to be reversed before
seeding, or the newest entries would be the first ones evicted.

Co-Authored-By: HAL 9000

Author: tnull

src/builder.rs

@@ -51,6 +51,7 @@ use crate::config::{
 	BitcoindRestClientConfig, Config, ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig,
 	TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL,
 	DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MIN_PROBE_AMOUNT_MSAT, PAYMENT_CACHE_CAPACITY,
+	PAYMENT_CACHE_WARMUP_COUNT,
 };
 use crate::connection::ConnectionManager;
 use crate::data_store::{KeepAllEntries, KeepLeastRecentlyUsed};
@@ -61,8 +62,8 @@ use crate::gossip::GossipSource;
 use crate::io::sqlite_store::SqliteStore;
 use crate::io::utils::{
 	open_or_migrate_fs_store, read_all_objects, read_event_queue,
-	read_external_pathfinding_scores_from_cache, read_network_graph, read_node_metrics,
-	read_output_sweeper, read_peer_info, read_scorer,
+	read_external_pathfinding_scores_from_cache, read_n_objects, read_network_graph,
+	read_node_metrics, read_output_sweeper, read_peer_info, read_scorer,
 };
 use crate::io::vss_store::VssStoreBuilder;
 use crate::io::{
@@ -1456,9 +1457,16 @@ fn build_with_store_internal(
 
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (node_metris_res, pending_payment_store_res, address_pool_res) =
-		runtime.block_on(async move {
+	let (payment_store_res, node_metris_res, pending_payment_store_res, address_pool_res) = runtime
+		.block_on(async move {
 			tokio::join!(
+				read_n_objects(
+					&*kv_store_ref,
+					PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+					PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+					PAYMENT_CACHE_WARMUP_COUNT,
+					Arc::clone(&logger_ref),
+				),
 				read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)),
 				read_all_objects(
 					&*kv_store_ref,
@@ -1483,17 +1491,23 @@ fn build_with_store_internal(
 		},
 	};
 
-	// The payment store caches a bounded number of payments and reads the rest back on demand, so
-	// we start it empty rather than paying to read a node's entire payment history at startup only
-	// to immediately drop all but the most recent entries.
-	let payment_store = Arc::new(PaymentStore::new(
-		Vec::new(),
-		KeepLeastRecentlyUsed::new(PAYMENT_CACHE_CAPACITY),
-		PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
-		PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
-		Arc::clone(&kv_store),
-		Arc::clone(&logger),
-	));
+	let payment_store = match payment_store_res {
+		Ok(payments) => Arc::new(PaymentStore::new(
+			// The read hands us the newest payments first, while the cache treats the objects it
+			// is seeded with as increasingly recently used. Reverse them, so that the newest
+			// payment is the last one to be evicted rather than the first.
+			payments.into_iter().rev().collect(),
+			KeepLeastRecentlyUsed::new(PAYMENT_CACHE_CAPACITY),
+			PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
+			PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE.to_string(),
+			Arc::clone(&kv_store),
+			Arc::clone(&logger),
+		)),
+		Err(e) => {
+			log_error!(logger, "Failed to read payment data 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 }) => {

src/config.rs

@@ -57,6 +57,13 @@ pub(crate) const DEFAULT_TX_BROADCAST_TIMEOUT_SECS: u64 = 10;
 // while still covering the recent payments a node actually works with.
 pub(crate) const PAYMENT_CACHE_CAPACITY: NonZeroUsize = NonZeroUsize::new(1000).unwrap();
 
+// The number of payments we read into the cache when starting up.
+//
+// This matches the storage backends' page size, so warming the cache costs a single page listing
+// and one batch of reads, and the first page of `Node::list_payments` is served without going to
+// the store at all. The remaining capacity fills as payments are used.
+pub(crate) const PAYMENT_CACHE_WARMUP_COUNT: NonZeroUsize = NonZeroUsize::new(50).unwrap();
+
 // The default {Esplora,Electrum} client timeout we're using.
 const DEFAULT_PER_REQUEST_TIMEOUT_SECS: u8 = 10;
 

src/data_store.rs

@@ -298,6 +298,10 @@ where
 	/// `objects` seeds the cache and must already be persisted under that namespace: under a
 	/// bounded policy any object beyond `cache_policy`'s capacity is dropped from memory
 	/// immediately, and is only recoverable by reading it back from the store.
+	///
+	/// They are taken in ascending order of recency, i.e., the last one given is treated as the
+	/// most recently used and is therefore the last to be evicted. Callers seeding from a
+	/// newest-first source have to reverse it.
 	pub(crate) fn new(
 		objects: Vec<SO>, cache_policy: P, primary_namespace: String, secondary_namespace: String,
 		kv_store: Arc<DynStore>, logger: L,
@@ -1648,6 +1652,30 @@ mod tests {
 		.is_err());
 	}
 
+	#[tokio::test]
+	async fn lru_seeding_treats_the_last_object_as_most_recently_used() {
+		// The builder relies on this to hand a newest-first read over in reverse: whichever
+		// objects are given last must be the ones that survive, or seeding the cache would
+		// preferentially throw away the newest entries.
+		let kv_store = in_memory_store();
+		let seed_store = new_data_store(Arc::clone(&kv_store), KeepAllEntries, Vec::new());
+		let mut objects = Vec::new();
+		for i in 0..5u8 {
+			let object = TestObject::new(test_id(i), [i; 3]);
+			seed_store.insert(object).await.unwrap();
+			objects.push(object);
+		}
+
+		let data_store = new_data_store(kv_store, keep_lru(2), objects.clone());
+
+		assert_eq!(2, data_store.cached_len());
+		assert!(data_store.is_cached(&objects[3].id));
+		assert!(data_store.is_cached(&objects[4].id));
+		for object in objects.iter().take(3) {
+			assert!(!data_store.is_cached(&object.id));
+		}
+	}
+
 	#[tokio::test]
 	async fn lru_seeding_trims_to_capacity() {
 		let kv_store = in_memory_store();