f Have sync event handler drop the queue lock during event handling

Author: tnull

lightning/src/onion_message/messenger.rs

@@ -1472,7 +1472,7 @@ where
 		{
 			let pending_intercepted_msgs_events = self.pending_intercepted_msgs_events.lock().unwrap();
 			intercepted_msgs = pending_intercepted_msgs_events.clone();
-			let mut pending_peer_connected_events = self.pending_peer_connected_events.lock().unwrap();
+			let pending_peer_connected_events = self.pending_peer_connected_events.lock().unwrap();
 			peer_connecteds = pending_peer_connected_events.clone();
 			#[cfg(debug_assertions)] {
 				for ev in pending_intercepted_msgs_events.iter() {
@@ -1482,14 +1482,45 @@ where
 					if let Event::OnionMessagePeerConnected { .. } = ev {} else { panic!(); }
 				}
 			}
-			pending_peer_connected_events.shrink_to(10); // Limit total heap usage
 		}
 
-		let mut res_iter = intercepted_msgs.into_iter().map(|ev| handler.handle_event(ev));
-		drop_handled_events_and_abort!(self, res_iter, self.pending_intercepted_msgs_events);
+		let mut handling_intercepted_msgs_failed = false;
+		let mut num_handled_intercepted_events = 0;
+		for ev in intercepted_msgs {
+			match handler.handle_event(ev) {
+				Ok(()) => num_handled_intercepted_events += 1,
+				Err(ReplayEvent ()) => {
+					handling_intercepted_msgs_failed = true;
+					break;
+				}
+			}
+		}
+
+		{
+			let mut pending_intercepted_msgs_events = self.pending_intercepted_msgs_events.lock().unwrap();
+			pending_intercepted_msgs_events.drain(..num_handled_intercepted_events);
+		}
 
-		let mut res_iter = peer_connecteds.into_iter().map(|ev| handler.handle_event(ev));
-		drop_handled_events_and_abort!(self, res_iter, self.pending_peer_connected_events);
+		if handling_intercepted_msgs_failed {
+			self.pending_events_processor.store(false, Ordering::Release);
+			return;
+		}
+
+		let mut num_handled_peer_connecteds = 0;
+		for ev in peer_connecteds {
+			match handler.handle_event(ev) {
+				Ok(()) => num_handled_peer_connecteds += 1,
+				Err(ReplayEvent ()) => {
+					break;
+				}
+			}
+		}
+
+		{
+			let mut pending_peer_connected_events = self.pending_peer_connected_events.lock().unwrap();
+			pending_peer_connected_events.drain(..num_handled_peer_connecteds);
+			pending_peer_connected_events.shrink_to(10); // Limit total heap usage
+		}
 
 		self.pending_events_processor.store(false, Ordering::Release);
 	}