@@ -2719,6 +2719,10 @@ pub struct ChannelManager<
27192719 /// A simple atomic flag to ensure only one task at a time can be processing events asynchronously.
27202720 pending_events_processor: AtomicBool,
27212721
2722+ /// A simple atomic flag to ensure only one task at a time can be processing HTLC forwards via
2723+ /// [`Self::process_pending_htlc_forwards`].
2724+ pending_htlc_forwards_processor: AtomicBool,
2725+
27222726 /// If we are running during init (either directly during the deserialization method or in
27232727 /// block connection methods which run after deserialization but before normal operation) we
27242728 /// cannot provide the user with [`ChannelMonitorUpdate`]s through the normal update flow -
@@ -3795,6 +3799,7 @@ where
37953799
37963800 pending_events: Mutex::new(VecDeque::new()),
37973801 pending_events_processor: AtomicBool::new(false),
3802+ pending_htlc_forwards_processor: AtomicBool::new(false),
37983803 pending_background_events: Mutex::new(Vec::new()),
37993804 total_consistency_lock: RwLock::new(()),
38003805 background_events_processed_since_startup: AtomicBool::new(false),
@@ -6324,9 +6329,19 @@ where
63246329 ///
63256330 /// Will regularly be called by the background processor.
63266331 pub fn process_pending_htlc_forwards(&self) {
6332+ if self
6333+ .pending_htlc_forwards_processor
6334+ .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
6335+ .is_err()
6336+ {
6337+ return;
6338+ }
6339+
63276340 let _persistence_guard = PersistenceNotifierGuard::optionally_notify(self, || {
63286341 self.internal_process_pending_htlc_forwards()
63296342 });
6343+
6344+ self.pending_htlc_forwards_processor.store(false, Ordering::Release);
63306345 }
63316346
63326347 // Returns whether or not we need to re-persist.
@@ -16302,6 +16317,7 @@ where
1630216317
1630316318 pending_events: Mutex::new(pending_events_read),
1630416319 pending_events_processor: AtomicBool::new(false),
16320+ pending_htlc_forwards_processor: AtomicBool::new(false),
1630516321 pending_background_events: Mutex::new(pending_background_events),
1630616322 total_consistency_lock: RwLock::new(()),
1630716323 background_events_processed_since_startup: AtomicBool::new(false),
0 commit comments