f maybe fix test on ci?

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3862,53 +3862,82 @@ fn test_claim_to_closed_channel_blocks_claimed_event() {
 }
 
 #[test]
-#[cfg(feature = "std")]
+#[cfg(all(feature = "std", not(target_os = "windows")))]
 fn test_single_channel_multiple_mpp() {
+	use std::sync::atomic::{AtomicBool, Ordering};
+
 	// Test what happens when we attempt to claim an MPP with many parts that came to us through
 	// the same channel with a synchronous persistence interface which has very high latency.
 	//
 	// Previously, if a `revoke_and_ack` came in while we were still running in
 	// `ChannelManager::claim_payment` we'd end up hanging waiting to apply a
 	// `ChannelMonitorUpdate` until after it completed. See the commit which introduced this test
 	// for more info.
-	let chanmon_cfgs = create_chanmon_cfgs(7);
-	let node_cfgs = create_node_cfgs(7, &chanmon_cfgs);
-	let node_chanmgrs = create_node_chanmgrs(7, &node_cfgs, &[None; 7]);
-	let mut nodes = create_network(7, &node_cfgs, &node_chanmgrs);
-
-	let node_5_id = nodes[5].node.get_our_node_id();
-	let node_6_id = nodes[6].node.get_our_node_id();
-
-	// Send an MPP payment in four parts along the path shown from top to bottom
-	//      0
-	//   1 2 3 4
-	//      5
-	//      6
+	let chanmon_cfgs = create_chanmon_cfgs(9);
+	let node_cfgs = create_node_cfgs(9, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(9, &node_cfgs, &[None; 9]);
+	let mut nodes = create_network(9, &node_cfgs, &node_chanmgrs);
+
+	let node_7_id = nodes[7].node.get_our_node_id();
+	let node_8_id = nodes[8].node.get_our_node_id();
+
+	// Send an MPP payment in six parts along the path shown from top to bottom
+	//        0
+	//   1 2 3 4 5 6
+	//        7
+	//        8
+	//
+	// We can in theory reproduce this issue with fewer channels/HTLCs, but getting this test
+	// robust is rather challenging. We rely on having the main test thread wait on locks held in
+	// the background `claim_funds` thread and unlocking when the `claim_funds` thread completes a
+	// single `ChannelMonitorUpdate`.
+	// This thread calls `get_and_clear_pending_msg_events()` and `handle_revoke_and_ack()`, both
+	// of which require `ChannelManager` locks, but we have to make sure this thread gets a chance
+	// to be blocked on the mutexes before we let the background thread wake `claim_funds` so that
+	// the mutex can switch to this main thread.
+	// This relies on our locks being fair, but also on our threads getting runtime during the test
+	// run, which can be pretty competitive. Thus we do a dumb dance to be as conservative as
+	// possible - we have a background thread which completes a `ChannelMonitorUpdate` (by sending
+	// into the `write_blocker` mpsc) but it doesn't run until a mpsc channel sends from this main
+	// thread to the background thread, and then we let it sleep a while before we send the
+	// `ChannelMonitorUpdate` unblocker.
+	// Further, we give ourselves two chances each time, needing 4 HTLCs just to unlock our two
+	// `ChannelManager` calls. We then need a few remaining HTLCs to actually trigger the bug, so
+	// we use 6 HTLCs.
+	// Finaly, we do not run this test on Winblowz because it, somehow, in 2025, does not implement
+	// actual preemptive multitasking and thinks that cooperative multitasking somehow is
+	// acceptable in the 21st century, let alone a quarter of the way into it.
+	const MAX_THREAD_INIT_TIME: std::time::Duration = std::time::Duration::from_secs(1);
 
 	create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 100_000, 0);
 	create_announced_chan_between_nodes_with_value(&nodes, 0, 2, 100_000, 0);
 	create_announced_chan_between_nodes_with_value(&nodes, 0, 3, 100_000, 0);
 	create_announced_chan_between_nodes_with_value(&nodes, 0, 4, 100_000, 0);
-	create_announced_chan_between_nodes_with_value(&nodes, 1, 5, 100_000, 0);
-	create_announced_chan_between_nodes_with_value(&nodes, 2, 5, 100_000, 0);
-	create_announced_chan_between_nodes_with_value(&nodes, 3, 5, 100_000, 0);
-	create_announced_chan_between_nodes_with_value(&nodes, 4, 5, 100_000, 0);
-	create_announced_chan_between_nodes_with_value(&nodes, 5, 6, 1_000_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 5, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 6, 100_000, 0);
 
-	let (mut route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(&nodes[0], nodes[6], 30_000_000);
+	create_announced_chan_between_nodes_with_value(&nodes, 1, 7, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 2, 7, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 3, 7, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 4, 7, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 5, 7, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 6, 7, 100_000, 0);
+	create_announced_chan_between_nodes_with_value(&nodes, 7, 8, 1_000_000, 0);
 
-	send_along_route_with_secret(&nodes[0], route, &[&[&nodes[1], &nodes[5], &nodes[6]], &[&nodes[2], &nodes[5], &nodes[6]], &[&nodes[3], &nodes[5], &nodes[6]], &[&nodes[4], &nodes[5], &nodes[6]]], 30_000_000, payment_hash, payment_secret);
+	let (mut route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(&nodes[0], nodes[8], 50_000_000);
+
+	send_along_route_with_secret(&nodes[0], route, &[&[&nodes[1], &nodes[7], &nodes[8]], &[&nodes[2], &nodes[7], &nodes[8]], &[&nodes[3], &nodes[7], &nodes[8]], &[&nodes[4], &nodes[7], &nodes[8]], &[&nodes[5], &nodes[7], &nodes[8]], &[&nodes[6], &nodes[7], &nodes[8]]], 50_000_000, payment_hash, payment_secret);
 
 	let (do_a_write, blocker) = std::sync::mpsc::sync_channel(0);
-	*nodes[6].chain_monitor.write_blocker.lock().unwrap() = Some(blocker);
+	*nodes[8].chain_monitor.write_blocker.lock().unwrap() = Some(blocker);
 
 	// Until we have std::thread::scoped we have to unsafe { turn off the borrow checker }.
 	// We do this by casting a pointer to a `TestChannelManager` to a pointer to a
 	// `TestChannelManager` with different (in this case 'static) lifetime.
 	// This is even suggested in the second example at
 	// https://doc.rust-lang.org/std/mem/fn.transmute.html#examples
 	let claim_node: &'static TestChannelManager<'static, 'static> =
-		unsafe { std::mem::transmute(nodes[6].node as &TestChannelManager) };
+		unsafe { std::mem::transmute(nodes[8].node as &TestChannelManager) };
 	let thrd = std::thread::spawn(move || {
 		// Initiate the claim in a background thread as it will immediately block waiting on the
 		// `write_blocker` we set above.
@@ -3924,85 +3953,131 @@ fn test_single_channel_multiple_mpp() {
 	// `claim_funds` is holding. Thus, we release a second write after a small sleep in the
 	// background to give `claim_funds` a chance to step forward, unblocking
 	// `get_and_clear_pending_msg_events`.
-	const MAX_THREAD_INIT_TIME: std::time::Duration = std::time::Duration::from_millis(10);
 	let do_a_write_background = do_a_write.clone();
+	let block_thrd2 = AtomicBool::new(true);
+	let block_thrd2_read: &'static AtomicBool = unsafe { std::mem::transmute(&block_thrd2) };
 	let thrd2 = std::thread::spawn(move || {
+		while block_thrd2_read.load(Ordering::Acquire) {
+			std::thread::yield_now();
+		}
+		std::thread::sleep(MAX_THREAD_INIT_TIME);
+		do_a_write_background.send(()).unwrap();
 		std::thread::sleep(MAX_THREAD_INIT_TIME);
 		do_a_write_background.send(()).unwrap();
 	});
-	let first_updates = get_htlc_update_msgs(&nodes[6], &nodes[5].node.get_our_node_id());
+	block_thrd2.store(false, Ordering::Release);
+	let first_updates = get_htlc_update_msgs(&nodes[8], &nodes[7].node.get_our_node_id());
 	thrd2.join().unwrap();
 
-	nodes[5].node.peer_disconnected(nodes[1].node.get_our_node_id());
-	nodes[5].node.peer_disconnected(nodes[2].node.get_our_node_id());
-	nodes[5].node.peer_disconnected(nodes[3].node.get_our_node_id());
-	nodes[5].node.peer_disconnected(nodes[4].node.get_our_node_id());
-
-	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &first_updates.update_fulfill_htlcs[0]);
-	check_added_monitors(&nodes[5], 1);
-	expect_payment_forwarded!(nodes[5], nodes[1], nodes[6], Some(1000), false, false);
-	nodes[5].node.handle_commitment_signed(node_6_id, &first_updates.commitment_signed);
-	check_added_monitors(&nodes[5], 1);
-	let (raa, cs) = get_revoke_commit_msgs(&nodes[5], &node_6_id);
+	// Disconnect node 6 from all its peers so it doesn't bother to fail the HTLCs back
+	nodes[7].node.peer_disconnected(nodes[1].node.get_our_node_id());
+	nodes[7].node.peer_disconnected(nodes[2].node.get_our_node_id());
+	nodes[7].node.peer_disconnected(nodes[3].node.get_our_node_id());
+	nodes[7].node.peer_disconnected(nodes[4].node.get_our_node_id());
+	nodes[7].node.peer_disconnected(nodes[5].node.get_our_node_id());
+	nodes[7].node.peer_disconnected(nodes[6].node.get_our_node_id());
+
+	nodes[7].node.handle_update_fulfill_htlc(node_8_id, &first_updates.update_fulfill_htlcs[0]);
+	check_added_monitors(&nodes[7], 1);
+	expect_payment_forwarded!(nodes[7], nodes[1], nodes[8], Some(1000), false, false);
+	nodes[7].node.handle_commitment_signed(node_8_id, &first_updates.commitment_signed);
+	check_added_monitors(&nodes[7], 1);
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[7], &node_8_id);
 
 	// Now, handle the `revoke_and_ack` from node 5. Note that `claim_funds` is still blocked on
 	// our peer lock, so we have to release a write to let it process.
+	// After this call completes, the channel previously would be locked up and should not be able
+	// to make further progress.
 	let do_a_write_background = do_a_write.clone();
+	let block_thrd3 = AtomicBool::new(true);
+	let block_thrd3_read: &'static AtomicBool = unsafe { std::mem::transmute(&block_thrd3) };
 	let thrd3 = std::thread::spawn(move || {
+		while block_thrd3_read.load(Ordering::Acquire) {
+			std::thread::yield_now();
+		}
+		std::thread::sleep(MAX_THREAD_INIT_TIME);
+		do_a_write_background.send(()).unwrap();
 		std::thread::sleep(MAX_THREAD_INIT_TIME);
 		do_a_write_background.send(()).unwrap();
 	});
-	nodes[6].node.handle_revoke_and_ack(node_5_id, &raa);
+	block_thrd3.store(false, Ordering::Release);
+	nodes[8].node.handle_revoke_and_ack(node_7_id, &raa);
 	thrd3.join().unwrap();
+	assert!(!thrd.is_finished());
 
 	let thrd4 = std::thread::spawn(move || {
-		std::thread::sleep(MAX_THREAD_INIT_TIME);
 		do_a_write.send(()).unwrap();
 		do_a_write.send(()).unwrap();
 	});
 
 	thrd4.join().unwrap();
 	thrd.join().unwrap();
 
-	expect_payment_claimed!(nodes[6], payment_hash, 30_000_000);
+	expect_payment_claimed!(nodes[8], payment_hash, 50_000_000);
 
-	// At the end, we should have 5 ChannelMonitorUpdates - 4 for HTLC claims, and one for the
+	// At the end, we should have 7 ChannelMonitorUpdates - 6 for HTLC claims, and one for the
 	// above `revoke_and_ack`.
-	check_added_monitors(&nodes[6], 5);
-
-	// Now drive everything to the end, at least as far as node 6 is concerned...
-	*nodes[6].chain_monitor.write_blocker.lock().unwrap() = None;
-	nodes[6].node.handle_commitment_signed(node_5_id, &cs);
-	check_added_monitors(&nodes[6], 1);
-
-	let (updates, raa) = get_updates_and_revoke(&nodes[6], &nodes[5].node.get_our_node_id());
-	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &updates.update_fulfill_htlcs[0]);
-	expect_payment_forwarded!(nodes[5], nodes[2], nodes[6], Some(1000), false, false);
-	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &updates.update_fulfill_htlcs[1]);
-	expect_payment_forwarded!(nodes[5], nodes[3], nodes[6], Some(1000), false, false);
-	nodes[5].node.handle_commitment_signed(node_6_id, &updates.commitment_signed);
-	nodes[5].node.handle_revoke_and_ack(node_6_id, &raa);
-	check_added_monitors(&nodes[5], 4);
-
-	let (raa, cs) = get_revoke_commit_msgs(&nodes[5], &node_6_id);
-
-	nodes[6].node.handle_revoke_and_ack(node_5_id, &raa);
-	nodes[6].node.handle_commitment_signed(node_5_id, &cs);
-	check_added_monitors(&nodes[6], 2);
-
-	let (updates, raa) = get_updates_and_revoke(&nodes[6], &nodes[5].node.get_our_node_id());
-	nodes[5].node.handle_update_fulfill_htlc(node_6_id, &updates.update_fulfill_htlcs[0]);
-	expect_payment_forwarded!(nodes[5], nodes[4], nodes[6], Some(1000), false, false);
-	nodes[5].node.handle_commitment_signed(node_6_id, &updates.commitment_signed);
-	nodes[5].node.handle_revoke_and_ack(node_6_id, &raa);
-	check_added_monitors(&nodes[5], 3);
-
-	let (raa, cs) = get_revoke_commit_msgs(&nodes[5], &node_6_id);
-	nodes[6].node.handle_revoke_and_ack(node_5_id, &raa);
-	nodes[6].node.handle_commitment_signed(node_5_id, &cs);
-	check_added_monitors(&nodes[6], 2);
-
-	let raa = get_event_msg!(nodes[6], MessageSendEvent::SendRevokeAndACK, node_5_id);
-	nodes[5].node.handle_revoke_and_ack(node_6_id, &raa);
-	check_added_monitors(&nodes[5], 1);
+	check_added_monitors(&nodes[8], 7);
+
+	// Now drive everything to the end, at least as far as node 7 is concerned...
+	*nodes[8].chain_monitor.write_blocker.lock().unwrap() = None;
+	nodes[8].node.handle_commitment_signed(node_7_id, &cs);
+	check_added_monitors(&nodes[8], 1);
+
+	let (updates, raa) = get_updates_and_revoke(&nodes[8], &nodes[7].node.get_our_node_id());
+
+	nodes[7].node.handle_update_fulfill_htlc(node_8_id, &updates.update_fulfill_htlcs[0]);
+	expect_payment_forwarded!(nodes[7], nodes[2], nodes[8], Some(1000), false, false);
+	nodes[7].node.handle_update_fulfill_htlc(node_8_id, &updates.update_fulfill_htlcs[1]);
+	expect_payment_forwarded!(nodes[7], nodes[3], nodes[8], Some(1000), false, false);
+	let mut next_source = 4;
+	if let Some(update) = updates.update_fulfill_htlcs.get(2) {
+		nodes[7].node.handle_update_fulfill_htlc(node_8_id, update);
+		expect_payment_forwarded!(nodes[7], nodes[4], nodes[8], Some(1000), false, false);
+		next_source += 1;
+	}
+
+	nodes[7].node.handle_commitment_signed(node_8_id, &updates.commitment_signed);
+	nodes[7].node.handle_revoke_and_ack(node_8_id, &raa);
+	if updates.update_fulfill_htlcs.get(2).is_some() {
+		check_added_monitors(&nodes[7], 5);
+	} else {
+		check_added_monitors(&nodes[7], 4);
+	}
+
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[7], &node_8_id);
+
+	nodes[8].node.handle_revoke_and_ack(node_7_id, &raa);
+	nodes[8].node.handle_commitment_signed(node_7_id, &cs);
+	check_added_monitors(&nodes[8], 2);
+
+	let (updates, raa) = get_updates_and_revoke(&nodes[8], &node_7_id);
+
+	nodes[7].node.handle_update_fulfill_htlc(node_8_id, &updates.update_fulfill_htlcs[0]);
+	expect_payment_forwarded!(nodes[7], nodes[next_source], nodes[8], Some(1000), false, false);
+	next_source += 1;
+	nodes[7].node.handle_update_fulfill_htlc(node_8_id, &updates.update_fulfill_htlcs[1]);
+	expect_payment_forwarded!(nodes[7], nodes[next_source], nodes[8], Some(1000), false, false);
+	next_source += 1;
+	if let Some(update) = updates.update_fulfill_htlcs.get(2) {
+		nodes[7].node.handle_update_fulfill_htlc(node_8_id, update);
+		expect_payment_forwarded!(nodes[7], nodes[next_source], nodes[8], Some(1000), false, false);
+	}
+
+	nodes[7].node.handle_commitment_signed(node_8_id, &updates.commitment_signed);
+	nodes[7].node.handle_revoke_and_ack(node_8_id, &raa);
+	if updates.update_fulfill_htlcs.get(2).is_some() {
+		check_added_monitors(&nodes[7], 5);
+	} else {
+		check_added_monitors(&nodes[7], 4);
+	}
+
+	let (raa, cs) = get_revoke_commit_msgs(&nodes[7], &node_8_id);
+	nodes[8].node.handle_revoke_and_ack(node_7_id, &raa);
+	nodes[8].node.handle_commitment_signed(node_7_id, &cs);
+	check_added_monitors(&nodes[8], 2);
+
+	let raa = get_event_msg!(nodes[8], MessageSendEvent::SendRevokeAndACK, node_7_id);
+	nodes[7].node.handle_revoke_and_ack(node_8_id, &raa);
+	check_added_monitors(&nodes[7], 1);
 }