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.

Author: tnull

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
-		});
+		}
 	}
 }
 

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()
 	}
 }

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>)> {