Let BackgroundProcessor drive HTLC forwarding

Previously, we'd require the user to manually call
`process_pending_htlc_forwards` as part of `PendingHTLCsForwardable`
event handling. Here, we rather move this responsibility to
`BackgroundProcessor`, which simplyfies the flow and allows us to
implement reasonable forwarding delays on our side rather than
delegating to users' implementations.

Note this also introduces batching rounds rather than calling
`process_pending_htlc_forwards` individually for each
`PendingHTLCsForwardable` event, which had been unintuitive anyways, as
subsequent `PendingHTLCsForwardable` could lead to overlapping batch
intervals, resulting in the shortest timespan 'winning' every time, as
`process_pending_htlc_forwards` would of course handle all pending HTLCs
at once.

Author: tnull

lightning-background-processor/src/fwd_batch.rs

@@ -0,0 +1,36 @@
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
+
+use core::time::Duration;
+
+pub(crate) struct BatchDelay {
+	next_batch_delay_millis: u16,
+}
+
+impl BatchDelay {
+	pub(crate) fn new() -> Self {
+		let next_batch_delay_millis = rand_batch_delay_millis();
+		Self { next_batch_delay_millis }
+	}
+
+	pub(crate) fn get(&self) -> Duration {
+		Duration::from_millis(self.next_batch_delay_millis as u64)
+	}
+
+	pub(crate) fn next(&mut self) -> Duration {
+		let next = rand_batch_delay_millis();
+		self.next_batch_delay_millis = next;
+		Duration::from_millis(next as u64)
+	}
+}
+
+fn rand_batch_delay_millis() -> u16 {
+	// TODO: actually randomize the result.
+	100
+}

lightning-background-processor/src/lib.rs

@@ -26,6 +26,10 @@ extern crate alloc;
 extern crate lightning;
 extern crate lightning_rapid_gossip_sync;
 
+mod fwd_batch;
+
+use fwd_batch::BatchDelay;
+
 use lightning::chain;
 use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use lightning::chain::chainmonitor::{ChainMonitor, Persist};
@@ -328,7 +332,7 @@ macro_rules! define_run_body {
 		$peer_manager: ident, $gossip_sync: ident,
 		$process_sweeper: expr,
 		$logger: ident, $scorer: ident, $loop_exit_check: expr, $await: expr, $get_timer: expr,
-		$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr,
+		$timer_elapsed: expr, $check_slow_await: expr, $time_fetch: expr, $batch_delay: expr,
 	) => { {
 		log_trace!($logger, "Calling ChannelManager's timer_tick_occurred on startup");
 		$channel_manager.get_cm().timer_tick_occurred();
@@ -345,6 +349,9 @@ macro_rules! define_run_body {
 		let mut have_pruned = false;
 		let mut have_decayed_scorer = false;
 
+		let mut cur_batch_delay = $batch_delay.get();
+		let mut last_forwards_processing_call = $get_timer(cur_batch_delay);
+
 		loop {
 			$process_channel_manager_events;
 			$process_chain_monitor_events;
@@ -369,6 +376,19 @@ macro_rules! define_run_body {
 				break;
 			}
 
+			if $timer_elapsed(&mut last_forwards_processing_call, cur_batch_delay) {
+				$channel_manager.get_cm().process_pending_htlc_forwards();
+				cur_batch_delay = $batch_delay.next();
+				last_forwards_processing_call = $get_timer(cur_batch_delay);
+			}
+
+			// Checke whether to exit the loop again, as some time might have passed since we
+			// checked above.
+			if $loop_exit_check {
+				log_trace!($logger, "Terminating background processor.");
+				break;
+			}
+
 			// We wait up to 100ms, but track how long it takes to detect being put to sleep,
 			// see `await_start`'s use below.
 			let mut await_start = None;
@@ -523,12 +543,14 @@ pub(crate) mod futures_util {
 		C: Future<Output = ()> + Unpin,
 		D: Future<Output = ()> + Unpin,
 		E: Future<Output = bool> + Unpin,
+		F: Future<Output = bool> + Unpin,
 	> {
 		pub a: A,
 		pub b: B,
 		pub c: C,
 		pub d: D,
 		pub e: E,
+		pub f: F,
 	}
 
 	pub(crate) enum SelectorOutput {
@@ -537,6 +559,7 @@ pub(crate) mod futures_util {
 		C,
 		D,
 		E(bool),
+		F(bool),
 	}
 
 	impl<
@@ -545,7 +568,8 @@ pub(crate) mod futures_util {
 			C: Future<Output = ()> + Unpin,
 			D: Future<Output = ()> + Unpin,
 			E: Future<Output = bool> + Unpin,
-		> Future for Selector<A, B, C, D, E>
+			F: Future<Output = bool> + Unpin,
+		> Future for Selector<A, B, C, D, E, F>
 	{
 		type Output = SelectorOutput;
 		fn poll(
@@ -581,6 +605,12 @@ pub(crate) mod futures_util {
 				},
 				Poll::Pending => {},
 			}
+			match Pin::new(&mut self.f).poll(ctx) {
+				Poll::Ready(res) => {
+					return Poll::Ready(SelectorOutput::F(res));
+				},
+				Poll::Pending => {},
+			}
 			Poll::Pending
 		}
 	}
@@ -863,6 +893,7 @@ where
 			event_handler(event).await
 		})
 	};
+	let mut batch_delay = BatchDelay::new();
 	define_run_body!(
 		persister,
 		chain_monitor,
@@ -901,7 +932,12 @@ where
 				b: chain_monitor.get_update_future(),
 				c: om_fut,
 				d: lm_fut,
-				e: sleeper(if mobile_interruptable_platform {
+				e: sleeper(if channel_manager.get_cm().needs_pending_htlc_processing() {
+					batch_delay.get()
+				} else {
+					Duration::MAX
+				}),
+				f: sleeper(if mobile_interruptable_platform {
 					Duration::from_millis(100)
 				} else {
 					FASTEST_TIMER
@@ -912,6 +948,9 @@ where
 				SelectorOutput::E(exit) => {
 					should_break = exit;
 				},
+				SelectorOutput::F(exit) => {
+					should_break = exit;
+				},
 			}
 		},
 		|t| sleeper(t),
@@ -928,6 +967,7 @@ where
 		},
 		mobile_interruptable_platform,
 		fetch_time,
+		batch_delay,
 	)
 }
 
@@ -1051,6 +1091,7 @@ impl BackgroundProcessor {
 				}
 				event_handler.handle_event(event)
 			};
+			let mut batch_delay = BatchDelay::new();
 			define_run_body!(
 				persister,
 				chain_monitor,
@@ -1094,7 +1135,13 @@ impl BackgroundProcessor {
 							&chain_monitor.get_update_future(),
 						),
 					};
-					sleeper.wait_timeout(Duration::from_millis(100));
+					let batch_delay = if channel_manager.get_cm().needs_pending_htlc_processing() {
+						batch_delay.get()
+					} else {
+						Duration::MAX
+					};
+					let fastest_timeout = batch_delay.min(Duration::from_millis(100));
+					sleeper.wait_timeout(fastest_timeout);
 				},
 				|_| Instant::now(),
 				|time: &Instant, dur| time.elapsed() > dur,
@@ -1107,6 +1154,7 @@ impl BackgroundProcessor {
 							.expect("Time should be sometime after 1970"),
 					)
 				},
+				batch_delay,
 			)
 		});
 		Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) }

lightning/src/ln/channelmanager.rs

@@ -6318,6 +6318,21 @@ where
 		}
 	}
 
+	/// Returns whether we have pending HTLC forwards that need to be processed via
+	/// [`Self::process_pending_htlc_forwards`].
+	pub fn needs_pending_htlc_processing(&self) -> bool {
+		if !self.forward_htlcs.lock().unwrap().is_empty() {
+			return true;
+		}
+		if !self.decode_update_add_htlcs.lock().unwrap().is_empty() {
+			return true;
+		}
+		if self.pending_outbound_payments.needs_abandon_or_retry() {
+			return true;
+		}
+		false
+	}
+
 	/// Processes HTLCs which are pending waiting on random forward delay.
 	///
 	/// Should only really ever be called in response to a PendingHTLCsForwardable event.

lightning/src/ln/outbound_payment.rs

@@ -1328,6 +1328,17 @@ impl OutboundPayments {
 		})
 	}
 
+	pub(super) fn needs_abandon_or_retry(&self) -> bool {
+		let outbounds = self.pending_outbound_payments.lock().unwrap();
+		outbounds.iter().any(|(_, pmt)| {
+			pmt.is_auto_retryable_now()
+				|| !pmt.is_auto_retryable_now()
+					&& pmt.remaining_parts() == 0
+					&& !pmt.is_fulfilled()
+					&& !pmt.is_awaiting_invoice()
+		})
+	}
+
 	#[rustfmt::skip]
 	fn find_initial_route<R: Deref, NS: Deref, IH, L: Deref>(
 		&self, payment_id: PaymentId, payment_hash: PaymentHash, recipient_onion: &RecipientOnionFields,