← Watch feed
Remove pruned LSPS2/LSPS5 peer state entries from the `KVStore`
Public commit record
73/100 · Adequate
What the developer wrote
Authored by Elias Rohrer
Remove pruned LSPS2/LSPS5 peer state entries from the `KVStore`
Previously, we'd persist peer states to the `KVStore`, but, while we
pruned them eventually from our in-memory state, we wouldn't remove it
from the `KVStore`.
Here, we change this and regularly prune and delete peer state entries
from the `KVStore`.
Note we still prune the state-internal data on peer disconnection, but
leave removal to our (BP-driven) async `persist` calls.
✓ Specific, descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context
What changed, and why it matters
This change fixes a cleanup bug in a Lightning liquidity service. Previously, when peer state was pruned from memory, the corresponding saved data was left behind in the persistent key-value store. Over time this could accumulate stale entries and potentially allow old or expired peer data to be reloaded after a restart. The patch now deletes those stale store entries during regular persistence cycles.
Recommended action
Review whether previously accumulated stale KVStore entries should be cleaned up on upgrade, and verify that the async `remove` failures are handled safely without corrupting the in-memory pruning decision.
Security signals we found
01
Stale persisted peer state not deleted from KVStore
02
Potential resurrection of expired/pruned peer state after restart
03
Accumulation of orphaned persistent entries
04
In-memory pruning and persistent-store pruning were inconsistent
Evidence from the diff
The commit removes pruned LSPS2/LSPS5 peer state entries from the KVStore. Previously peer_disconnected and prune_peer_state only dropped entries from the in-memory per_peer_state map, leaving a TODO about KVStore removal. The patch refactors the persist path to: (1) prune expired request/webhook state, (2) identify prunable peers, (3) persist peers that need it, and (4) call kv_store.remove(...) for prunable peers. It also removes the now-redundant prune_peer_state method and the best-block hook that invoked it.
Changed components
lightning-liquidity/src/lsps2/service.rslightning-liquidity/src/lsps5/service.rslightning-liquidity/src/manager.rsKVStore persistence layer for LSPS2/LSPS5 peer stateInspect captured patch +79 / −47
diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs
index 8bf8517..ba3f54e 100644
--- a/lightning-liquidity/src/lsps2/service.rs
+++ b/lightning-liquidity/src/lsps2/service.rs
@@ -514,7 +514,7 @@ impl PeerState {
// We abort the flow, and prune any data kept.
self.intercept_scid_by_channel_id.retain(|_, iscid| intercept_scid != iscid);
self.intercept_scid_by_user_channel_id.retain(|_, iscid| intercept_scid != iscid);
- // TODO: Remove peer state entry from the KVStore
+ self.needs_persist |= true;
return false;
}
true
@@ -1645,44 +1645,53 @@ where
// TODO: We should eventually persist in parallel, however, when we do, we probably want to
// introduce some batching to upper-bound the number of requests inflight at any given
// time.
- let need_persist: Vec<PublicKey> = {
- let outer_state_lock = self.per_peer_state.read().unwrap();
- outer_state_lock
- .iter()
- .filter_map(|(k, v)| if v.lock().unwrap().needs_persist { Some(*k) } else { None })
- .collect()
- };
+
+ let mut need_remove = Vec::new();
+ 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| {
+ 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();
+ if is_prunable {
+ need_remove.push(*counterparty_node_id);
+ } else if peer_state_lock.needs_persist {
+ need_persist.push(*counterparty_node_id);
+ }
+ !is_prunable
+ });
+ }
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 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,
+ true,
+ )
+ .await?;
+ }
+
Ok(())
}
pub(crate) fn peer_disconnected(&self, counterparty_node_id: PublicKey) {
- let mut outer_state_lock = self.per_peer_state.write().unwrap();
- let is_prunable =
- if let Some(inner_state_lock) = outer_state_lock.get(&counterparty_node_id) {
- let mut peer_state_lock = inner_state_lock.lock().unwrap();
- peer_state_lock.prune_expired_request_state();
- peer_state_lock.is_prunable()
- } else {
- return;
- };
- if is_prunable {
- outer_state_lock.remove(&counterparty_node_id);
- }
- }
-
- #[allow(clippy::bool_comparison)]
- pub(crate) fn prune_peer_state(&self) {
- let mut outer_state_lock = self.per_peer_state.write().unwrap();
- outer_state_lock.retain(|_, inner_state_lock| {
+ let outer_state_lock = self.per_peer_state.write().unwrap();
+ if let Some(inner_state_lock) = outer_state_lock.get(&counterparty_node_id) {
let mut peer_state_lock = inner_state_lock.lock().unwrap();
+ // We clean up the peer state, but leave removing the peer entry to the prune logic in
+ // `persist` which removes it from the store.
peer_state_lock.prune_expired_request_state();
- peer_state_lock.is_prunable() == false
- });
+ }
}
}
diff --git a/lightning-liquidity/src/lsps5/service.rs b/lightning-liquidity/src/lsps5/service.rs
index 5753dc8..2eb5e5f 100644
--- a/lightning-liquidity/src/lsps5/service.rs
+++ b/lightning-liquidity/src/lsps5/service.rs
@@ -242,18 +242,41 @@ where
// TODO: We should eventually persist in parallel, however, when we do, we probably want to
// introduce some batching to upper-bound the number of requests inflight at any given
// time.
- let need_persist: Vec<PublicKey> = {
- let outer_state_lock = self.per_peer_state.read().unwrap();
- outer_state_lock
- .iter()
- .filter_map(|(k, v)| if v.needs_persist { Some(*k) } else { None })
- .collect()
+ let mut need_remove = Vec::new();
+ let mut need_persist = Vec::new();
+ {
+ let mut outer_state_lock = self.per_peer_state.write().unwrap();
+ self.check_prune_stale_webhooks(&mut outer_state_lock);
+
+ outer_state_lock.retain(|client_id, peer_state| {
+ let is_prunable = peer_state.is_prunable();
+ let has_open_channel = self.client_has_open_channel(client_id);
+ if is_prunable && !has_open_channel {
+ need_remove.push(*client_id);
+ } 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 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,
+ true,
+ )
+ .await?;
+ }
+
Ok(())
}
@@ -269,14 +292,11 @@ where
});
if should_prune {
- outer_state_lock.retain(|client_id, peer_state| {
- if self.client_has_open_channel(client_id) {
- // Don't prune clients with open channels
- return true;
- }
- // TODO: Remove peer state entry from the KVStore
- !peer_state.prune_stale_webhooks(now)
- });
+ for (_, peer_state) in outer_state_lock.iter_mut() {
+ // Prune stale webhooks, but leave removal of the peers states to the prune logic
+ // in `persist` which will remove it from the store.
+ peer_state.prune_stale_webhooks(now)
+ }
*last_pruning = Some(now);
}
}
@@ -732,11 +752,17 @@ impl PeerState {
}
// Returns whether the entire state is empty and can be pruned.
- fn prune_stale_webhooks(&mut self, now: LSPSDateTime) -> bool {
+ fn prune_stale_webhooks(&mut self, now: LSPSDateTime) {
self.webhooks.retain(|(_, webhook)| {
- now.duration_since(&webhook.last_used) < MIN_WEBHOOK_RETENTION_DAYS
+ let should_prune = now.duration_since(&webhook.last_used) >= MIN_WEBHOOK_RETENTION_DAYS;
+ if should_prune {
+ self.needs_persist |= true;
+ }
+ !should_prune
});
+ }
+ fn is_prunable(&mut self) -> bool {
self.webhooks.is_empty()
}
}
diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs
index b927d41..0ea5436 100644
--- a/lightning-liquidity/src/manager.rs
+++ b/lightning-liquidity/src/manager.rs
@@ -1015,9 +1015,6 @@ where
*self.best_block.write().unwrap() = Some(new_best_block);
// TODO: Call best_block_updated on all sub-modules that require it, e.g., LSPS1MessageHandler.
- if let Some(lsps2_service_handler) = self.lsps2_service_handler.as_ref() {
- lsps2_service_handler.prune_peer_state();
- }
}
fn get_relevant_txids(&self) -> Vec<(bitcoin::Txid, u32, Option<bitcoin::BlockHash>)> {
Risk score
Our methodology →Why this scored 35/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.