Public commit record
50/100 · Thin
What the developer wrote
Authored by Elias Rohrer
Add `EventQueue` persistence
We add simple `persist` call to `EventQueue` that persists it under a
`event_queue` key.
✓ Descriptive subject✓ Provides an explanatory body
What changed, and why it matters
This commit adds the ability to save (persist) the LiquidityManager's event queue to disk so events survive application restarts. It is a feature commit, not a security patch. Most event types are explicitly documented as not persisted, while LSPS2 'OpenChannel' and LSPS5 'SendWebhookNotification' events are now saved and replayed after restart. The code includes safety notes warning callers to make their channel-opening logic idempotent because persisted events may be replayed.
Recommended action
Review downstream callers of LSPS2ServiceEvent::OpenChannel and LSPS5ServiceEvent::SendWebhookNotification to confirm they implement idempotent handling before relying on persisted-event replay. Ensure the KVStore implementation used provides atomic/consistent writes for the new event_queue key. No immediate patch is required; this is a feature addition with documented operational caveats.
Security signals we found
01
New persistence surface introduced for in-memory event queue
02
Replay of persisted events after restart could lead to duplicate channel opens if callers do not implement idempotency
03
Serialization skips unknown odd-type TLVs, which is the expected forward-compatibility behavior but means unknown events are silently lost on reload
04
No input validation changes; persistence relies on existing KVStore abstraction
05
Documentation explicitly warns about idempotency requirement for OpenChannel replay
Evidence from the diff
The change threads a KVStore through EventQueue and all handlers that use it, adds serialization for LSPS2ServiceEvent and LSPS5ServiceEvent (and LSPS5 WebhookNotification/Method), and implements EventQueue::persist() plus a deserializer that skips unknown odd-type TLVs. Persisted events are limited to LSPS2ServiceEvent::OpenChannel and LSPS5ServiceEvent::SendWebhookNotification; other events are dropped on serialization and documented as non-persistent. The commit also exposes CollectionLength in lightning::util::ser for use by the new serialization code.
Changed components
lightning-liquidity/src/events/event_queue.rslightning-liquidity/src/manager.rslightning-liquidity/src/persist.rslightning-liquidity/src/lsps2/event.rslightning-liquidity/src/lsps2/service.rslightning-liquidity/src/lsps5/event.rslightning-liquidity/src/lsps5/msgs.rslightning-liquidity/src/lsps5/service.rslightning/src/util/ser.rsInspect captured patch +343 / −68
diff --git a/lightning-liquidity/src/events/event_queue.rs b/lightning-liquidity/src/events/event_queue.rs
index f59d34e..1c8282f 100644
--- a/lightning-liquidity/src/events/event_queue.rs
+++ b/lightning-liquidity/src/events/event_queue.rs
@@ -1,24 +1,45 @@
use super::LiquidityEvent;
+
+use crate::lsps2::event::LSPS2ServiceEvent;
+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::collections::VecDeque;
use alloc::vec::Vec;
use core::future::Future;
+use core::ops::Deref;
use core::task::{Poll, Waker};
+use lightning::ln::msgs::DecodeError;
+use lightning::util::persist::KVStore;
+use lightning::util::ser::{
+ BigSize, CollectionLength, FixedLengthReader, Readable, Writeable, Writer,
+};
+
/// The maximum queue size we allow before starting to drop events.
pub const MAX_EVENT_QUEUE_SIZE: usize = 1000;
-pub(crate) struct EventQueue {
+pub(crate) struct EventQueue<K: Deref + Clone>
+where
+ K::Target: KVStore,
+{
queue: Arc<Mutex<VecDeque<LiquidityEvent>>>,
waker: Arc<Mutex<Option<Waker>>>,
#[cfg(feature = "std")]
condvar: Arc<crate::sync::Condvar>,
+ kv_store: K,
}
-impl EventQueue {
- pub fn new() -> Self {
+impl<K: Deref + Clone> EventQueue<K>
+where
+ K::Target: KVStore,
+{
+ pub fn new(kv_store: K) -> Self {
let queue = Arc::new(Mutex::new(VecDeque::new()));
let waker = Arc::new(Mutex::new(None));
Self {
@@ -26,6 +47,7 @@ impl EventQueue {
waker,
#[cfg(feature = "std")]
condvar: Arc::new(crate::sync::Condvar::new()),
+ kv_store,
}
}
@@ -67,16 +89,35 @@ impl EventQueue {
}
// Returns an [`EventQueueNotifierGuard`] that will notify about new event when dropped.
- pub fn notifier(&self) -> EventQueueNotifierGuard<'_> {
+ pub fn notifier(&self) -> EventQueueNotifierGuard<'_, K> {
EventQueueNotifierGuard(self)
}
+
+ pub async fn persist(&self) -> Result<(), lightning::io::Error> {
+ 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,
+ )
+ .await
+ }
}
// A guard type that will notify about new events when dropped.
#[must_use]
-pub(crate) struct EventQueueNotifierGuard<'a>(&'a EventQueue);
-
-impl<'a> EventQueueNotifierGuard<'a> {
+pub(crate) struct EventQueueNotifierGuard<'a, K: Deref + Clone>(&'a EventQueue<K>)
+where
+ K::Target: KVStore;
+
+impl<'a, K: Deref + Clone> EventQueueNotifierGuard<'a, K>
+where
+ K::Target: KVStore,
+{
pub fn enqueue<E: Into<LiquidityEvent>>(&self, event: E) {
let mut queue = self.0.queue.lock().unwrap();
if queue.len() < MAX_EVENT_QUEUE_SIZE {
@@ -87,7 +128,10 @@ impl<'a> EventQueueNotifierGuard<'a> {
}
}
-impl<'a> Drop for EventQueueNotifierGuard<'a> {
+impl<'a, K: Deref + Clone> Drop for EventQueueNotifierGuard<'a, K>
+where
+ K::Target: KVStore,
+{
fn drop(&mut self) {
let should_notify = !self.0.queue.lock().unwrap().is_empty();
@@ -122,6 +166,91 @@ 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, DecodeError> {
+ let len: CollectionLength = Readable::read(reader)?;
+ let mut queue = VecDeque::with_capacity(len.0 as usize);
+ for _ in 0..len.0 {
+ let event = match Readable::read(reader)? {
+ 0u8 => {
+ let ev = Readable::read(reader)?;
+ LiquidityEvent::LSPS2Service(ev)
+ },
+ 2u8 => {
+ let ev = Readable::read(reader)?;
+ LiquidityEvent::LSPS5Service(ev)
+ },
+ 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)?;
+ continue;
+ },
+ _ => return Err(DecodeError::InvalidValue),
+ };
+ 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> {
+ let maybe_process_event = |event: &LiquidityEvent,
+ writer: Option<&mut W>|
+ -> Result<bool, lightning::io::Error> {
+ match event {
+ LiquidityEvent::LSPS2Service(event) => {
+ if matches!(event, LSPS2ServiceEvent::GetInfo { .. })
+ || matches!(event, LSPS2ServiceEvent::BuyRequest { .. })
+ {
+ // Skip persisting GetInfoRequest and BuyRequest events as we prune the pending
+ // request state currently anyways.
+ Ok(false)
+ } else {
+ if let Some(writer) = writer {
+ 0u8.write(writer)?;
+ event.write(writer)?;
+ }
+ Ok(true)
+ }
+ },
+ LiquidityEvent::LSPS5Service(event) => {
+ if let Some(writer) = writer {
+ 2u8.write(writer)?;
+ event.write(writer)?;
+ }
+ Ok(true)
+ },
+ _ => Ok(false),
+ }
+ };
+
+ let mut persisted_events_len = 0;
+ for e in self.0.iter() {
+ if maybe_process_event(e, None)? {
+ persisted_events_len += 1;
+ }
+ }
+
+ CollectionLength(persisted_events_len).write(writer)?;
+ for e in self.0.iter() {
+ maybe_process_event(e, Some(writer))?;
+ }
+ Ok(())
+ }
+}
+
#[cfg(test)]
mod tests {
#[tokio::test]
@@ -131,10 +260,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();
diff --git a/lightning-liquidity/src/lsps0/client.rs b/lightning-liquidity/src/lsps0/client.rs
index f7e01b3..2efadd4 100644
--- a/lightning-liquidity/src/lsps0/client.rs
+++ b/lightning-liquidity/src/lsps0/client.rs
@@ -18,28 +18,31 @@ use crate::utils;
use lightning::ln::msgs::{ErrorAction, LightningError};
use lightning::sign::EntropySource;
use lightning::util::logger::Level;
+use lightning::util::persist::KVStore;
use bitcoin::secp256k1::PublicKey;
use core::ops::Deref;
/// A message handler capable of sending and handling bLIP-50 / LSPS0 messages.
-pub struct LSPS0ClientHandler<ES: Deref>
+pub struct LSPS0ClientHandler<ES: Deref, K: Deref + Clone>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
entropy_source: ES,
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
}
-impl<ES: Deref> LSPS0ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPS0ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
/// Returns a new instance of [`LSPS0ClientHandler`].
pub(crate) fn new(
- entropy_source: ES, pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>,
+ entropy_source: ES, pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue<K>>,
) -> Self {
Self { entropy_source, pending_messages, pending_events }
}
@@ -86,9 +89,10 @@ where
}
}
-impl<ES: Deref> LSPSProtocolMessageHandler for LSPS0ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPSProtocolMessageHandler for LSPS0ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
type ProtocolMessage = LSPS0Message;
const PROTOCOL_NUMBER: Option<u16> = None;
@@ -113,10 +117,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 +132,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,
diff --git a/lightning-liquidity/src/lsps0/event.rs b/lightning-liquidity/src/lsps0/event.rs
index 97a3a95..4141b51 100644
--- a/lightning-liquidity/src/lsps0/event.rs
+++ b/lightning-liquidity/src/lsps0/event.rs
@@ -14,6 +14,8 @@ use alloc::vec::Vec;
use bitcoin::secp256k1::PublicKey;
/// An event which an bLIP-50 / LSPS0 client may want to take some action in response to.
+///
+/// **Note: ** This event will *not* be persisted across restarts.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LSPS0ClientEvent {
/// Information from the LSP about the protocols they support.
diff --git a/lightning-liquidity/src/lsps1/client.rs b/lightning-liquidity/src/lsps1/client.rs
index 45008ba..5b9d373 100644
--- a/lightning-liquidity/src/lsps1/client.rs
+++ b/lightning-liquidity/src/lsps1/client.rs
@@ -25,6 +25,7 @@ use crate::sync::{Arc, Mutex, RwLock};
use lightning::ln::msgs::{ErrorAction, LightningError};
use lightning::sign::EntropySource;
use lightning::util::logger::Level;
+use lightning::util::persist::KVStore;
use bitcoin::secp256k1::PublicKey;
use bitcoin::Address;
@@ -46,25 +47,27 @@ struct PeerState {
}
/// The main object allowing to send and receive bLIP-51 / LSPS1 messages.
-pub struct LSPS1ClientHandler<ES: Deref>
+pub struct LSPS1ClientHandler<ES: Deref, K: Deref + Clone>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
entropy_source: ES,
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
config: LSPS1ClientConfig,
}
-impl<ES: Deref> LSPS1ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPS1ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
/// Constructs an `LSPS1ClientHandler`.
pub(crate) fn new(
- entropy_source: ES, pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>,
- config: LSPS1ClientConfig,
+ entropy_source: ES, pending_messages: Arc<MessageQueue>,
+ pending_events: Arc<EventQueue<K>>, config: LSPS1ClientConfig,
) -> Self {
Self {
entropy_source,
@@ -429,9 +432,10 @@ where
}
}
-impl<ES: Deref> LSPSProtocolMessageHandler for LSPS1ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPSProtocolMessageHandler for LSPS1ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
type ProtocolMessage = LSPS1Message;
const PROTOCOL_NUMBER: Option<u16> = Some(1);
diff --git a/lightning-liquidity/src/lsps1/event.rs b/lightning-liquidity/src/lsps1/event.rs
index 508a5a4..fdf3fc5 100644
--- a/lightning-liquidity/src/lsps1/event.rs
+++ b/lightning-liquidity/src/lsps1/event.rs
@@ -25,6 +25,8 @@ pub enum LSPS1ClientEvent {
/// You must check whether LSP supports the parameters the client wants and then call
/// [`LSPS1ClientHandler::create_order`] to place an order.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ClientHandler::request_supported_options`]: crate::lsps1::client::LSPS1ClientHandler::request_supported_options
/// [`LSPS1ClientHandler::create_order`]: crate::lsps1::client::LSPS1ClientHandler::create_order
SupportedOptionsReady {
@@ -43,6 +45,8 @@ pub enum LSPS1ClientEvent {
/// A request previously issued via [`LSPS1ClientHandler::request_supported_options`]
/// failed as the LSP returned an error response.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ClientHandler::request_supported_options`]: crate::lsps1::client::LSPS1ClientHandler::request_supported_options
SupportedOptionsRequestFailed {
/// The identifier of the issued bLIP-51 / LSPS1 `get_info` request, as returned by
@@ -66,6 +70,8 @@ pub enum LSPS1ClientEvent {
/// call [`LSPS1ClientHandler::check_order_status`] with the order id
/// to get information from LSP about progress of the order.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ClientHandler::check_order_status`]: crate::lsps1::client::LSPS1ClientHandler::check_order_status
OrderCreated {
/// The identifier of the issued bLIP-51 / LSPS1 `create_order` request, as returned by
@@ -90,6 +96,8 @@ pub enum LSPS1ClientEvent {
///
/// Will be emitted in response to calling [`LSPS1ClientHandler::check_order_status`].
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ClientHandler::check_order_status`]: crate::lsps1::client::LSPS1ClientHandler::check_order_status
OrderStatus {
/// The identifier of the issued bLIP-51 / LSPS1 `get_order` request, as returned by
@@ -113,6 +121,8 @@ pub enum LSPS1ClientEvent {
/// A request previously issued via [`LSPS1ClientHandler::create_order`] or [`LSPS1ClientHandler::check_order_status`].
/// failed as the LSP returned an error response.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ClientHandler::create_order`]: crate::lsps1::client::LSPS1ClientHandler::create_order
/// [`LSPS1ClientHandler::check_order_status`]: crate::lsps1::client::LSPS1ClientHandler::check_order_status
OrderRequestFailed {
@@ -142,6 +152,8 @@ pub enum LSPS1ServiceEvent {
/// send order parameters including the details regarding the
/// payment and order id for this order for the client.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ServiceHandler::send_payment_details`]: crate::lsps1::service::LSPS1ServiceHandler::send_payment_details
RequestForPaymentDetails {
/// An identifier that must be passed to [`LSPS1ServiceHandler::send_payment_details`].
@@ -160,6 +172,8 @@ pub enum LSPS1ServiceEvent {
/// You must call [`LSPS1ServiceHandler::update_order_status`] to update the client
/// regarding the status of the payment and order.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS1ServiceHandler::update_order_status`]: crate::lsps1::service::LSPS1ServiceHandler::update_order_status
CheckPaymentConfirmation {
/// An identifier that must be passed to [`LSPS1ServiceHandler::update_order_status`].
@@ -172,6 +186,8 @@ pub enum LSPS1ServiceEvent {
order_id: LSPS1OrderId,
},
/// If error is encountered, refund the amount if paid by the client.
+ ///
+ /// **Note: ** This event will *not* be persisted across restarts.
Refund {
/// An identifier.
request_id: LSPSRequestId,
diff --git a/lightning-liquidity/src/lsps1/service.rs b/lightning-liquidity/src/lsps1/service.rs
index 1b4bdf5..8afea1b 100644
--- a/lightning-liquidity/src/lsps1/service.rs
+++ b/lightning-liquidity/src/lsps1/service.rs
@@ -36,6 +36,7 @@ use lightning::ln::msgs::{ErrorAction, LightningError};
use lightning::sign::EntropySource;
use lightning::util::errors::APIError;
use lightning::util::logger::Level;
+use lightning::util::persist::KVStore;
use bitcoin::secp256k1::PublicKey;
@@ -131,32 +132,35 @@ impl PeerState {
}
/// The main object allowing to send and receive bLIP-51 / LSPS1 messages.
-pub struct LSPS1ServiceHandler<ES: Deref, CM: Deref + Clone, C: Deref>
+pub struct LSPS1ServiceHandler<ES: Deref, CM: Deref + Clone, C: Deref, K: Deref + Clone>
where
ES::Target: EntropySource,
CM::Target: AChannelManager,
C::Target: Filter,
+ K::Target: KVStore,
{
entropy_source: ES,
channel_manager: CM,
chain_source: Option<C>,
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
config: LSPS1ServiceConfig,
}
-impl<ES: Deref, CM: Deref + Clone, C: Deref> LSPS1ServiceHandler<ES, CM, C>
+impl<ES: Deref, CM: Deref + Clone, C: Deref, K: Deref + Clone> LSPS1ServiceHandler<ES, CM, C, K>
where
ES::Target: EntropySource,
CM::Target: AChannelManager,
C::Target: Filter,
ES::Target: EntropySource,
+ K::Target: KVStore,
{
/// Constructs a `LSPS1ServiceHandler`.
pub(crate) fn new(
- entropy_source: ES, pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>,
- channel_manager: CM, chain_source: Option<C>, config: LSPS1ServiceConfig,
+ entropy_source: ES, pending_messages: Arc<MessageQueue>,
+ pending_events: Arc<EventQueue<K>>, channel_manager: CM, chain_source: Option<C>,
+ config: LSPS1ServiceConfig,
) -> Self {
Self {
entropy_source,
@@ -417,12 +421,13 @@ where
}
}
-impl<ES: Deref, CM: Deref + Clone, C: Deref> LSPSProtocolMessageHandler
- for LSPS1ServiceHandler<ES, CM, C>
+impl<ES: Deref, CM: Deref + Clone, C: Deref, K: Deref + Clone> LSPSProtocolMessageHandler
+ for LSPS1ServiceHandler<ES, CM, C, K>
where
ES::Target: EntropySource,
CM::Target: AChannelManager,
C::Target: Filter,
+ K::Target: KVStore,
{
type ProtocolMessage = LSPS1Message;
const PROTOCOL_NUMBER: Option<u16> = Some(1);
diff --git a/lightning-liquidity/src/lsps2/client.rs b/lightning-liquidity/src/lsps2/client.rs
index 7008d42..71b2a2b 100644
--- a/lightning-liquidity/src/lsps2/client.rs
+++ b/lightning-liquidity/src/lsps2/client.rs
@@ -10,6 +10,7 @@
//! Contains the main bLIP-52 / LSPS2 client object, [`LSPS2ClientHandler`].
use alloc::string::{String, ToString};
+use lightning::util::persist::KVStore;
use core::default::Default;
use core::ops::Deref;
@@ -67,25 +68,27 @@ impl PeerState {
/// opened. Please refer to the [`bLIP-52 / LSPS2 specification`] for more information.
///
/// [`bLIP-52 / LSPS2 specification`]: https://github.com/lightning/blips/blob/master/blip-0052.md#trust-models
-pub struct LSPS2ClientHandler<ES: Deref>
+pub struct LSPS2ClientHandler<ES: Deref, K: Deref + Clone>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
entropy_source: ES,
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
config: LSPS2ClientConfig,
}
-impl<ES: Deref> LSPS2ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPS2ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
/// Constructs an `LSPS2ClientHandler`.
pub(crate) fn new(
- entropy_source: ES, pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>,
- config: LSPS2ClientConfig,
+ entropy_source: ES, pending_messages: Arc<MessageQueue>,
+ pending_events: Arc<EventQueue<K>>, config: LSPS2ClientConfig,
) -> Self {
Self {
entropy_source,
@@ -366,9 +369,10 @@ where
}
}
-impl<ES: Deref> LSPSProtocolMessageHandler for LSPS2ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPSProtocolMessageHandler for LSPS2ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
type ProtocolMessage = LSPS2Message;
const PROTOCOL_NUMBER: Option<u16> = Some(2);
diff --git a/lightning-liquidity/src/lsps2/event.rs b/lightning-liquidity/src/lsps2/event.rs
index f738dc0..29cc577 100644
--- a/lightning-liquidity/src/lsps2/event.rs
+++ b/lightning-liquidity/src/lsps2/event.rs
@@ -16,6 +16,8 @@ use alloc::vec::Vec;
use bitcoin::secp256k1::PublicKey;
+use lightning::impl_writeable_tlv_based_enum;
+
/// An event which an LSPS2 client should take some action in response to.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LSPS2ClientEvent {
@@ -24,6 +26,8 @@ pub enum LSPS2ClientEvent {
/// You must call [`LSPS2ClientHandler::select_opening_params`] with the fee parameter
/// you want to use if you wish to proceed opening a channel.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS2ClientHandler::select_opening_params`]: crate::lsps2::client::LSPS2ClientHandler::select_opening_params
OpeningParametersReady {
/// The identifier of the issued bLIP-52 / LSPS2 `get_info` request, as returned by
@@ -44,6 +48,8 @@ pub enum LSPS2ClientEvent {
///
/// When the invoice is paid, the LSP will open a channel with the previously agreed upon
/// parameters to you.
+ ///
+ /// **Note: ** This event will *not* be persisted across restarts.
InvoiceParametersReady {
/// The identifier of the issued bLIP-52 / LSPS2 `buy` request, as returned by
/// [`LSPS2ClientHandler::select_opening_params`].
@@ -64,6 +70,8 @@ pub enum LSPS2ClientEvent {
/// A request previously issued via [`LSPS2ClientHandler::request_opening_params`]
/// failed as the LSP returned an error response.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS2ClientHandler::request_opening_params`]: crate::lsps2::client::LSPS2ClientHandler::request_opening_params
GetInfoFailed {
/// The identifier of the issued LSPS2 `get_info` request, as returned by
@@ -81,6 +89,8 @@ pub enum LSPS2ClientEvent {
/// A request previously issued via [`LSPS2ClientHandler::select_opening_params`]
/// failed as the LSP returned an error response.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS2ClientHandler::select_opening_params`]: crate::lsps2::client::LSPS2ClientHandler::select_opening_params
BuyRequestFailed {
/// The identifier of the issued LSPS2 `buy` request, as returned by
@@ -108,6 +118,8 @@ pub enum LSPS2ServiceEvent {
/// If an unrecognized or stale token is provided you can use
/// `[LSPS2ServiceHandler::invalid_token_provided`] to error the request.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`LSPS2ServiceHandler::opening_fee_params_generated`]: crate::lsps2::service::LSPS2ServiceHandler::opening_fee_params_generated
/// [`LSPS2ServiceHandler::invalid_token_provided`]: crate::lsps2::service::LSPS2ServiceHandler::invalid_token_provided
GetInfo {
@@ -130,6 +142,8 @@ pub enum LSPS2ServiceEvent {
/// [`ChannelManager::get_intercept_scid`] for them to use and then call
/// [`LSPS2ServiceHandler::invoice_parameters_generated`].
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`ChannelManager::get_intercept_scid`]: lightning::ln::channelmanager::ChannelManager::get_intercept_scid
///
/// [`LSPS2ServiceHandler::invoice_parameters_generated`]: crate::lsps2::service::LSPS2ServiceHandler::invoice_parameters_generated
@@ -147,6 +161,11 @@ pub enum LSPS2ServiceEvent {
},
/// You should open a channel using [`ChannelManager::create_channel`].
///
+ /// **Note: ** As this event is persisted and might get replayed after restart, you'll need to
+ /// ensure channel creation idempotency. I.e., please check if you already created a
+ /// corresponding channel based on the given `their_network_key` and `intercept_scid` and
+ /// ignore this event in case you did.
+ ///
/// [`ChannelManager::create_channel`]: lightning::ln::channelmanager::ChannelManager::create_channel
OpenChannel {
/// The node to open channel with.
@@ -161,3 +180,24 @@ pub enum LSPS2ServiceEvent {
intercept_scid: u64,
},
}
+
+impl_writeable_tlv_based_enum!(LSPS2ServiceEvent,
+ (0, GetInfo) => {
+ (0, request_id, required),
+ (2, counterparty_node_id, required),
+ (4, token, option),
+ },
+ (2, BuyRequest) => {
+ (0, request_id, required),
+ (2, counterparty_node_id, required),
+ (4, opening_fee_params, required),
+ (6, payment_size_msat, option),
+ },
+ (4, OpenChannel) => {
+ (0, their_network_key, required),
+ (2, amt_to_forward_msat, required),
+ (4, opening_fee_msat, required),
+ (6, user_channel_id, required),
+ (8, intercept_scid, required),
+ }
+);
diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs
index 2244fbf..9e8073f 100644
--- a/lightning-liquidity/src/lsps2/service.rs
+++ b/lightning-liquidity/src/lsps2/service.rs
@@ -577,7 +577,7 @@ where
channel_manager: CM,
kv_store: K,
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
peer_by_intercept_scid: RwLock<HashMap<u64, PublicKey>>,
peer_by_channel_id: RwLock<HashMap<ChannelId, PublicKey>>,
@@ -592,8 +592,8 @@ where
{
/// Constructs a `LSPS2ServiceHandler`.
pub(crate) fn new(
- pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>, channel_manager: CM,
- kv_store: K, config: LSPS2ServiceConfig,
+ pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue<K>>,
+ channel_manager: CM, kv_store: K, config: LSPS2ServiceConfig,
) -> Self {
Self {
pending_messages,
diff --git a/lightning-liquidity/src/lsps5/client.rs b/lightning-liquidity/src/lsps5/client.rs
index 2e90545..e464e88 100644
--- a/lightning-liquidity/src/lsps5/client.rs
+++ b/lightning-liquidity/src/lsps5/client.rs
@@ -33,6 +33,7 @@ use lightning::util::logger::Level;
use alloc::collections::VecDeque;
use alloc::string::String;
+use lightning::util::persist::KVStore;
use core::ops::Deref;
@@ -124,25 +125,27 @@ impl PeerState {
/// [`lsps5.list_webhooks`]: super::msgs::LSPS5Request::ListWebhooks
/// [`lsps5.remove_webhook`]: super::msgs::LSPS5Request::RemoveWebhook
/// [`LSPS5Validator`]: super::validator::LSPS5Validator
-pub struct LSPS5ClientHandler<ES: Deref>
+pub struct LSPS5ClientHandler<ES: Deref, K: Deref + Clone>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
entropy_source: ES,
per_peer_state: RwLock<HashMap<PublicKey, Mutex<PeerState>>>,
_config: LSPS5ClientConfig,
}
-impl<ES: Deref> LSPS5ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPS5ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
/// Constructs an `LSPS5ClientHandler`.
pub(crate) fn new(
- entropy_source: ES, pending_messages: Arc<MessageQueue>, pending_events: Arc<EventQueue>,
- _config: LSPS5ClientConfig,
+ entropy_source: ES, pending_messages: Arc<MessageQueue>,
+ pending_events: Arc<EventQueue<K>>, _config: LSPS5ClientConfig,
) -> Self {
Self {
pending_messages,
@@ -423,9 +426,10 @@ where
}
}
-impl<ES: Deref> LSPSProtocolMessageHandler for LSPS5ClientHandler<ES>
+impl<ES: Deref, K: Deref + Clone> LSPSProtocolMessageHandler for LSPS5ClientHandler<ES, K>
where
ES::Target: EntropySource,
+ K::Target: KVStore,
{
type ProtocolMessage = LSPS5Message;
const PROTOCOL_NUMBER: Option<u16> = Some(5);
@@ -444,6 +448,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,
@@ -459,15 +465,17 @@ mod tests {
}
fn setup_test_client() -> (
- LSPS5ClientHandler<Arc<UniqueTestEntropy>>,
+ LSPS5ClientHandler<Arc<UniqueTestEntropy>, Arc<KVStoreSyncWrapper<Arc<TestStore>>>>,
Arc<MessageQueue>,
- Arc<EventQueue>,
+ Arc<EventQueue<Arc<KVStoreSyncWrapper<Arc<TestStore>>>>>,
PublicKey,
PublicKey,
) {
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),
diff --git a/lightning-liquidity/src/lsps5/event.rs b/lightning-liquidity/src/lsps5/event.rs
index f401c0e..a9c1052 100644
--- a/lightning-liquidity/src/lsps5/event.rs
+++ b/lightning-liquidity/src/lsps5/event.rs
@@ -13,6 +13,8 @@ use crate::lsps0::ser::LSPSRequestId;
use alloc::string::String;
use alloc::vec::Vec;
use bitcoin::secp256k1::PublicKey;
+
+use lightning::impl_writeable_tlv_based_enum;
use lightning::util::hash_tables::HashMap;
use super::msgs::LSPS5AppName;
@@ -37,6 +39,8 @@ pub enum LSPS5ServiceEvent {
/// when received by the client. The client verifies this signature using
/// [`validate`], which guards against replay attacks and tampering.
///
+ /// **Note: ** This event will be persisted across restarts.
+ ///
/// [`validate`]: super::validator::LSPS5Validator::validate
/// [`url`]: super::msgs::LSPS5WebhookUrl
/// [`notification`]: super::msgs::WebhookNotification
@@ -70,6 +74,16 @@ pub enum LSPS5ServiceEvent {
},
}
+impl_writeable_tlv_based_enum!(LSPS5ServiceEvent,
+ (0, SendWebhookNotification) => {
+ (0, counterparty_node_id, required),
+ (2, app_name, required),
+ (4, url, required),
+ (6, notification, required),
+ (8, headers, required),
+ }
+);
+
/// An event which an LSPS5 client should take some action in response to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LSPS5ClientEvent {
@@ -82,6 +96,8 @@ pub enum LSPS5ClientEvent {
/// the LSP will also emit a [`SendWebhookNotification`] event with a [`webhook_registered`] notification
/// to notify the client about this registration.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`lsps5.set_webhook`]: super::msgs::LSPS5Request::SetWebhook
/// [`SendWebhookNotification`]: super::event::LSPS5ServiceEvent::SendWebhookNotification
/// [`webhook_registered`]: super::msgs::WebhookNotificationMethod::LSPS5WebhookRegistered
@@ -117,6 +133,8 @@ pub enum LSPS5ClientEvent {
/// - Maximum number of webhooks per client has been reached (error [`TooManyWebhooks`]). Remove a webhook before
/// registering a new one.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`lsps5.set_webhook`]: super::msgs::LSPS5Request::SetWebhook
/// [`app_name`]: super::msgs::LSPS5AppName
/// [`url`]: super::msgs::LSPS5WebhookUrl
@@ -170,6 +188,8 @@ pub enum LSPS5ClientEvent {
/// After this event, the app_name is free to be reused for a new webhook
/// registration if desired.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`lsps5.remove_webhook`]: super::msgs::LSPS5Request::RemoveWebhook
WebhookRemoved {
/// The node id of the LSP that confirmed the removal.
@@ -191,6 +211,8 @@ pub enum LSPS5ClientEvent {
/// (error code [`LSPS5_APP_NAME_NOT_FOUND_ERROR_CODE`]), which indicates
/// the given [`app_name`] was not found in the LSP's registration database.
///
+ /// **Note: ** This event will *not* be persisted across restarts.
+ ///
/// [`lsps5.remove_webhook`]: super::msgs::LSPS5Request::RemoveWebhook
/// [`AppNameNotFound`]: super::msgs::LSPS5ProtocolError::AppNameNotFound
/// [`LSPS5ProtocolError::AppNameNotFound`]: super::msgs::LSPS5ProtocolError::AppNameNotFound
diff --git a/lightning-liquidity/src/lsps5/msgs.rs b/lightning-liquidity/src/lsps5/msgs.rs
index f1ef06d..341dfcd 100644
--- a/lightning-liquidity/src/lsps5/msgs.rs
+++ b/lightning-liquidity/src/lsps5/msgs.rs
@@ -18,6 +18,7 @@ use super::url_utils::LSPSUrl;
use lightning::ln::msgs::DecodeError;
use lightning::util::ser::{Readable, Writeable};
+use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum};
use lightning_types::string::UntrustedString;
use serde::de::{self, Deserializer, MapAccess, Visitor};
@@ -522,6 +523,16 @@ pub enum WebhookNotificationMethod {
LSPS5OnionMessageIncoming,
}
+impl_writeable_tlv_based_enum!(WebhookNotificationMethod,
+ (0, LSPS5WebhookRegistered) => {},
+ (2, LSPS5PaymentIncoming) => {},
+ (4, LSPS5ExpirySoon) => {
+ (0, timeout, required),
+ },
+ (6, LSPS5LiquidityManagementRequest) => {},
+ (8, LSPS5OnionMessageIncoming) => {},
+);
+
/// Webhook notification payload.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WebhookNotification {
@@ -672,6 +683,10 @@ impl<'de> Deserialize<'de> for WebhookNotification {
}
}
+impl_writeable_tlv_based!(WebhookNotification, {
+ (0, method, required),
+});
+
/// An LSPS5 protocol request.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LSPS5Request {
diff --git a/lightning-liquidity/src/lsps5/service.rs b/lightning-liquidity/src/lsps5/service.rs
index ff1a27c..a4a636c 100644
--- a/lightning-liquidity/src/lsps5/service.rs
+++ b/lightning-liquidity/src/lsps5/service.rs
@@ -132,7 +132,7 @@ where
{
config: LSPS5ServiceConfig,
per_peer_state: RwLock<HashMap<PublicKey, PeerState>>,
- event_queue: Arc<EventQueue>,
+ event_queue: Arc<EventQueue<K>>,
pending_messages: Arc<MessageQueue>,
time_provider: TP,
channel_manager: CM,
@@ -150,7 +150,7 @@ where
{
/// Constructs a `LSPS5ServiceHandler` using the given time provider.
pub(crate) fn new_with_time_provider(
- event_queue: Arc<EventQueue>, pending_messages: Arc<MessageQueue>, channel_manager: CM,
+ event_queue: Arc<EventQueue<K>>, pending_messages: Arc<MessageQueue>, channel_manager: CM,
kv_store: K, node_signer: NS, config: LSPS5ServiceConfig, time_provider: TP,
) -> Self {
assert!(config.max_webhooks_per_client > 0, "`max_webhooks_per_client` must be > 0");
diff --git a/lightning-liquidity/src/manager.rs b/lightning-liquidity/src/manager.rs
index 86343d5..e3e059a 100644
--- a/lightning-liquidity/src/manager.rs
+++ b/lightning-liquidity/src/manager.rs
@@ -292,19 +292,19 @@ pub struct LiquidityManager<
TP::Target: TimeProvider,
{
pending_messages: Arc<MessageQueue>,
- pending_events: Arc<EventQueue>,
+ pending_events: Arc<EventQueue<K>>,
request_id_to_method_map: Mutex<HashMap<LSPSRequestId, LSPSMethod>>,
// We ignore peers if they send us bogus data.
ignored_peers: RwLock<HashSet<PublicKey>>,
- lsps0_client_handler: LSPS0ClientHandler<ES>,
+ lsps0_client_handler: LSPS0ClientHandler<ES, K>,
lsps0_service_handler: Option<LSPS0ServiceHandler>,
#[cfg(lsps1_service)]
- lsps1_service_handler: Option<LSPS1ServiceHandler<ES, CM, C>>,
- lsps1_client_handler: Option<LSPS1ClientHandler<ES>>,
+ lsps1_service_handler: Option<LSPS1ServiceHandler<ES, CM, C, K>>,
+ lsps1_client_handler: Option<LSPS1ClientHandler<ES, K>>,
lsps2_service_handler: Option<LSPS2ServiceHandler<CM, K>>,
- lsps2_client_handler: Option<LSPS2ClientHandler<ES>>,
+ lsps2_client_handler: Option<LSPS2ClientHandler<ES, K>>,
lsps5_service_handler: Option<LSPS5ServiceHandler<CM, NS, K, TP>>,
- lsps5_client_handler: Option<LSPS5ClientHandler<ES>>,
+ lsps5_client_handler: Option<LSPS5ClientHandler<ES, K>>,
service_config: Option<LiquidityServiceConfig>,
_client_config: Option<LiquidityClientConfig>,
best_block: RwLock<Option<BestBlock>>,
@@ -377,7 +377,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(kv_store.clone()));
let ignored_peers = RwLock::new(new_hash_set());
let mut supported_protocols = Vec::new();
@@ -454,7 +454,7 @@ where
#[cfg(lsps1_service)]
let lsps1_service_handler = service_config.as_ref().and_then(|config| {
if let Some(number) =
- <LSPS1ServiceHandler<ES, CM, C> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
+ <LSPS1ServiceHandler<ES, CM, C, K> as LSPSProtocolMessageHandler>::PROTOCOL_NUMBER
{
supported_protocols.push(number);
}
@@ -504,7 +504,7 @@ where
}
/// Returns a reference to the LSPS0 client-side handler.
- pub fn lsps0_client_handler(&self) -> &LSPS0ClientHandler<ES> {
+ pub fn lsps0_client_handler(&self) -> &LSPS0ClientHandler<ES, K> {
&self.lsps0_client_handler
}
@@ -517,13 +517,13 @@ where
///
/// The returned handler allows to initiate the LSPS1 client-side flow, i.e., allows to request
/// channels from the configured LSP.
- pub fn lsps1_client_handler(&self) -> Option<&LSPS1ClientHandler<ES>> {
+ pub fn lsps1_client_handler(&self) -> Option<&LSPS1ClientHandler<ES, K>> {
self.lsps1_client_handler.as_ref()
}
/// Returns a reference to the LSPS1 server-side handler.
#[cfg(lsps1_service)]
- pub fn lsps1_service_handler(&self) -> Option<&LSPS1ServiceHandler<ES, CM, C>> {
+ pub fn lsps1_service_handler(&self) -> Option<&LSPS1ServiceHandler<ES, CM, C, K>> {
self.lsps1_service_handler.as_ref()
}
@@ -532,7 +532,7 @@ where
/// The returned handler allows to initiate the LSPS2 client-side flow. That is, it allows to
/// retrieve all necessary data to create 'just-in-time' invoices that, when paid, will have
/// the configured LSP open a 'just-in-time' channel.
- pub fn lsps2_client_handler(&self) -> Option<&LSPS2ClientHandler<ES>> {
+ pub fn lsps2_client_handler(&self) -> Option<&LSPS2ClientHandler<ES, K>> {
self.lsps2_client_handler.as_ref()
}
@@ -546,7 +546,7 @@ where
/// Returns a reference to the LSPS5 client-side handler.
///
/// The returned handler allows to initiate the LSPS5 client-side flow. That is, it allows to
- pub fn lsps5_client_handler(&self) -> Option<&LSPS5ClientHandler<ES>> {
+ pub fn lsps5_client_handler(&self) -> Option<&LSPS5ClientHandler<ES, K>> {
self.lsps5_client_handler.as_ref()
}
@@ -620,6 +620,8 @@ where
/// be called manually if it's not utilized.
pub async fn persist(&self) -> Result<(), lightning::io::Error> {
// TODO: We should eventually persist in parallel.
+ self.pending_events.persist().await?;
+
if let Some(lsps2_service_handler) = self.lsps2_service_handler.as_ref() {
lsps2_service_handler.persist().await?;
}
@@ -1095,7 +1097,7 @@ where
/// Returns a reference to the LSPS0 client-side handler.
///
/// Wraps [`LiquidityManager::lsps0_client_handler`].
- pub fn lsps0_client_handler(&self) -> &LSPS0ClientHandler<ES> {
+ pub fn lsps0_client_handler(&self) -> &LSPS0ClientHandler<ES, Arc<KVStoreSyncWrapper<KS>>> {
self.inner.lsps0_client_handler()
}
@@ -1109,7 +1111,9 @@ where
/// Returns a reference to the LSPS1 client-side handler.
///
/// Wraps [`LiquidityManager::lsps1_client_handler`].
- pub fn lsps1_client_handler(&self) -> Option<&LSPS1ClientHandler<ES>> {
+ pub fn lsps1_client_handler(
+ &self,
+ ) -> Option<&LSPS1ClientHandler<ES, Arc<KVStoreSyncWrapper<KS>>>> {
self.inner.lsps1_client_handler()
}
@@ -1117,14 +1121,18 @@ where
///
/// Wraps [`LiquidityManager::lsps1_service_handler`].
#[cfg(lsps1_service)]
- pub fn lsps1_service_handler(&self) -> Option<&LSPS1ServiceHandler<ES, CM, C>> {
+ pub fn lsps1_service_handler(
+ &self,
+ ) -> Option<&LSPS1ServiceHandler<ES, CM, C, Arc<KVStoreSyncWrapper<KS>>>> {
self.inner.lsps1_service_handler()
}
/// Returns a reference to the LSPS2 client-side handler.
///
/// Wraps [`LiquidityManager::lsps2_client_handler`].
- pub fn lsps2_client_handler(&self) -> Option<&LSPS2ClientHandler<ES>> {
+ pub fn lsps2_client_handler(
+ &self,
+ ) -> Option<&LSPS2ClientHandler<ES, Arc<KVStoreSyncWrapper<KS>>>> {
self.inner.lsps2_client_handler()
}
@@ -1140,7 +1148,9 @@ where
/// Returns a reference to the LSPS5 client-side handler.
///
/// Wraps [`LiquidityManager::lsps5_client_handler`].
- pub fn lsps5_client_handler(&self) -> Option<&LSPS5ClientHandler<ES>> {
+ pub fn lsps5_client_handler(
+ &self,
+ ) -> Option<&LSPS5ClientHandler<ES, Arc<KVStoreSyncWrapper<KS>>>> {
self.inner.lsps5_client_handler()
}
diff --git a/lightning-liquidity/src/persist.rs b/lightning-liquidity/src/persist.rs
index f90b3ed..8b62220 100644
--- a/lightning-liquidity/src/persist.rs
+++ b/lightning-liquidity/src/persist.rs
@@ -14,6 +14,16 @@
/// [`LiquidityManager`]: crate::LiquidityManager
pub const LIQUIDITY_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "lightning_liquidity_state";
+/// 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
diff --git a/lightning/src/util/ser.rs b/lightning/src/util/ser.rs
index aa2d105..2578af0 100644
--- a/lightning/src/util/ser.rs
+++ b/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> {
Risk score
Our methodology →Why this scored 34/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.