Extract chanmon harness peer links

diff --git a/fuzz/src/chanmon_consistency.rs b/fuzz/src/chanmon_consistency.rs
index 3a44cff..7030f4e 100644
--- a/fuzz/src/chanmon_consistency.rs
+++ b/fuzz/src/chanmon_consistency.rs
@@ -1429,6 +1429,20 @@ impl EventQueues {
 		}
 	}
 
+	fn clear_link(&mut self, link: &PeerLink) {
+		match (link.node_a, link.node_b) {
+			(0, 1) | (1, 0) => {
+				self.ab.clear();
+				self.ba.clear();
+			},
+			(1, 2) | (2, 1) => {
+				self.bc.clear();
+				self.cb.clear();
+			},
+			_ => panic!("unsupported link"),
+		}
+	}
+
 	fn drain_on_disconnect(&mut self, edge_node: usize, nodes: &[HarnessNode<'_>; 3]) {
 		match edge_node {
 			0 => {
@@ -1474,6 +1488,109 @@ impl EventQueues {
 	}
 }
 
+struct PeerLink {
+	node_a: usize,
+	node_b: usize,
+	channel_ids: [ChannelId; 3],
+	disconnected: bool,
+}
+
+impl PeerLink {
+	fn new(node_a: usize, node_b: usize, channel_ids: [ChannelId; 3]) -> Self {
+		Self { node_a, node_b, channel_ids, disconnected: false }
+	}
+
+	fn first_channel_id(&self) -> ChannelId {
+		self.channel_ids[0]
+	}
+
+	fn channel_ids(&self) -> &[ChannelId; 3] {
+		&self.channel_ids
+	}
+
+	fn complete_all_monitor_updates(&self, nodes: &[HarnessNode<'_>; 3]) {
+		for id in &self.channel_ids {
+			nodes[self.node_a].complete_all_monitor_updates(id);
+			nodes[self.node_b].complete_all_monitor_updates(id);
+		}
+	}
+
+	fn complete_monitor_updates_for_node(
+		&self, node_idx: usize, nodes: &[HarnessNode<'_>; 3], selector: MonitorUpdateSelector,
+	) {
+		assert!(node_idx == self.node_a || node_idx == self.node_b);
+		for id in &self.channel_ids {
+			nodes[node_idx].complete_monitor_update(id, selector);
+		}
+	}
+
+	fn disconnect(&mut self, nodes: &[HarnessNode<'_>; 3], queues: &mut EventQueues) {
+		if self.disconnected {
+			return;
+		}
+		let node_a_id = nodes[self.node_a].get_our_node_id();
+		let node_b_id = nodes[self.node_b].get_our_node_id();
+		nodes[self.node_a].peer_disconnected(node_b_id);
+		nodes[self.node_b].peer_disconnected(node_a_id);
+		self.disconnected = true;
+		let edge_node = if self.node_a == 1 {
+			self.node_b
+		} else if self.node_b == 1 {
+			self.node_a
+		} else {
+			panic!("unsupported link topology")
+		};
+		queues.drain_on_disconnect(edge_node, nodes);
+		queues.clear_link(self);
+	}
+
+	fn reconnect(&mut self, nodes: &[HarnessNode<'_>; 3]) {
+		if !self.disconnected {
+			return;
+		}
+		let node_a_id = nodes[self.node_a].get_our_node_id();
+		let node_b_id = nodes[self.node_b].get_our_node_id();
+		let init_b = Init {
+			features: nodes[self.node_b].init_features(),
+			networks: None,
+			remote_network_address: None,
+		};
+		nodes[self.node_a].peer_connected(node_b_id, &init_b, true).unwrap();
+		let init_a = Init {
+			features: nodes[self.node_a].init_features(),
+			networks: None,
+			remote_network_address: None,
+		};
+		nodes[self.node_b].peer_connected(node_a_id, &init_a, false).unwrap();
+		self.disconnected = false;
+	}
+
+	fn disconnect_for_reload(
+		&mut self, restarted_node: usize, nodes: &[HarnessNode<'_>; 3], queues: &mut EventQueues,
+	) {
+		if self.disconnected {
+			return;
+		}
+		assert!(restarted_node == self.node_a || restarted_node == self.node_b);
+
+		let remaining_node = if restarted_node == self.node_a { self.node_b } else { self.node_a };
+		let restarted_node_id = nodes[restarted_node].get_our_node_id();
+		nodes[remaining_node].peer_disconnected(restarted_node_id);
+		self.disconnected = true;
+
+		if remaining_node == 1 {
+			queues.route_from_middle(
+				nodes[1].get_and_clear_pending_msg_events(),
+				Some(restarted_node),
+				nodes,
+			);
+		} else {
+			nodes[remaining_node].get_and_clear_pending_msg_events();
+		}
+		queues.clear_link(self);
+	}
+}
+
 fn build_node_config(chan_type: ChanType) -> UserConfig {
 	let mut config = UserConfig::default();
 	config.channel_config.forwarding_fee_proportional_millionths = 0;
@@ -1836,14 +1953,14 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 		let node_c_chans = nodes[2].list_usable_channels();
 		[node_c_chans[0].channel_id, node_c_chans[1].channel_id, node_c_chans[2].channel_id]
 	};
+	let mut ab_link = PeerLink::new(0, 1, chan_ab_ids);
+	let mut bc_link = PeerLink::new(1, 2, chan_bc_ids);
 	// Keep old names for backward compatibility in existing code
-	let chan_a_id = chan_ab_ids[0];
-	let chan_b_id = chan_bc_ids[0];
+	let chan_a_id = ab_link.first_channel_id();
+	let chan_b_id = bc_link.first_channel_id();
 
 	let mut p_ctr: u64 = 0;
 
-	let mut peers_ab_disconnected = false;
-	let mut peers_bc_disconnected = false;
 	let mut queues = EventQueues::new();
 
 	for node in &mut nodes {
@@ -2347,80 +2464,30 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 			0x06 => nodes[2].set_persistence_style(ChannelMonitorUpdateStatus::Completed),
 
 			0x08 => {
-				for id in &chan_ab_ids {
+				for id in ab_link.channel_ids() {
 					nodes[0].complete_all_monitor_updates(id);
 				}
 			},
 			0x09 => {
-				for id in &chan_ab_ids {
+				for id in ab_link.channel_ids() {
 					nodes[1].complete_all_monitor_updates(id);
 				}
 			},
 			0x0a => {
-				for id in &chan_bc_ids {
+				for id in bc_link.channel_ids() {
 					nodes[1].complete_all_monitor_updates(id);
 				}
 			},
 			0x0b => {
-				for id in &chan_bc_ids {
+				for id in bc_link.channel_ids() {
 					nodes[2].complete_all_monitor_updates(id);
 				}
 			},
 
-			0x0c => {
-				if !peers_ab_disconnected {
-					nodes[0].peer_disconnected(nodes[1].get_our_node_id());
-					nodes[1].peer_disconnected(nodes[0].get_our_node_id());
-					peers_ab_disconnected = true;
-					queues.drain_on_disconnect(0, &nodes);
-					queues.ab.clear();
-					queues.ba.clear();
-				}
-			},
-			0x0d => {
-				if !peers_bc_disconnected {
-					nodes[1].peer_disconnected(nodes[2].get_our_node_id());
-					nodes[2].peer_disconnected(nodes[1].get_our_node_id());
-					peers_bc_disconnected = true;
-					queues.drain_on_disconnect(2, &nodes);
-					queues.bc.clear();
-					queues.cb.clear();
-				}
-			},
-			0x0e => {
-				if peers_ab_disconnected {
-					let init_1 = Init {
-						features: nodes[1].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[0].peer_connected(nodes[1].get_our_node_id(), &init_1, true).unwrap();
-					let init_0 = Init {
-						features: nodes[0].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[1].peer_connected(nodes[0].get_our_node_id(), &init_0, false).unwrap();
-					peers_ab_disconnected = false;
-				}
-			},
-			0x0f => {
-				if peers_bc_disconnected {
-					let init_2 = Init {
-						features: nodes[2].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[1].peer_connected(nodes[2].get_our_node_id(), &init_2, true).unwrap();
-					let init_1 = Init {
-						features: nodes[1].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[2].peer_connected(nodes[1].get_our_node_id(), &init_1, false).unwrap();
-					peers_bc_disconnected = false;
-				}
-			},
+			0x0c => ab_link.disconnect(&nodes, &mut queues),
+			0x0d => bc_link.disconnect(&nodes, &mut queues),
+			0x0e => ab_link.reconnect(&nodes),
+			0x0f => bc_link.reconnect(&nodes),
 
 			0x10 => process_msg_noret!(0, true, ProcessMessages::AllMessages),
 			0x11 => process_msg_noret!(0, false, ProcessMessages::AllMessages),
@@ -2618,52 +2685,20 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 			0xb0 | 0xb1 | 0xb2 => {
 				// Restart node A, picking among the in-flight `ChannelMonitor`s to use based on
 				// the value of `v` we're matching.
-				if !peers_ab_disconnected {
-					nodes[1].peer_disconnected(nodes[0].get_our_node_id());
-					peers_ab_disconnected = true;
-					queues.route_from_middle(
-						nodes[1].get_and_clear_pending_msg_events(),
-						Some(0),
-						&nodes,
-					);
-					queues.ab.clear();
-					queues.ba.clear();
-				}
+				ab_link.disconnect_for_reload(0, &nodes, &mut queues);
 				nodes[0].reload(v, &out, &router, chan_type);
 			},
 			0xb3..=0xbb => {
 				// Restart node B, picking among the in-flight `ChannelMonitor`s to use based on
 				// the value of `v` we're matching.
-				if !peers_ab_disconnected {
-					nodes[0].peer_disconnected(nodes[1].get_our_node_id());
-					peers_ab_disconnected = true;
-					nodes[0].get_and_clear_pending_msg_events();
-					queues.ab.clear();
-					queues.ba.clear();
-				}
-				if !peers_bc_disconnected {
-					nodes[2].peer_disconnected(nodes[1].get_our_node_id());
-					peers_bc_disconnected = true;
-					nodes[2].get_and_clear_pending_msg_events();
-					queues.bc.clear();
-					queues.cb.clear();
-				}
+				ab_link.disconnect_for_reload(1, &nodes, &mut queues);
+				bc_link.disconnect_for_reload(1, &nodes, &mut queues);
 				nodes[1].reload(v, &out, &router, chan_type);
 			},
 			0xbc | 0xbd | 0xbe => {
 				// Restart node C, picking among the in-flight `ChannelMonitor`s to use based on
 				// the value of `v` we're matching.
-				if !peers_bc_disconnected {
-					nodes[1].peer_disconnected(nodes[2].get_our_node_id());
-					peers_bc_disconnected = true;
-					queues.route_from_middle(
-						nodes[1].get_and_clear_pending_msg_events(),
-						Some(2),
-						&nodes,
-					);
-					queues.bc.clear();
-					queues.cb.clear();
-				}
+				bc_link.disconnect_for_reload(2, &nodes, &mut queues);
 				nodes[2].reload(v, &out, &router, chan_type);
 			},
 
@@ -2734,67 +2769,43 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 			},
 
 			0xf0 => {
-				for id in &chan_ab_ids {
-					nodes[0].complete_monitor_update(id, MonitorUpdateSelector::First);
-				}
+				ab_link.complete_monitor_updates_for_node(0, &nodes, MonitorUpdateSelector::First)
 			},
 			0xf1 => {
-				for id in &chan_ab_ids {
-					nodes[0].complete_monitor_update(id, MonitorUpdateSelector::Second);
-				}
+				ab_link.complete_monitor_updates_for_node(0, &nodes, MonitorUpdateSelector::Second)
 			},
 			0xf2 => {
-				for id in &chan_ab_ids {
-					nodes[0].complete_monitor_update(id, MonitorUpdateSelector::Last);
-				}
+				ab_link.complete_monitor_updates_for_node(0, &nodes, MonitorUpdateSelector::Last)
 			},
 
 			0xf4 => {
-				for id in &chan_ab_ids {
-					nodes[1].complete_monitor_update(id, MonitorUpdateSelector::First);
-				}
+				ab_link.complete_monitor_updates_for_node(1, &nodes, MonitorUpdateSelector::First)
 			},
 			0xf5 => {
-				for id in &chan_ab_ids {
-					nodes[1].complete_monitor_update(id, MonitorUpdateSelector::Second);
-				}
+				ab_link.complete_monitor_updates_for_node(1, &nodes, MonitorUpdateSelector::Second)
 			},
 			0xf6 => {
-				for id in &chan_ab_ids {
-					nodes[1].complete_monitor_update(id, MonitorUpdateSelector::Last);
-				}
+				ab_link.complete_monitor_updates_for_node(1, &nodes, MonitorUpdateSelector::Last)
 			},
 
 			0xf8 => {
-				for id in &chan_bc_ids {
-					nodes[1].complete_monitor_update(id, MonitorUpdateSelector::First);
-				}
+				bc_link.complete_monitor_updates_for_node(1, &nodes, MonitorUpdateSelector::First)
 			},
 			0xf9 => {
-				for id in &chan_bc_ids {
-					nodes[1].complete_monitor_update(id, MonitorUpdateSelector::Second);
-				}
+				bc_link.complete_monitor_updates_for_node(1, &nodes, MonitorUpdateSelector::Second)
 			},
 			0xfa => {
-				for id in &chan_bc_ids {
-					nodes[1].complete_monitor_update(id, MonitorUpdateSelector::Last);
-				}
+				bc_link.complete_monitor_updates_for_node(1, &nodes, MonitorUpdateSelector::Last)
 			},
 
 			0xfc => {
-				for id in &chan_bc_ids {
-					nodes[2].complete_monitor_update(id, MonitorUpdateSelector::First);
-				}
+				bc_link.complete_monitor_updates_for_node(2, &nodes, MonitorUpdateSelector::First)
 			},
 			0xfd => {
-				for id in &chan_bc_ids {
-					nodes[2].complete_monitor_update(id, MonitorUpdateSelector::Second);
-				}
+				bc_link.complete_monitor_updates_for_node(2, &nodes, MonitorUpdateSelector::Second)
 			},
 			0xfe => {
-				for id in &chan_bc_ids {
-					nodes[2].complete_monitor_update(id, MonitorUpdateSelector::Last);
-				}
+				bc_link.complete_monitor_updates_for_node(2, &nodes, MonitorUpdateSelector::Last)
 			},
 
 			0xff => {
@@ -2802,36 +2813,8 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 				// after we resolve all pending events.
 
 				// First, make sure peers are all connected to each other
-				if peers_ab_disconnected {
-					let init_1 = Init {
-						features: nodes[1].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[0].peer_connected(nodes[1].get_our_node_id(), &init_1, true).unwrap();
-					let init_0 = Init {
-						features: nodes[0].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[1].peer_connected(nodes[0].get_our_node_id(), &init_0, false).unwrap();
-					peers_ab_disconnected = false;
-				}
-				if peers_bc_disconnected {
-					let init_2 = Init {
-						features: nodes[2].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[1].peer_connected(nodes[2].get_our_node_id(), &init_2, true).unwrap();
-					let init_1 = Init {
-						features: nodes[1].init_features(),
-						networks: None,
-						remote_network_address: None,
-					};
-					nodes[2].peer_connected(nodes[1].get_our_node_id(), &init_1, false).unwrap();
-					peers_bc_disconnected = false;
-				}
+				ab_link.reconnect(&nodes);
+				bc_link.reconnect(&nodes);
 
 				for op in SUPPORTED_SIGNER_OPS {
 					nodes[0].keys_manager.enable_op_for_all_signers(op);
@@ -2850,14 +2833,8 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 								panic!("It may take may iterations to settle the state, but it should not take forever");
 							}
 							// Next, make sure no monitor updates are pending
-							for id in &chan_ab_ids {
-								nodes[0].complete_all_monitor_updates(id);
-								nodes[1].complete_all_monitor_updates(id);
-							}
-							for id in &chan_bc_ids {
-								nodes[1].complete_all_monitor_updates(id);
-								nodes[2].complete_all_monitor_updates(id);
-							}
+							ab_link.complete_all_monitor_updates(&nodes);
+							bc_link.complete_all_monitor_updates(&nodes);
 							// Then, make sure any current forwards make their way to their destination
 							if process_msg_events!(0, false, ProcessMessages::AllMessages) {
 								last_pass_no_updates = false;
@@ -2934,13 +2911,13 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 				}
 
 				// Finally, make sure that at least one end of each channel can make a substantial payment
-				for &chan_id in &chan_ab_ids {
+				for &chan_id in ab_link.channel_ids() {
 					assert!(
 						send(0, 1, chan_id, 10_000_000, &mut p_ctr)
 							|| send(1, 0, chan_id, 10_000_000, &mut p_ctr)
 					);
 				}
-				for &chan_id in &chan_bc_ids {
+				for &chan_id in bc_link.channel_ids() {
 					assert!(
 						send(1, 2, chan_id, 10_000_000, &mut p_ctr)
 							|| send(2, 1, chan_id, 10_000_000, &mut p_ctr)