fuzz: support initial async monitor persistence in chanmon_consistency

Read the first byte of fuzz input to determine initial monitor styles
for each node (bit 0 = node A, bit 1 = node B, bit 2 = node C). When
set, the node starts with InProgress persistence mode from the
beginning.

This allows fuzzing the async persistence path during initial channel
creation, not just after reload. The make_channel macro now completes
pending monitor updates after watch_channel calls to allow the channel
handshake to proceed.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: joostjager

fuzz/src/chanmon_consistency.rs

@@ -299,8 +299,10 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 				persisted_monitor: ser.0,
 				pending_monitors: Vec::new(),
 			},
-			Ok(chain::ChannelMonitorUpdateStatus::InProgress) => {
-				panic!("The test currently doesn't test initial-persistence via the async pipeline")
+			Ok(chain::ChannelMonitorUpdateStatus::InProgress) => LatestMonitorState {
+				persisted_monitor_id: monitor_id,
+				persisted_monitor: Vec::new(),
+				pending_monitors: vec![(monitor_id, ser.0)],
 			},
 			Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
 			Err(()) => panic!(),
@@ -706,6 +708,26 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 	let broadcast = Arc::new(TestBroadcaster {});
 	let router = FuzzRouter {};
 
+	// Read initial monitor styles from fuzz input (1 byte: 2 bits per node)
+	let initial_mon_styles = if !data.is_empty() { data[0] } else { 0 };
+	let mon_style = [
+		RefCell::new(if initial_mon_styles & 0b01 != 0 {
+			ChannelMonitorUpdateStatus::InProgress
+		} else {
+			ChannelMonitorUpdateStatus::Completed
+		}),
+		RefCell::new(if initial_mon_styles & 0b10 != 0 {
+			ChannelMonitorUpdateStatus::InProgress
+		} else {
+			ChannelMonitorUpdateStatus::Completed
+		}),
+		RefCell::new(if initial_mon_styles & 0b100 != 0 {
+			ChannelMonitorUpdateStatus::InProgress
+		} else {
+			ChannelMonitorUpdateStatus::Completed
+		}),
+	];
+
 	macro_rules! make_node {
 		($node_id: expr, $fee_estimator: expr) => {{
 			let logger: Arc<dyn Logger> =
@@ -725,7 +747,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				logger.clone(),
 				$fee_estimator.clone(),
 				Arc::new(TestPersister {
-					update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed),
+					update_ret: Mutex::new(mon_style[$node_id as usize].borrow().clone()),
 				}),
 				Arc::clone(&keys_manager),
 			));
@@ -762,9 +784,6 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 		}};
 	}
 
-	let default_mon_style = RefCell::new(ChannelMonitorUpdateStatus::Completed);
-	let mon_style = [default_mon_style.clone(), default_mon_style.clone(), default_mon_style];
-
 	let reload_node = |ser: &Vec<u8>,
 	                   node_id: u8,
 	                   old_monitors: &TestChainMonitor,
@@ -860,8 +879,21 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 	};
 
 	let mut channel_txn = Vec::new();
+	macro_rules! complete_all_pending_monitor_updates {
+		($monitor: expr) => {{
+			for (channel_id, state) in $monitor.latest_monitors.lock().unwrap().iter_mut() {
+				for (id, data) in state.pending_monitors.drain(..) {
+					$monitor.chain_monitor.channel_monitor_updated(*channel_id, id).unwrap();
+					if id >= state.persisted_monitor_id {
+						state.persisted_monitor_id = id;
+						state.persisted_monitor = data;
+					}
+				}
+			}
+		}};
+	}
 	macro_rules! make_channel {
-		($source: expr, $dest: expr, $dest_keys_manager: expr, $chan_id: expr) => {{
+		($source: expr, $dest: expr, $source_monitor: expr, $dest_monitor: expr, $dest_keys_manager: expr, $chan_id: expr) => {{
 			let init_dest = Init {
 				features: $dest.init_features(),
 				networks: None,
@@ -965,12 +997,14 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 				}
 			};
 			$dest.handle_funding_created($source.get_our_node_id(), &funding_created);
+			// Complete any pending monitor updates for dest after watch_channel
+			complete_all_pending_monitor_updates!($dest_monitor);
 
-			let funding_signed = {
+			let (funding_signed, channel_id) = {
 				let events = $dest.get_and_clear_pending_msg_events();
 				assert_eq!(events.len(), 1);
 				if let MessageSendEvent::SendFundingSigned { ref msg, .. } = events[0] {
-					msg.clone()
+					(msg.clone(), msg.channel_id.clone())
 				} else {
 					panic!("Wrong event type");
 				}
@@ -984,19 +1018,22 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			}
 
 			$source.handle_funding_signed($dest.get_our_node_id(), &funding_signed);
+			// Complete any pending monitor updates for source after watch_channel
+			complete_all_pending_monitor_updates!($source_monitor);
+
 			let events = $source.get_and_clear_pending_events();
 			assert_eq!(events.len(), 1);
-			let channel_id = if let events::Event::ChannelPending {
+			if let events::Event::ChannelPending {
 				ref counterparty_node_id,
-				ref channel_id,
+				channel_id: ref event_channel_id,
 				..
 			} = events[0]
 			{
 				assert_eq!(counterparty_node_id, &$dest.get_our_node_id());
-				channel_id.clone()
+				assert_eq!(*event_channel_id, channel_id);
 			} else {
 				panic!("Wrong event type");
-			};
+			}
 
 			channel_id
 		}};
@@ -1087,8 +1124,8 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 
 	let mut nodes = [node_a, node_b, node_c];
 
-	let chan_1_id = make_channel!(nodes[0], nodes[1], keys_manager_b, 0);
-	let chan_2_id = make_channel!(nodes[1], nodes[2], keys_manager_c, 1);
+	let chan_1_id = make_channel!(nodes[0], nodes[1], monitor_a, monitor_b, keys_manager_b, 0);
+	let chan_2_id = make_channel!(nodes[1], nodes[2], monitor_b, monitor_c, keys_manager_c, 1);
 
 	for node in nodes.iter() {
 		confirm_txn!(node);
@@ -1124,7 +1161,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 		}};
 	}
 
-	let mut read_pos = 0;
+	let mut read_pos = 1; // First byte was consumed for initial mon_style
 	macro_rules! get_slice {
 		($len: expr) => {{
 			let slice_len = $len as usize;