Add EventQueue persistence

We add simple `persist` call to `EventQueue` that persists it under a
`event_queue` key.

Author: tnull

lightning-liquidity/src/events/event_queue.rs

@@ -1,12 +1,23 @@
 use super::LiquidityEvent;
+
+use crate::persist::{
+	LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_KEY,
+	LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+	LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+};
 use crate::sync::{Arc, Mutex};
 
+use alloc::boxed::Box;
 use alloc::collections::VecDeque;
 use alloc::vec::Vec;
 
 use core::future::Future;
+use core::pin::Pin;
 use core::task::{Poll, Waker};
 
+use lightning::util::persist::KVStore;
+use lightning::util::ser::{CollectionLength, MaybeReadable, Readable, Writeable, Writer};
+
 /// The maximum queue size we allow before starting to drop events.
 pub const MAX_EVENT_QUEUE_SIZE: usize = 1000;
 
@@ -15,17 +26,19 @@ pub(crate) struct EventQueue {
 	waker: Arc<Mutex<Option<Waker>>>,
 	#[cfg(feature = "std")]
 	condvar: Arc<crate::sync::Condvar>,
+	kv_store: Arc<dyn KVStore + Send + Sync>,
 }
 
 impl EventQueue {
-	pub fn new() -> Self {
+	pub fn new(kv_store: Arc<dyn KVStore + Send + Sync>) -> Self {
 		let queue = Arc::new(Mutex::new(VecDeque::new()));
 		let waker = Arc::new(Mutex::new(None));
 		Self {
 			queue,
 			waker,
 			#[cfg(feature = "std")]
 			condvar: Arc::new(crate::sync::Condvar::new()),
+			kv_store,
 		}
 	}
 
@@ -70,6 +83,20 @@ impl EventQueue {
 	pub fn notifier(&self) -> EventQueueNotifierGuard<'_> {
 		EventQueueNotifierGuard(self)
 	}
+
+	pub fn persist(
+		&self,
+	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + Send>> {
+		let queue = self.queue.lock().unwrap();
+		let encoded = EventQueueSerWrapper(&queue).encode();
+
+		self.kv_store.write(
+			LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
+			LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+			LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_KEY,
+			encoded,
+		)
+	}
 }
 
 // A guard type that will notify about new events when dropped.
@@ -122,6 +149,35 @@ impl Future for EventFuture {
 	}
 }
 
+pub(crate) struct EventQueueDeserWrapper(pub VecDeque<LiquidityEvent>);
+
+impl Readable for EventQueueDeserWrapper {
+	fn read<R: lightning::io::Read>(
+		reader: &mut R,
+	) -> Result<Self, lightning::ln::msgs::DecodeError> {
+		let len: CollectionLength = Readable::read(reader)?;
+		let mut queue = VecDeque::with_capacity(len.0 as usize);
+		for _ in 0..len.0 {
+			if let Some(event) = MaybeReadable::read(reader)? {
+				queue.push_back(event);
+			}
+		}
+		Ok(Self(queue))
+	}
+}
+
+struct EventQueueSerWrapper<'a>(&'a VecDeque<LiquidityEvent>);
+
+impl Writeable for EventQueueSerWrapper<'_> {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), lightning::io::Error> {
+		CollectionLength(self.0.len() as u64).write(writer)?;
+		for e in self.0.iter() {
+			e.write(writer)?;
+		}
+		Ok(())
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	#[tokio::test]
@@ -131,10 +187,13 @@ mod tests {
 		use crate::lsps0::event::LSPS0ClientEvent;
 		use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
 		use core::sync::atomic::{AtomicU16, Ordering};
+		use lightning::util::persist::KVStoreSyncWrapper;
+		use lightning::util::test_utils::TestStore;
 		use std::sync::Arc;
 		use std::time::Duration;
 
-		let event_queue = Arc::new(EventQueue::new());
+		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
+		let event_queue = Arc::new(EventQueue::new(kv_store));
 		assert_eq!(event_queue.next_event(), None);
 
 		let secp_ctx = Secp256k1::new();

lightning-liquidity/src/lsps0/client.rs

@@ -113,10 +113,12 @@ where
 
 #[cfg(test)]
 mod tests {
-
 	use alloc::string::ToString;
 	use alloc::sync::Arc;
 
+	use lightning::util::persist::KVStoreSyncWrapper;
+	use lightning::util::test_utils::TestStore;
+
 	use crate::lsps0::ser::{LSPSMessage, LSPSRequestId};
 	use crate::tests::utils::{self, TestEntropy};
 
@@ -126,7 +128,8 @@ mod tests {
 	fn test_list_protocols() {
 		let pending_messages = Arc::new(MessageQueue::new());
 		let entropy_source = Arc::new(TestEntropy {});
-		let event_queue = Arc::new(EventQueue::new());
+		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
+		let event_queue = Arc::new(EventQueue::new(kv_store));
 
 		let lsps0_handler = Arc::new(LSPS0ClientHandler::new(
 			entropy_source,

lightning-liquidity/src/lsps5/client.rs

@@ -444,6 +444,8 @@ mod tests {
 	use crate::{lsps0::ser::LSPSRequestId, lsps5::msgs::SetWebhookResponse};
 	use bitcoin::{key::Secp256k1, secp256k1::SecretKey};
 	use core::sync::atomic::{AtomicU64, Ordering};
+	use lightning::util::persist::KVStoreSyncWrapper;
+	use lightning::util::test_utils::TestStore;
 
 	struct UniqueTestEntropy {
 		counter: AtomicU64,
@@ -467,7 +469,9 @@ mod tests {
 	) {
 		let test_entropy_source = Arc::new(UniqueTestEntropy { counter: AtomicU64::new(2) });
 		let message_queue = Arc::new(MessageQueue::new());
-		let event_queue = Arc::new(EventQueue::new());
+
+		let kv_store = Arc::new(KVStoreSyncWrapper(Arc::new(TestStore::new(false))));
+		let event_queue = Arc::new(EventQueue::new(kv_store));
 		let client = LSPS5ClientHandler::new(
 			test_entropy_source,
 			Arc::clone(&message_queue),

lightning-liquidity/src/manager.rs

@@ -329,7 +329,7 @@ where
 		client_config: Option<LiquidityClientConfig>, time_provider: TP,
 	) -> Self {
 		let pending_messages = Arc::new(MessageQueue::new());
-		let pending_events = Arc::new(EventQueue::new());
+		let pending_events = Arc::new(EventQueue::new(Arc::clone(&kv_store)));
 		let ignored_peers = RwLock::new(new_hash_set());
 
 		let mut supported_protocols = Vec::new();
@@ -574,6 +574,9 @@ where
 		&self,
 	) -> Pin<Box<dyn StdFuture<Output = Result<(), lightning::io::Error>> + Send>> {
 		let mut futures = Vec::new();
+
+		futures.push(self.pending_events.persist());
+
 		if let Some(lsps2_service_handler) = self.lsps2_service_handler.as_ref() {
 			futures.push(lsps2_service_handler.persist());
 		}

lightning-liquidity/src/persist.rs

@@ -14,6 +14,16 @@
 /// [`LiquidityManager`]: crate::LiquidityManager
 pub const LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "liquidity";
 
+/// The secondary namespace under which the [`LiquidityManager`] event queue will be persisted.
+///
+/// [`LiquidityManager`]: crate::LiquidityManager
+pub const LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+
+/// The key under which the [`LiquidityManager`] event queue will be persisted.
+///
+/// [`LiquidityManager`]: crate::LiquidityManager
+pub const LIQUIDITY_MANAGER_EVENT_QUEUE_PERSISTENCE_KEY: &str = "event_queue";
+
 /// The secondary namespace under which the [`LSPS2ServiceHandler`] data will be persisted.
 ///
 /// [`LSPS2ServiceHandler`]: crate::lsps2::service::LSPS2ServiceHandler

lightning/src/util/ser.rs

@@ -553,7 +553,7 @@ impl Readable for BigSize {
 /// To ensure we only have one valid encoding per value, we add 0xffff to values written as eight
 /// bytes. Thus, 0xfffe is serialized as 0xfffe, whereas 0xffff is serialized as
 /// 0xffff0000000000000000 (i.e. read-eight-bytes then zero).
-struct CollectionLength(pub u64);
+pub struct CollectionLength(pub u64);
 impl Writeable for CollectionLength {
 	#[inline]
 	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), io::Error> {