EventQueue: load the persisted queue at start

orange-sdk/src/event.rs

@@ -9,8 +9,10 @@ use ldk_node::lightning::events::{ClosureReason, PaymentFailureReason};
 use ldk_node::lightning::ln::types::ChannelId;
 use ldk_node::lightning::util::logger::Logger as _;
 use ldk_node::lightning::util::persist::KVStore;
-use ldk_node::lightning::util::ser::{Writeable, Writer};
-use ldk_node::lightning::{impl_writeable_tlv_based_enum, log_debug, log_error, log_warn};
+use ldk_node::lightning::util::ser::{Readable, Writeable, Writer};
+use ldk_node::lightning::{
+	impl_writeable_tlv_based_enum, log_debug, log_error, log_info, log_warn,
+};
 use ldk_node::lightning_types::payment::{PaymentHash, PaymentPreimage};
 use ldk_node::payment::{ConfirmationStatus, PaymentKind};
 use ldk_node::{CustomTlvRecord, UserChannelId};
@@ -218,14 +220,70 @@ pub struct EventQueue {
 }
 
 impl EventQueue {
-	pub(crate) fn new(
+	/// Builds the queue over `kv_store`, starting from the events a previous run left unhandled.
+	///
+	/// The queue is persisted under [`EVENT_QUEUE_PERSISTENCE_KEY`] after every push and every
+	/// [`EventQueue::event_handled`], so the stored copy is exactly the set of events the user has
+	/// not acknowledged yet. Loading it means an event emitted shortly before a shutdown, or one
+	/// the user never got to handle, is delivered on the next run instead of being dropped.
+	///
+	/// Delivery is at-least-once: an LDK event is persisted here before it is acknowledged to
+	/// ldk-node, so a crash between the two delivers it again after a restart, once from this
+	/// queue and once from ldk-node's own replay. Consumers must tolerate a repeat, as consumers
+	/// of ldk-node's own queue already must: it is reloaded at build time and an event stays at
+	/// its front until `event_handled`.
+	///
+	/// A missing key is a fresh wallet and starts empty. Any other read error, or a stored queue
+	/// that does not decode, fails the build: the wallet must not silently start without events
+	/// it owes the user.
+	pub(crate) async fn new(
 		kv_store: Arc<dyn DynStore>, tx_metadata: TxMetadataStore, logger: Arc<Logger>,
-	) -> Self {
-		let queue = Arc::new(Mutex::new(VecDeque::new()));
+	) -> Result<Self, ldk_node::lightning::io::Error> {
+		let persisted = match KVStore::read(
+			kv_store.as_ref(),
+			EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_KEY,
+		)
+		.await
+		{
+			Ok(bytes) => {
+				EventQueueDeserWrapper::read(&mut &bytes[..])
+					.map_err(|e| {
+						log_error!(logger.as_ref(), "Failed to deserialize event queue: {}", e);
+						ldk_node::lightning::io::Error::new(
+							ldk_node::lightning::io::ErrorKind::InvalidData,
+							"Failed to deserialize EventQueue",
+						)
+					})?
+					.0
+			},
+			Err(e) if e.kind() == ldk_node::lightning::io::ErrorKind::NotFound => VecDeque::new(),
+			Err(e) => {
+				log_error!(logger.as_ref(), "Failed to read event queue from store: {}", e);
+				return Err(e);
+			},
+		};
+		if !persisted.is_empty() {
+			log_info!(
+				logger.as_ref(),
+				"Loaded {} unhandled event(s) from the store",
+				persisted.len()
+			);
+		}
+		let queue = Arc::new(Mutex::new(persisted));
 		let pending_mpp_events = Arc::new(Mutex::new(HashMap::new()));
 		let waker = Arc::new(Mutex::new(None));
 		let rebalance_enabled = RebalanceEnabledCache::new(Arc::clone(&kv_store));
-		Self { queue, pending_mpp_events, waker, kv_store, rebalance_enabled, tx_metadata, logger }
+		Ok(Self {
+			queue,
+			pending_mpp_events,
+			waker,
+			kv_store,
+			rebalance_enabled,
+			tx_metadata,
+			logger,
+		})
 	}
 
 	pub(crate) async fn get_rebalance_enabled(&self) -> bool {
@@ -402,7 +460,11 @@ mod tests {
 	use crate::logging::LoggerType;
 	use crate::store::{PaymentId, PaymentType, TxMetadata, TxMetadataStore, TxType};
 	use ldk_node::io::sqlite_store::SqliteStore;
+	use ldk_node::lightning::io;
+	use ldk_node::lightning::util::persist::{PageToken, PaginatedListResponse};
+	use std::future::Future;
 	use std::path::PathBuf;
+	use std::pin::Pin;
 	use std::time::{Duration, UNIX_EPOCH};
 
 	fn temp_sqlite_store() -> (PathBuf, Arc<dyn DynStore>) {
@@ -440,7 +502,9 @@ mod tests {
 			store,
 			tx_metadata.clone(),
 			Arc::new(Logger::new(&LoggerType::LogFacade).expect("logger")),
-		);
+		)
+		.await
+		.expect("event queue");
 
 		let payment_hash = PaymentHash([3u8; 32]);
 		let surface_id = PaymentId::Trusted([7u8; 32]);
@@ -484,6 +548,113 @@ mod tests {
 			})
 		);
 	}
+
+	fn failed_payment(tag: u8) -> Event {
+		Event::PaymentFailed {
+			payment_id: PaymentId::SelfCustodial([tag; 32]),
+			payment_hash: Some(PaymentHash([tag; 32])),
+			reason: None,
+		}
+	}
+
+	/// The queue is persisted after every push and every acknowledgement; a queue built over the
+	/// same store must start from that copy, or every event still unhandled at a shutdown is lost.
+	#[tokio::test]
+	async fn unhandled_events_survive_a_restart() {
+		let (_path, store) = temp_sqlite_store();
+		let logger = Arc::new(Logger::new(&LoggerType::LogFacade).expect("logger"));
+		let tx_metadata = TxMetadataStore::new(Arc::clone(&store)).await;
+
+		let queue = EventQueue::new(Arc::clone(&store), tx_metadata.clone(), Arc::clone(&logger))
+			.await
+			.expect("event queue");
+		queue.add_event(failed_payment(1)).await.expect("push");
+		queue.add_event(failed_payment(2)).await.expect("push");
+		drop(queue);
+
+		let queue = EventQueue::new(Arc::clone(&store), tx_metadata.clone(), Arc::clone(&logger))
+			.await
+			.expect("event queue");
+		assert_eq!(queue.next_event().await, Some(failed_payment(1)));
+		queue.event_handled().await.expect("ack");
+		assert_eq!(queue.next_event().await, Some(failed_payment(2)));
+		drop(queue);
+
+		// The acknowledgement was persisted too: only the second event comes back.
+		let queue =
+			EventQueue::new(Arc::clone(&store), tx_metadata, logger).await.expect("event queue");
+		assert_eq!(queue.next_event().await, Some(failed_payment(2)));
+		queue.event_handled().await.expect("ack");
+		assert_eq!(queue.next_event().await, None);
+	}
+
+	/// A stored queue that does not decode fails the build instead of starting empty.
+	#[tokio::test]
+	async fn an_undecodable_stored_queue_fails_the_build() {
+		let (_path, store) = temp_sqlite_store();
+		// One event announced, then a variant id no `Event` has.
+		KVStore::write(
+			store.as_ref(),
+			EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
+			EVENT_QUEUE_PERSISTENCE_KEY,
+			vec![0x00, 0x01, 0xff],
+		)
+		.await
+		.expect("write");
+		let logger = Arc::new(Logger::new(&LoggerType::LogFacade).expect("logger"));
+		let tx_metadata = TxMetadataStore::new(Arc::clone(&store)).await;
+		let err = EventQueue::new(store, tx_metadata, logger).await.err().expect("must fail");
+		assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+	}
+
+	/// A store that cannot be read at all; everything else goes to the real store.
+	struct ReadFailsStore(Arc<dyn DynStore>);
+
+	impl DynStore for ReadFailsStore {
+		fn read_async(
+			&self, _p: &str, _s: &str, _k: &str,
+		) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, io::Error>> + Send + 'static>> {
+			Box::pin(async { Err(io::Error::new(io::ErrorKind::Other, "store down")) })
+		}
+
+		fn write_async(
+			&self, p: &str, s: &str, k: &str, buf: Vec<u8>,
+		) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send + 'static>> {
+			self.0.write_async(p, s, k, buf)
+		}
+
+		fn remove_async(
+			&self, p: &str, s: &str, k: &str, lazy: bool,
+		) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send + 'static>> {
+			self.0.remove_async(p, s, k, lazy)
+		}
+
+		fn list_async(
+			&self, p: &str, s: &str,
+		) -> Pin<Box<dyn Future<Output = Result<Vec<String>, io::Error>> + Send + 'static>> {
+			self.0.list_async(p, s)
+		}
+
+		fn list_paginated_async(
+			&self, p: &str, s: &str, page_token: Option<PageToken>,
+		) -> Pin<Box<dyn Future<Output = Result<PaginatedListResponse, io::Error>> + Send + 'static>>
+		{
+			self.0.list_paginated_async(p, s, page_token)
+		}
+	}
+
+	/// A store that cannot be read fails the build with the store's own error; the queue never
+	/// starts empty over a store it could not read.
+	#[tokio::test]
+	async fn an_unreadable_store_fails_the_build() {
+		let (_path, sqlite) = temp_sqlite_store();
+		let tx_metadata = TxMetadataStore::new(Arc::clone(&sqlite)).await;
+		let store: Arc<dyn DynStore> = Arc::new(ReadFailsStore(sqlite));
+		let logger = Arc::new(Logger::new(&LoggerType::LogFacade).expect("logger"));
+		let err = EventQueue::new(store, tx_metadata, logger).await.err().expect("must fail");
+		assert_eq!(err.kind(), io::ErrorKind::Other);
+	}
 }
 
 struct EventQueueSerWrapper<'a>(&'a VecDeque<Event>);
@@ -498,6 +669,21 @@ impl Writeable for EventQueueSerWrapper<'_> {
 	}
 }
 
+struct EventQueueDeserWrapper(VecDeque<Event>);
+
+impl Readable for EventQueueDeserWrapper {
+	fn read<R: ldk_node::lightning::io::Read>(
+		reader: &mut R,
+	) -> Result<Self, ldk_node::lightning::ln::msgs::DecodeError> {
+		let len: u16 = Readable::read(reader)?;
+		let mut queue = VecDeque::with_capacity(len as usize);
+		for _ in 0..len {
+			queue.push_back(Readable::read(reader)?);
+		}
+		Ok(Self(queue))
+	}
+}
+
 struct EventFuture {
 	event_queue: Arc<Mutex<VecDeque<Event>>>,
 	waker: Arc<Mutex<Option<Waker>>>,

orange-sdk/src/lib.rs

@@ -640,8 +640,9 @@ impl Wallet {
 
 		let tx_metadata = TxMetadataStore::new(Arc::clone(&store)).await;
 
-		let event_queue =
-			Arc::new(EventQueue::new(Arc::clone(&store), tx_metadata.clone(), Arc::clone(&logger)));
+		let event_queue = Arc::new(
+			EventQueue::new(Arc::clone(&store), tx_metadata.clone(), Arc::clone(&logger)).await?,
+		);
 
 		// Cashu must init before LDK Node because CashuKvDatabase does
 		// synchronous SQLite reads that deadlock with LDK Node's background

orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs

@@ -36,7 +36,11 @@ async fn open_cashu(
 	let logger = Arc::new(crate::logging::Logger::new(&config.logger_type).unwrap());
 	let runtime = Arc::new(crate::runtime::Runtime::new(logger.clone()).unwrap());
 	let metadata = crate::store::TxMetadataStore::new(store.clone()).await;
-	let events = Arc::new(crate::EventQueue::new(store.clone(), metadata.clone(), logger.clone()));
+	let events = Arc::new(
+		crate::EventQueue::new(store.clone(), metadata.clone(), logger.clone())
+			.await
+			.expect("event queue"),
+	);
 	Cashu::init(&config, cashu_config, store, events, metadata, logger, runtime).await
 }