Improve chanmon_consistency fuzz target performance

fuzz/src/chanmon_consistency.rs

@@ -244,16 +244,14 @@ struct LatestMonitorState {
 	/// which we haven't yet completed. We're allowed to reload with those as well, at least until
 	/// they're completed.
 	persisted_monitor_id: u64,
-	/// The latest serialized `ChannelMonitor` that we told LDK we persisted.
-	persisted_monitor: Vec<u8>,
-	/// A set of (monitor id, serialized `ChannelMonitor`)s which we're currently "persisting",
+	/// The latest `ChannelMonitor` that we told LDK we persisted.
+	persisted_monitor: channelmonitor::ChannelMonitor<TestChannelSigner>,
+	/// A set of (monitor id, `ChannelMonitor`)s which we're currently "persisting",
 	/// from LDK's perspective.
-	pending_monitors: Vec<(u64, Vec<u8>)>,
+	pending_monitors: Vec<(u64, channelmonitor::ChannelMonitor<TestChannelSigner>)>,
 }
 
 struct TestChainMonitor {
-	pub logger: Arc<dyn Logger>,
-	pub keys: Arc<KeyProvider>,
 	pub persister: Arc<TestPersister>,
 	pub chain_monitor: Arc<
 		chainmonitor::ChainMonitor<
@@ -277,15 +275,13 @@ impl TestChainMonitor {
 			chain_monitor: Arc::new(chainmonitor::ChainMonitor::new(
 				None,
 				broadcaster,
-				logger.clone(),
+				logger,
 				feeest,
 				Arc::clone(&persister),
 				Arc::clone(&keys),
 				keys.get_peer_storage_key(),
 				false,
 			)),
-			logger,
-			keys,
 			persister,
 			latest_monitors: Mutex::new(new_hash_map()),
 		}
@@ -295,20 +291,22 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 	fn watch_channel(
 		&self, channel_id: ChannelId, monitor: channelmonitor::ChannelMonitor<TestChannelSigner>,
 	) -> Result<chain::ChannelMonitorUpdateStatus, ()> {
-		let mut ser = VecWriter(Vec::new());
-		monitor.write(&mut ser).unwrap();
 		let monitor_id = monitor.get_latest_update_id();
 		let res = self.chain_monitor.watch_channel(channel_id, monitor);
+		let mon = self.persister.take_latest_monitor(&channel_id);
 		let state = match res {
 			Ok(chain::ChannelMonitorUpdateStatus::Completed) => LatestMonitorState {
 				persisted_monitor_id: monitor_id,
-				persisted_monitor: ser.0,
+				persisted_monitor: mon,
 				pending_monitors: Vec::new(),
 			},
-			Ok(chain::ChannelMonitorUpdateStatus::InProgress) => LatestMonitorState {
-				persisted_monitor_id: monitor_id,
-				persisted_monitor: Vec::new(),
-				pending_monitors: vec![(monitor_id, ser.0)],
+			Ok(chain::ChannelMonitorUpdateStatus::InProgress) => {
+				let persisted_monitor = mon.clone();
+				LatestMonitorState {
+					persisted_monitor_id: monitor_id,
+					persisted_monitor,
+					pending_monitors: vec![(monitor_id, mon)],
+				}
 			},
 			Ok(chain::ChannelMonitorUpdateStatus::UnrecoverableError) => panic!(),
 			Err(()) => panic!(),
@@ -324,37 +322,15 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 	) -> chain::ChannelMonitorUpdateStatus {
 		let mut map_lock = self.latest_monitors.lock().unwrap();
 		let map_entry = map_lock.get_mut(&channel_id).expect("Didn't have monitor on update call");
-		let latest_monitor_data = map_entry
-			.pending_monitors
-			.last()
-			.as_ref()
-			.map(|(_, data)| data)
-			.unwrap_or(&map_entry.persisted_monitor);
-		let deserialized_monitor =
-			<(BlockHash, channelmonitor::ChannelMonitor<TestChannelSigner>)>::read(
-				&mut &latest_monitor_data[..],
-				(&*self.keys, &*self.keys),
-			)
-			.unwrap()
-			.1;
-		deserialized_monitor
-			.update_monitor(
-				update,
-				&&TestBroadcaster { txn_broadcasted: RefCell::new(Vec::new()) },
-				&&FuzzEstimator { ret_val: atomic::AtomicU32::new(253) },
-				&self.logger,
-			)
-			.unwrap();
-		let mut ser = VecWriter(Vec::new());
-		deserialized_monitor.write(&mut ser).unwrap();
 		let res = self.chain_monitor.update_channel(channel_id, update);
+		let mon = self.persister.take_latest_monitor(&channel_id);
 		match res {
 			chain::ChannelMonitorUpdateStatus::Completed => {
 				map_entry.persisted_monitor_id = update.update_id;
-				map_entry.persisted_monitor = ser.0;
+				map_entry.persisted_monitor = mon;
 			},
 			chain::ChannelMonitorUpdateStatus::InProgress => {
-				map_entry.pending_monitors.push((update.update_id, ser.0));
+				map_entry.pending_monitors.push((update.update_id, mon));
 			},
 			chain::ChannelMonitorUpdateStatus::UnrecoverableError => panic!(),
 		}
@@ -864,9 +840,8 @@ fn assert_action_timeout_awaiting_response(action: &msgs::ErrorAction) {
 
 #[inline]
 pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
-	data: &[u8], underlying_out: Out, anchors: bool,
+	data: &[u8], out: Out, anchors: bool,
 ) {
-	let out = SearchingOutput::new(underlying_out);
 	let broadcast_a = Arc::new(TestBroadcaster { txn_broadcasted: RefCell::new(Vec::new()) });
 	let broadcast_b = Arc::new(TestBroadcaster { txn_broadcasted: RefCell::new(Vec::new()) });
 	let broadcast_c = Arc::new(TestBroadcaster { txn_broadcasted: RefCell::new(Vec::new()) });
@@ -915,9 +890,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 				$broadcaster.clone(),
 				logger.clone(),
 				$fee_estimator.clone(),
-				Arc::new(TestPersister {
-					update_ret: Mutex::new(mon_style[$node_id as usize].borrow().clone()),
-				}),
+				Arc::new(TestPersister::new(mon_style[$node_id as usize].borrow().clone())),
 				Arc::clone(&keys_manager),
 			));
 
@@ -967,9 +940,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 			broadcaster.clone(),
 			logger.clone(),
 			Arc::clone(fee_estimator),
-			Arc::new(TestPersister {
-				update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed),
-			}),
+			Arc::new(TestPersister::new(ChannelMonitorUpdateStatus::Completed)),
 			Arc::clone(keys),
 		));
 
@@ -984,7 +955,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 		let mut monitors = new_hash_map();
 		let mut old_monitors = old_monitors.latest_monitors.lock().unwrap();
 		for (channel_id, mut prev_state) in old_monitors.drain() {
-			let (mon_id, serialized_mon) = if use_old_mons % 3 == 0 {
+			let (mon_id, mon) = if use_old_mons % 3 == 0 {
 				// Reload with the oldest `ChannelMonitor` (the one that we already told
 				// `ChannelManager` we finished persisting).
 				(prev_state.persisted_monitor_id, prev_state.persisted_monitor)
@@ -1000,14 +971,17 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 			// Use a different value of `use_old_mons` if we have another monitor (only for node B)
 			// by shifting `use_old_mons` one in base-3.
 			use_old_mons /= 3;
-			let mon = <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
-				&mut &serialized_mon[..],
+			// Serialize and deserialize the monitor to verify round-trip correctness.
+			let mut ser = VecWriter(Vec::new());
+			mon.write(&mut ser).unwrap();
+			let (_, deserialized_mon) = <(BlockHash, ChannelMonitor<TestChannelSigner>)>::read(
+				&mut &ser.0[..],
 				(&**keys, &**keys),
 			)
 			.expect("Failed to read monitor");
-			monitors.insert(channel_id, mon.1);
+			monitors.insert(channel_id, deserialized_mon);
 			// Update the latest `ChannelMonitor` state to match what we just told LDK.
-			prev_state.persisted_monitor = serialized_mon;
+			prev_state.persisted_monitor = mon;
 			prev_state.persisted_monitor_id = mon_id;
 			// Wipe any `ChannelMonitor`s which we never told LDK we finished persisting,
 			// considering them discarded. LDK should replay these for us as they're stored in
@@ -1793,11 +1767,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 							// Can be generated as a result of calling `timer_tick_occurred` enough
 							// times while peers are disconnected
 						},
-						_ => if out.may_fail.load(atomic::Ordering::Acquire) {
-							return;
-						} else {
-							panic!("Unhandled message event {:?}", event)
-						},
+						_ => panic!("Unhandled message event {:?}", event),
 					}
 					if $limit_events != ProcessMessages::AllMessages {
 						break;
@@ -1837,13 +1807,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 							MessageSendEvent::HandleError { ref action, .. } => {
 								assert_action_timeout_awaiting_response(action);
 							},
-							_ => {
-								if out.may_fail.load(atomic::Ordering::Acquire) {
-									return;
-								} else {
-									panic!("Unhandled message event")
-								}
-							},
+							_ => panic!("Unhandled message event"),
 						}
 					}
 					push_excess_b_events!(
@@ -1865,13 +1829,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 							MessageSendEvent::HandleError { ref action, .. } => {
 								assert_action_timeout_awaiting_response(action);
 							},
-							_ => {
-								if out.may_fail.load(atomic::Ordering::Acquire) {
-									return;
-								} else {
-									panic!("Unhandled message event")
-								}
-							},
+							_ => panic!("Unhandled message event"),
 						}
 					}
 					push_excess_b_events!(
@@ -1980,13 +1938,7 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 						},
 						events::Event::SpliceFailed { .. } => {},
 
-						_ => {
-							if out.may_fail.load(atomic::Ordering::Acquire) {
-								return;
-							} else {
-								panic!("Unhandled event")
-							}
-						},
+						_ => panic!("Unhandled event"),
 					}
 				}
 				while nodes[$node].needs_pending_htlc_processing() {
@@ -2004,10 +1956,11 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 
 		let complete_first = |v: &mut Vec<_>| if !v.is_empty() { Some(v.remove(0)) } else { None };
 		let complete_second = |v: &mut Vec<_>| if v.len() > 1 { Some(v.remove(1)) } else { None };
+		type PendingMonitors = Vec<(u64, channelmonitor::ChannelMonitor<TestChannelSigner>)>;
 		let complete_monitor_update =
 			|monitor: &Arc<TestChainMonitor>,
 			 chan_funding,
-			 compl_selector: &dyn Fn(&mut Vec<(u64, Vec<u8>)>) -> Option<(u64, Vec<u8>)>| {
+			 compl_selector: &dyn Fn(&mut PendingMonitors) -> Option<(u64, channelmonitor::ChannelMonitor<TestChannelSigner>)>| {
 				if let Some(state) = monitor.latest_monitors.lock().unwrap().get_mut(chan_funding) {
 					assert!(
 						state.pending_monitors.windows(2).all(|pair| pair[0].0 < pair[1].0),
@@ -2809,27 +2762,6 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 	}
 }
 
-/// We actually have different behavior based on if a certain log string has been seen, so we have
-/// to do a bit more tracking.
-#[derive(Clone)]
-struct SearchingOutput<O: Output> {
-	output: O,
-	may_fail: Arc<atomic::AtomicBool>,
-}
-impl<O: Output> Output for SearchingOutput<O> {
-	fn locked_write(&self, data: &[u8]) {
-		// We hit a design limitation of LN state machine (see CONCURRENT_INBOUND_HTLC_FEE_BUFFER)
-		if std::str::from_utf8(data).unwrap().contains("Outbound update_fee HTLC buffer overflow - counterparty should force-close this channel") {
-			self.may_fail.store(true, atomic::Ordering::Release);
-		}
-		self.output.locked_write(data)
-	}
-}
-impl<O: Output> SearchingOutput<O> {
-	pub fn new(output: O) -> Self {
-		Self { output, may_fail: Arc::new(atomic::AtomicBool::new(false)) }
-	}
-}
 
 pub fn chanmon_consistency_test<Out: Output + MaybeSend + MaybeSync>(data: &[u8], out: Out) {
 	do_test(data, out.clone(), false);

fuzz/src/full_stack.rs

@@ -598,7 +598,7 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger + MaybeSend + MaybeSync>
 		broadcast.clone(),
 		Arc::clone(&logger),
 		fee_est.clone(),
-		Arc::new(TestPersister { update_ret: Mutex::new(ChannelMonitorUpdateStatus::Completed) }),
+		Arc::new(TestPersister::new(ChannelMonitorUpdateStatus::Completed)),
 		Arc::clone(&keys_manager),
 		keys_manager.get_peer_storage_key(),
 		false,

fuzz/src/utils/test_logger.rs

@@ -8,6 +8,7 @@
 // licenses.
 
 use lightning::util::logger::{Logger, Record};
+use std::any::TypeId;
 use std::io::Write;
 use std::sync::{Arc, Mutex};
 
@@ -66,6 +67,9 @@ impl<'a, Out: Output> Write for LockedWriteAdapter<'a, Out> {
 
 impl<Out: Output> Logger for TestLogger<Out> {
 	fn log(&self, record: Record) {
+		if TypeId::of::<Out>() == TypeId::of::<DevNull>() {
+			return;
+		}
 		writeln!(LockedWriteAdapter(&self.out), "{:<6} {}", self.id, record).unwrap();
 	}
 }

fuzz/src/utils/test_persister.rs

@@ -1,25 +1,45 @@
 use lightning::chain;
 use lightning::chain::{chainmonitor, channelmonitor};
+use lightning::ln::types::ChannelId;
 use lightning::util::persist::MonitorName;
 use lightning::util::test_channel_signer::TestChannelSigner;
 
+use std::collections::HashMap;
 use std::sync::Mutex;
 
 pub struct TestPersister {
 	pub update_ret: Mutex<chain::ChannelMonitorUpdateStatus>,
+	latest_monitors: Mutex<HashMap<ChannelId, channelmonitor::ChannelMonitor<TestChannelSigner>>>,
+}
+impl TestPersister {
+	pub fn new(update_ret: chain::ChannelMonitorUpdateStatus) -> Self {
+		Self {
+			update_ret: Mutex::new(update_ret),
+			latest_monitors: Mutex::new(HashMap::new()),
+		}
+	}
+
+	pub fn take_latest_monitor(
+		&self, channel_id: &ChannelId,
+	) -> channelmonitor::ChannelMonitor<TestChannelSigner> {
+		self.latest_monitors.lock().unwrap().remove(channel_id)
+			.expect("Persister should have monitor for channel")
+	}
 }
 impl chainmonitor::Persist<TestChannelSigner> for TestPersister {
 	fn persist_new_channel(
 		&self, _monitor_name: MonitorName,
-		_data: &channelmonitor::ChannelMonitor<TestChannelSigner>,
+		data: &channelmonitor::ChannelMonitor<TestChannelSigner>,
 	) -> chain::ChannelMonitorUpdateStatus {
+		self.latest_monitors.lock().unwrap().insert(data.channel_id(), data.clone());
 		self.update_ret.lock().unwrap().clone()
 	}
 
 	fn update_persisted_channel(
 		&self, _monitor_name: MonitorName, _update: Option<&channelmonitor::ChannelMonitorUpdate>,
-		_data: &channelmonitor::ChannelMonitor<TestChannelSigner>,
+		data: &channelmonitor::ChannelMonitor<TestChannelSigner>,
 	) -> chain::ChannelMonitorUpdateStatus {
+		self.latest_monitors.lock().unwrap().insert(data.channel_id(), data.clone());
 		self.update_ret.lock().unwrap().clone()
 	}