diff --git a/orange-sdk/src/event.rs b/orange-sdk/src/event.rs index c00f053..39ed68c 100644 --- a/orange-sdk/src/event.rs +++ b/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, tx_metadata: TxMetadataStore, logger: Arc, - ) -> Self { - let queue = Arc::new(Mutex::new(VecDeque::new())); + ) -> Result { + 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) { @@ -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); + + impl DynStore for ReadFailsStore { + fn read_async( + &self, _p: &str, _s: &str, _k: &str, + ) -> Pin, 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, + ) -> Pin> + Send + 'static>> { + self.0.write_async(p, s, k, buf) + } + + fn remove_async( + &self, p: &str, s: &str, k: &str, lazy: bool, + ) -> Pin> + Send + 'static>> { + self.0.remove_async(p, s, k, lazy) + } + + fn list_async( + &self, p: &str, s: &str, + ) -> Pin, io::Error>> + Send + 'static>> { + self.0.list_async(p, s) + } + + fn list_paginated_async( + &self, p: &str, s: &str, page_token: Option, + ) -> Pin> + 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 = 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); @@ -498,6 +669,21 @@ impl Writeable for EventQueueSerWrapper<'_> { } } +struct EventQueueDeserWrapper(VecDeque); + +impl Readable for EventQueueDeserWrapper { + fn read( + reader: &mut R, + ) -> Result { + 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>>, waker: Arc>>, diff --git a/orange-sdk/src/lib.rs b/orange-sdk/src/lib.rs index b51d72d..b243641 100644 --- a/orange-sdk/src/lib.rs +++ b/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 diff --git a/orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs b/orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs index e3236d1..1b828a9 100644 --- a/orange-sdk/src/trusted_wallet/cashu/cashu_store_upgrade_tests.rs +++ b/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 }