f Run async event processing in loop until we're done

Author: tnull

lightning/src/ln/channelmanager.rs

@@ -5779,35 +5779,40 @@ where
 	pub async fn process_pending_events_async<Future: core::future::Future, H: Fn(Event) -> Future>(
 		&self, handler: H
 	) {
-		if self.pending_events_processor.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed).is_err() {
-			return;
-		}
+		let mut processed_all_events = false;
+		while !processed_all_events {
+			if self.pending_events_processor.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed).is_err() {
+				return;
+			}
 
-		let mut result = NotifyOption::SkipPersist;
+			let mut result = NotifyOption::SkipPersist;
 
-		// TODO: This behavior should be documented. It's unintuitive that we query
-		// ChannelMonitors when clearing other events.
-		if self.process_pending_monitor_events() {
-			result = NotifyOption::DoPersist;
-		}
+			// TODO: This behavior should be documented. It's unintuitive that we query
+			// ChannelMonitors when clearing other events.
+			if self.process_pending_monitor_events() {
+				result = NotifyOption::DoPersist;
+			}
 
-		let pending_events = self.pending_events.lock().unwrap().clone();
-		let num_events = pending_events.len();
-		if !pending_events.is_empty() {
-			result = NotifyOption::DoPersist;
-		}
+			let pending_events = self.pending_events.lock().unwrap().clone();
+			let num_events = pending_events.len();
+			if !pending_events.is_empty() {
+				result = NotifyOption::DoPersist;
+			}
 
-		for event in pending_events {
-			handler(event).await;
-		}
+			for event in pending_events {
+				handler(event).await;
+			}
 
-		self.pending_events.lock().unwrap().drain(..num_events);
+			let mut pending_events = self.pending_events.lock().unwrap().clone();
+			pending_events.drain(..num_events);
+			processed_all_events = pending_events.is_empty();
 
-		if result == NotifyOption::DoPersist {
-			self.persistence_notifier.notify();
-		}
+			if result == NotifyOption::DoPersist {
+				self.persistence_notifier.notify();
+			}
 
-		self.pending_events_processor.store(false, Ordering::Release);
+			self.pending_events_processor.store(false, Ordering::Release);
+		}
 	}
 }