From 9cac53c1c0f813da88c46f5ae97e12f4977bbdf0 Mon Sep 17 00:00:00 2001 From: Enigbe Date: Tue, 7 Apr 2026 11:27:40 +0100 Subject: [PATCH 1/5] Implement tiered storage This commit adds `TierStore`, a tiered `KVStore` implementation that routes node persistence across three storage roles: - a primary store for durable, authoritative data - an optional backup store for a second durable copy of primary-backed data - an optional ephemeral store for rebuildable cached data such as the network graph and scorer TierStore routes ephemeral cache data to the ephemeral store when configured, while durable data remains primary+backup. Reads and lists do not consult the backup store during normal operation. For primary+backup writes and removals, this implementation treats the backup store as part of the persistence success path rather than as a best-effort background mirror. Earlier designs used asynchronous backup queueing to avoid blocking the primary path, but that weakens the durability contract by allowing primary success to be reported before backup persistence has completed. TierStore now issues primary and backup operations together and only returns success once both complete. This gives callers a clearer persistence guarantee when a backup store is configured: acknowledged primary+backup mutations have been attempted against both durable stores. The tradeoff is that dual-store operations are not atomic across stores, so an error may still be returned after one store has already been updated. Additionally, adds unit coverage for the current contract, including: - basic read/write/remove/list persistence - routing of ephemeral data away from the primary store - backup participation in the foreground success path for writes and removals --- src/io/in_memory_store.rs | 2 +- src/io/mod.rs | 1 + src/io/test_utils.rs | 2 +- src/io/tier_store.rs | 1450 +++++++++++++++++++++++++++++++++++++ 4 files changed, 1453 insertions(+), 2 deletions(-) create mode 100644 src/io/tier_store.rs diff --git a/src/io/in_memory_store.rs b/src/io/in_memory_store.rs index 156fef3a38..82418e7d5d 100644 --- a/src/io/in_memory_store.rs +++ b/src/io/in_memory_store.rs @@ -15,7 +15,7 @@ use lightning::util::persist::{ KVStore, MigratableKVStore, PageToken, PaginatedKVStore, PaginatedListResponse, }; -const IN_MEMORY_PAGE_SIZE: usize = 50; +pub(crate) const IN_MEMORY_PAGE_SIZE: usize = 50; pub struct InMemoryStore { persisted_bytes: Mutex>>>, diff --git a/src/io/mod.rs b/src/io/mod.rs index a01aa59a83..f0bbaee468 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -12,6 +12,7 @@ pub mod postgres_store; pub mod sqlite_store; #[cfg(test)] pub(crate) mod test_utils; +pub(crate) mod tier_store; pub(crate) mod utils; pub mod vss_store; diff --git a/src/io/test_utils.rs b/src/io/test_utils.rs index aadb4b79a8..fa9b3e8cae 100644 --- a/src/io/test_utils.rs +++ b/src/io/test_utils.rs @@ -159,7 +159,7 @@ impl chainmonitor::Persist const EXPECTED_UPDATES_PER_PAYMENT: u64 = 5; -pub(crate) use in_memory_store::InMemoryStore; +pub(crate) use in_memory_store::{InMemoryStore, IN_MEMORY_PAGE_SIZE}; pub(crate) fn random_storage_path() -> PathBuf { let mut temp_path = std::env::temp_dir(); diff --git a/src/io/tier_store.rs b/src/io/tier_store.rs new file mode 100644 index 0000000000..3bc3e2590d --- /dev/null +++ b/src/io/tier_store.rs @@ -0,0 +1,1450 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. +#![allow(dead_code)] // TODO: Temporal warning silencer. Will be removed in later commit. + +use std::collections::HashMap; +use std::future::Future; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use lightning::util::persist::{ + KVStore, PageToken, PaginatedKVStore, PaginatedListResponse, NETWORK_GRAPH_PERSISTENCE_KEY, + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_KEY, + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, +}; +use lightning::{io, log_error}; +use tokio::sync::Mutex as TokioMutex; + +use crate::io::utils::check_namespace_key_validity; +use crate::logger::{LdkLogger, Logger}; +use crate::types::DynStore; + +/// A 3-tiered [`KVStore`] implementation that routes data across +/// storage backends that may be local or remote: +/// - a primary store for durable, authoritative persistence, +/// - an optional backup store that maintains an additional durable copy of +/// primary-backed data, and +/// - an optional ephemeral store for non-critical, rebuildable cached data. +/// +/// When a backup store is configured, writes and removals for primary-backed data +/// are issued to the primary and backup stores concurrently and only succeed once +/// both stores complete successfully. +/// +/// Reads and lists do not consult the backup store during normal operation. +/// Ephemeral data is read from and written to the ephemeral store when configured. +/// +/// Note that dual-store writes and removals are not atomic across the primary and +/// backup stores. If one store succeeds and the other fails, the operation +/// returns an error even though one store may already reflect the change. +pub(crate) struct TierStore { + inner: Arc, +} + +impl TierStore { + pub fn new(primary_store: Arc, logger: Arc) -> Self { + let inner = Arc::new(TierStoreInner::new(primary_store, Arc::clone(&logger))); + + Self { inner } + } + + /// Configures a backup store for primary-backed data. + /// + /// Once set, writes and removals targeting the primary tier succeed only if both + /// the primary and backup stores succeed. The two operations are issued + /// concurrently, and any failure is returned to the caller. + /// + /// Note: dual-store writes/removals are not atomic. An error may be returned + /// after the primary store has already been updated if the backup store fails. + /// + /// The backup store is not consulted for normal reads or lists. + pub fn set_backup_store(&mut self, backup: Arc) { + debug_assert_eq!(Arc::strong_count(&self.inner), 1); + + let inner = Arc::get_mut(&mut self.inner).expect( + "TierStore should not be shared during configuration. No other references should exist", + ); + + inner.backup_store = Some(backup); + } + + /// Configures the ephemeral store for non-critical, rebuildable data. + /// + /// When configured, selected cache-like data is routed to this store instead of + /// the primary store. + pub fn set_ephemeral_store(&mut self, ephemeral: Arc) { + debug_assert_eq!(Arc::strong_count(&self.inner), 1); + + let inner = Arc::get_mut(&mut self.inner).expect( + "TierStore should not be shared during configuration. No other references should exist", + ); + + inner.ephemeral_store = Some(ephemeral); + } +} + +impl KVStore for TierStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, io::Error>> + 'static + Send { + let inner = Arc::clone(&self.inner); + + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + + async move { inner.read_internal(primary_namespace, secondary_namespace, key).await } + } + + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> impl Future> + 'static + Send { + let inner = Arc::clone(&self.inner); + let locking_key = inner.build_locking_key(primary_namespace, secondary_namespace, key); + let (lock_ref, version) = inner.get_new_version_and_lock_ref(locking_key.clone()); + + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + + async move { + inner + .write_internal( + primary_namespace, + secondary_namespace, + key, + buf, + lock_ref, + locking_key, + version, + ) + .await + } + } + + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> impl Future> + 'static + Send { + let inner = Arc::clone(&self.inner); + let locking_key = inner.build_locking_key(primary_namespace, secondary_namespace, key); + let (lock_ref, version) = inner.get_new_version_and_lock_ref(locking_key.clone()); + + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + let key = key.to_string(); + + async move { + inner + .remove_internal( + primary_namespace, + secondary_namespace, + key, + lazy, + lock_ref, + locking_key, + version, + ) + .await + } + } + + fn list( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> impl Future, io::Error>> + 'static + Send { + let inner = Arc::clone(&self.inner); + + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + + async move { inner.list_internal(primary_namespace, secondary_namespace).await } + } +} + +struct TierStoreInner { + /// The authoritative store for durable data. + primary_store: Arc, + /// The store used for non-critical, rebuildable cached data. + ephemeral_store: Option>, + /// An optional second durable store for primary-backed data. + backup_store: Option>, + /// Per-key locks for serializing primary+backup operations and skipping stale writes. + locks: Mutex>>>, + next_write_version: AtomicU64, + logger: Arc, +} + +impl TierStoreInner { + /// Creates a tier store with the primary data store. + pub fn new(primary_store: Arc, logger: Arc) -> Self { + Self { + primary_store, + ephemeral_store: None, + backup_store: None, + locks: Mutex::new(HashMap::new()), + next_write_version: AtomicU64::new(1), + logger, + } + } + + fn get_new_version_and_lock_ref(&self, locking_key: String) -> (Arc>, u64) { + let version = self.next_write_version.fetch_add(1, Ordering::Relaxed); + if version == u64::MAX { + panic!("TierStore version counter overflowed"); + } + + let mut locks = self.locks.lock().expect("lock"); + let lock_ref = + Arc::clone(locks.entry(locking_key).or_insert_with(|| Arc::new(TokioMutex::new(0)))); + + (lock_ref, version) + } + + fn clean_locks(&self, lock_ref: &Arc>, locking_key: String) { + let mut locks = self.locks.lock().expect("lock"); + let strong_count = Arc::strong_count(lock_ref); + debug_assert!(strong_count >= 2, "Unexpected TierStore lock strong count"); + if strong_count == 2 { + locks.remove(&locking_key); + } + } + + fn build_locking_key( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> String { + if primary_namespace.is_empty() { + key.to_owned() + } else { + format!("{}#{}#{}", primary_namespace, secondary_namespace, key) + } + } + + /// Reads from the primary data store. + async fn read_primary( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> io::Result> { + match KVStore::read( + self.primary_store.as_ref(), + primary_namespace, + secondary_namespace, + key, + ) + .await + { + Ok(data) => Ok(data), + Err(e) => Err(e), + } + } + + /// Lists keys from the primary data store. + async fn list_primary( + &self, primary_namespace: &str, secondary_namespace: &str, + ) -> io::Result> { + match KVStore::list(self.primary_store.as_ref(), primary_namespace, secondary_namespace) + .await + { + Ok(keys) => Ok(keys), + Err(e) => { + log_error!( + self.logger, + "Failed to list from primary store for namespace {}/{}: {}.", + primary_namespace, + secondary_namespace, + e + ); + Err(e) + }, + } + } + + async fn write_primary_backup_async( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> io::Result<()> { + if let Some(backup_store) = self.backup_store.as_ref() { + let primary_fut = KVStore::write( + self.primary_store.as_ref(), + primary_namespace, + secondary_namespace, + key, + buf.clone(), + ); + + let backup_fut = KVStore::write( + backup_store.as_ref(), + primary_namespace, + secondary_namespace, + key, + buf, + ); + + let (primary_res, backup_res) = tokio::join!(primary_fut, backup_fut); + + self.handle_primary_backup_results( + "write", + primary_namespace, + secondary_namespace, + key, + primary_res, + backup_res, + ) + } else { + KVStore::write( + self.primary_store.as_ref(), + primary_namespace, + secondary_namespace, + key, + buf, + ) + .await + } + } + + async fn remove_primary_backup_async( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> io::Result<()> { + let primary_fut = KVStore::remove( + self.primary_store.as_ref(), + primary_namespace, + secondary_namespace, + key, + lazy, + ); + + if let Some(backup_store) = self.backup_store.as_ref() { + let backup_fut = KVStore::remove( + backup_store.as_ref(), + primary_namespace, + secondary_namespace, + key, + lazy, + ); + + let (primary_res, backup_res) = tokio::join!(primary_fut, backup_fut); + + self.handle_primary_backup_results( + "removal", + primary_namespace, + secondary_namespace, + key, + primary_res, + backup_res, + ) + } else { + primary_fut.await + } + } + + async fn execute_locked_write( + &self, lock_ref: Arc>, locking_key: String, version: u64, callback: F, + ) -> io::Result<()> + where + F: FnOnce() -> Fut, + Fut: Future>, + { + let res = { + let mut last_written_version = lock_ref.lock().await; + + if version <= *last_written_version { + Ok(()) + } else { + let res = callback().await; + if res.is_ok() { + *last_written_version = version; + } + res + } + }; + + self.clean_locks(&lock_ref, locking_key); + res + } + + async fn read_internal( + &self, primary_namespace: String, secondary_namespace: String, key: String, + ) -> io::Result> { + check_namespace_key_validity( + primary_namespace.as_str(), + secondary_namespace.as_str(), + Some(key.as_str()), + "read", + )?; + + if is_ephemeral_cached_key(&primary_namespace, &secondary_namespace, &key) { + if let Some(eph_store) = self.ephemeral_store.as_ref() { + // We don't retry ephemeral-store reads here. Local failures are treated as + // terminal for this access path rather than falling back to another store. + return KVStore::read( + eph_store.as_ref(), + &primary_namespace, + &secondary_namespace, + &key, + ) + .await; + } + } + + self.read_primary(&primary_namespace, &secondary_namespace, &key).await + } + + async fn write_internal( + &self, primary_namespace: String, secondary_namespace: String, key: String, buf: Vec, + lock_ref: Arc>, locking_key: String, version: u64, + ) -> io::Result<()> { + check_namespace_key_validity( + primary_namespace.as_str(), + secondary_namespace.as_str(), + Some(key.as_str()), + "write", + )?; + + if is_ephemeral_cached_key(&primary_namespace, &secondary_namespace, &key) { + if let Some(eph_store) = self.ephemeral_store.as_ref() { + let eph_store = Arc::clone(eph_store); + return self + .execute_locked_write(lock_ref, locking_key, version, || async move { + KVStore::write( + eph_store.as_ref(), + primary_namespace.as_str(), + secondary_namespace.as_str(), + key.as_str(), + buf, + ) + .await + }) + .await; + } + } + + self.execute_locked_write(lock_ref, locking_key, version, || async move { + self.write_primary_backup_async( + primary_namespace.as_str(), + secondary_namespace.as_str(), + key.as_str(), + buf, + ) + .await + }) + .await + } + + async fn remove_internal( + &self, primary_namespace: String, secondary_namespace: String, key: String, lazy: bool, + lock_ref: Arc>, locking_key: String, version: u64, + ) -> io::Result<()> { + check_namespace_key_validity( + primary_namespace.as_str(), + secondary_namespace.as_str(), + Some(key.as_str()), + "remove", + )?; + + if is_ephemeral_cached_key(&primary_namespace, &secondary_namespace, &key) { + if let Some(eph_store) = self.ephemeral_store.as_ref() { + let eph_store = Arc::clone(eph_store); + return self + .execute_locked_write(lock_ref, locking_key, version, || async move { + KVStore::remove( + eph_store.as_ref(), + primary_namespace.as_str(), + secondary_namespace.as_str(), + key.as_str(), + lazy, + ) + .await + }) + .await; + } + } + + self.execute_locked_write(lock_ref, locking_key, version, || async move { + self.remove_primary_backup_async( + primary_namespace.as_str(), + secondary_namespace.as_str(), + key.as_str(), + lazy, + ) + .await + }) + .await + } + + async fn list_internal( + &self, primary_namespace: String, secondary_namespace: String, + ) -> io::Result> { + check_namespace_key_validity( + primary_namespace.as_str(), + secondary_namespace.as_str(), + None, + "list", + )?; + + let mut keys = self.list_primary(&primary_namespace, &secondary_namespace).await?; + + self.apply_ephemeral_overlay(&primary_namespace, &secondary_namespace, &mut keys, true) + .await?; + + Ok(keys) + } + + async fn list_paginated_internal( + &self, primary_namespace: String, secondary_namespace: String, + page_token: Option, + ) -> io::Result { + check_namespace_key_validity( + primary_namespace.as_str(), + secondary_namespace.as_str(), + None, + "list_paginated", + )?; + + let mut response = PaginatedKVStore::list_paginated( + self.primary_store.as_ref(), + &primary_namespace, + &secondary_namespace, + page_token, + ) + .await?; + + // Filter stale primary copies of ephemeral-cached keys from every page, and append the + // live ephemeral overlay only once primary pagination is exhausted (`next_page_token` + // is `None`). + // + // Because filtering can drop entries from a full page, a non-terminal page can come back + // shorter than the nominal page size: callers must not treat `next_page_token.is_some()` + // as a guarantee of a full page. Conversely, appending the overlay onto an already + // full-sized terminal page can make it up to `MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE` + // entries larger than the nominal page size. Both effects are bounded by that constant. + let append_live = response.next_page_token.is_none(); + self.apply_ephemeral_overlay( + &primary_namespace, + &secondary_namespace, + &mut response.keys, + append_live, + ) + .await?; + + Ok(response) + } + + /// Reconciles a set of keys already listed from the primary store with the ephemeral store. + /// + /// This keeps `list`/`list_paginated` consistent with the key-level routing in + /// `read`/`write`/`remove`: once an ephemeral store is configured it is authoritative for + /// ephemeral-cached keys (`network_graph`/`scorer`). Any copy of such a key still held by the + /// primary store is a stale leftover from before the ephemeral store was configured, so we + /// drop it here; when `append_live` is set we then append the live ephemeral copy. + /// + /// The ephemeral store is only consulted for namespaces that can actually contain an + /// ephemeral-cached key (see [`namespace_may_hold_ephemeral_cached_key`]). For every other + /// namespace this is a no-op, so listing durable, primary-backed data never depends on the + /// optional ephemeral backend being reachable. + async fn apply_ephemeral_overlay( + &self, primary_namespace: &str, secondary_namespace: &str, keys: &mut Vec, + append_live: bool, + ) -> io::Result<()> { + let Some(eph_store) = self.ephemeral_store.as_ref() else { + return Ok(()); + }; + + if !namespace_may_hold_ephemeral_cached_key(primary_namespace) { + return Ok(()); + } + + keys.retain(|key| !is_ephemeral_cached_key(primary_namespace, secondary_namespace, key)); + + if !append_live { + return Ok(()); + } + + // We don't retry ephemeral-store lists here. Local failures are treated as terminal for + // this access path rather than falling back to another store. + let cached_keys: Vec = + KVStore::list(eph_store.as_ref(), primary_namespace, secondary_namespace) + .await? + .into_iter() + .filter(|key| is_ephemeral_cached_key(primary_namespace, secondary_namespace, key)) + .collect(); + + debug_assert!( + cached_keys.len() <= MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE, + "ephemeral-cached key overlay ({} keys) exceeded the bound this pagination \ + design assumes (see MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE)", + cached_keys.len(), + ); + + for key in cached_keys { + // Guards against `list` on the ephemeral store itself returning a duplicate entry. + if !keys.contains(&key) { + keys.push(key); + } + } + + Ok(()) + } + + fn handle_primary_backup_results( + &self, op: &str, primary_namespace: &str, secondary_namespace: &str, key: &str, + primary_res: io::Result<()>, backup_res: io::Result<()>, + ) -> io::Result<()> { + match (primary_res, backup_res) { + (Ok(()), Ok(())) => Ok(()), + (Err(primary_err), Ok(())) => { + log_error!( + self.logger, + "Primary {} failed after backup {} succeeded for key {}/{}/{}; primary and backup may have diverged: {}", + op, + op, + primary_namespace, + secondary_namespace, + key, + primary_err + ); + Err(primary_err) + }, + (Ok(()), Err(backup_err)) => { + log_error!( + self.logger, + "Backup {} failed after primary {} succeeded for key {}/{}/{}; primary and backup may have diverged: {}", + op, + op, + primary_namespace, + secondary_namespace, + key, + backup_err + ); + Err(backup_err) + }, + (Err(primary_err), Err(backup_err)) => { + log_error!( + self.logger, + "Primary and backup {}s both failed for key {}/{}/{}: primary={}, backup={}", + op, + primary_namespace, + secondary_namespace, + key, + primary_err, + backup_err + ); + Err(primary_err) + }, + } + } +} + +/// The maximum number of distinct keys [`is_ephemeral_cached_key`] can ever match for a +/// single namespace pair -- one per matched key literal (`network_graph`, `scorer`). +/// +/// `apply_ephemeral_overlay` reads the ephemeral-cached overlay unpaginated and appends +/// it onto the terminal primary page in a single shot, reusing primary's own page token +/// unmodified rather than tracking a cross-store cursor. That's only sound while this +/// stays small. If a future key is added to `is_ephemeral_cached_key`, bump this constant +/// to match, and re-examine whether `list_paginated_internal` still needs revisiting -- a +/// larger or unbounded ephemeral-cached set can silently blow past the nominal page size +/// and can't correctly report `next_page_token` for a partial overlay. +const MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE: usize = 2; + +fn is_ephemeral_cached_key(pn: &str, sn: &str, key: &str) -> bool { + matches!( + (pn, sn, key), + (NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, _, NETWORK_GRAPH_PERSISTENCE_KEY) + | (SCORER_PERSISTENCE_PRIMARY_NAMESPACE, _, SCORER_PERSISTENCE_KEY) + ) +} + +/// Whether a primary namespace can ever contain a key that [`is_ephemeral_cached_key`] matches. +/// +/// Listing paths use this to gate the (optional) ephemeral-store lookup: for any namespace that +/// cannot hold an ephemeral-cached key, the overlay is skipped entirely, so durable, +/// primary-backed listings neither pay for an extra ephemeral list nor fail when the ephemeral +/// backend is unreachable. +fn namespace_may_hold_ephemeral_cached_key(pn: &str) -> bool { + pn == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE || pn == SCORER_PERSISTENCE_PRIMARY_NAMESPACE +} + +#[cfg(test)] +mod tests { + use std::future::Future; + use std::panic::RefUnwindSafe; + use std::path::PathBuf; + use std::sync::Arc; + + use lightning::util::logger::Level; + use lightning::util::persist::{ + CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + }; + use lightning_persister::fs_store::v2::FilesystemStoreV2; + + use super::*; + use crate::io::test_utils::{ + do_read_write_remove_list_persist, random_storage_path, InMemoryStore, IN_MEMORY_PAGE_SIZE, + }; + use crate::io::tier_store::TierStore; + use crate::logger::Logger; + use crate::types::{DynStore, DynStoreWrapper}; + + impl RefUnwindSafe for TierStore {} + + struct CleanupDir(PathBuf); + impl Drop for CleanupDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + fn setup_tier_store(primary_store: Arc, logger: Arc) -> TierStore { + TierStore::new(primary_store, logger) + } + + /// A store whose `list`/`list_paginated` always fail while every other operation is delegated + /// to an inner [`InMemoryStore`]. Used to prove that a failing ephemeral list does not sink a + /// listing for a namespace that can never hold an ephemeral-cached key. + struct FailingListStore { + inner: InMemoryStore, + } + + impl FailingListStore { + fn new() -> Self { + Self { inner: InMemoryStore::new() } + } + } + + impl KVStore for FailingListStore { + fn read( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, + ) -> impl Future, io::Error>> + 'static + Send { + KVStore::read(&self.inner, primary_namespace, secondary_namespace, key) + } + fn write( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, + ) -> impl Future> + 'static + Send { + KVStore::write(&self.inner, primary_namespace, secondary_namespace, key, buf) + } + fn remove( + &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, + ) -> impl Future> + 'static + Send { + KVStore::remove(&self.inner, primary_namespace, secondary_namespace, key, lazy) + } + fn list( + &self, _primary_namespace: &str, _secondary_namespace: &str, + ) -> impl Future, io::Error>> + 'static + Send { + async { Err(io::Error::new(io::ErrorKind::Other, "list failed")) } + } + } + + impl PaginatedKVStore for FailingListStore { + fn list_paginated( + &self, _primary_namespace: &str, _secondary_namespace: &str, + _page_token: Option, + ) -> impl Future> + 'static + Send { + async { Err(io::Error::new(io::ErrorKind::Other, "list_paginated failed")) } + } + } + + #[tokio::test] + async fn write_read_list_remove() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let tier = setup_tier_store(primary_store, logger); + + do_read_write_remove_list_persist(&tier).await; + } + + #[tokio::test] + async fn ephemeral_routing() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("ephemeral")).unwrap())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + let data = vec![42u8; 32]; + + tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + data.clone(), + ) + .await + .unwrap(); + + tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + data.clone(), + ) + .await + .unwrap(); + + let primary_read_ng = primary_store + .read( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + ) + .await; + let ephemeral_read_ng = ephemeral_store + .read( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + ) + .await; + + let primary_read_cm = primary_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ) + .await; + let ephemeral_read_cm = ephemeral_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ) + .await; + + assert!(primary_read_ng.is_err()); + assert_eq!(ephemeral_read_ng.unwrap(), data); + + assert!(ephemeral_read_cm.is_err()); + assert_eq!(primary_read_cm.unwrap(), data); + } + + #[tokio::test] + async fn list_discovers_durable_keys_alongside_ephemeral_cache() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("ephemeral")).unwrap())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + // A durable root-namespace key, routed to primary since it isn't ephemeral-cached. + tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + vec![1u8; 32], + ) + .await + .unwrap(); + + // The ephemeral-cached key, routed to the ephemeral store. + tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + vec![2u8; 32], + ) + .await + .unwrap(); + + // A decoy sitting in the ephemeral store under an unrelated namespace. This must + // never leak into a listing for that namespace just because an ephemeral + // store happens to be configured. + ephemeral_store + .write( + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + "ephemeral-decoy", + vec![3u8; 32], + ) + .await + .unwrap(); + + // This is `list("", "")`: CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE and + // NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE are the same empty string, so both + // keys live in the exact namespace. + let root_keys = KVStore::list( + &tier, + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + ) + .await + .unwrap(); + + // The durable primary-backed key and the ephemeral-cached key must both be + // discoverable from a single call. + assert!(root_keys.contains(&CHANNEL_MANAGER_PERSISTENCE_KEY.to_string())); + assert!(root_keys.contains(&NETWORK_GRAPH_PERSISTENCE_KEY.to_string())); + + let monitor_keys = KVStore::list( + &tier, + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + ) + .await + .unwrap(); + + // The unrelated-namespace decoy sitting in the ephemeral store must not leak + // into a listing for a namespace it was never routed to. + assert!(!monitor_keys.contains(&"ephemeral-decoy".to_string())); + } + + #[tokio::test] + async fn list_paginated_routes_to_selected_tier() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("ephemeral")).unwrap())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + tier.write( + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + "monitor-key", + vec![1u8; 32], + ) + .await + .unwrap(); + + // This decoy uses the same namespace but the opposite physical store, so it + // would show up if paginated listing routed to the wrong tier. + ephemeral_store + .write( + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + "ephemeral-decoy", + vec![2u8; 32], + ) + .await + .unwrap(); + + // This key shares the network graph's namespace tuple ("", "") but is not + // itself an ephemeral-cached key, standing in for durable root-namespace data + // such as `manager`/`output_sweeper`/`peers`. It must still be listed even + // though the ephemeral store is configured and authoritative for + // `network_graph`/`scorer` specifically. + primary_store + .write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + "other-root-namespace-key", + vec![3u8; 32], + ) + .await + .unwrap(); + + tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + vec![4u8; 32], + ) + .await + .unwrap(); + + let primary_response = PaginatedKVStore::list_paginated( + &tier, + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + None, + ) + .await + .unwrap(); + assert_eq!(primary_response.keys, vec!["monitor-key".to_string()]); + + let ephemeral_response = PaginatedKVStore::list_paginated( + &tier, + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + None, + ) + .await + .unwrap(); + + // The durable root-namespace key surfaces from the primary store, and the + // ephemeral-cached `network_graph` key is appended once primary's pagination + // is exhausted. + assert_eq!( + ephemeral_response.keys, + vec!["other-root-namespace-key".to_string(), NETWORK_GRAPH_PERSISTENCE_KEY.to_string()] + ); + } + + #[tokio::test] + async fn list_paginated_filters_stale_primary_copies_across_every_page() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let ephemeral_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + // Exactly IN_MEMORY_PAGE_SIZE unrelated, durable root-namespace keys written + // directly to primary, so a single additional entry (the stale copy below) + // pushes the namespace to exactly two pages. + for i in 0..IN_MEMORY_PAGE_SIZE { + primary_store + .write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + &format!("filler-{i:02}"), + vec![0u8; 32], + ) + .await + .unwrap(); + } + + // A stale copy of `network_graph`, written directly to primary as if it had + // been persisted there before the ephemeral store was configured. Writing it + // last gives it the highest creation order, guaranteeing it lands on the + // *first* page rather than the last -- filtering only the terminal page + // would miss it entirely. + primary_store + .write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + vec![1u8; 32], + ) + .await + .unwrap(); + + // The live, authoritative copy, written through the tier so it is routed to + // the ephemeral store. + tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + vec![2u8; 32], + ) + .await + .unwrap(); + + let page_one = PaginatedKVStore::list_paginated( + &tier, + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + None, + ) + .await + .unwrap(); + + // The stale primary copy must be filtered out of the first page even though + // the walk isn't done yet -- it should never be visible once an ephemeral + // store is configured. + let expected_page_one: Vec = + (1..IN_MEMORY_PAGE_SIZE).rev().map(|i| format!("filler-{i:02}")).collect(); + assert_eq!(page_one.keys, expected_page_one); + assert!(page_one.next_page_token.is_some()); + + let page_two = PaginatedKVStore::list_paginated( + &tier, + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + page_one.next_page_token, + ) + .await + .unwrap(); + + // The final page carries the one remaining filler key plus the ephemeral + // store's live copy, appended exactly once now that primary pagination is + // exhausted. + assert_eq!( + page_two.keys, + vec!["filler-00".to_string(), NETWORK_GRAPH_PERSISTENCE_KEY.to_string()] + ); + assert!(page_two.next_page_token.is_none()); + + // Across the full walk, `network_graph` appears exactly once -- this is the + // property that would have caught the duplicate-key bug. + let occurrences = page_one + .keys + .iter() + .chain(page_two.keys.iter()) + .filter(|k| k.as_str() == NETWORK_GRAPH_PERSISTENCE_KEY) + .count(); + assert_eq!(occurrences, 1); + } + + #[tokio::test] + async fn list_unrelated_namespace_survives_ephemeral_list_failure() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + // An ephemeral store whose `list`/`list_paginated` always fail. + let ephemeral_store: Arc = Arc::new(DynStoreWrapper(FailingListStore::new())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + // A durable key in a namespace that can never hold an ephemeral-cached key. + tier.write( + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + "monitor-key", + vec![1u8; 32], + ) + .await + .unwrap(); + + // Listing that namespace must not consult (or depend on) the ephemeral store, so it + // succeeds even though the ephemeral list would fail. + let monitor_keys = KVStore::list( + &tier, + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + ) + .await + .unwrap(); + assert_eq!(monitor_keys, vec!["monitor-key".to_string()]); + + // The paginated path is gated identically. + let monitor_page = PaginatedKVStore::list_paginated( + &tier, + CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, + None, + ) + .await + .unwrap(); + assert_eq!(monitor_page.keys, vec!["monitor-key".to_string()]); + + // Sanity check the fixture: listing a namespace that *can* hold an ephemeral-cached key + // still surfaces the ephemeral failure rather than silently swallowing it. + assert!(KVStore::list( + &tier, + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + ) + .await + .is_err()); + } + + #[tokio::test] + async fn list_filters_stale_primary_copy_when_ephemeral_missing() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let ephemeral_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + // A durable root-namespace key that must always be discoverable. + tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + vec![1u8; 32], + ) + .await + .unwrap(); + + // A stale copy of `network_graph` sitting in primary as if it had been persisted there + // before the ephemeral store was configured. The ephemeral store holds no copy, so + // `read` routes to ephemeral and would fail for this key. + primary_store + .write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + vec![2u8; 32], + ) + .await + .unwrap(); + + let root_keys = KVStore::list( + &tier, + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + ) + .await + .unwrap(); + + // The durable key is still listed, and the stale primary copy of `network_graph` is + // filtered out so `list` cannot report a key that `read` would fail to fetch -- matching + // the paginated path. + assert!(root_keys.contains(&CHANNEL_MANAGER_PERSISTENCE_KEY.to_string())); + assert!(!root_keys.contains(&NETWORK_GRAPH_PERSISTENCE_KEY.to_string())); + } + + #[tokio::test] + async fn primary_backed_writes_preserve_latest_call_order() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let tier = setup_tier_store(primary_store, logger); + + let old_data = vec![1u8; 32]; + let new_data = vec![2u8; 32]; + + let old_write = tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + old_data, + ); + let new_write = tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + new_data.clone(), + ); + + new_write.await.unwrap(); + old_write.await.unwrap(); + + // Stale data doesn't overwrite latest + let persisted = tier + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ) + .await + .unwrap(); + assert_eq!(persisted, new_data); + } + + #[tokio::test] + async fn ephemeral_writes_preserve_latest_call_order() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(primary_store, logger); + + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("ephemeral")).unwrap())); + tier.set_ephemeral_store(ephemeral_store); + + let old_data = vec![1u8; 32]; + let new_data = vec![2u8; 32]; + + let old_write = tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + old_data, + ); + let new_write = tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + new_data.clone(), + ); + + new_write.await.unwrap(); + old_write.await.unwrap(); + + let persisted = tier + .read( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + ) + .await + .unwrap(); + assert_eq!(persisted, new_data); + } + + #[tokio::test] + async fn ephemeral_removes_preserve_latest_call_order() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(primary_store, logger); + + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("ephemeral")).unwrap())); + tier.set_ephemeral_store(ephemeral_store); + + let data = vec![2u8; 32]; + + let stale_remove = tier.remove( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + true, + ); + let new_write = tier.write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + data.clone(), + ); + + new_write.await.unwrap(); + stale_remove.await.unwrap(); + + let persisted = tier + .read( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + ) + .await + .unwrap(); + assert_eq!(persisted, data); + } + + #[tokio::test] + async fn backup_write_is_part_of_success_path() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let backup_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("backup")).unwrap())); + tier.set_backup_store(Arc::clone(&backup_store)); + + let data = vec![42u8; 32]; + + tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + data.clone(), + ) + .await + .unwrap(); + + let primary_read = primary_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ) + .await; + let backup_read = backup_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ) + .await; + + assert_eq!(primary_read.unwrap(), data); + assert_eq!(backup_read.unwrap(), data); + } + + #[tokio::test] + async fn backup_remove_is_part_of_success_path() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let backup_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("backup")).unwrap())); + tier.set_backup_store(Arc::clone(&backup_store)); + + let data = vec![42u8; 32]; + let key = CHANNEL_MANAGER_PERSISTENCE_KEY; + + tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + key, + data, + ) + .await + .unwrap(); + + tier.remove( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + key, + true, + ) + .await + .unwrap(); + + let primary_read = primary_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + key, + ) + .await; + let backup_read = backup_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + key, + ) + .await; + + assert!(primary_read.is_err()); + assert!(backup_read.is_err()); + } +} From 372f2df9faf39437e0c5d82db0426fcb67362022 Mon Sep 17 00:00:00 2001 From: Enigbe Date: Tue, 7 Apr 2026 19:17:22 +0100 Subject: [PATCH 2/5] Integrate TierStore into NodeBuilder Add native builder support for configuring ephemeral storage and a local SQLite backup mirror. Wrap the primary store in TierStore during node construction and create configured secondary stores using dedicated SQLite database files. Implement paginated listing through TierStore and update filesystem-backed tests to use FilesystemStoreV2. Add full-cycle integration coverage verifying durable backup mirroring. --- src/builder.rs | 89 ++++++++++++++++++++++++++++++++- src/io/sqlite_store/mod.rs | 4 ++ src/io/tier_store.rs | 16 +++++- tests/common/mod.rs | 6 +-- tests/integration_tests_rust.rs | 72 ++++++++++++++++++++++++++ 5 files changed, 181 insertions(+), 6 deletions(-) diff --git a/src/builder.rs b/src/builder.rs index f117800996..87dd1dce4e 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -58,6 +58,7 @@ use crate::event::EventQueue; use crate::fee_estimator::OnchainFeeEstimator; use crate::gossip::GossipSource; use crate::io::sqlite_store::SqliteStore; +use crate::io::tier_store::TierStore; 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, @@ -156,6 +157,12 @@ impl std::fmt::Debug for LogWriterConfig { } } +#[derive(Default, Debug)] +struct TierStoreConfig { + ephemeral_storage_dir_path: Option, + backup_storage_dir_path: Option, +} + /// An error encountered during building a [`Node`]. /// /// [`Node`]: crate::Node @@ -309,6 +316,7 @@ pub struct NodeBuilder { liquidity_source_config: Option, log_writer_config: Option, async_payments_role: Option, + tier_store_config: Option, runtime_handle: Option, pathfinding_scores_sync_config: Option, probing_config: Option, @@ -327,6 +335,7 @@ impl NodeBuilder { let gossip_source_config = None; let liquidity_source_config = None; let log_writer_config = None; + let tier_store_config = None; let runtime_handle = None; let pathfinding_scores_sync_config = None; let probing_config = None; @@ -336,6 +345,7 @@ impl NodeBuilder { gossip_source_config, liquidity_source_config, log_writer_config, + tier_store_config, runtime_handle, async_payments_role: None, pathfinding_scores_sync_config, @@ -661,6 +671,41 @@ impl NodeBuilder { self } + /// Configures a local SQLite backup store for disaster recovery. + /// + /// When building with tiered storage, a SQLite store will be created at the + /// given directory path using [`SQLITE_BACKUP_DB_FILE_NAME`] as its database + /// file name. It receives a second durable copy of data written to the + /// primary store. + /// + /// Writes and removals for primary-backed data only succeed once both the + /// primary and backup SQLite stores complete successfully. + /// + /// If not set, durable data will be stored only in the primary store. + /// + /// [`SQLITE_BACKUP_DB_FILE_NAME`]: crate::io::sqlite_store::SQLITE_BACKUP_DB_FILE_NAME + #[cfg(not(feature = "uniffi"))] + pub fn set_backup_storage_dir_path(&mut self, backup_storage_dir_path: String) -> &mut Self { + let tier_store_config = self.tier_store_config.get_or_insert(TierStoreConfig::default()); + tier_store_config.backup_storage_dir_path = Some(backup_storage_dir_path.into()); + self + } + + /// Configures the ephemeral storage directory path for non-critical, frequently-accessed data. + /// + /// When set, a local SQLite store is created at this path for ephemeral data like + /// the network graph and scorer. Data stored here can be rebuilt if lost. + /// + /// If not set, non-critical data will be stored in the primary store. + #[cfg(not(feature = "uniffi"))] + pub fn set_ephemeral_storage_dir_path( + &mut self, ephemeral_storage_dir_path: String, + ) -> &mut Self { + let tier_store_config = self.tier_store_config.get_or_insert(TierStoreConfig::default()); + tier_store_config.ephemeral_storage_dir_path = Some(ephemeral_storage_dir_path.into()); + self + } + /// Builds a [`Node`] instance with a [`SqliteStore`] backend and according to the options /// previously configured. pub fn build(&self, node_entropy: NodeEntropy) -> Result { @@ -862,11 +907,18 @@ impl NodeBuilder { } /// Builds a [`Node`] instance according to the options previously configured. + /// + /// The provided `kv_store` will be used as the primary storage backend. Optionally, + /// an ephemeral store for frequently-accessed non-critical data (e.g., network graph, scorer) + /// and a local SQLite backup store for disaster recovery can be configured via + /// [`set_ephemeral_storage_dir_path`] and [`set_backup_storage_dir_path`]. + /// + /// [`set_ephemeral_storage_dir_path`]: Self::set_ephemeral_storage_dir_path + /// [`set_backup_storage_dir_path`]: Self::set_backup_storage_dir_path pub fn build_with_store( &self, node_entropy: NodeEntropy, kv_store: S, ) -> Result { let logger = setup_logger(&self.log_writer_config, &self.config)?; - self.build_with_store_and_logger(node_entropy, kv_store, logger) } @@ -891,6 +943,39 @@ impl NodeBuilder { fn build_with_store_runtime_and_logger( &self, node_entropy: NodeEntropy, kv_store: S, runtime: Arc, logger: Arc, ) -> Result { + let ts_config = self.tier_store_config.as_ref(); + let primary_store = Arc::new(DynStoreWrapper(kv_store)); + let mut tier_store = TierStore::new(primary_store, Arc::clone(&logger)); + if let Some(config) = ts_config { + if let Some(ephemeral_storage_dir_path) = config.ephemeral_storage_dir_path.as_ref() { + let ephemeral_store = SqliteStore::new( + ephemeral_storage_dir_path.clone(), + Some(io::sqlite_store::SQLITE_EPHEMERAL_DB_FILE_NAME.to_string()), + Some(io::sqlite_store::KV_TABLE_NAME.to_string()), + ) + .map_err(|e| { + log_error!(logger, "Failed to setup ephemeral SQLite store: {}", e); + BuildError::KVStoreSetupFailed + })?; + let ephemeral_store: Arc = Arc::new(DynStoreWrapper(ephemeral_store)); + tier_store.set_ephemeral_store(ephemeral_store); + } + + if let Some(backup_storage_dir_path) = config.backup_storage_dir_path.as_ref() { + let backup_store = SqliteStore::new( + backup_storage_dir_path.clone(), + Some(io::sqlite_store::SQLITE_BACKUP_DB_FILE_NAME.to_string()), + Some(io::sqlite_store::KV_TABLE_NAME.to_string()), + ) + .map_err(|e| { + log_error!(logger, "Failed to setup backup SQLite store: {}", e); + BuildError::KVStoreSetupFailed + })?; + let backup_store: Arc = Arc::new(DynStoreWrapper(backup_store)); + tier_store.set_backup_store(backup_store); + } + } + let seed_bytes = node_entropy.to_seed_bytes(); let config = Arc::new(self.config.clone()); @@ -905,7 +990,7 @@ impl NodeBuilder { seed_bytes, runtime, logger, - Arc::new(DynStoreWrapper(kv_store)), + Arc::new(DynStoreWrapper(tier_store)), ) } } diff --git a/src/io/sqlite_store/mod.rs b/src/io/sqlite_store/mod.rs index 2587220598..34edd2bc21 100644 --- a/src/io/sqlite_store/mod.rs +++ b/src/io/sqlite_store/mod.rs @@ -26,6 +26,10 @@ mod migrations; /// LDK Node's database file name. pub const SQLITE_DB_FILE_NAME: &str = "ldk_node_data.sqlite"; +/// LDK Node's backup database file name. +pub const SQLITE_BACKUP_DB_FILE_NAME: &str = "ldk_node_data_backup.sqlite"; +/// LDK Node's ephemeral database file name. +pub const SQLITE_EPHEMERAL_DB_FILE_NAME: &str = "ldk_node_data_ephemeral.sqlite"; /// LDK Node's table in which we store all data. pub const KV_TABLE_NAME: &str = "ldk_node_data"; diff --git a/src/io/tier_store.rs b/src/io/tier_store.rs index 3bc3e2590d..f7f3deb8e9 100644 --- a/src/io/tier_store.rs +++ b/src/io/tier_store.rs @@ -4,7 +4,6 @@ // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license , at your option. You may not use this file except in // accordance with one or both of these licenses. -#![allow(dead_code)] // TODO: Temporal warning silencer. Will be removed in later commit. use std::collections::HashMap; use std::future::Future; @@ -163,6 +162,21 @@ impl KVStore for TierStore { } } +impl PaginatedKVStore for TierStore { + fn list_paginated( + &self, primary_namespace: &str, secondary_namespace: &str, page_token: Option, + ) -> impl Future> + 'static + Send { + let inner = Arc::clone(&self.inner); + + let primary_namespace = primary_namespace.to_string(); + let secondary_namespace = secondary_namespace.to_string(); + + async move { + inner.list_paginated_internal(primary_namespace, secondary_namespace, page_token).await + } + } +} + struct TierStoreInner { /// The authoritative store for durable data. primary_store: Arc, diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 1fbbaad7e2..194a31b711 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -54,7 +54,7 @@ use lightning::ln::msgs::SocketAddress; use lightning::routing::gossip::NodeAlias; use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse}; use lightning_invoice::{Bolt11InvoiceDescription, Description}; -use lightning_persister::fs_store::v1::FilesystemStore; +use lightning_persister::fs_store::v2::FilesystemStoreV2; use lightning_types::payment::{PaymentHash, PaymentPreimage}; use logging::TestLogWriter; use rand::distr::Alphanumeric; @@ -1845,7 +1845,7 @@ impl PaginatedKVStore for TestSyncStore { struct TestSyncStoreInner { serializer: tokio::sync::RwLock<()>, test_store: InMemoryStore, - fs_store: FilesystemStore, + fs_store: FilesystemStoreV2, sqlite_store: SqliteStore, } @@ -1854,7 +1854,7 @@ impl TestSyncStoreInner { let serializer = tokio::sync::RwLock::new(()); let mut fs_dir = dest_dir.clone(); fs_dir.push("fs_store"); - let fs_store = FilesystemStore::new(fs_dir); + let fs_store = FilesystemStoreV2::new(fs_dir).unwrap(); let mut sql_dir = dest_dir.clone(); sql_dir.push("sqlite_store"); let sqlite_store = SqliteStore::new( diff --git a/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index 2c36c2b159..76f04ee91f 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -34,6 +34,8 @@ use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; use ldk_node::config::{AsyncPaymentsRole, EsploraSyncConfig, DEFAULT_FULL_SCAN_STOP_GAP}; use ldk_node::entropy::NodeEntropy; +#[cfg(not(feature = "uniffi"))] +use ldk_node::io::sqlite_store::SqliteStore; use ldk_node::liquidity::LSPS2ServiceConfig; use ldk_node::payment::{ ConfirmationStatus, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, @@ -4460,3 +4462,73 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { cheap.stop().unwrap(); expensive.stop().unwrap(); } + +// Builder backup-store configuration is not yet exposed via FFI (see #871) +#[cfg(not(feature = "uniffi"))] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn builder_configures_sqlite_backup_store() { + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let chain_source = random_chain_source(&bitcoind, &electrsd); + + let mut config_a = random_config(); + config_a.store_type = TestStoreType::Sqlite; + let primary_dir = config_a.node_config.storage_dir_path.clone(); + let backup_dir = common::random_storage_path(); + + // Build node_a with backup storage configured + setup_builder!(builder_a, config_a.node_config.clone()); + builder_a.set_chain_source_esplora( + format!("http://{}", electrsd.esplora_url.as_ref().unwrap()), + None, + ); + builder_a.set_filesystem_logger(None, None); + builder_a.set_backup_storage_dir_path(backup_dir.to_str().unwrap().to_owned()); + + let node_a = builder_a.build(config_a.node_entropy.into()).unwrap(); + node_a.start().unwrap(); + assert!(node_a.status().is_running); + assert!(node_a.status().latest_fee_rate_cache_update_timestamp.is_some()); + + let config_b = random_config(); + let node_b = setup_node(&chain_source, config_b); + + do_channel_full_cycle( + node_a, + node_b, + &bitcoind.client, + &electrsd.client, + false, + true, + true, + false, + ) + .await; + + let primary_store = SqliteStore::new( + primary_dir.into(), + Some(ldk_node::io::sqlite_store::SQLITE_DB_FILE_NAME.to_string()), + Some(ldk_node::io::sqlite_store::KV_TABLE_NAME.to_string()), + ) + .unwrap(); + + let backup_store = SqliteStore::new( + backup_dir, + Some(ldk_node::io::sqlite_store::SQLITE_BACKUP_DB_FILE_NAME.to_string()), + Some(ldk_node::io::sqlite_store::KV_TABLE_NAME.to_string()), + ) + .unwrap(); + + for (pn, sn, key) in [ + ("bdk_wallet", "", "descriptor"), + ("bdk_wallet", "", "change_descriptor"), + ("bdk_wallet", "", "network"), + ("", "", "node_metrics"), + ("", "", "events"), + ("", "", "peers"), + ] { + let primary = primary_store.read(pn, sn, key).await.unwrap(); + let backup = backup_store.read(pn, sn, key).await.unwrap(); + + assert_eq!(backup, primary, "backup mismatch for {pn}/{sn}/{key}"); + } +} From 29fa6fb08ccf018018d94e94718dc4b4f9930912 Mon Sep 17 00:00:00 2001 From: Enigbe Date: Mon, 10 Aug 2026 15:21:36 +0100 Subject: [PATCH 3/5] fixup! Implement tiered storage List keys exclusively from the primary store until PaginatedKVStore provides ordering and cursor semantics that permit a correct cross-tier merge. Remove the ephemeral overlay logic and update the tests to document the resulting primary-only listing behavior. --- src/io/test_utils.rs | 2 +- src/io/tier_store.rs | 280 +++++++++---------------------------------- 2 files changed, 57 insertions(+), 225 deletions(-) diff --git a/src/io/test_utils.rs b/src/io/test_utils.rs index fa9b3e8cae..aadb4b79a8 100644 --- a/src/io/test_utils.rs +++ b/src/io/test_utils.rs @@ -159,7 +159,7 @@ impl chainmonitor::Persist const EXPECTED_UPDATES_PER_PAYMENT: u64 = 5; -pub(crate) use in_memory_store::{InMemoryStore, IN_MEMORY_PAGE_SIZE}; +pub(crate) use in_memory_store::InMemoryStore; pub(crate) fn random_storage_path() -> PathBuf { let mut temp_path = std::env::temp_dir(); diff --git a/src/io/tier_store.rs b/src/io/tier_store.rs index f7f3deb8e9..2e02fde4f6 100644 --- a/src/io/tier_store.rs +++ b/src/io/tier_store.rs @@ -35,6 +35,9 @@ use crate::types::DynStore; /// /// Reads and lists do not consult the backup store during normal operation. /// Ephemeral data is read from and written to the ephemeral store when configured. +/// Unpaginated listings expose the logical contents of the primary and ephemeral +/// stores, while paginated listings expose only the primary store until cross-tier +/// pagination semantics are defined. /// /// Note that dual-store writes and removals are not atomic across the primary and /// backup stores. If one store succeeds and the other fails, the operation @@ -496,8 +499,31 @@ impl TierStoreInner { let mut keys = self.list_primary(&primary_namespace, &secondary_namespace).await?; - self.apply_ephemeral_overlay(&primary_namespace, &secondary_namespace, &mut keys, true) - .await?; + let Some(ephemeral_store) = self.ephemeral_store.as_ref() else { + return Ok(keys); + }; + + if primary_namespace != NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE + && primary_namespace != SCORER_PERSISTENCE_PRIMARY_NAMESPACE + { + return Ok(keys); + } + + // The ephemeral store is authoritative for keys routed there. Exclude stale + // primary copies, then add only routed keys from the ephemeral store. + keys.retain(|key| !is_ephemeral_cached_key(&primary_namespace, &secondary_namespace, key)); + + let ephemeral_keys = + KVStore::list(ephemeral_store.as_ref(), &primary_namespace, &secondary_namespace) + .await?; + + for key in ephemeral_keys { + if is_ephemeral_cached_key(&primary_namespace, &secondary_namespace, &key) + && !keys.contains(&key) + { + keys.push(key); + } + } Ok(keys) } @@ -513,89 +539,16 @@ impl TierStoreInner { "list_paginated", )?; - let mut response = PaginatedKVStore::list_paginated( + // TODO(@enigbe): Merge listings across tiers once `PaginatedKVStore` provides a + // shared ordering and cursor contract that permits a correct cross-store merge. + // Until then, listings intentionally expose only the primary tier. + PaginatedKVStore::list_paginated( self.primary_store.as_ref(), &primary_namespace, &secondary_namespace, page_token, ) - .await?; - - // Filter stale primary copies of ephemeral-cached keys from every page, and append the - // live ephemeral overlay only once primary pagination is exhausted (`next_page_token` - // is `None`). - // - // Because filtering can drop entries from a full page, a non-terminal page can come back - // shorter than the nominal page size: callers must not treat `next_page_token.is_some()` - // as a guarantee of a full page. Conversely, appending the overlay onto an already - // full-sized terminal page can make it up to `MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE` - // entries larger than the nominal page size. Both effects are bounded by that constant. - let append_live = response.next_page_token.is_none(); - self.apply_ephemeral_overlay( - &primary_namespace, - &secondary_namespace, - &mut response.keys, - append_live, - ) - .await?; - - Ok(response) - } - - /// Reconciles a set of keys already listed from the primary store with the ephemeral store. - /// - /// This keeps `list`/`list_paginated` consistent with the key-level routing in - /// `read`/`write`/`remove`: once an ephemeral store is configured it is authoritative for - /// ephemeral-cached keys (`network_graph`/`scorer`). Any copy of such a key still held by the - /// primary store is a stale leftover from before the ephemeral store was configured, so we - /// drop it here; when `append_live` is set we then append the live ephemeral copy. - /// - /// The ephemeral store is only consulted for namespaces that can actually contain an - /// ephemeral-cached key (see [`namespace_may_hold_ephemeral_cached_key`]). For every other - /// namespace this is a no-op, so listing durable, primary-backed data never depends on the - /// optional ephemeral backend being reachable. - async fn apply_ephemeral_overlay( - &self, primary_namespace: &str, secondary_namespace: &str, keys: &mut Vec, - append_live: bool, - ) -> io::Result<()> { - let Some(eph_store) = self.ephemeral_store.as_ref() else { - return Ok(()); - }; - - if !namespace_may_hold_ephemeral_cached_key(primary_namespace) { - return Ok(()); - } - - keys.retain(|key| !is_ephemeral_cached_key(primary_namespace, secondary_namespace, key)); - - if !append_live { - return Ok(()); - } - - // We don't retry ephemeral-store lists here. Local failures are treated as terminal for - // this access path rather than falling back to another store. - let cached_keys: Vec = - KVStore::list(eph_store.as_ref(), primary_namespace, secondary_namespace) - .await? - .into_iter() - .filter(|key| is_ephemeral_cached_key(primary_namespace, secondary_namespace, key)) - .collect(); - - debug_assert!( - cached_keys.len() <= MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE, - "ephemeral-cached key overlay ({} keys) exceeded the bound this pagination \ - design assumes (see MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE)", - cached_keys.len(), - ); - - for key in cached_keys { - // Guards against `list` on the ephemeral store itself returning a duplicate entry. - if !keys.contains(&key) { - keys.push(key); - } - } - - Ok(()) + .await } fn handle_primary_backup_results( @@ -647,18 +600,6 @@ impl TierStoreInner { } } -/// The maximum number of distinct keys [`is_ephemeral_cached_key`] can ever match for a -/// single namespace pair -- one per matched key literal (`network_graph`, `scorer`). -/// -/// `apply_ephemeral_overlay` reads the ephemeral-cached overlay unpaginated and appends -/// it onto the terminal primary page in a single shot, reusing primary's own page token -/// unmodified rather than tracking a cross-store cursor. That's only sound while this -/// stays small. If a future key is added to `is_ephemeral_cached_key`, bump this constant -/// to match, and re-examine whether `list_paginated_internal` still needs revisiting -- a -/// larger or unbounded ephemeral-cached set can silently blow past the nominal page size -/// and can't correctly report `next_page_token` for a partial overlay. -const MAX_EPHEMERAL_CACHED_KEYS_PER_NAMESPACE: usize = 2; - fn is_ephemeral_cached_key(pn: &str, sn: &str, key: &str) -> bool { matches!( (pn, sn, key), @@ -667,16 +608,6 @@ fn is_ephemeral_cached_key(pn: &str, sn: &str, key: &str) -> bool { ) } -/// Whether a primary namespace can ever contain a key that [`is_ephemeral_cached_key`] matches. -/// -/// Listing paths use this to gate the (optional) ephemeral-store lookup: for any namespace that -/// cannot hold an ephemeral-cached key, the overlay is skipped entirely, so durable, -/// primary-backed listings neither pay for an extra ephemeral list nor fail when the ephemeral -/// backend is unreachable. -fn namespace_may_hold_ephemeral_cached_key(pn: &str) -> bool { - pn == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE || pn == SCORER_PERSISTENCE_PRIMARY_NAMESPACE -} - #[cfg(test)] mod tests { use std::future::Future; @@ -696,7 +627,7 @@ mod tests { use super::*; use crate::io::test_utils::{ - do_read_write_remove_list_persist, random_storage_path, InMemoryStore, IN_MEMORY_PAGE_SIZE, + do_read_write_remove_list_persist, random_storage_path, InMemoryStore, }; use crate::io::tier_store::TierStore; use crate::logger::Logger; @@ -849,7 +780,7 @@ mod tests { } #[tokio::test] - async fn list_discovers_durable_keys_alongside_ephemeral_cache() { + async fn list_exposes_primary_and_routed_ephemeral_keys() { let base_dir = random_storage_path(); let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); @@ -896,6 +827,15 @@ mod tests { ) .await .unwrap(); + ephemeral_store + .write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + "ephemeral-root-decoy", + vec![4u8; 32], + ) + .await + .unwrap(); // This is `list("", "")`: CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE and // NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE are the same empty string, so both @@ -908,10 +848,11 @@ mod tests { .await .unwrap(); - // The durable primary-backed key and the ephemeral-cached key must both be - // discoverable from a single call. + // Unpaginated listing exposes the logical view across both tiers without leaking + // unrelated keys from the ephemeral store. assert!(root_keys.contains(&CHANNEL_MANAGER_PERSISTENCE_KEY.to_string())); assert!(root_keys.contains(&NETWORK_GRAPH_PERSISTENCE_KEY.to_string())); + assert!(!root_keys.contains(&"ephemeral-root-decoy".to_string())); let monitor_keys = KVStore::list( &tier, @@ -927,7 +868,7 @@ mod tests { } #[tokio::test] - async fn list_paginated_routes_to_selected_tier() { + async fn list_paginated_only_exposes_primary_keys() { let base_dir = random_storage_path(); let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); @@ -997,80 +938,7 @@ mod tests { .unwrap(); assert_eq!(primary_response.keys, vec!["monitor-key".to_string()]); - let ephemeral_response = PaginatedKVStore::list_paginated( - &tier, - NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, - None, - ) - .await - .unwrap(); - - // The durable root-namespace key surfaces from the primary store, and the - // ephemeral-cached `network_graph` key is appended once primary's pagination - // is exhausted. - assert_eq!( - ephemeral_response.keys, - vec!["other-root-namespace-key".to_string(), NETWORK_GRAPH_PERSISTENCE_KEY.to_string()] - ); - } - - #[tokio::test] - async fn list_paginated_filters_stale_primary_copies_across_every_page() { - let base_dir = random_storage_path(); - let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); - let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); - - let _cleanup = CleanupDir(base_dir.clone()); - - let primary_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); - let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); - - let ephemeral_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); - tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); - - // Exactly IN_MEMORY_PAGE_SIZE unrelated, durable root-namespace keys written - // directly to primary, so a single additional entry (the stale copy below) - // pushes the namespace to exactly two pages. - for i in 0..IN_MEMORY_PAGE_SIZE { - primary_store - .write( - NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, - &format!("filler-{i:02}"), - vec![0u8; 32], - ) - .await - .unwrap(); - } - - // A stale copy of `network_graph`, written directly to primary as if it had - // been persisted there before the ephemeral store was configured. Writing it - // last gives it the highest creation order, guaranteeing it lands on the - // *first* page rather than the last -- filtering only the terminal page - // would miss it entirely. - primary_store - .write( - NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_KEY, - vec![1u8; 32], - ) - .await - .unwrap(); - - // The live, authoritative copy, written through the tier so it is routed to - // the ephemeral store. - tier.write( - NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_KEY, - vec![2u8; 32], - ) - .await - .unwrap(); - - let page_one = PaginatedKVStore::list_paginated( + let root_response = PaginatedKVStore::list_paginated( &tier, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, @@ -1079,45 +947,11 @@ mod tests { .await .unwrap(); - // The stale primary copy must be filtered out of the first page even though - // the walk isn't done yet -- it should never be visible once an ephemeral - // store is configured. - let expected_page_one: Vec = - (1..IN_MEMORY_PAGE_SIZE).rev().map(|i| format!("filler-{i:02}")).collect(); - assert_eq!(page_one.keys, expected_page_one); - assert!(page_one.next_page_token.is_some()); - - let page_two = PaginatedKVStore::list_paginated( - &tier, - NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, - page_one.next_page_token, - ) - .await - .unwrap(); - - // The final page carries the one remaining filler key plus the ephemeral - // store's live copy, appended exactly once now that primary pagination is - // exhausted. - assert_eq!( - page_two.keys, - vec!["filler-00".to_string(), NETWORK_GRAPH_PERSISTENCE_KEY.to_string()] - ); - assert!(page_two.next_page_token.is_none()); - - // Across the full walk, `network_graph` appears exactly once -- this is the - // property that would have caught the duplicate-key bug. - let occurrences = page_one - .keys - .iter() - .chain(page_two.keys.iter()) - .filter(|k| k.as_str() == NETWORK_GRAPH_PERSISTENCE_KEY) - .count(); - assert_eq!(occurrences, 1); + assert_eq!(root_response.keys, vec!["other-root-namespace-key".to_string()]); } #[tokio::test] - async fn list_unrelated_namespace_survives_ephemeral_list_failure() { + async fn listings_only_consult_ephemeral_store_for_routed_namespaces() { let base_dir = random_storage_path(); let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); @@ -1152,7 +986,7 @@ mod tests { .unwrap(); assert_eq!(monitor_keys, vec!["monitor-key".to_string()]); - // The paginated path is gated identically. + // The paginated path always exposes only the primary store. let monitor_page = PaginatedKVStore::list_paginated( &tier, CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, @@ -1163,8 +997,7 @@ mod tests { .unwrap(); assert_eq!(monitor_page.keys, vec!["monitor-key".to_string()]); - // Sanity check the fixture: listing a namespace that *can* hold an ephemeral-cached key - // still surfaces the ephemeral failure rather than silently swallowing it. + // An unpaginated root listing must consult the authoritative ephemeral store. assert!(KVStore::list( &tier, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, @@ -1175,7 +1008,7 @@ mod tests { } #[tokio::test] - async fn list_filters_stale_primary_copy_when_ephemeral_missing() { + async fn list_hides_stale_primary_copy_when_ephemeral_key_is_missing() { let base_dir = random_storage_path(); let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); @@ -1219,9 +1052,8 @@ mod tests { .await .unwrap(); - // The durable key is still listed, and the stale primary copy of `network_graph` is - // filtered out so `list` cannot report a key that `read` would fail to fetch -- matching - // the paginated path. + // The ephemeral store is authoritative for routed keys, so its missing entry hides + // the stale primary copy. assert!(root_keys.contains(&CHANNEL_MANAGER_PERSISTENCE_KEY.to_string())); assert!(!root_keys.contains(&NETWORK_GRAPH_PERSISTENCE_KEY.to_string())); } From 9290e8b6444b4e7246930c752dc0c00e5db68deb Mon Sep 17 00:00:00 2001 From: Enigbe Date: Mon, 10 Aug 2026 16:59:14 +0100 Subject: [PATCH 4/5] fixup! Implement tiered storage Route the external pathfinding scores cache through the ephemeral store, alongside the network graph and scorer, as it is rebuildable cached data. Add coverage for writing, reading, and removing the cache through the ephemeral tier. --- src/io/tier_store.rs | 68 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 66 insertions(+), 2 deletions(-) diff --git a/src/io/tier_store.rs b/src/io/tier_store.rs index 2e02fde4f6..dc5f069497 100644 --- a/src/io/tier_store.rs +++ b/src/io/tier_store.rs @@ -18,7 +18,7 @@ use lightning::util::persist::{ use lightning::{io, log_error}; use tokio::sync::Mutex as TokioMutex; -use crate::io::utils::check_namespace_key_validity; +use crate::io::utils::{check_namespace_key_validity, EXTERNAL_PATHFINDING_SCORES_CACHE_KEY}; use crate::logger::{LdkLogger, Logger}; use crate::types::DynStore; @@ -605,6 +605,7 @@ fn is_ephemeral_cached_key(pn: &str, sn: &str, key: &str) -> bool { (pn, sn, key), (NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, _, NETWORK_GRAPH_PERSISTENCE_KEY) | (SCORER_PERSISTENCE_PRIMARY_NAMESPACE, _, SCORER_PERSISTENCE_KEY) + | (SCORER_PERSISTENCE_PRIMARY_NAMESPACE, _, EXTERNAL_PATHFINDING_SCORES_CACHE_KEY) ) } @@ -621,7 +622,7 @@ mod tests { CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE, - NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE, }; use lightning_persister::fs_store::v2::FilesystemStoreV2; @@ -779,6 +780,69 @@ mod tests { assert_eq!(primary_read_cm.unwrap(), data); } + #[tokio::test] + async fn external_pathfinding_scores_cache_routes_to_ephemeral_store() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir.clone()); + + let primary_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("primary")).unwrap())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FilesystemStoreV2::new(base_dir.join("ephemeral")).unwrap())); + tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); + + let data = vec![42u8; 32]; + tier.write( + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, + SCORER_PERSISTENCE_SECONDARY_NAMESPACE, + EXTERNAL_PATHFINDING_SCORES_CACHE_KEY, + data.clone(), + ) + .await + .unwrap(); + + assert!(primary_store + .read( + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, + SCORER_PERSISTENCE_SECONDARY_NAMESPACE, + EXTERNAL_PATHFINDING_SCORES_CACHE_KEY, + ) + .await + .is_err()); + assert_eq!( + tier.read( + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, + SCORER_PERSISTENCE_SECONDARY_NAMESPACE, + EXTERNAL_PATHFINDING_SCORES_CACHE_KEY, + ) + .await + .unwrap(), + data + ); + + tier.remove( + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, + SCORER_PERSISTENCE_SECONDARY_NAMESPACE, + EXTERNAL_PATHFINDING_SCORES_CACHE_KEY, + false, + ) + .await + .unwrap(); + assert!(ephemeral_store + .read( + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, + SCORER_PERSISTENCE_SECONDARY_NAMESPACE, + EXTERNAL_PATHFINDING_SCORES_CACHE_KEY, + ) + .await + .is_err()); + } + #[tokio::test] async fn list_exposes_primary_and_routed_ephemeral_keys() { let base_dir = random_storage_path(); From 9f20edf83592509c23899dfd0370bd1b91e9925d Mon Sep 17 00:00:00 2001 From: Enigbe Date: Mon, 10 Aug 2026 23:24:22 +0100 Subject: [PATCH 5/5] fixup! Implement tiered storage Record the attempted write version even when a multi-store peration fails, as one store may already contain the newer state. This prevents a previously started write that completes later from overwriting a newer primary write whose corresponding backup write failed. Add a regression test covering a successful primary write paired with a failed backup write. --- src/io/tier_store.rs | 131 ++++++++++++++++++++++++++++++++++++------- 1 file changed, 112 insertions(+), 19 deletions(-) diff --git a/src/io/tier_store.rs b/src/io/tier_store.rs index dc5f069497..2a6b047cce 100644 --- a/src/io/tier_store.rs +++ b/src/io/tier_store.rs @@ -367,9 +367,9 @@ impl TierStoreInner { Ok(()) } else { let res = callback().await; - if res.is_ok() { - *last_written_version = version; - } + // A failed multi-store operation may still have updated one of its stores. We record + // the attempted version regardless so an older operation cannot overwrite newer state. + *last_written_version = version; res } }; @@ -614,6 +614,7 @@ mod tests { use std::future::Future; use std::panic::RefUnwindSafe; use std::path::PathBuf; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use lightning::util::logger::Level; @@ -647,20 +648,25 @@ mod tests { TierStore::new(primary_store, logger) } - /// A store whose `list`/`list_paginated` always fail while every other operation is delegated - /// to an inner [`InMemoryStore`]. Used to prove that a failing ephemeral list does not sink a - /// listing for a namespace that can never hold an ephemeral-cached key. - struct FailingListStore { + enum FailureMode { + List, + Write { attempts: Arc }, + } + + /// A store that injects a selected failure while delegating other operations to an inner + /// [`InMemoryStore`]. + struct FailingStore { inner: InMemoryStore, + failure_mode: FailureMode, } - impl FailingListStore { - fn new() -> Self { - Self { inner: InMemoryStore::new() } + impl FailingStore { + fn new(failure_mode: FailureMode) -> Self { + Self { inner: InMemoryStore::new(), failure_mode } } } - impl KVStore for FailingListStore { + impl KVStore for FailingStore { fn read( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, ) -> impl Future, io::Error>> + 'static + Send { @@ -669,7 +675,18 @@ mod tests { fn write( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec, ) -> impl Future> + 'static + Send { - KVStore::write(&self.inner, primary_namespace, secondary_namespace, key, buf) + let write = if let FailureMode::Write { attempts } = &self.failure_mode { + attempts.fetch_add(1, Ordering::Relaxed); + None + } else { + Some(KVStore::write(&self.inner, primary_namespace, secondary_namespace, key, buf)) + }; + async move { + match write { + Some(write) => write.await, + None => Err(io::Error::new(io::ErrorKind::Other, "write failed")), + } + } } fn remove( &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, @@ -677,18 +694,43 @@ mod tests { KVStore::remove(&self.inner, primary_namespace, secondary_namespace, key, lazy) } fn list( - &self, _primary_namespace: &str, _secondary_namespace: &str, + &self, primary_namespace: &str, secondary_namespace: &str, ) -> impl Future, io::Error>> + 'static + Send { - async { Err(io::Error::new(io::ErrorKind::Other, "list failed")) } + let list = match &self.failure_mode { + FailureMode::List => None, + FailureMode::Write { .. } => { + Some(KVStore::list(&self.inner, primary_namespace, secondary_namespace)) + }, + }; + async move { + match list { + Some(list) => list.await, + None => Err(io::Error::new(io::ErrorKind::Other, "list failed")), + } + } } } - impl PaginatedKVStore for FailingListStore { + impl PaginatedKVStore for FailingStore { fn list_paginated( - &self, _primary_namespace: &str, _secondary_namespace: &str, - _page_token: Option, + &self, primary_namespace: &str, secondary_namespace: &str, + page_token: Option, ) -> impl Future> + 'static + Send { - async { Err(io::Error::new(io::ErrorKind::Other, "list_paginated failed")) } + let list = match &self.failure_mode { + FailureMode::List => None, + FailureMode::Write { .. } => Some(PaginatedKVStore::list_paginated( + &self.inner, + primary_namespace, + secondary_namespace, + page_token, + )), + }; + async move { + match list { + Some(list) => list.await, + None => Err(io::Error::new(io::ErrorKind::Other, "list_paginated failed")), + } + } } } @@ -1026,7 +1068,8 @@ mod tests { let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); // An ephemeral store whose `list`/`list_paginated` always fail. - let ephemeral_store: Arc = Arc::new(DynStoreWrapper(FailingListStore::new())); + let ephemeral_store: Arc = + Arc::new(DynStoreWrapper(FailingStore::new(FailureMode::List))); tier.set_ephemeral_store(Arc::clone(&ephemeral_store)); // A durable key in a namespace that can never hold an ephemeral-cached key. @@ -1165,6 +1208,56 @@ mod tests { assert_eq!(persisted, new_data); } + #[tokio::test] + async fn failed_newer_backup_write_still_supersedes_older_write() { + let base_dir = random_storage_path(); + let log_path = base_dir.join("tier_store_test.log").to_string_lossy().into_owned(); + let logger = Arc::new(Logger::new_fs_writer(log_path, Level::Trace).unwrap()); + + let _cleanup = CleanupDir(base_dir); + + let primary_store: Arc = Arc::new(DynStoreWrapper(InMemoryStore::new())); + let mut tier = setup_tier_store(Arc::clone(&primary_store), logger); + + let backup_write_attempts = Arc::new(AtomicUsize::new(0)); + let backup_store: Arc = + Arc::new(DynStoreWrapper(FailingStore::new(FailureMode::Write { + attempts: Arc::clone(&backup_write_attempts), + }))); + tier.set_backup_store(backup_store); + + let old_data = vec![1u8; 32]; + let new_data = vec![2u8; 32]; + let old_write = tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + old_data, + ); + let new_write = tier.write( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + new_data.clone(), + ); + + // The primary write succeeds, but the same newer write fails on the backup. + assert!(new_write.await.is_err()); + // The older operation must be treated as stale even though the newer operation failed. + old_write.await.unwrap(); + + let persisted = primary_store + .read( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ) + .await + .unwrap(); + assert_eq!(persisted, new_data); + assert_eq!(backup_write_attempts.load(Ordering::Relaxed), 1); + } + #[tokio::test] async fn ephemeral_writes_preserve_latest_call_order() { let base_dir = random_storage_path();