lightning-liquidity persistence: Add serialization logic for services and event queue

fuzz/src/lsps_message.rs

@@ -21,7 +21,7 @@ use lightning::util::test_utils::{
 };
 
 use lightning_liquidity::lsps0::ser::LSPS_MESSAGE_TYPE_ID;
-use lightning_liquidity::LiquidityManager;
+use lightning_liquidity::LiquidityManagerSync;
 
 use core::time::Duration;
 
@@ -77,15 +77,16 @@ pub fn do_test(data: &[u8]) {
 		genesis_block.header.time,
 	));
 
-	let liquidity_manager = Arc::new(LiquidityManager::new(
+	let liquidity_manager = Arc::new(LiquidityManagerSync::new(
 		Arc::clone(&keys_manager),
 		Arc::clone(&keys_manager),
 		Arc::clone(&manager),
 		None::<Arc<dyn Filter + Send + Sync>>,
 		None,
+		kv_store,
 		None,
 		None,
-	));
+	).unwrap());
 	let mut reader = data;
 	if let Ok(Some(msg)) = liquidity_manager.read(LSPS_MESSAGE_TYPE_ID, &mut reader) {
 		let secp = Secp256k1::signing_only();

lightning-background-processor/Cargo.toml

@@ -31,6 +31,7 @@ possiblyrandom = { version = "0.2", path = "../possiblyrandom", default-features
 tokio = { version = "1.35", features = [ "macros", "rt", "rt-multi-thread", "sync", "time" ] }
 lightning = { version = "0.2.0", path = "../lightning", features = ["_test_utils"] }
 lightning-invoice = { version = "0.34.0", path = "../lightning-invoice" }
+lightning-liquidity = { version = "0.2.0", path = "../lightning-liquidity", default-features = false, features = ["_test_utils"] }
 lightning-persister = { version = "0.2.0", path = "../lightning-persister" }
 
 [lints]

lightning-background-processor/src/lib.rs

@@ -69,6 +69,8 @@ use lightning::util::wakers::Sleeper;
 use lightning_rapid_gossip_sync::RapidGossipSync;
 
 use lightning_liquidity::ALiquidityManager;
+#[cfg(feature = "std")]
+use lightning_liquidity::ALiquidityManagerSync;
 
 use core::ops::Deref;
 use core::time::Duration;
@@ -1350,7 +1352,7 @@ impl BackgroundProcessor {
 		CM::Target: AChannelManager,
 		OM::Target: AOnionMessenger,
 		PM::Target: APeerManager,
-		LM::Target: ALiquidityManager,
+		LM::Target: ALiquidityManagerSync,
 		D::Target: ChangeDestinationSourceSync,
 		O::Target: 'static + OutputSpender,
 		K::Target: 'static + KVStoreSync,
@@ -1693,7 +1695,7 @@ mod tests {
 	use lightning::util::test_utils;
 	use lightning::{get_event, get_event_msg};
 	use lightning_liquidity::utils::time::DefaultTimeProvider;
-	use lightning_liquidity::LiquidityManager;
+	use lightning_liquidity::{ALiquidityManagerSync, LiquidityManagerSync};
 	use lightning_persister::fs_store::FilesystemStore;
 	use lightning_rapid_gossip_sync::RapidGossipSync;
 	use std::collections::VecDeque;
@@ -1790,7 +1792,7 @@ mod tests {
 		IgnoringMessageHandler,
 	>;
 
-	type LM = LiquidityManager<
+	type LM = LiquidityManagerSync<
 		Arc<KeysManager>,
 		Arc<KeysManager>,
 		Arc<ChannelManager>,
@@ -2242,15 +2244,19 @@ mod tests {
 				Arc::clone(&logger),
 				Arc::clone(&keys_manager),
 			));
-			let liquidity_manager = Arc::new(LiquidityManager::new(
-				Arc::clone(&keys_manager),
-				Arc::clone(&keys_manager),
-				Arc::clone(&manager),
-				None,
-				None,
-				None,
-				None,
-			));
+			let liquidity_manager = Arc::new(
+				LiquidityManagerSync::new(
+					Arc::clone(&keys_manager),
+					Arc::clone(&keys_manager),
+					Arc::clone(&manager),
+					None,
+					None,
+					Arc::clone(&kv_store) as Arc<dyn KVStoreSync + Sync + Send>,
+					None,
+					None,
+				)
+				.unwrap(),
+			);
 			let node = Node {
 				node: manager,
 				p2p_gossip_sync,
@@ -2627,7 +2633,7 @@ mod tests {
 			Some(Arc::clone(&nodes[0].messenger)),
 			nodes[0].rapid_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			Some(Arc::clone(&nodes[0].liquidity_manager)),
+			Some(nodes[0].liquidity_manager.get_lm_async()),
 			Some(nodes[0].sweeper.sweeper_async()),
 			Arc::clone(&nodes[0].logger),
 			Some(Arc::clone(&nodes[0].scorer)),
@@ -3136,7 +3142,7 @@ mod tests {
 			Some(Arc::clone(&nodes[0].messenger)),
 			nodes[0].rapid_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			Some(Arc::clone(&nodes[0].liquidity_manager)),
+			Some(nodes[0].liquidity_manager.get_lm_async()),
 			Some(nodes[0].sweeper.sweeper_async()),
 			Arc::clone(&nodes[0].logger),
 			Some(Arc::clone(&nodes[0].scorer)),
@@ -3351,7 +3357,7 @@ mod tests {
 			Some(Arc::clone(&nodes[0].messenger)),
 			nodes[0].no_gossip_sync(),
 			Arc::clone(&nodes[0].peer_manager),
-			Some(Arc::clone(&nodes[0].liquidity_manager)),
+			Some(nodes[0].liquidity_manager.get_lm_async()),
 			Some(nodes[0].sweeper.sweeper_async()),
 			Arc::clone(&nodes[0].logger),
 			Some(Arc::clone(&nodes[0].scorer)),

lightning-liquidity/Cargo.toml

@@ -18,11 +18,13 @@ default = ["std", "time"]
 std = ["lightning/std"]
 time = ["std"]
 backtrace = ["dep:backtrace"]
+_test_utils = []
 
 [dependencies]
 lightning = { version = "0.2.0", path = "../lightning", default-features = false }
 lightning-types = { version = "0.3.0", path = "../lightning-types", default-features = false }
 lightning-invoice = { version = "0.34.0", path = "../lightning-invoice", default-features = false, features = ["serde"] }
+lightning-macros = { version = "0.2", path = "../lightning-macros" }
 
 bitcoin = { version = "0.32.2", default-features = false, features = ["serde"] }
 

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 {
-		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,
 			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(VecDeque::new(), kv_store));
 		assert_eq!(event_queue.next_event(), None);
 
 		let secp_ctx = Secp256k1::new();

lightning-liquidity/src/events/mod.rs

@@ -17,14 +17,20 @@
 
 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;
 use crate::lsps2;
 use crate::lsps5;
 
+use lightning::io;
+use lightning::ln::msgs::DecodeError;
+use lightning::util::ser::{
+	BigSize, FixedLengthReader, MaybeReadable, Readable, Writeable, Writer,
+};
+
 /// An event which you should probably take some action in response to.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum LiquidityEvent {
@@ -87,3 +93,53 @@ impl From<lsps5::event::LSPS5ServiceEvent> for LiquidityEvent {
 		Self::LSPS5Service(event)
 	}
 }
+
+impl Writeable for LiquidityEvent {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), io::Error> {
+		match self {
+			Self::LSPS0Client(_) => {},
+			Self::LSPS1Client(_) => {},
+			#[cfg(lsps1_service)]
+			Self::LSPS1Service(_) => {},
+			Self::LSPS2Client(_) => {},
+			Self::LSPS2Service(event) => {
+				0u8.write(writer)?;
+				event.write(writer)?;
+			},
+			Self::LSPS5Client(_) => {},
+			Self::LSPS5Service(event) => {
+				2u8.write(writer)?;
+				event.write(writer)?;
+			},
+		}
+		Ok(())
+	}
+}
+
+impl MaybeReadable for LiquidityEvent {
+	fn read<R: io::Read>(reader: &mut R) -> Result<Option<Self>, DecodeError> {
+		match Readable::read(reader)? {
+			0u8 => {
+				let event = Readable::read(reader)?;
+				Ok(Some(LiquidityEvent::LSPS2Service(event)))
+			},
+			2u8 => {
+				let event = Readable::read(reader)?;
+				Ok(Some(LiquidityEvent::LSPS5Service(event)))
+			},
+			x if x % 2 == 1 => {
+				// If the event is of unknown type, assume it was written with `write_tlv_fields`,
+				// which prefixes the whole thing with a length BigSize. Because the event is
+				// odd-type unknown, we should treat it as `Ok(None)` even if it has some TLV
+				// fields that are even. Thus, we avoid using `read_tlv_fields` and simply read
+				// exactly the number of bytes specified, ignoring them entirely.
+				let tlv_len: BigSize = Readable::read(reader)?;
+				FixedLengthReader::new(reader, tlv_len.0)
+					.eat_remaining()
+					.map_err(|_| DecodeError::ShortRead)?;
+				Ok(None)
+			},
+			_ => Err(DecodeError::InvalidValue),
+		}
+	}
+}

lightning-liquidity/src/lib.rs

@@ -65,6 +65,7 @@ pub mod lsps2;
 pub mod lsps5;
 mod manager;
 pub mod message_queue;
+pub mod persist;
 #[allow(dead_code)]
 #[allow(unused_imports)]
 mod sync;
@@ -73,5 +74,6 @@ mod tests;
 pub mod utils;
 
 pub use manager::{
-	ALiquidityManager, LiquidityClientConfig, LiquidityManager, LiquidityServiceConfig,
+	ALiquidityManager, ALiquidityManagerSync, LiquidityClientConfig, LiquidityManager,
+	LiquidityManagerSync, LiquidityServiceConfig,
 };

lightning-liquidity/src/lsps0/client.rs

@@ -113,10 +113,13 @@ where
 
 #[cfg(test)]
 mod tests {
-
+	use alloc::collections::VecDeque;
 	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 +129,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(VecDeque::new(), kv_store));
 
 		let lsps0_handler = Arc::new(LSPS0ClientHandler::new(
 			entropy_source,