Block the mon update removing a preimage until upstream mon writes

When we forward a payment and receive an `update_fulfill_htlc`
message from the downstream channel, we immediately claim the HTLC
on the upstream channel, before even doing a `commitment_signed`
dance on the downstream channel. This implies that our
`ChannelMonitorUpdate`s "go out" in the right order - first we
ensure we'll get our money by writing the preimage down, then we
write the update that resolves giving money on the downstream node.

This is safe as long as `ChannelMonitorUpdate`s complete in the
order in which they are generated, but of course looking forward we
want to support asynchronous updates, which may complete in any
order.

Thus, here, we enforce the correct ordering by blocking the
downstream `ChannelMonitorUpdate` until the upstream one completes.
Like the `PaymentSent` event handling we do so only for the
`revoke_and_ack` `ChannelMonitorUpdate`, ensuring the
preimage-containing upstream update has a full RTT to complete
before we actually manage to slow anything down.

Author: TheBlueMatt

lightning/src/ln/chanmon_update_fail_tests.rs

@@ -3021,18 +3021,27 @@ fn test_blocked_chan_preimage_release() {
 	check_added_monitors(&nodes[1], 1); // We generate only a preimage monitor update
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
-	// Finish the CS dance between nodes[0] and nodes[1].
-	commitment_signed_dance!(nodes[1], nodes[0], as_htlc_fulfill_updates.commitment_signed, false);
+	// Finish the CS dance between nodes[0] and nodes[1]. Note that until the final RAA CS is held
+	// until the full set of `ChannelMonitorUpdate`s on the nodes[1] <-> nodes[2] channel are
+	// complete, while the preimage that we care about ensuring is on disk did make it there above,
+	// the holding logic doesn't care about the type of update, it just cares that there is one.
+	nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_htlc_fulfill_updates.commitment_signed);
+	check_added_monitors(&nodes[1], 1);
+	let (a, raa) = do_main_commitment_signed_dance(&nodes[1], &nodes[0], false);
+	assert!(a.is_none());
+
+	nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &raa);
 	check_added_monitors(&nodes[1], 0);
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
 	let events = nodes[1].node.get_and_clear_pending_events();
 	assert_eq!(events.len(), 3);
 	if let Event::PaymentSent { .. } = events[0] {} else { panic!(); }
 	if let Event::PaymentPathSuccessful { .. } = events[2] {} else { panic!(); }
 	if let Event::PaymentForwarded { .. } = events[1] {} else { panic!(); }
 
-	// The event processing should release the last RAA update.
-	check_added_monitors(&nodes[1], 1);
+	// The event processing should release the last RAA updates on both channels.
+	check_added_monitors(&nodes[1], 2);
 
 	// When we fetch the next update the message getter will generate the next update for nodes[2],
 	// generating a further monitor update.
@@ -3043,3 +3052,123 @@ fn test_blocked_chan_preimage_release() {
 	commitment_signed_dance!(nodes[2], nodes[1], bs_htlc_fulfill_updates.commitment_signed, false);
 	expect_payment_sent(&nodes[2], payment_preimage_2, None, true, true);
 }
+
+fn do_test_inverted_mon_completion_order(complete_bc_commitment_dance: bool) {
+	// When we forward a payment and receive an `update_fulfill_htlc` message from the downstream
+	// channel, we immediately claim the HTLC on the upstream channel, before even doing a
+	// `commitment_signed` dance on the downstream channel. This implies that our
+	// `ChannelMonitorUpdate`s "go out" in the right order - first we ensure we'll get our money,
+	// then we write the update that resolves giving money on the downstream node. This is safe as
+	// long as `ChannelMonitorUpdate`s complete in the order in which they are generated, but of
+	// course this may not be the case. For asynchronous update writes, we have to ensure monitor
+	// updates can block each other, preventing the inversion all together.
+	let chanmon_cfgs = create_chanmon_cfgs(3);
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+
+	let persister;
+	let new_chain_monitor;
+	let nodes_1_deserialized;
+
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	let chan_id_ab = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+	let chan_id_bc = create_announced_chan_between_nodes(&nodes, 1, 2).2;
+
+	// Route a payment from A, through B, to C, then claim it on C. Once we pass B the
+	// `update_fulfill_htlc` we have a monitor update for both of B's channels. We complete the one
+	// on the B<->C channel but leave the A<->B monitor update pending, then reload B.
+	let (payment_preimage, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 100_000);
+
+	let mon_ab = get_monitor!(nodes[1], chan_id_ab).encode();
+
+	nodes[2].node.claim_funds(payment_preimage);
+	check_added_monitors(&nodes[2], 1);
+	expect_payment_claimed!(nodes[2], payment_hash, 100_000);
+
+	chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+	let cs_updates = get_htlc_update_msgs(&nodes[2], &nodes[1].node.get_our_node_id());
+	nodes[1].node.handle_update_fulfill_htlc(&nodes[2].node.get_our_node_id(), &cs_updates.update_fulfill_htlcs[0]);
+
+	// B generates a new monitor update for the A <-> B channel, but doesn't send the new messages
+	// for it since the monitor update is marked in-progress.
+	check_added_monitors(&nodes[1], 1);
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+	// Now step the Commitment Signed Dance between B and C forward a bit (or fully), ensuring we
+	// won't get the preimage when the nodes reconnect, at which point we have to ensure we get it
+	// from the ChannelMonitor.
+	nodes[1].node.handle_commitment_signed(&nodes[2].node.get_our_node_id(), &cs_updates.commitment_signed);
+	check_added_monitors(&nodes[1], 1);
+	if complete_bc_commitment_dance {
+		let (bs_revoke_and_ack, bs_commitment_signed) = get_revoke_commit_msgs!(nodes[1], nodes[2].node.get_our_node_id());
+		nodes[2].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_revoke_and_ack);
+		check_added_monitors(&nodes[2], 1);
+		nodes[2].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_commitment_signed);
+		check_added_monitors(&nodes[2], 1);
+		let cs_raa = get_event_msg!(nodes[2], MessageSendEvent::SendRevokeAndACK, nodes[1].node.get_our_node_id());
+
+		// At this point node B still hasn't persisted the `ChannelMonitorUpdate` with the
+		// preimage in the A <-> B channel, which will prevent it from persisting the
+		// `ChannelMonitorUpdate` here to avoid "losing" the preimage.
+		nodes[1].node.handle_revoke_and_ack(&nodes[2].node.get_our_node_id(), &cs_raa);
+		check_added_monitors(&nodes[1], 0);
+		assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+	}
+
+	// Now reload node B
+	let manager_b = nodes[1].node.encode();
+
+	let mon_bc = get_monitor!(nodes[1], chan_id_bc).encode();
+	reload_node!(nodes[1], &manager_b, &[&mon_ab, &mon_bc], persister, new_chain_monitor, nodes_1_deserialized);
+
+	nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
+	nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
+
+	// If we used the latest ChannelManager to reload from, we should have both channels still
+	// live. The B <-> C channel's final RAA ChannelMonitorUpdate must still be blocked as
+	// before - the ChannelMonitorUpdate for the A <-> B channel hasn't completed.
+	// When we call `timer_tick_occurred` we will get that monitor update back, which we'll
+	// complete after reconnecting to our peers.
+	persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress);
+	nodes[1].node.timer_tick_occurred();
+	check_added_monitors(&nodes[1], 1);
+	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+
+	// Now reconnect B to both A and C. If the B <-> C commitment signed dance wasn't run to
+	// the end go ahead and do that, though the -2 in `reconnect_nodes` indicates that we
+	// expect to *not* receive the final RAA ChannelMonitorUpdate.
+	if complete_bc_commitment_dance {
+		reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
+	} else {
+		reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, -2), (0, 0), (0, 0), (0, 0), (0, 0), (false, true));
+	}
+
+	reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
+
+	// (Finally) complete the A <-> B ChannelMonitorUpdate, ensuring the preimage is durably on
+	// disk in the proper ChannelMonitor, unblocking the B <-> C ChannelMonitor updating
+	// process.
+	let (outpoint, _, ab_update_id) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_id_ab).unwrap().clone();
+	nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(outpoint, ab_update_id).unwrap();
+
+	// When we fetch B's HTLC update messages here (now that the ChannelMonitorUpdate has
+	// completed), it will also release the final RAA ChannelMonitorUpdate on the B <-> C
+	// channel.
+	let bs_updates = get_htlc_update_msgs(&nodes[1], &nodes[0].node.get_our_node_id());
+	check_added_monitors(&nodes[1], 1);
+
+	nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_updates.update_fulfill_htlcs[0]);
+	do_commitment_signed_dance(&nodes[0], &nodes[1], &bs_updates.commitment_signed, false, false);
+
+	expect_payment_forwarded!(nodes[1], &nodes[0], &nodes[2], Some(1_000), false, false);
+
+	// Finally, check that the payment was, ultimately, seen as sent by node A.
+	expect_payment_sent(&nodes[0], payment_preimage, None, true, true);
+}
+
+#[test]
+fn test_inverted_mon_completion_order() {
+	do_test_inverted_mon_completion_order(true);
+	do_test_inverted_mon_completion_order(false);
+}

lightning/src/ln/channelmanager.rs

@@ -4504,9 +4504,12 @@ where
 		self.pending_outbound_payments.finalize_claims(sources, &self.pending_events);
 	}
 
-	fn claim_funds_internal(&self, source: HTLCSource, payment_preimage: PaymentPreimage, forwarded_htlc_value_msat: Option<u64>, from_onchain: bool, next_channel_outpoint: OutPoint) {
+	fn claim_funds_internal(&self, source: HTLCSource, payment_preimage: PaymentPreimage, forwarded_htlc_value_msat: Option<u64>, from_onchain: bool, next_channel_counterparty_node_id: Option<PublicKey>, next_channel_outpoint: OutPoint) {
 		match source {
 			HTLCSource::OutboundRoute { session_priv, payment_id, path, .. } => {
+				if let Some(pubkey) = next_channel_counterparty_node_id {
+					debug_assert_eq!(pubkey, path.hops[0].pubkey);
+				}
 				let ev_completion_action = EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
 					channel_funding_outpoint: next_channel_outpoint,
 					counterparty_node_id: path.hops[0].pubkey,
@@ -4533,7 +4536,17 @@ where
 									next_channel_id: Some(next_channel_outpoint.to_channel_id()),
 									outbound_amount_forwarded_msat: forwarded_htlc_value_msat,
 								},
-								downstream_counterparty_and_funding_outpoint: None,
+								downstream_counterparty_and_funding_outpoint:
+									if let Some(node_id) = next_channel_counterparty_node_id {
+										Some((node_id, next_channel_outpoint, completed_blocker))
+									} else {
+										// We can only get `None` here if we are processing a
+										// `ChannelMonitor`-originated event, in which case we
+										// don't care about ensuring we wake the downstream
+										// channel's monitor updating - the channel is already
+										// closed.
+										None
+									},
 							})
 						} else { None }
 					});
@@ -5246,13 +5259,27 @@ where
 			match peer_state.channel_by_id.entry(msg.channel_id) {
 				hash_map::Entry::Occupied(mut chan) => {
 					let res = try_chan_entry!(self, chan.get_mut().update_fulfill_htlc(&msg), chan);
+					if let HTLCSource::PreviousHopData(prev_hop) = &res.0 {
+						peer_state.actions_blocking_raa_monitor_updates.entry(msg.channel_id)
+							.or_insert_with(Vec::new)
+							.push(RAAMonitorUpdateBlockingAction::from_prev_hop_data(&prev_hop));
+					}
+					// Note that we do not need to push an `actions_blocking_raa_monitor_updates`
+					// entry here, even though we *do* need to block the next RAA coming in from
+					// generating a monitor update which we let fly. We do this instead in the
+					// `claim_funds_internal` by attaching a `ReleaseRAAChannelMonitorUpdate`
+					// action to the event generated when we "claim" the sent payment. This is
+					// guaranteed to all complete before we process the RAA even though there is no
+					// lock held through that point as we aren't allowed to see another P2P message
+					// from the counterparty until we return, but `claim_funds_internal` runs
+					// first.
 					funding_txo = chan.get().get_funding_txo().expect("We won't accept a fulfill until funded");
 					res
 				},
 				hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close(format!("Got a message for a channel from the wrong node! No such channel for the passed counterparty_node_id {}", counterparty_node_id), msg.channel_id))
 			}
 		};
-		self.claim_funds_internal(htlc_source, msg.payment_preimage.clone(), Some(forwarded_htlc_value), false, funding_txo);
+		self.claim_funds_internal(htlc_source, msg.payment_preimage.clone(), Some(forwarded_htlc_value), false, Some(*counterparty_node_id), funding_txo);
 		Ok(())
 	}
 
@@ -5459,12 +5486,10 @@ where
 			match peer_state.channel_by_id.entry(msg.channel_id) {
 				hash_map::Entry::Occupied(mut chan) => {
 					let funding_txo = chan.get().get_funding_txo();
-					let mon_update_blocked = self.pending_events.lock().unwrap().iter().any(|(_, action)| {
-						action == &Some(EventCompletionAction::ReleaseRAAChannelMonitorUpdate {
-							channel_funding_outpoint: funding_txo.expect("We won't accept an RAA until funded"),
-							counterparty_node_id: *counterparty_node_id,
-						})
-					});
+					let mon_update_blocked = self.raa_monitor_updates_held(
+						&peer_state.actions_blocking_raa_monitor_updates,
+						chan.get().get_funding_txo().expect("We won't accept an RAA until funded"),
+						*counterparty_node_id);
 					let (htlcs_to_fail, monitor_update_opt) = try_chan_entry!(self,
 						chan.get_mut().revoke_and_ack(&msg, &self.logger, mon_update_blocked), chan);
 					let res = if let Some(monitor_update) = monitor_update_opt {
@@ -5645,7 +5670,7 @@ where
 					MonitorEvent::HTLCEvent(htlc_update) => {
 						if let Some(preimage) = htlc_update.payment_preimage {
 							log_trace!(self.logger, "Claiming HTLC with preimage {} from our monitor", log_bytes!(preimage.0));
-							self.claim_funds_internal(htlc_update.source, preimage, htlc_update.htlc_value_satoshis.map(|v| v * 1000), true, funding_outpoint);
+							self.claim_funds_internal(htlc_update.source, preimage, htlc_update.htlc_value_satoshis.map(|v| v * 1000), true, counterparty_node_id, funding_outpoint);
 						} else {
 							log_trace!(self.logger, "Failing HTLC with hash {} from our monitor", log_bytes!(htlc_update.payment_hash.0));
 							let receiver = HTLCDestination::NextHopChannel { node_id: counterparty_node_id, channel_id: funding_outpoint.to_channel_id() };

lightning/src/ln/functional_test_utils.rs

@@ -74,6 +74,20 @@ pub fn mine_transactions<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, txn: &[&Tra
 	let height = node.best_block_info().1 + 1;
 	confirm_transactions_at(node, txn, height);
 }
+/// Mine a single block containing the given transaction without extra checks which may impact
+/// ChannelManager state.
+pub fn mine_transaction_without_checks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, tx: &Transaction) {
+	let height = node.best_block_info().1 + 1;
+	let mut block = Block {
+		header: BlockHeader { version: 0x20000000, prev_blockhash: node.best_block_hash(), merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 },
+		txdata: Vec::new(),
+	};
+	for _ in 0..*node.network_chan_count.borrow() { // Make sure we don't end up with channels at the same short id by offsetting by chan_count
+		block.txdata.push(Transaction { version: 0, lock_time: PackedLockTime::ZERO, input: Vec::new(), output: Vec::new() });
+	}
+	block.txdata.push((*tx).clone());
+	do_connect_block_without_checks(node, block, false);
+}
 /// Mine the given transaction at the given height, mining blocks as required to build to that
 /// height
 ///
@@ -202,19 +216,19 @@ pub fn connect_blocks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, depth: u32) ->
 	assert!(depth >= 1);
 	for i in 1..depth {
 		let prev_blockhash = block.header.block_hash();
-		do_connect_block(node, block, skip_intermediaries);
+		do_connect_block_with_checks(node, block, skip_intermediaries);
 		block = Block {
 			header: BlockHeader { version: 0x20000000, prev_blockhash, merkle_root: TxMerkleNode::all_zeros(), time: height + i, bits: 42, nonce: 42 },
 			txdata: vec![],
 		};
 	}
 	let hash = block.header.block_hash();
-	do_connect_block(node, block, false);
+	do_connect_block_with_checks(node, block, false);
 	hash
 }
 
 pub fn connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: &Block) {
-	do_connect_block(node, block.clone(), false);
+	do_connect_block_with_checks(node, block.clone(), false);
 }
 
 fn call_claimable_balances<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>) {
@@ -224,8 +238,14 @@ fn call_claimable_balances<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>) {
 	}
 }
 
-fn do_connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool) {
+fn do_connect_block_with_checks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool) {
+	call_claimable_balances(node);
+	do_connect_block_without_checks(node, block, skip_intermediaries);
 	call_claimable_balances(node);
+	node.node.test_process_background_events();
+}
+
+fn do_connect_block_without_checks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool) {
 	let height = node.best_block_info().1 + 1;
 	#[cfg(feature = "std")] {
 		eprintln!("Connecting block using Block Connection Style: {:?}", *node.connect_style.borrow());
@@ -280,8 +300,6 @@ fn do_connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, sk
 			}
 		}
 	}
-	call_claimable_balances(node);
-	node.node.test_process_background_events();
 }
 
 pub fn disconnect_blocks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, count: u32) {
@@ -1644,20 +1662,7 @@ pub fn do_commitment_signed_dance(node_a: &Node<'_, '_, '_>, node_b: &Node<'_, '
 	check_added_monitors!(node_a, 1);
 
 	// If this commitment signed dance was due to a claim, don't check for an RAA monitor update.
-	let got_claim = node_a.node.pending_events.lock().unwrap().iter().any(|(ev, action)| {
-		let matching_action = if let Some(channelmanager::EventCompletionAction::ReleaseRAAChannelMonitorUpdate
-			{ channel_funding_outpoint, counterparty_node_id }) = action
-		{
-			if channel_funding_outpoint.to_channel_id() == commitment_signed.channel_id {
-				assert_eq!(*counterparty_node_id, node_b.node.get_our_node_id());
-				true
-			} else { false }
-		} else { false };
-		if matching_action {
-			if let Event::PaymentSent { .. } = ev {} else { panic!(); }
-		}
-		matching_action
-	});
+	let got_claim = node_a.node.test_raa_monitor_updates_held(node_b.node.get_our_node_id(), commitment_signed.channel_id);
 	if fail_backwards { assert!(!got_claim); }
 	commitment_signed_dance!(node_a, node_b, (), fail_backwards, true, false, got_claim);
 
@@ -2964,7 +2969,7 @@ pub fn reconnect_nodes<'a, 'b, 'c>(node_a: &Node<'a, 'b, 'c>, node_b: &Node<'a,
 		}
 		if pending_htlc_adds.1 != 0 || pending_htlc_claims.1 != 0 || pending_htlc_fails.1 != 0 || pending_cell_htlc_claims.1 != 0 || pending_cell_htlc_fails.1 != 0 {
 			let commitment_update = chan_msgs.2.unwrap();
-			if pending_htlc_adds.1 != -1 { // We use -1 to denote a response commitment_signed
+			if pending_htlc_adds.1 > 0 { // We use -1/-2 to denote a response commitment_signed
 				assert_eq!(commitment_update.update_add_htlcs.len(), pending_htlc_adds.1 as usize);
 			}
 			assert_eq!(commitment_update.update_fulfill_htlcs.len(), pending_htlc_claims.1 + pending_cell_htlc_claims.1);
@@ -2980,7 +2985,7 @@ pub fn reconnect_nodes<'a, 'b, 'c>(node_a: &Node<'a, 'b, 'c>, node_b: &Node<'a,
 				node_b.node.handle_update_fail_htlc(&node_a.node.get_our_node_id(), &update_fail);
 			}
 
-			if pending_htlc_adds.1 != -1 { // We use -1 to denote a response commitment_signed
+			if pending_htlc_adds.1 >= 0 { // We use -1/-2 to denote a response commitment_signed
 				commitment_signed_dance!(node_b, node_a, commitment_update.commitment_signed, false);
 			} else {
 				node_b.node.handle_commitment_signed(&node_a.node.get_our_node_id(), &commitment_update.commitment_signed);
@@ -2989,7 +2994,7 @@ pub fn reconnect_nodes<'a, 'b, 'c>(node_a: &Node<'a, 'b, 'c>, node_b: &Node<'a,
 				// No commitment_signed so get_event_msg's assert(len == 1) passes
 				node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &bs_revoke_and_ack);
 				assert!(node_a.node.get_and_clear_pending_msg_events().is_empty());
-				check_added_monitors!(node_a, 1);
+				check_added_monitors(node_a, if pending_htlc_adds.1 == -1 { 1 } else { 0 });
 			}
 		} else {
 			assert!(chan_msgs.2.is_none());