Upper-bound the event queue size

We add a size limit on the event queue, after which we'll just start
dropping events to ensure we could never OOM.

Additionally, we document the requirement that users need to handle
generated events ASAP.

Author: tnull

lightning-liquidity/src/events.rs

@@ -24,6 +24,9 @@ use crate::sync::{Arc, Mutex};
 use core::future::Future;
 use core::task::{Poll, Waker};
 
+/// The maximum queue size we allow before starting to drop events.
+pub const MAX_EVENT_QUEUE_SIZE: usize = 1000;
+
 pub(crate) struct EventQueue {
 	queue: Arc<Mutex<VecDeque<Event>>>,
 	waker: Arc<Mutex<Option<Waker>>>,
@@ -47,7 +50,11 @@ impl EventQueue {
 	pub fn enqueue(&self, event: Event) {
 		{
 			let mut queue = self.queue.lock().unwrap();
-			queue.push_back(event);
+			if queue.len() < MAX_EVENT_QUEUE_SIZE {
+				queue.push_back(event);
+			} else {
+				return;
+			}
 		}
 
 		if let Some(waker) = self.waker.lock().unwrap().take() {

lightning-liquidity/src/manager.rs

@@ -378,6 +378,12 @@ where {
 	/// Blocks the current thread until next event is ready and returns it.
 	///
 	/// Typically you would spawn a thread or task that calls this in a loop.
+	///
+	/// **Note**: Users must handle events as soon as possible to avoid an increased event queue
+	/// memory footprint. We will start dropping any generated events after
+	/// [`MAX_EVENT_QUEUE_SIZE`] has been reached.
+	///
+	/// [`MAX_EVENT_QUEUE_SIZE`]: crate::events::MAX_EVENT_QUEUE_SIZE
 	#[cfg(feature = "std")]
 	pub fn wait_next_event(&self) -> Event {
 		self.pending_events.wait_next_event()
@@ -386,20 +392,38 @@ where {
 	/// Returns `Some` if an event is ready.
 	///
 	/// Typically you would spawn a thread or task that calls this in a loop.
+	///
+	/// **Note**: Users must handle events as soon as possible to avoid an increased event queue
+	/// memory footprint. We will start dropping any generated events after
+	/// [`MAX_EVENT_QUEUE_SIZE`] has been reached.
+	///
+	/// [`MAX_EVENT_QUEUE_SIZE`]: crate::events::MAX_EVENT_QUEUE_SIZE
 	pub fn next_event(&self) -> Option<Event> {
 		self.pending_events.next_event()
 	}
 
 	/// Asynchronously polls the event queue and returns once the next event is ready.
 	///
 	/// Typically you would spawn a thread or task that calls this in a loop.
+	///
+	/// **Note**: Users must handle events as soon as possible to avoid an increased event queue
+	/// memory footprint. We will start dropping any generated events after
+	/// [`MAX_EVENT_QUEUE_SIZE`] has been reached.
+	///
+	/// [`MAX_EVENT_QUEUE_SIZE`]: crate::events::MAX_EVENT_QUEUE_SIZE
 	pub async fn next_event_async(&self) -> Event {
 		self.pending_events.next_event_async().await
 	}
 
 	/// Returns and clears all events without blocking.
 	///
 	/// Typically you would spawn a thread or task that calls this in a loop.
+	///
+	/// **Note**: Users must handle events as soon as possible to avoid an increased event queue
+	/// memory footprint. We will start dropping any generated events after
+	/// [`MAX_EVENT_QUEUE_SIZE`] has been reached.
+	///
+	/// [`MAX_EVENT_QUEUE_SIZE`]: crate::events::MAX_EVENT_QUEUE_SIZE
 	pub fn get_and_clear_pending_events(&self) -> Vec<Event> {
 		self.pending_events.get_and_clear_pending_events()
 	}