Read persisted event queue state in LiquidityManager::new

We read any previously-persisted state upon construction of
`LiquidityManager`.

Author: tnull

lightning-liquidity/src/events/event_queue.rs

@@ -30,8 +30,8 @@ pub(crate) struct EventQueue {
 }
 
 impl EventQueue {
-	pub fn new(kv_store: Arc<dyn KVStore + Send + Sync>) -> Self {
-		let queue = Arc::new(Mutex::new(VecDeque::new()));
+	pub fn new(queue: VecDeque<LiquidityEvent>, kv_store: Arc<dyn KVStore + Send + Sync>) -> Self {
+		let queue = Arc::new(Mutex::new(queue));
 		let waker = Arc::new(Mutex::new(None));
 		Self {
 			queue,
@@ -193,7 +193,7 @@ mod tests {
 		use std::time::Duration;
 
 		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
-		let event_queue = Arc::new(EventQueue::new(kv_store));
+		let event_queue = Arc::new(EventQueue::new(VecDeque::new(), kv_store));
 		assert_eq!(event_queue.next_event(), None);
 
 		let secp_ctx = Secp256k1::new();

lightning-liquidity/src/events/mod.rs

@@ -17,8 +17,8 @@
 
 mod event_queue;
 
-pub(crate) use event_queue::EventQueue;
 pub use event_queue::MAX_EVENT_QUEUE_SIZE;
+pub(crate) use event_queue::{EventQueue, EventQueueDeserWrapper};
 
 use crate::lsps0;
 use crate::lsps1;

lightning-liquidity/src/lsps0/client.rs

@@ -113,6 +113,7 @@ where
 
 #[cfg(test)]
 mod tests {
+	use alloc::collections::VecDeque;
 	use alloc::string::ToString;
 	use alloc::sync::Arc;
 
@@ -129,7 +130,7 @@ mod tests {
 		let pending_messages = Arc::new(MessageQueue::new());
 		let entropy_source = Arc::new(TestEntropy {});
 		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
-		let event_queue = Arc::new(EventQueue::new(kv_store));
+		let event_queue = Arc::new(EventQueue::new(VecDeque::new(), kv_store));
 
 		let lsps0_handler = Arc::new(LSPS0ClientHandler::new(
 			entropy_source,

lightning-liquidity/src/lsps5/client.rs

@@ -471,7 +471,7 @@ mod tests {
 		let message_queue = Arc::new(MessageQueue::new());
 
 		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
-		let event_queue = Arc::new(EventQueue::new(kv_store));
+		let event_queue = Arc::new(EventQueue::new(VecDeque::new(), kv_store));
 		let client = LSPS5ClientHandler::new(
 			test_entropy_source,
 			Arc::clone(&message_queue),

lightning-liquidity/src/manager.rs

@@ -24,7 +24,9 @@ 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, read_lsps5_service_peer_states};
+use crate::persist::{
+	read_event_queue, read_lsps2_service_peer_states, read_lsps5_service_peer_states,
+};
 
 use crate::lsps1::client::{LSPS1ClientConfig, LSPS1ClientHandler};
 use crate::lsps1::msgs::LSPS1Message;
@@ -337,7 +339,9 @@ where
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
 	) -> Result<Self, lightning::io::Error> {
 		let pending_messages = Arc::new(MessageQueue::new());
-		let pending_events = Arc::new(EventQueue::new(Arc::clone(&kv_store)));
+
+		let persisted_queue = read_event_queue(Arc::clone(&kv_store)).await?.unwrap_or_default();
+		let pending_events = Arc::new(EventQueue::new(persisted_queue, Arc::clone(&kv_store)));
 		let ignored_peers = RwLock::new(new_hash_set());
 
 		let mut supported_protocols = Vec::new();

lightning-liquidity/src/persist.rs

@@ -9,6 +9,7 @@
 
 //! Types and utils for persistence.
 
+use crate::events::{EventQueueDeserWrapper, LiquidityEvent};
 use crate::lsps2::service::PeerState as LSPS2ServicePeerState;
 use crate::lsps5::service::PeerState as LSPS5ServicePeerState;
 
@@ -19,6 +20,7 @@ use lightning::util::ser::Readable;
 use bitcoin::secp256k1::PublicKey;
 
 use alloc::vec::Vec;
+use alloc::collections::VecDeque;
 
 use core::ops::Deref;
 use core::str::FromStr;
@@ -48,6 +50,40 @@ 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_event_queue<K: Deref>(
+	kv_store: K,
+) -> Result<Option<VecDeque<LiquidityEvent>>, lightning::io::Error>
+where
+	K::Target: KVStore,
+{
+	let read_fut = kv_store.read(
+		LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+		LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+		LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_KEY,
+	);
+
+	let mut reader = match read_fut.await {
+		Ok(r) => Cursor::new(r),
+		Err(e) => {
+			if e.kind() == lightning::io::ErrorKind::NotFound {
+				// Key wasn't found, no error but first time running.
+				return Ok(None);
+			} else {
+				return Err(e);
+			}
+		},
+	};
+
+	let queue: EventQueueDeserWrapper = Readable::read(&mut reader).map_err(|_| {
+		lightning::io::Error::new(
+			lightning::io::ErrorKind::InvalidData,
+			"Failed to deserialize liquidity event queue",
+		)
+	})?;
+
+	Ok(Some(queue.0))
+}
+
 pub(crate) async fn read_lsps2_service_peer_states<K: Deref>(
 	kv_store: K,
 ) -> Result<Vec<(PublicKey, LSPS2ServicePeerState)>, lightning::io::Error>