Let BackgroundProcessor drive HTLC forwarding

fuzz/src/chanmon_consistency.rs

@@ -85,7 +85,7 @@ use bitcoin::secp256k1::{self, Message, PublicKey, Scalar, Secp256k1, SecretKey}
 use lightning::util::dyn_signer::DynSigner;
 
 use std::cell::RefCell;
-use std::cmp::{self, Ordering};
+use std::cmp;
 use std::mem;
 use std::sync::atomic;
 use std::sync::{Arc, Mutex};
@@ -1304,28 +1304,6 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				// deduplicate the calls here.
 				let mut claim_set = new_hash_map();
 				let mut events = nodes[$node].get_and_clear_pending_events();
-				// Sort events so that PendingHTLCsForwardable get processed last. This avoids a
-				// case where we first process a PendingHTLCsForwardable, then claim/fail on a
-				// PaymentClaimable, claiming/failing two HTLCs, but leaving a just-generated
-				// PaymentClaimable event for the second HTLC in our pending_events (and breaking
-				// our claim_set deduplication).
-				events.sort_by(|a, b| {
-					if let events::Event::PaymentClaimable { .. } = a {
-						if let events::Event::PendingHTLCsForwardable { .. } = b {
-							Ordering::Less
-						} else {
-							Ordering::Equal
-						}
-					} else if let events::Event::PendingHTLCsForwardable { .. } = a {
-						if let events::Event::PaymentClaimable { .. } = b {
-							Ordering::Greater
-						} else {
-							Ordering::Equal
-						}
-					} else {
-						Ordering::Equal
-					}
-				});
 				let had_events = !events.is_empty();
 				for event in events.drain(..) {
 					match event {
@@ -1352,9 +1330,6 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 						},
 						events::Event::PaymentForwarded { .. } if $node == 1 => {},
 						events::Event::ChannelReady { .. } => {},
-						events::Event::PendingHTLCsForwardable { .. } => {
-							nodes[$node].process_pending_htlc_forwards();
-						},
 						events::Event::HTLCHandlingFailed { .. } => {},
 						_ => {
 							if out.may_fail.load(atomic::Ordering::Acquire) {
@@ -1365,6 +1340,9 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 						},
 					}
 				}
+				while nodes[$node].needs_pending_htlc_processing() {
+					nodes[$node].process_pending_htlc_forwards();
+				}
 				had_events
 			}};
 		}
@@ -1806,8 +1784,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 								last_pass_no_updates = false;
 								continue;
 							}
-							// ...making sure any pending PendingHTLCsForwardable events are handled and
-							// payments claimed.
+							// ...making sure any payments are claimed.
 							if process_events!(0, false) {
 								last_pass_no_updates = false;
 								continue;

fuzz/src/full_stack.rs

@@ -635,7 +635,6 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 	let mut loss_detector =
 		MoneyLossDetector::new(&peers, channelmanager.clone(), monitor.clone(), peer_manager);
 
-	let mut should_forward = false;
 	let mut payments_received: Vec<PaymentHash> = Vec::new();
 	let mut intercepted_htlcs: Vec<InterceptId> = Vec::new();
 	let mut payments_sent: u16 = 0;
@@ -785,10 +784,7 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 				}
 			},
 			7 => {
-				if should_forward {
-					channelmanager.process_pending_htlc_forwards();
-					should_forward = false;
-				}
+				channelmanager.process_pending_htlc_forwards();
 			},
 			8 => {
 				for payment in payments_received.drain(..) {
@@ -1004,9 +1000,6 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 					//TODO: enhance by fetching random amounts from fuzz input?
 					payments_received.push(payment_hash);
 				},
-				Event::PendingHTLCsForwardable { .. } => {
-					should_forward = true;
-				},
 				Event::HTLCIntercepted { intercept_id, .. } => {
 					if !intercepted_htlcs.contains(&intercept_id) {
 						intercepted_htlcs.push(intercept_id);

lightning-background-processor/Cargo.toml

@@ -25,6 +25,7 @@ bitcoin-io = { version = "0.1.2", default-features = false }
 lightning = { version = "0.2.0", path = "../lightning", default-features = false }
 lightning-rapid-gossip-sync = { version = "0.2.0", path = "../lightning-rapid-gossip-sync", default-features = false }
 lightning-liquidity = { version = "0.2.0", path = "../lightning-liquidity", default-features = false }
+possiblyrandom = { version = "0.2", path = "../possiblyrandom", default-features = false }
 
 [dev-dependencies]
 tokio = { version = "1.35", features = [ "macros", "rt", "rt-multi-thread", "sync", "time" ] }