Time out incomplete MPP payments in chanmon_consistency

diff --git a/fuzz/src/chanmon_consistency.rs b/fuzz/src/chanmon_consistency.rs
index abaa92d..d1f6094 100644
--- a/fuzz/src/chanmon_consistency.rs
+++ b/fuzz/src/chanmon_consistency.rs
@@ -58,7 +58,7 @@ use lightning::ln::channelmanager::{
 use lightning::ln::functional_test_utils::*;
 use lightning::ln::inbound_payment::ExpandedKey;
 use lightning::ln::msgs::{
-	BaseMessageHandler, ChannelMessageHandler, CommitmentUpdate, Init, MessageSendEvent,
+	self, BaseMessageHandler, ChannelMessageHandler, CommitmentUpdate, Init, MessageSendEvent,
 	UpdateAddHTLC,
 };
 use lightning::ln::outbound_payment::RecipientOnionFields;
@@ -843,6 +843,17 @@ fn send_mpp_hop_payment(
 	}
 }
 
+#[inline]
+fn assert_action_timeout_awaiting_response(action: &msgs::ErrorAction) {
+	// Since sending/receiving messages may be delayed, `timer_tick_occurred` may cause a node to
+	// disconnect their counterparty if they're expecting a timely response.
+	assert!(matches!(
+		action,
+		msgs::ErrorAction::DisconnectPeerWithWarning { msg }
+		if msg.data.contains("Disconnecting due to timeout awaiting response")
+	));
+}
+
 #[inline]
 pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 	data: &[u8], underlying_out: Out, anchors: bool,
@@ -1424,8 +1435,12 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 						},
 						MessageSendEvent::SendChannelReady { .. } => continue,
 						MessageSendEvent::SendAnnouncementSignatures { .. } => continue,
-						MessageSendEvent::SendChannelUpdate { ref node_id, ref msg } => {
-							assert_eq!(msg.contents.channel_flags & 2, 0); // The disable bit must never be set!
+						MessageSendEvent::SendChannelUpdate { ref node_id, .. } => {
+							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
+							*node_id == a_id
+						},
+						MessageSendEvent::HandleError { ref action, ref node_id } => {
+							assert_action_timeout_awaiting_response(action);
 							if Some(*node_id) == expect_drop_id { panic!("peer_disconnected should drop msgs bound for the disconnected peer"); }
 							*node_id == a_id
 						},
@@ -1638,20 +1653,21 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 								}
 							}
 						},
+						MessageSendEvent::HandleError { ref action, .. } => {
+							assert_action_timeout_awaiting_response(action);
+						},
 						MessageSendEvent::SendChannelReady { .. } => {
 							// Can be generated as a reestablish response
 						},
 						MessageSendEvent::SendAnnouncementSignatures { .. } => {
 							// Can be generated as a reestablish response
 						},
-						MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
-							// When we reconnect we will resend a channel_update to make sure our
-							// counterparty has the latest parameters for receiving payments
-							// through us. We do, however, check that the message does not include
-							// the "disabled" bit, as we should never ever have a channel which is
-							// disabled when we send such an update (or it may indicate channel
-							// force-close which we should detect as an error).
-							assert_eq!(msg.contents.channel_flags & 2, 0);
+						MessageSendEvent::SendChannelUpdate { .. } => {
+							// Can be generated as a reestablish response
+						},
+						MessageSendEvent::BroadcastChannelUpdate { .. } => {
+							// 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;
@@ -1693,8 +1709,9 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 							MessageSendEvent::SendStfu { .. } => {},
 							MessageSendEvent::SendChannelReady { .. } => {},
 							MessageSendEvent::SendAnnouncementSignatures { .. } => {},
-							MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
-								assert_eq!(msg.contents.channel_flags & 2, 0); // The disable bit must never be set!
+							MessageSendEvent::SendChannelUpdate { .. } => {},
+							MessageSendEvent::HandleError { ref action, .. } => {
+								assert_action_timeout_awaiting_response(action);
 							},
 							_ => {
 								if out.may_fail.load(atomic::Ordering::Acquire) {
@@ -1720,8 +1737,9 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 							MessageSendEvent::SendStfu { .. } => {},
 							MessageSendEvent::SendChannelReady { .. } => {},
 							MessageSendEvent::SendAnnouncementSignatures { .. } => {},
-							MessageSendEvent::SendChannelUpdate { ref msg, .. } => {
-								assert_eq!(msg.contents.channel_flags & 2, 0); // The disable bit must never be set!
+							MessageSendEvent::SendChannelUpdate { .. } => {},
+							MessageSendEvent::HandleError { ref action, .. } => {
+								assert_action_timeout_awaiting_response(action);
 							},
 							_ => {
 								if out.may_fail.load(atomic::Ordering::Acquire) {
@@ -2195,11 +2213,11 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 				if fee_est_a.ret_val.fetch_add(250, atomic::Ordering::AcqRel) + 250 > max_feerate {
 					fee_est_a.ret_val.store(max_feerate, atomic::Ordering::Release);
 				}
-				nodes[0].maybe_update_chan_fees();
+				nodes[0].timer_tick_occurred();
 			},
 			0x81 => {
 				fee_est_a.ret_val.store(253, atomic::Ordering::Release);
-				nodes[0].maybe_update_chan_fees();
+				nodes[0].timer_tick_occurred();
 			},
 
 			0x84 => {
@@ -2210,11 +2228,11 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 				if fee_est_b.ret_val.fetch_add(250, atomic::Ordering::AcqRel) + 250 > max_feerate {
 					fee_est_b.ret_val.store(max_feerate, atomic::Ordering::Release);
 				}
-				nodes[1].maybe_update_chan_fees();
+				nodes[1].timer_tick_occurred();
 			},
 			0x85 => {
 				fee_est_b.ret_val.store(253, atomic::Ordering::Release);
-				nodes[1].maybe_update_chan_fees();
+				nodes[1].timer_tick_occurred();
 			},
 
 			0x88 => {
@@ -2225,11 +2243,11 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 				if fee_est_c.ret_val.fetch_add(250, atomic::Ordering::AcqRel) + 250 > max_feerate {
 					fee_est_c.ret_val.store(max_feerate, atomic::Ordering::Release);
 				}
-				nodes[2].maybe_update_chan_fees();
+				nodes[2].timer_tick_occurred();
 			},
 			0x89 => {
 				fee_est_c.ret_val.store(253, atomic::Ordering::Release);
-				nodes[2].maybe_update_chan_fees();
+				nodes[2].timer_tick_occurred();
 			},
 
 			0xa0 => {
@@ -2798,6 +2816,14 @@ pub fn do_test<Out: Output + MaybeSend + MaybeSync>(
 
 				process_all_events!();
 
+				// Since MPP payments are supported, we wait until we fully settle the state of all
+				// channels to see if we have any committed HTLC parts of an MPP payment that need
+				// to be failed back.
+				for node in &nodes {
+					node.timer_tick_occurred();
+				}
+				process_all_events!();
+
 				// Verify no payments are stuck - all should have resolved
 				for (idx, pending) in pending_payments.borrow().iter().enumerate() {
 					assert!(
diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
index 0f1916a..dd38374 100644
--- a/lightning/src/ln/channel.rs
+++ b/lightning/src/ln/channel.rs
@@ -3030,7 +3030,7 @@ pub(crate) enum QuiescentAction {
 		contribution: FundingContribution,
 		locktime: LockTime,
 	},
-	#[cfg(any(test, fuzzing))]
+	#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 	DoNothing,
 }
 
@@ -3039,7 +3039,7 @@ pub(crate) enum StfuResponse {
 	SpliceInit(msgs::SpliceInit),
 }
 
-#[cfg(any(test, fuzzing))]
+#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 impl_writeable_tlv_based_enum_upgradable!(QuiescentAction,
 	(0, DoNothing) => {},
 	(2, Splice) => {
@@ -3048,7 +3048,7 @@ impl_writeable_tlv_based_enum_upgradable!(QuiescentAction,
 	},
 	{1, LegacySplice} => (),
 );
-#[cfg(not(any(test, fuzzing)))]
+#[cfg(not(any(test, fuzzing, feature = "_test_utils")))]
 impl_writeable_tlv_based_enum_upgradable!(QuiescentAction,
 	(2, Splice) => {
 		(0, contribution, required),
@@ -7066,7 +7066,7 @@ where
 							contributed_outputs: outputs,
 						})
 					},
-					#[cfg(any(test, fuzzing))]
+					#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 					Some(quiescent_action) => {
 						self.quiescent_action = Some(quiescent_action);
 						None
@@ -13569,7 +13569,7 @@ where
 					let splice_init = self.send_splice_init_internal(context, ChangeStrategy::FromCoinSelection);
 					return Ok(Some(StfuResponse::SpliceInit(splice_init)));
 				},
-				#[cfg(any(test, fuzzing))]
+				#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 				Some(QuiescentAction::DoNothing) => {
 					// In quiescence test we want to just hang out here, letting the test manually
 					// leave quiescence.
@@ -13612,7 +13612,7 @@ where
 		Ok(None)
 	}
 
-	#[cfg(any(test, fuzzing))]
+	#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 	#[rustfmt::skip]
 	pub fn exit_quiescence(&mut self) -> bool {
 		// Make sure we either finished the quiescence handshake and are quiescent, or we never
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index 869a431..051bda3 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -55,7 +55,7 @@ use crate::events::{
 };
 use crate::events::{FundingInfo, PaidBolt12Invoice};
 use crate::ln::chan_utils::selected_commitment_sat_per_1000_weight;
-#[cfg(any(test, fuzzing))]
+#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 use crate::ln::channel::QuiescentAction;
 use crate::ln::channel::{
 	self, hold_time_since, Channel, ChannelError, ChannelUpdateStatus, DisconnectResult,
@@ -3047,7 +3047,10 @@ const _CHECK_CLTV_EXPIRY_OFFCHAIN: () = assert!(
 );
 
 /// The number of ticks of [`ChannelManager::timer_tick_occurred`] until expiry of incomplete MPPs
+#[cfg(not(any(fuzzing, test, feature = "_test_utils")))]
 pub(crate) const MPP_TIMEOUT_TICKS: u8 = 3;
+#[cfg(any(fuzzing, test, feature = "_test_utils"))]
+pub(crate) const MPP_TIMEOUT_TICKS: u8 = 1;
 
 /// The number of ticks of [`ChannelManager::timer_tick_occurred`] where a peer is disconnected
 /// until we mark the channel disabled and gossip the update.
@@ -8313,39 +8316,6 @@ impl<
 		NotifyOption::DoPersist
 	}
 
-	#[cfg(any(test, fuzzing, feature = "_externalize_tests"))]
-	/// In chanmon_consistency we want to sometimes do the channel fee updates done in
-	/// timer_tick_occurred, but we can't generate the disabled channel updates as it considers
-	/// these a fuzz failure (as they usually indicate a channel force-close, which is exactly what
-	/// it wants to detect). Thus, we have a variant exposed here for its benefit.
-	#[rustfmt::skip]
-	pub fn maybe_update_chan_fees(&self) {
-		PersistenceNotifierGuard::optionally_notify(self, || {
-			let mut should_persist = NotifyOption::SkipPersistNoEvents;
-			let mut feerate_cache = new_hash_map();
-
-			let per_peer_state = self.per_peer_state.read().unwrap();
-			for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
-				let mut peer_state_lock = peer_state_mutex.lock().unwrap();
-				let peer_state = &mut *peer_state_lock;
-				for (chan_id, chan) in peer_state.channel_by_id.iter_mut()
-					.filter_map(|(chan_id, chan)| chan.as_funded_mut().map(|chan| (chan_id, chan)))
-				{
-					let channel_type = chan.funding.get_channel_type();
-					let new_feerate = feerate_cache.get(channel_type).copied().or_else(|| {
-						let feerate = selected_commitment_sat_per_1000_weight(&self.fee_estimator, &channel_type);
-						feerate_cache.insert(channel_type.clone(), feerate);
-						Some(feerate)
-					}).unwrap();
-					let chan_needs_persist = self.update_channel_fee(chan_id, chan, new_feerate);
-					if chan_needs_persist == NotifyOption::DoPersist { should_persist = NotifyOption::DoPersist; }
-				}
-			}
-
-			should_persist
-		});
-	}
-
 	/// Performs actions which should happen on startup and roughly once per minute thereafter.
 	///
 	/// This currently includes:
@@ -13351,7 +13321,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		}
 	}
 
-	#[cfg(any(test, fuzzing))]
+	#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 	#[rustfmt::skip]
 	pub fn maybe_propose_quiescence(&self, counterparty_node_id: &PublicKey, channel_id: &ChannelId) -> Result<(), APIError> {
 		let mut result = Ok(());
@@ -13408,7 +13378,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 		result
 	}
 
-	#[cfg(any(test, fuzzing))]
+	#[cfg(any(test, fuzzing, feature = "_test_utils"))]
 	#[rustfmt::skip]
 	pub fn exit_quiescence(&self, counterparty_node_id: &PublicKey, channel_id: &ChannelId) -> Result<bool, APIError> {
 		let _read_guard = self.total_consistency_lock.read().unwrap();
diff --git a/lightning/src/ln/update_fee_tests.rs b/lightning/src/ln/update_fee_tests.rs
index 24ae852..423d27b 100644
--- a/lightning/src/ln/update_fee_tests.rs
+++ b/lightning/src/ln/update_fee_tests.rs
@@ -1089,9 +1089,13 @@ pub fn do_cannot_afford_on_holding_cell_release(
 		*feerate_lock = target_feerate;
 	}
 
-	// Put the update fee into the holding cell of node 0
-
-	nodes[0].node.maybe_update_chan_fees();
+	// Put the update fee into the holding cell of node 0. We use quiescence as an easy way to force
+	// the update into the holding cell.
+	nodes[0].node.maybe_propose_quiescence(&node_b_id, &chan_id).unwrap();
+	let stfu = get_event_msg!(nodes[0], MessageSendEvent::SendStfu, node_b_id);
+	nodes[0].node.timer_tick_occurred();
+	assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
+	check_added_monitors(&nodes[0], 0);
 
 	// While the update_fee is in the holding cell, add an inbound HTLC
 
@@ -1132,11 +1136,17 @@ pub fn do_cannot_afford_on_holding_cell_release(
 		panic!();
 	}
 
-	// Release the update_fee from its holding cell
+	// Release the update_fee from its holding cell by completing the quiescence handshake.
+	nodes[1].node.handle_stfu(node_a_id, &stfu);
+	let stfu = get_event_msg!(nodes[1], MessageSendEvent::SendStfu, node_a_id);
+	nodes[0].node.handle_stfu(node_b_id, &stfu);
+	let _ = nodes[0].node.exit_quiescence(&node_b_id, &chan_id);
+	let _ = nodes[1].node.exit_quiescence(&node_a_id, &chan_id);
 	let mut events = nodes[0].node.get_and_clear_pending_msg_events();
 	if can_afford {
 		// We could afford the update_fee, sanity check everything
 		assert_eq!(events.len(), 1);
+		check_added_monitors(&nodes[0], 1);
 		if let MessageSendEvent::UpdateHTLCs { node_id, channel_id, updates } =
 			events.pop().unwrap()
 		{