Read persisted LSPS2 service state in `LiquidityManager::new`
What changed, and why it matters
This commit changes how the Lightning Dev Kit liquidity manager starts up: it now reads previously saved LSPS2 service state from disk when created. Previously, the manager started with empty in-memory state even if data had been saved, which could cause it to lose track of active liquidity requests and routing information after a restart. The change also makes the constructors return errors if the saved data is corrupt or inconsistent, rather than silently starting fresh. This is a reliability and data-integrity improvement, not a typical remote-exploitable vulnerability.
Treat this as a bug-fix/data-integrity hardening commit. Review downstream callers that now receive a Result to ensure errors are handled rather than unwrapped in production. Verify that the KVStore read in the sync wrapper cannot legitimately return Pending, or add a proper blocking mechanism if it can. Confirm that persisted state is written atomically to avoid the newly-detected inconsistent duplicates.
Security signals we found
Constructor now reads persisted state instead of starting with empty peer state
Adds duplicate-key validation for intercept SCIDs and channel IDs when reloading state
Makes constructors fallible so corrupt/inconsistent persisted data is reported rather than ignored
Sync wrapper uses dummy waker and unreachable! on Pending, which assumes the async read will complete synchronously
Evidence from the diff
The patch makes LiquidityManager::new and LiquidityManager::new_with_custom_time_provider async and fallible (Result). On construction they call a new helper read_lsps2_service_peer_states, which enumerates the LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE entries in the supplied KVStore, deserializes each stored PeerState, and returns a map keyed by PublicKey. LSPS2ServiceHandler::new now accepts that map and rebuilds its peer_by_intercept_scid and peer_by_channel_id reverse indexes, returning an InvalidData error if duplicate intercept SCIDs or channel IDs are detected. Sync wrappers poll the async constructor with a dummy waker and propagate errors. Callers are updated to unwrap() the Result in tests/fuzz. The change closes a state-loss window on restart and adds validation against inconsistent persisted data.
Changed components
lightning-liquidity/src/manager.rslightning-liquidity/src/lsps2/service.rslightning-liquidity/src/persist.rslightning-background-processor/src/lib.rsfuzz/src/lsps_message.rslightning-liquidity/tests/common/mod.rsInspect captured patch +166 / −41
diff --git a/fuzz/src/lsps_message.rs b/fuzz/src/lsps_message.rs
index 2b5ee68..2bc83c3 100644
--- a/fuzz/src/lsps_message.rs
+++ b/fuzz/src/lsps_message.rs
@@ -86,7 +86,7 @@ pub fn do_test(data: &[u8]) {
kv_store,
None,
None,
- ));
+ ).unwrap());
let mut reader = data;
if let Ok(Some(msg)) = liquidity_manager.read(LSPS_MESSAGE_TYPE_ID, &mut reader) {
let secp = Secp256k1::signing_only();
diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index f10a3e2..ef6e82a 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -2370,16 +2370,19 @@ mod tests {
Arc::clone(&logger),
Arc::clone(&keys_manager),
));
- let liquidity_manager = Arc::new(LiquidityManagerSync::new(
- Arc::clone(&keys_manager),
- Arc::clone(&keys_manager),
- Arc::clone(&manager),
- None,
- None,
- Arc::clone(&kv_store),
- None,
- None,
- ));
+ let liquidity_manager = Arc::new(
+ LiquidityManagerSync::new(
+ Arc::clone(&keys_manager),
+ Arc::clone(&keys_manager),
+ Arc::clone(&manager),
+ None,
+ None,
+ Arc::clone(&kv_store),
+ None,
+ None,
+ )
+ .unwrap(),
+ );
let node = Node {
node: manager,
p2p_gossip_sync,
diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs
index 9e8073f..aa3f0e7 100644
--- a/lightning-liquidity/src/lsps2/service.rs
+++ b/lightning-liquidity/src/lsps2/service.rs
@@ -465,7 +465,7 @@ impl OutboundJITChannel {
}
}
-struct PeerState {
+pub(crate) struct PeerState {
outbound_channels_by_intercept_scid: HashMap<u64, OutboundJITChannel>,
intercept_scid_by_user_channel_id: HashMap<u128, u64>,
intercept_scid_by_channel_id: HashMap<ChannelId, u64>,
@@ -592,20 +592,48 @@ where
{
/// Constructs a `LSPS2ServiceHandler`.
pub(crate) fn new(
- pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue<K>>,
- channel_manager: CM, kv_store: K, config: LSPS2ServiceConfig,
- ) -> Self {
- Self {
+ per_peer_state: HashMap<PublicKey, Mutex<PeerState>>, pending_messages: Arc<MessageQueue>,
+ pending_events: Arc<EventQueue<K>>, channel_manager: CM, kv_store: K,
+ config: LSPS2ServiceConfig,
+ ) -> Result<Self, lightning::io::Error> {
+ let mut peer_by_intercept_scid = new_hash_map();
+ let mut peer_by_channel_id = new_hash_map();
+ for (node_id, peer_state) in per_peer_state.iter() {
+ let peer_state_lock = peer_state.lock().unwrap();
+ for (intercept_scid, _) in peer_state_lock.outbound_channels_by_intercept_scid.iter() {
+ let res = peer_by_intercept_scid.insert(*intercept_scid, *node_id);
+ debug_assert!(res.is_none(), "Intercept SCIDs should never collide");
+ if res.is_some() {
+ return Err(lightning::io::Error::new(
+ lightning::io::ErrorKind::InvalidData,
+ "Failed to read LSPS2 peer state due to data inconsistencies: Intercept SCIDs should never collide",
+ ));
+ }
+ }
+
+ for (channel_id, _) in peer_state_lock.intercept_scid_by_channel_id.iter() {
+ let res = peer_by_channel_id.insert(*channel_id, *node_id);
+ debug_assert!(res.is_none(), "Channel IDs should never collide");
+ if res.is_some() {
+ return Err(lightning::io::Error::new(
+ lightning::io::ErrorKind::InvalidData,
+ "Failed to read LSPS2 peer state due to data inconsistencies: Channel IDs should never collide",
+ ));
+ }
+ }
+ }
+
+ Ok(Self {
pending_messages,
pending_events,
- per_peer_state: RwLock::new(new_hash_map()),
- peer_by_intercept_scid: RwLock::new(new_hash_map()),
- peer_by_channel_id: RwLock::new(new_hash_map()),
+ per_peer_state: RwLock::new(per_peer_state),
+ peer_by_intercept_scid: RwLock::new(peer_by_intercept_scid),
+ peer_by_channel_id: RwLock::new(peer_by_channel_id),
total_pending_requests: AtomicUsize::new(0),
channel_manager,
kv_store,
config,
- }
+ })
}
/// Returns a reference to the used config.
diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs
index e3e059a..76403f1 100644
--- a/lightning-liquidity/src/manager.rs
+++ b/lightning-liquidity/src/manager.rs
@@ -24,6 +24,7 @@ use crate::lsps5::client::{LSPS5ClientConfig, LSPS5ClientHandler};
use crate::lsps5::msgs::LSPS5Message;
use crate::lsps5::service::{LSPS5ServiceConfig, LSPS5ServiceHandler};
use crate::message_queue::MessageQueue;
+use crate::persist::read_lsps2_service_peer_states;
use crate::lsps1::client::{LSPS1ClientConfig, LSPS1ClientHandler};
use crate::lsps1::msgs::LSPS1Message;
@@ -327,12 +328,14 @@ where
K::Target: KVStore,
{
/// Constructor for the [`LiquidityManager`] using the default system clock
- pub fn new(
+ ///
+ /// Will read persisted service states from the given [`KVStore`].
+ pub async fn new(
entropy_source: ES, node_signer: NS, channel_manager: CM, chain_source: Option<C>,
chain_params: Option<ChainParameters>, kv_store: K,
service_config: Option<LiquidityServiceConfig>,
client_config: Option<LiquidityClientConfig>,
- ) -> Self {
+ ) -> Result<Self, lightning::io::Error> {
let time_provider = Arc::new(DefaultTimeProvider);
Self::new_with_custom_time_provider(
entropy_source,
@@ -345,6 +348,7 @@ where
client_config,
time_provider,
)
+ .await
}
}
@@ -366,16 +370,18 @@ where
{
/// Constructor for the [`LiquidityManager`] with a custom time provider.
///
+ /// Will read persisted service states from the given [`KVStore`].
+ ///
/// This should be used on non-std platforms where access to the system time is not
/// available.
/// Sets up the required protocol message handlers based on the given
/// [`LiquidityClientConfig`] and [`LiquidityServiceConfig`].
- pub fn new_with_custom_time_provider(
+ pub async fn new_with_custom_time_provider(
entropy_source: ES, node_signer: NS, channel_manager: CM, chain_source: Option<C>,
chain_params: Option<ChainParameters>, kv_store: K,
service_config: Option<LiquidityServiceConfig>,
client_config: Option<LiquidityClientConfig>, time_provider: TP,
- ) -> Self {
+ ) -> Result<Self, lightning::io::Error> {
let pending_messages = Arc::new(MessageQueue::new());
let pending_events = Arc::new(EventQueue::new(kv_store.clone()));
let ignored_peers = RwLock::new(new_hash_set());
@@ -392,22 +398,30 @@ where
)
})
});
- let lsps2_service_handler = service_config.as_ref().and_then(|config| {
- config.lsps2_service_config.as_ref().map(|config| {
+
+ let lsps2_service_handler = if let Some(service_config) = service_config.as_ref() {
+ if let Some(lsps2_service_config) = service_config.lsps2_service_config.as_ref() {
if let Some(number) =
<LSPS2ServiceHandler<CM, K> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
{
supported_protocols.push(number);
}
- LSPS2ServiceHandler::new(
+
+ let peer_states = read_lsps2_service_peer_states(kv_store.clone()).await?;
+ Some(LSPS2ServiceHandler::new(
+ peer_states,
Arc::clone(&pending_messages),
Arc::clone(&pending_events),
channel_manager.clone(),
kv_store.clone(),
- config.clone(),
- )
- })
- });
+ lsps2_service_config.clone(),
+ )?)
+ } else {
+ None
+ }
+ } else {
+ None
+ };
let lsps5_client_handler = client_config.as_ref().and_then(|config| {
config.lsps5_client_config.as_ref().map(|config| {
@@ -482,7 +496,7 @@ where
None
};
- Self {
+ Ok(Self {
pending_messages,
pending_events,
request_id_to_method_map: Mutex::new(new_hash_map()),
@@ -500,7 +514,7 @@ where
_client_config: client_config,
best_block: RwLock::new(chain_params.map(|chain_params| chain_params.best_block)),
_chain_source: chain_source,
- }
+ })
}
/// Returns a reference to the LSPS0 client-side handler.
@@ -1038,9 +1052,10 @@ where
chain_params: Option<ChainParameters>, kv_store_sync: KS,
service_config: Option<LiquidityServiceConfig>,
client_config: Option<LiquidityClientConfig>,
- ) -> Self {
+ ) -> Result<Self, lightning::io::Error> {
let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
- let inner = Arc::new(LiquidityManager::new(
+
+ let mut fut = Box::pin(LiquidityManager::new(
entropy_source,
node_signer,
channel_manager,
@@ -1050,7 +1065,17 @@ where
service_config,
client_config,
));
- Self { inner }
+
+ let mut waker = dummy_waker();
+ let mut ctx = task::Context::from_waker(&mut waker);
+ let inner = match fut.as_mut().poll(&mut ctx) {
+ task::Poll::Ready(result) => result,
+ task::Poll::Pending => {
+ // In a sync context, we can't wait for the future to complete.
+ unreachable!("LiquidityManager::new should not be pending in a sync context");
+ },
+ }?;
+ Ok(Self { inner: Arc::new(inner) })
}
}
@@ -1078,9 +1103,9 @@ where
chain_params: Option<ChainParameters>, kv_store_sync: KS,
service_config: Option<LiquidityServiceConfig>,
client_config: Option<LiquidityClientConfig>, time_provider: TP,
- ) -> Self {
+ ) -> Result<Self, lightning::io::Error> {
let kv_store = Arc::new(KVStoreSyncWrapper(kv_store_sync));
- let inner = Arc::new(LiquidityManager::new_with_custom_time_provider(
+ let mut fut = Box::pin(LiquidityManager::new_with_custom_time_provider(
entropy_source,
node_signer,
channel_manager,
@@ -1091,7 +1116,17 @@ where
client_config,
time_provider,
));
- Self { inner }
+
+ let mut waker = dummy_waker();
+ let mut ctx = task::Context::from_waker(&mut waker);
+ let inner = match fut.as_mut().poll(&mut ctx) {
+ task::Poll::Ready(result) => result,
+ task::Poll::Pending => {
+ // In a sync context, we can't wait for the future to complete.
+ unreachable!("LiquidityManager::new should not be pending in a sync context");
+ },
+ }?;
+ Ok(Self { inner: Arc::new(inner) })
}
/// Returns a reference to the LSPS0 client-side handler.
diff --git a/lightning-liquidity/src/persist.rs b/lightning-liquidity/src/persist.rs
index 8b62220..1f7e60d 100644
--- a/lightning-liquidity/src/persist.rs
+++ b/lightning-liquidity/src/persist.rs
@@ -9,6 +9,19 @@
//! Types and utils for persistence.
+use crate::lsps2::service::PeerState as LSPS2ServicePeerState;
+use crate::prelude::{new_hash_map, HashMap};
+use crate::sync::Mutex;
+
+use lightning::io::Cursor;
+use lightning::util::persist::KVStore;
+use lightning::util::ser::Readable;
+
+use bitcoin::secp256k1::PublicKey;
+
+use core::ops::Deref;
+use core::str::FromStr;
+
/// The primary namespace under which the [`LiquidityManager`] will be persisted.
///
/// [`LiquidityManager`]: crate::LiquidityManager
@@ -33,3 +46,47 @@ pub const LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps2_service";
///
/// [`LSPS5ServiceHandler`]: crate::lsps5::service::LSPS5ServiceHandler
pub const LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "lsps5_service";
+
+pub(crate) async fn read_lsps2_service_peer_states<K: Deref>(
+ kv_store: K,
+) -> Result<HashMap<PublicKey, Mutex<LSPS2ServicePeerState>>, lightning::io::Error>
+where
+ K::Target: KVStore,
+{
+ let mut res = new_hash_map();
+
+ for stored_key in kv_store
+ .list(
+ LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+ LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+ )
+ .await?
+ {
+ let mut reader = Cursor::new(
+ kv_store
+ .read(
+ LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+ LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+ &stored_key,
+ )
+ .await?,
+ );
+
+ let peer_state = LSPS2ServicePeerState::read(&mut reader).map_err(|_| {
+ lightning::io::Error::new(
+ lightning::io::ErrorKind::InvalidData,
+ "Failed to deserialize LSPS2 peer state",
+ )
+ })?;
+
+ let key = PublicKey::from_str(&stored_key).map_err(|_| {
+ lightning::io::Error::new(
+ lightning::io::ErrorKind::InvalidData,
+ "Failed to deserialize stored key entry",
+ )
+ })?;
+
+ res.insert(key, Mutex::new(peer_state));
+ }
+ Ok(res)
+}
diff --git a/lightning-liquidity/tests/common/mod.rs b/lightning-liquidity/tests/common/mod.rs
index c7ec116..751de1e 100644
--- a/lightning-liquidity/tests/common/mod.rs
+++ b/lightning-liquidity/tests/common/mod.rs
@@ -38,7 +38,8 @@ pub(crate) fn create_service_and_client_nodes<'a, 'b, 'c>(
Some(service_config),
None,
Arc::clone(&time_provider),
- );
+ )
+ .unwrap();
let client_kv_store = Arc::new(TestStore::new(false));
let client_lm = LiquidityManagerSync::new_with_custom_time_provider(
@@ -51,7 +52,8 @@ pub(crate) fn create_service_and_client_nodes<'a, 'b, 'c>(
None,
Some(client_config),
time_provider,
- );
+ )
+ .unwrap();
let mut iter = nodes.into_iter();
let service_node = LiquidityNode::new(iter.next().unwrap(), service_lm);
Why this scored 30/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.