Fail HTLCs from late counterparty commitment updates after funding spend

diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs
index 215b48e..396ee27 100644
--- a/lightning/src/chain/chainmonitor.rs
+++ b/lightning/src/chain/chainmonitor.rs
@@ -1279,15 +1279,17 @@ where
 	/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
 	pub fn flush(&self, count: usize, logger: &L) {
 		let _guard = self.flush_lock.lock().unwrap();
-		if count > 0 {
-			log_info!(logger, "Flushing up to {} monitor operations", count);
+		if count == 0 {
+			return;
 		}
+		log_info!(logger, "Flushing up to {} monitor operations", count);
 		for _ in 0..count {
 			let mut queue = self.pending_ops.lock().unwrap();
 			let op = match queue.pop_front() {
 				Some(op) => op,
 				None => {
 					debug_assert!(false, "flush count exceeded queue length");
+					log_error!(logger, "flush count exceeded queue length");
 					return;
 				},
 			};
@@ -1341,6 +1343,10 @@ where
 				},
 			}
 		}
+
+		// A flushed monitor update may have generated new events, so assume we have
+		// some and wake the event processor.
+		self.event_notifier.notify();
 	}
 }
 
diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs
index 4b8fdd6..8e7b603 100644
--- a/lightning/src/chain/channelmonitor.rs
+++ b/lightning/src/chain/channelmonitor.rs
@@ -1378,8 +1378,8 @@ pub(crate) struct ChannelMonitorImpl<Signer: EcdsaChannelSigner> {
 	/// In-memory only HTLC ids used to track upstream HTLCs that have been failed backwards due to
 	/// a downstream channel force-close remaining unconfirmed by the time the upstream timeout
 	/// expires. This is used to tell us we already generated an event to fail this HTLC back
-	/// during a previous block scan.
-	failed_back_htlc_ids: HashSet<SentHTLCId>,
+	/// during a previous block scan. Not serialized.
+	pub(crate) failed_back_htlc_ids: HashSet<SentHTLCId>,
 
 	// The auxiliary HTLC data associated with a holder commitment transaction. This includes
 	// non-dust HTLC sources, along with dust HTLCs and their sources. Note that this assumes any
@@ -4299,6 +4299,55 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 
 		self.latest_update_id = updates.update_id;
 
+		// If a counterparty commitment update was applied while the funding output has already
+		// been spent on-chain, fail back the outbound HTLCs from the update. This handles the
+		// race where a monitor update is dispatched before the channel force-closes but only
+		// applied after the commitment transaction confirms.
+		for update in updates.updates.iter() {
+			match update {
+				ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo {
+					htlc_outputs, ..
+				} => {
+					// Only outbound HTLCs have a source; inbound ones are `None`
+					// and skipped by the `filter_map`.
+					self.fail_htlcs_from_update_after_funding_spend(
+						htlc_outputs.iter().filter_map(|(htlc, source)| {
+							source.as_ref().map(|s| (&**s, htlc.payment_hash, htlc.amount_msat))
+						}),
+						logger,
+					);
+				},
+				ChannelMonitorUpdateStep::LatestCounterpartyCommitment {
+					commitment_txs, htlc_data,
+				} => {
+					// On a counterparty commitment, `offered=false` means offered by
+					// us (outbound). `nondust_htlc_sources` contains sources only for
+					// these outbound nondust HTLCs, matching the filter order.
+					debug_assert_eq!(
+						commitment_txs[0].nondust_htlcs().iter()
+							.filter(|htlc| !htlc.offered).count(),
+						htlc_data.nondust_htlc_sources.len(),
+					);
+					let nondust = commitment_txs[0]
+						.nondust_htlcs()
+						.iter()
+						.filter(|htlc| !htlc.offered)
+						.zip(htlc_data.nondust_htlc_sources.iter())
+						.map(|(htlc, source)| (source, htlc.payment_hash, htlc.amount_msat));
+					// Only outbound dust HTLCs have a source; inbound ones are `None`
+					// and skipped by the `filter_map`.
+					let dust = htlc_data.dust_htlcs.iter().filter_map(|(htlc, source)| {
+						source.as_ref().map(|s| (s, htlc.payment_hash, htlc.amount_msat))
+					});
+					self.fail_htlcs_from_update_after_funding_spend(
+						nondust.chain(dust),
+						logger,
+					);
+				},
+				_ => {},
+			}
+		}
+
 		// Refuse updates after we've detected a spend onchain (or if the channel was otherwise
 		// closed), but only if the update isn't the kind of update we expect to see after channel
 		// closure.
@@ -4345,6 +4394,121 @@ impl<Signer: EcdsaChannelSigner> ChannelMonitorImpl<Signer> {
 		self.funding_spend_seen || self.lockdown_from_offchain || self.holder_tx_signed
 	}
 
+	/// Given outbound HTLCs from a counterparty commitment update, checks if the funding output
+	/// has been spent on-chain. If so, creates `OnchainEvent::HTLCUpdate` entries to fail back
+	/// HTLCs that weren't already known to the monitor.
+	///
+	/// This handles the race where a `ChannelMonitorUpdate` with a new counterparty commitment
+	/// is dispatched (e.g., via deferred writes) before the channel force-closes, but only
+	/// applied to the in-memory monitor after the commitment transaction has already confirmed.
+	///
+	/// Only truly new HTLCs (not present in any previously-known commitment) need to be failed
+	/// here. HTLCs that were already tracked by the monitor will be handled by the existing
+	/// `fail_unbroadcast_htlcs` logic when the spending transaction confirms.
+	fn fail_htlcs_from_update_after_funding_spend<'a, L: Logger>(
+		&mut self, htlcs: impl Iterator<Item = (&'a HTLCSource, PaymentHash, u64)>,
+		logger: &WithContext<L>,
+	) {
+		let pending_spend_entry = self
+			.onchain_events_awaiting_threshold_conf
+			.iter()
+			.find(|event| matches!(event.event, OnchainEvent::FundingSpendConfirmation { .. }))
+			.map(|entry| (entry.txid, entry.transaction.clone(), entry.height, entry.block_hash));
+		if self.funding_spend_confirmed.is_none() && pending_spend_entry.is_none() {
+			return;
+		}
+
+		// Check HTLC sources against all previously-known commitments to find truly new
+		// ones. After the update has been applied, `prev_counterparty_commitment_txid` holds
+		// what was `current` before this update, so it represents the already-known
+		// counterparty state. HTLCs already present in any of these will be handled by
+		// `fail_unbroadcast_htlcs` when the spending transaction confirms.
+		let is_source_known = |source: &HTLCSource| {
+			if let Some(ref txid) = self.funding.prev_counterparty_commitment_txid {
+				if let Some(htlc_list) = self.funding.counterparty_claimable_outpoints.get(txid) {
+					if htlc_list.iter().any(|(_, s)| s.as_ref().map(|s| s.as_ref()) == Some(source))
+					{
+						return true;
+					}
+				}
+			}
+			// Note that we don't care about the case where a counterparty sent us a fresh local commitment transaction
+			// post-closure (with the `ChannelManager` still operating the channel). First of all we only care about
+			// resolving outbound HTLCs, which fundamentally have to be initiated by us. However we also don't mind
+			// looking at the current holder commitment transaction's HTLCs as any fresh outbound HTLCs will have to
+			// first come in a locally-initiated update to the counterparty's commitment transaction which we can, by
+			// refusing to apply the update, prevent the counterparty from ever seeing (as no messages can be sent until
+			// the monitor is updated). Thus, the HTLCs we care about can never appear in the holder commitment
+			// transaction.
+			if holder_commitment_htlcs!(self, CURRENT_WITH_SOURCES).any(|(_, s)| s == Some(source))
+			{
+				return true;
+			}
+			if let Some(mut iter) = holder_commitment_htlcs!(self, PREV_WITH_SOURCES) {
+				if iter.any(|(_, s)| s == Some(source)) {
+					return true;
+				}
+			}
+			false
+		};
+		for (source, payment_hash, amount_msat) in htlcs {
+			if is_source_known(source) {
+				continue;
+			}
+			if self.counterparty_fulfilled_htlcs.get(&SentHTLCId::from_source(source)).is_some() {
+				continue;
+			}
+			let htlc_value_satoshis = Some(amount_msat / 1000);
+			let logger = WithContext::from(logger, None, None, Some(payment_hash));
+			// Defensively mark the HTLC as failed back so the expiry-based failure
+			// path in `block_connected` doesn't generate a duplicate `HTLCUpdate`
+			// event for the same source.
+			self.failed_back_htlc_ids.insert(SentHTLCId::from_source(source));
+			if let Some(confirmed_txid) = self.funding_spend_confirmed {
+				// Funding spend already confirmed past ANTI_REORG_DELAY: resolve immediately.
+				log_trace!(
+					logger,
+					"Failing HTLC from late counterparty commitment update immediately \
+					(funding spend already confirmed)"
+				);
+				self.pending_monitor_events.push(MonitorEvent::HTLCEvent(HTLCUpdate {
+					payment_hash,
+					payment_preimage: None,
+					source: source.clone(),
+					htlc_value_satoshis,
+				}));
+				self.htlcs_resolved_on_chain.push(IrrevocablyResolvedHTLC {
+					commitment_tx_output_idx: None,
+					resolving_txid: Some(confirmed_txid),
+					resolving_tx: None,
+					payment_preimage: None,
+				});
+			} else {
+				// Funding spend still awaiting ANTI_REORG_DELAY: queue the failure.
+				let (txid, transaction, height, block_hash) = pending_spend_entry.clone().unwrap();
+				let entry = OnchainEventEntry {
+					txid,
+					transaction,
+					height,
+					block_hash,
+					event: OnchainEvent::HTLCUpdate {
+						source: source.clone(),
+						payment_hash,
+						htlc_value_satoshis,
+						commitment_tx_output_idx: None,
+					},
+				};
+				log_trace!(
+					logger,
+					"Failing HTLC from late counterparty commitment update, \
+					waiting for confirmation (at height {})",
+					entry.confirmation_threshold()
+				);
+				self.onchain_events_awaiting_threshold_conf.push(entry);
+			}
+		}
+	}
+
 	fn get_latest_update_id(&self) -> u64 {
 		self.latest_update_id
 	}
@@ -6834,7 +6998,7 @@ mod tests {
 	use bitcoin::{Sequence, Witness};
 
 	use crate::chain::chaininterface::LowerBoundedFeeEstimator;
-	use crate::events::ClosureReason;
+	use crate::events::{ClosureReason, Event};
 
 	use super::ChannelMonitorUpdateStep;
 	use crate::chain::channelmonitor::{ChannelMonitor, WithChannelMonitor};
@@ -6957,8 +7121,21 @@ mod tests {
 		check_spends!(htlc_txn[1], broadcast_tx);
 
 		check_closed_broadcast(&nodes[1], 1, true);
-		check_closed_event(&nodes[1], 1, ClosureReason::CommitmentTxConfirmed, &[nodes[0].node.get_our_node_id()], 100000);
-		check_added_monitors(&nodes[1], 1);
+		if !use_local_txn {
+			// When the counterparty commitment confirms, FundingSpendConfirmation matures
+			// immediately (no CSV delay), so funding_spend_confirmed is set. The new payment's
+			// commitment update then triggers immediate HTLC failure, generating payment events
+			// alongside the channel close event.
+			let events = nodes[1].node.get_and_clear_pending_events();
+			assert_eq!(events.len(), 3);
+			assert!(events.iter().any(|e| matches!(e, Event::PaymentPathFailed { .. })));
+			assert!(events.iter().any(|e| matches!(e, Event::PaymentFailed { .. })));
+			assert!(events.iter().any(|e| matches!(e, Event::ChannelClosed { .. })));
+			check_added_monitors(&nodes[1], 2);
+		} else {
+			check_closed_event(&nodes[1], 1, ClosureReason::CommitmentTxConfirmed, &[nodes[0].node.get_our_node_id()], 100000);
+			check_added_monitors(&nodes[1], 1);
+		}
 	}
 
 	#[test]
diff --git a/lightning/src/ln/chanmon_update_fail_tests.rs b/lightning/src/ln/chanmon_update_fail_tests.rs
index a92af3e..623d028 100644
--- a/lightning/src/ln/chanmon_update_fail_tests.rs
+++ b/lightning/src/ln/chanmon_update_fail_tests.rs
@@ -48,6 +48,7 @@ use crate::util::test_utils;
 use crate::prelude::*;
 use crate::sync::{Arc, Mutex};
 use bitcoin::hashes::Hash;
+use core::sync::atomic::Ordering;
 
 #[test]
 fn test_monitor_and_persister_update_fail() {
@@ -5171,3 +5172,246 @@ fn test_mpp_claim_to_holding_cell() {
 	expect_payment_claimable!(nodes[3], paymnt_hash_2, payment_secret_2, 400_000);
 	claim_payment(&nodes[2], &[&nodes[3]], preimage_2);
 }
+
+fn do_test_late_counterparty_commitment_update_after_funding_spend(fully_confirmed: bool) {
+	// Tests that when a ChannelMonitorUpdate containing a new counterparty commitment (with an
+	// outbound HTLC) is applied to a monitor that has already seen the funding output spent
+	// on-chain, the HTLC is properly failed back.
+	//
+	// This exercises the race condition where:
+	// 1. A sends an HTLC to B, creating a monitor update with LatestCounterpartyCommitmentTXInfo
+	// 2. In deferred-write mode, this update is queued but not applied to the in-memory monitor
+	// 3. B's commitment transaction (without the HTLC) is broadcast and confirmed
+	// 4. The queued update is flushed, applying the counterparty commitment to the monitor
+	// 5. The monitor detects the funding spend and fails the HTLC
+	//
+	// When `fully_confirmed` is true, ANTI_REORG_DELAY has fully passed before the flush, so
+	// funding_spend_confirmed is set. Otherwise, the FundingSpendConfirmation entry is still
+	// pending in onchain_events_awaiting_threshold_conf.
+
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs_deferred(2, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let node_b_id = nodes[1].node.get_our_node_id();
+
+	let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+
+	// Get B's commitment transaction before any HTLCs are added. This is the transaction that
+	// will be mined on-chain, simulating B broadcasting while A's monitor update is pending.
+	let bs_commitment_tx = get_local_commitment_txn!(nodes[1], chan_id);
+	assert_eq!(bs_commitment_tx.len(), 1);
+
+	// Pause auto-flush on A so that the monitor update from send_payment is queued but NOT
+	// applied to the in-memory monitor.
+	nodes[0].chain_monitor.pause_flush.store(true, Ordering::Release);
+
+	// Send a payment from A to B. The ChannelManager creates a LatestCounterpartyCommitmentTXInfo
+	// monitor update, but in deferred mode with pause_flush it remains queued.
+	let (route, payment_hash, _, payment_secret) =
+		get_route_and_payment_hash!(nodes[0], nodes[1], 1_000_000);
+	let payment_id = PaymentId(payment_hash.0);
+	nodes[0]
+		.node
+		.send_payment_with_route(
+			route,
+			payment_hash,
+			RecipientOnionFields::secret_only(payment_secret, 1_000_000),
+			payment_id,
+		)
+		.unwrap();
+	check_added_monitors(&nodes[0], 1);
+
+	// Mine B's (old) commitment transaction on A and advance blocks. When fully_confirmed,
+	// advance past ANTI_REORG_DELAY so FundingSpendConfirmation is consumed and
+	// funding_spend_confirmed is set. Otherwise, stop one block short so the entry remains
+	// in onchain_events_awaiting_threshold_conf.
+	mine_transaction(&nodes[0], &bs_commitment_tx[0]);
+	let extra_blocks = if fully_confirmed { ANTI_REORG_DELAY - 1 } else { ANTI_REORG_DELAY - 2 };
+	connect_blocks(&nodes[0], extra_blocks);
+
+	if fully_confirmed {
+		// The channel close event, error message, and ChannelForceClosed monitor update were
+		// generated during block connection. Consume them before flushing.
+		check_closed_event(
+			&nodes[0],
+			1,
+			ClosureReason::CommitmentTxConfirmed,
+			&[node_b_id],
+			100000,
+		);
+		check_closed_broadcast(&nodes[0], 1, true);
+		check_added_monitors(&nodes[0], 1);
+	}
+
+	// Flush the queued monitor updates. This applies the LatestCounterpartyCommitmentTXInfo
+	// (and ChannelForceClosed) to the monitor, which triggers fail_htlcs_from_update_after_
+	// funding_spend to create OnchainEvent::HTLCUpdate entries for the HTLC.
+	nodes[0].chain_monitor.pause_flush.store(false, Ordering::Release);
+	let pending_count = nodes[0].chain_monitor.chain_monitor.pending_operation_count();
+	nodes[0].chain_monitor.chain_monitor.flush(pending_count, &nodes[0].logger);
+
+	if !fully_confirmed {
+		// The channel close event, error message, and ChannelForceClosed monitor update were
+		// generated during block connection.
+		check_closed_event(
+			&nodes[0],
+			1,
+			ClosureReason::CommitmentTxConfirmed,
+			&[node_b_id],
+			100000,
+		);
+		check_closed_broadcast(&nodes[0], 1, true);
+		check_added_monitors(&nodes[0], 1);
+	}
+
+	// Advance ANTI_REORG_DELAY blocks so the OnchainEvent::HTLCUpdate entries (created at
+	// best_block.height during the flush) mature into MonitorEvent::HTLCEvent.
+	connect_blocks(&nodes[0], ANTI_REORG_DELAY);
+
+	// The ChannelManager processes the MonitorEvent::HTLCEvent and fails the payment.
+	expect_payment_failed_conditions(
+		&nodes[0],
+		payment_hash,
+		false,
+		PaymentFailedConditions::new(),
+	);
+	// The payment failure generates a ReleasePaymentComplete monitor update.
+	check_added_monitors(&nodes[0], 1);
+}
+
+#[test]
+fn test_late_counterparty_commitment_update_after_funding_spend() {
+	do_test_late_counterparty_commitment_update_after_funding_spend(false);
+}
+
+#[test]
+fn test_late_counterparty_commitment_update_after_funding_spend_fully_confirmed() {
+	do_test_late_counterparty_commitment_update_after_funding_spend(true);
+}
+
+fn do_test_late_counterparty_commitment_update_after_holder_commitment_spend(dust: bool) {
+	// Tests that when the confirmed spending transaction is a holder commitment, HTLCs that
+	// have non-dust outputs in the holder commitment are NOT failed by
+	// fail_htlcs_from_update_after_funding_spend (they'll be resolved on-chain via
+	// HTLC-timeout), while HTLCs only present in the late counterparty commitment update ARE
+	// failed.
+	//
+	// When `dust` is true, HTLC Y is a dust amount, verifying that dust HTLCs in late
+	// counterparty commitment updates are also correctly failed.
+	//
+	// Setup:
+	// 1. Route HTLC X from A to B (fully committed in both holder and counterparty commitments)
+	// 2. Grab A's holder commitment (which contains HTLC X)
+	// 3. Pause flush, then send HTLC Y from A to B (counterparty commitment update is queued)
+	// 4. Mine A's holder commitment (contains X but not Y)
+	// 5. Flush the queued update (contains both X and Y)
+	// 6. Verify: X is not failed by our code (on-chain output), Y is failed
+	// 7. Drive HTLC X to resolution via the on-chain HTLC-timeout path
+
+	let chanmon_cfgs = create_chanmon_cfgs(2);
+	let node_cfgs = create_node_cfgs_deferred(2, &chanmon_cfgs);
+	// Use legacy (non-anchor) channels so that the HTLC-timeout transaction is broadcast
+	// directly by the monitor rather than going through the BumpTransaction event path.
+	let legacy_cfg = test_legacy_channel_config();
+	let node_chanmgrs =
+		create_node_chanmgrs(2, &node_cfgs, &[Some(legacy_cfg.clone()), Some(legacy_cfg)]);
+	let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+
+	let node_b_id = nodes[1].node.get_our_node_id();
+
+	let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+
+	// Route HTLC X fully (committed in both commitments).
+	let (_, payment_hash_x, ..) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000);
+
+	// Get A's holder commitment which now contains HTLC X. For legacy (non-anchor) channels,
+	// the HTLC-timeout transaction is also returned.
+	let as_txn = get_local_commitment_txn!(nodes[0], chan_id);
+	let as_commitment_tx = &as_txn[0];
+	// Verify HTLC X is present as a non-dust output (commitment has HTLC-timeout tx too).
+	assert!(as_txn.len() >= 2, "Expected commitment + HTLC-timeout tx, got {}", as_txn.len());
+
+	// Pause flush so the next monitor update is queued.
+	nodes[0].chain_monitor.pause_flush.store(true, Ordering::Release);
+
+	// Send HTLC Y. When `dust` is true, 1000 msat (1 sat) is well below the dust limit and
+	// will not appear as an output in any commitment transaction. When false, 2_000_000 msat
+	// is non-dust. Either way, the LatestCounterpartyCommitmentTXInfo update (containing both
+	// X and Y) is queued in deferred mode.
+	let htlc_y_amount = if dust { 1_000 } else { 2_000_000 };
+	let (route_y, payment_hash_y, _, payment_secret_y) =
+		get_route_and_payment_hash!(nodes[0], nodes[1], htlc_y_amount);
+	let payment_id_y = PaymentId(payment_hash_y.0);
+	nodes[0]
+		.node
+		.send_payment_with_route(
+			route_y,
+			payment_hash_y,
+			RecipientOnionFields::secret_only(payment_secret_y, htlc_y_amount),
+			payment_id_y,
+		)
+		.unwrap();
+	check_added_monitors(&nodes[0], 1);
+
+	// Mine A's holder commitment (contains X but not Y).
+	mine_transaction(&nodes[0], as_commitment_tx);
+	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 2);
+
+	// Flush the queued monitor updates.
+	nodes[0].chain_monitor.pause_flush.store(false, Ordering::Release);
+	let pending_count = nodes[0].chain_monitor.chain_monitor.pending_operation_count();
+	nodes[0].chain_monitor.chain_monitor.flush(pending_count, &nodes[0].logger);
+
+	check_closed_event(&nodes[0], 1, ClosureReason::CommitmentTxConfirmed, &[node_b_id], 100000);
+	check_closed_broadcast(&nodes[0], 1, true);
+	check_added_monitors(&nodes[0], 1);
+
+	// Advance ANTI_REORG_DELAY blocks so OnchainEvent::HTLCUpdate entries mature.
+	connect_blocks(&nodes[0], ANTI_REORG_DELAY);
+
+	// HTLC Y should be failed by our code. HTLC X has an on-chain output in the holder
+	// commitment and will be resolved via the HTLC-timeout path.
+	expect_payment_failed_conditions(
+		&nodes[0],
+		payment_hash_y,
+		false,
+		PaymentFailedConditions::new(),
+	);
+	check_added_monitors(&nodes[0], 1);
+
+	// Verify HTLC X was NOT failed (no payment failure event for it at this point).
+	assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
+
+	// Drive HTLC X to resolution via the on-chain HTLC-timeout path. Connect blocks until we
+	// pass the CLTV expiry so the monitor broadcasts the HTLC-timeout transaction.
+	connect_blocks(&nodes[0], TEST_FINAL_CLTV);
+	let as_htlc_timeout_claim =
+		nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
+	assert_eq!(as_htlc_timeout_claim.len(), 1);
+	check_spends!(as_htlc_timeout_claim[0], as_commitment_tx);
+
+	// Mine the HTLC-timeout transaction and wait for ANTI_REORG_DELAY.
+	mine_transaction(&nodes[0], &as_htlc_timeout_claim[0]);
+	connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
+
+	// HTLC X should now be resolved on-chain.
+	expect_payment_failed_conditions(
+		&nodes[0],
+		payment_hash_x,
+		false,
+		PaymentFailedConditions::new(),
+	);
+	check_added_monitors(&nodes[0], 1);
+}
+
+#[test]
+fn test_late_counterparty_commitment_update_after_holder_commitment_spend() {
+	do_test_late_counterparty_commitment_update_after_holder_commitment_spend(false);
+}
+
+#[test]
+fn test_late_counterparty_commitment_update_after_holder_commitment_spend_dust() {
+	do_test_late_counterparty_commitment_update_after_holder_commitment_spend(true);
+}
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
index 3541c82..d31c16c 100644
--- a/lightning/src/util/test_utils.rs
+++ b/lightning/src/util/test_utils.rs
@@ -518,6 +518,10 @@ pub struct TestChainMonitor<'a> {
 	pub expect_monitor_round_trip_fail: Mutex<Option<ChannelId>>,
 	#[cfg(feature = "std")]
 	pub write_blocker: Mutex<Option<std::sync::mpsc::Receiver<()>>>,
+	/// When set to `true`, `release_pending_monitor_events` will not auto-flush pending
+	/// deferred operations. This allows tests to control exactly when queued monitor updates
+	/// are applied to the in-memory monitor.
+	pub pause_flush: AtomicBool,
 }
 impl<'a> TestChainMonitor<'a> {
 	pub fn new(
@@ -577,6 +581,7 @@ impl<'a> TestChainMonitor<'a> {
 			expect_monitor_round_trip_fail: Mutex::new(None),
 			#[cfg(feature = "std")]
 			write_blocker: Mutex::new(None),
+			pause_flush: AtomicBool::new(false),
 		}
 	}
 
@@ -701,12 +706,18 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		)
 		.unwrap()
 		.1;
+		// failed_back_htlc_ids is an in-memory-only dedup guard that is intentionally not
+		// serialized. Copy it to the deserialized monitor for the comparison, then clear
+		// it so it doesn't leak into the rest of the test.
+		let failed_back = monitor.inner.lock().unwrap().failed_back_htlc_ids.clone();
+		new_monitor.inner.lock().unwrap().failed_back_htlc_ids = failed_back;
 		if let Some(chan_id) = self.expect_monitor_round_trip_fail.lock().unwrap().take() {
 			assert_eq!(chan_id, channel_id);
 			assert!(new_monitor != *monitor);
 		} else {
 			assert!(new_monitor == *monitor);
 		}
+		new_monitor.inner.lock().unwrap().failed_back_htlc_ids.clear();
 		self.added_monitors.lock().unwrap().push((channel_id, new_monitor));
 		update_res
 	}
@@ -718,8 +729,10 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		// completion events. When not in deferred mode the queue is empty so this only
 		// costs a lock acquisition. It ensures standard test helpers (route_payment, etc.)
 		// work with deferred chain monitors.
-		let count = self.chain_monitor.pending_operation_count();
-		self.chain_monitor.flush(count, &self.logger);
+		if !self.pause_flush.load(Ordering::Acquire) {
+			let count = self.chain_monitor.pending_operation_count();
+			self.chain_monitor.flush(count, &self.logger);
+		}
 		return self.chain_monitor.release_pending_monitor_events();
 	}
 }