Fix races when removing per-peer state from KVStore in LSPS2/5
What the developer wrote
Authored by Matt Corallo
Fix races when removing per-peer state from KVStore in LSPS2/5
If we note that a peer should be removed in LSPS2/5 handling, we
need to make sure that the peer wasn't re-added between dropping
its state in memory and going to remove its state from disk. If it
is, we need to overwrite the current on-disk state instead.
What changed, and why it matters
This patch fixes a race condition in the Lightning Dev Kit's liquidity service plugins (LSPS2 and LSPS5). Previously, when the code decided a peer's state should be removed from disk, it could drop the in-memory lock before deleting the stored data. In that gap, another operation could re-add state for the same peer. The removal could then delete the newly-written state, or an outdated write could overwrite newer data, leading to lost or inconsistent peer records. The fix holds the appropriate locks until the storage write or remove operation is started, and re-checks whether the peer is still removable before deleting anything. If the peer was re-added, it now forces a fresh persist instead of removing.
Review and merge the patch. After deployment, monitor for any persisted-state inconsistencies in LSPS2/5 services and consider adding tests that interleave peer disconnect/reconnect with the periodic prune/persist task to prevent regression.
Security signals we found
Race condition between in-memory state pruning and on-disk state removal
TOCTOU window between dropping peer state and KVStore remove/write
Potential loss or inconsistency of persisted peer state
Lock scope tightened to cover the start of async storage operations
Re-validation of prunability under write lock before deletion
Fallback to re-persist when peer state is re-added concurrently
Evidence from the diff
The commit addresses TOCTOU-style races in lightning-liquidity/src/lsps2/service.rs and lightning-liquidity/src/lsps5/service.rs around per_peer_state persistence. In both files, the old logic used a write lock to prune the in-memory map and then, after releasing it, issued async kv_store.remove calls. Concurrent persist_peer_state calls only held a read lock while beginning an async kv_store.write, so a remove could interleave with a re-added peer’s write, causing either stale writes to survive or fresh writes to be deleted. The patch: (1) begins kv_store.write while still holding the read lock in LSPS2 and the write lock in LSPS5; (2) scans for prunable peers under a read lock and only takes the write lock when actually removing; (3) re-checks prunability under the write lock before removal; (4) if the peer is no longer prunable, sets needs_persist = true and persists instead of deleting. is_prunable is also changed to take &self in LSPS5 to support read-lock usage.
Changed components
lightning-liquidity/src/lsps2/service.rslightning-liquidity/src/lsps5/service.rsKVStore persistence layer for LSPS2/5 peer stateInspect captured patch +97 / −42
diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs
index 2d727ac..2ded321 100644
--- a/lightning-liquidity/src/lsps2/service.rs
+++ b/lightning-liquidity/src/lsps2/service.rs
@@ -1603,7 +1603,7 @@ where
) -> Result<(), lightning::io::Error> {
let fut = {
let outer_state_lock = self.per_peer_state.read().unwrap();
- let encoded = match outer_state_lock.get(&counterparty_node_id) {
+ match outer_state_lock.get(&counterparty_node_id) {
None => {
// We dropped the peer state by now.
return Ok(());
@@ -1615,18 +1615,19 @@ where
return Ok(());
} else {
peer_state_lock.needs_persist = false;
- peer_state_lock.encode()
+ let key = counterparty_node_id.to_string();
+ let encoded = peer_state_lock.encode();
+ // Begin the write with the entry lock held. This avoids racing with
+ // potentially-in-flight `persist` calls writing state for the same peer.
+ self.kv_store.write(
+ LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+ LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+ &key,
+ encoded,
+ )
}
},
- };
- let key = counterparty_node_id.to_string();
-
- self.kv_store.write(
- LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
- LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
- &key,
- encoded,
- )
+ }
};
fut.await.map_err(|e| {
@@ -1648,8 +1649,10 @@ where
let mut need_persist = Vec::new();
{
- let mut outer_state_lock = self.per_peer_state.write().unwrap();
- outer_state_lock.retain(|counterparty_node_id, inner_state_lock| {
+ // First build a list of peers to persist and prune with the read lock. This allows
+ // us to avoid the write lock unless we actually need to remove a node.
+ let outer_state_lock = self.per_peer_state.read().unwrap();
+ for (counterparty_node_id, inner_state_lock) in outer_state_lock.iter() {
let mut peer_state_lock = inner_state_lock.lock().unwrap();
peer_state_lock.prune_expired_request_state();
let is_prunable = peer_state_lock.is_prunable();
@@ -1658,8 +1661,7 @@ where
} else if peer_state_lock.needs_persist {
need_persist.push(*counterparty_node_id);
}
- !is_prunable
- });
+ }
}
for counterparty_node_id in need_persist.into_iter() {
@@ -1668,14 +1670,39 @@ where
}
for counterparty_node_id in need_remove {
- let key = counterparty_node_id.to_string();
- self.kv_store
- .remove(
- LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
- LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
- &key,
- )
- .await?;
+ let mut future_opt = None;
+ {
+ // We need to take the `per_peer_state` write lock to remove an entry, but also
+ // have to hold it until after the `remove` call returns (but not through
+ // future completion) to ensure that writes for the peer's state are
+ // well-ordered with other `persist_peer_state` calls even across the removal
+ // itself.
+ let mut per_peer_state = self.per_peer_state.write().unwrap();
+ if let Entry::Occupied(mut entry) = per_peer_state.entry(counterparty_node_id) {
+ let state = entry.get_mut().get_mut().unwrap();
+ if state.is_prunable() {
+ entry.remove();
+ let key = counterparty_node_id.to_string();
+ future_opt = Some(self.kv_store.remove(
+ LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+ LSPS2_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+ &key,
+ ));
+ } else {
+ // If the peer got new state, force a re-persist of the current state.
+ state.needs_persist = true;
+ }
+ } else {
+ // This should never happen, we can only have one `persist` call
+ // in-progress at once and map entries are only removed by it.
+ debug_assert!(false);
+ }
+ }
+ if let Some(future) = future_opt {
+ future.await?;
+ } else {
+ self.persist_peer_state(counterparty_node_id).await?;
+ }
}
Ok(())
diff --git a/lightning-liquidity/src/lsps5/service.rs b/lightning-liquidity/src/lsps5/service.rs
index 4439130..da2a90c 100644
--- a/lightning-liquidity/src/lsps5/service.rs
+++ b/lightning-liquidity/src/lsps5/service.rs
@@ -20,6 +20,7 @@ use crate::message_queue::MessageQueue;
use crate::persist::{
LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
};
+use crate::prelude::hash_map::Entry;
use crate::prelude::*;
use crate::sync::{Arc, Mutex, RwLock, RwLockWriteGuard};
use crate::utils::time::TimeProvider;
@@ -220,6 +221,8 @@ where
let key = counterparty_node_id.to_string();
+ // Begin the write with the `per_peer_state` write lock held to avoid racing with
+ // potentially-in-flight `persist` calls writing state for the same peer.
self.kv_store.write(
LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
@@ -244,11 +247,12 @@ where
// time.
let mut need_remove = Vec::new();
let mut need_persist = Vec::new();
+
+ self.check_prune_stale_webhooks(&mut self.per_peer_state.write().unwrap());
{
- let mut outer_state_lock = self.per_peer_state.write().unwrap();
- self.check_prune_stale_webhooks(&mut outer_state_lock);
+ let outer_state_lock = self.per_peer_state.read().unwrap();
- outer_state_lock.retain(|client_id, peer_state| {
+ for (client_id, peer_state) in outer_state_lock.iter() {
let is_prunable = peer_state.is_prunable();
let has_open_channel = self.client_has_open_channel(client_id);
if is_prunable && !has_open_channel {
@@ -256,24 +260,48 @@ where
} else if peer_state.needs_persist {
need_persist.push(*client_id);
}
- !is_prunable || has_open_channel
- });
- };
+ }
+ }
- for counterparty_node_id in need_persist.into_iter() {
- debug_assert!(!need_remove.contains(&counterparty_node_id));
- self.persist_peer_state(counterparty_node_id).await?;
+ for client_id in need_persist.into_iter() {
+ debug_assert!(!need_remove.contains(&client_id));
+ self.persist_peer_state(client_id).await?;
}
- for counterparty_node_id in need_remove {
- let key = counterparty_node_id.to_string();
- self.kv_store
- .remove(
- LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
- LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
- &key,
- )
- .await?;
+ for client_id in need_remove {
+ let mut future_opt = None;
+ {
+ // We need to take the `per_peer_state` write lock to remove an entry, but also
+ // have to hold it until after the `remove` call returns (but not through
+ // future completion) to ensure that writes for the peer's state are
+ // well-ordered with other `persist_peer_state` calls even across the removal
+ // itself.
+ let mut per_peer_state = self.per_peer_state.write().unwrap();
+ if let Entry::Occupied(mut entry) = per_peer_state.entry(client_id) {
+ let state = entry.get_mut();
+ if state.is_prunable() && !self.client_has_open_channel(&client_id) {
+ entry.remove();
+ let key = client_id.to_string();
+ future_opt = Some(self.kv_store.remove(
+ LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+ LSPS5_SERVICE_PERSISTENCE_SECONDARY_NAMESPACE,
+ &key,
+ ));
+ } else {
+ // If the peer was re-added, force a re-persist of the current state.
+ state.needs_persist = true;
+ }
+ } else {
+ // This should never happen, we can only have one `persist` call
+ // in-progress at once and map entries are only removed by it.
+ debug_assert!(false);
+ }
+ }
+ if let Some(future) = future_opt {
+ future.await?;
+ } else {
+ self.persist_peer_state(client_id).await?;
+ }
}
Ok(())
@@ -761,7 +789,7 @@ impl PeerState {
});
}
- fn is_prunable(&mut self) -> bool {
+ fn is_prunable(&self) -> bool {
self.webhooks.is_empty()
}
}
Why this scored 61/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.