Implement delay channel closing in ChannelManager until maturation

Add pruning of waiting-conf channel closing at block_disconnect

Fix tests broken by introduced change

Author: Antoine Riard

src/ln/channelmanager.rs

@@ -262,6 +262,12 @@ pub(super) struct ChannelHolder {
 	/// Messages to send to peers - pushed to in the same lock that they are generated in (except
 	/// for broadcast messages, where ordering isn't as strict).
 	pub(super) pending_msg_events: Vec<events::MessageSendEvent>,
+	/// Due to reorgs, we want to be sure that commitment txn has enough depth
+	/// (HTLC_FAIL_ANTI_REORG_DELAY).
+	/// before to close channel. Doing it to early would block us to pass backward
+	/// preimage in case of successful claim on onward channel. Key is block
+	/// height timer, value is array of channels id, waiting their closing.
+	pub(super) channel_closing_waiting_threshold_conf: HashMap<u32, Vec<[u8; 32]>>,
 }
 pub(super) struct MutChannelHolder<'a> {
 	pub(super) by_id: &'a mut HashMap<[u8; 32], Channel>,
@@ -270,6 +276,7 @@ pub(super) struct MutChannelHolder<'a> {
 	pub(super) forward_htlcs: &'a mut HashMap<u64, Vec<HTLCForwardInfo>>,
 	pub(super) claimable_htlcs: &'a mut HashMap<PaymentHash, Vec<(u64, HTLCPreviousHopData)>>,
 	pub(super) pending_msg_events: &'a mut Vec<events::MessageSendEvent>,
+	pub(super) channel_closing_waiting_threshold_conf: &'a mut HashMap<u32, Vec<[u8; 32]>>,
 }
 impl ChannelHolder {
 	pub(super) fn borrow_parts(&mut self) -> MutChannelHolder {
@@ -280,6 +287,7 @@ impl ChannelHolder {
 			forward_htlcs: &mut self.forward_htlcs,
 			claimable_htlcs: &mut self.claimable_htlcs,
 			pending_msg_events: &mut self.pending_msg_events,
+			channel_closing_waiting_threshold_conf: &mut self.channel_closing_waiting_threshold_conf,
 		}
 	}
 }
@@ -553,6 +561,7 @@ impl ChannelManager {
 				forward_htlcs: HashMap::new(),
 				claimable_htlcs: HashMap::new(),
 				pending_msg_events: Vec::new(),
+				channel_closing_waiting_threshold_conf: HashMap::new(),
 			}),
 			our_network_key: keys_manager.get_node_secret(),
 
@@ -2404,7 +2413,8 @@ impl ChainListener for ChannelManager {
 			let channel_state = channel_lock.borrow_parts();
 			let short_to_id = channel_state.short_to_id;
 			let pending_msg_events = channel_state.pending_msg_events;
-			channel_state.by_id.retain(|_, channel| {
+			let mut channel_closings = Vec::new();
+			channel_state.by_id.retain(|channel_id, channel| {
 				let chan_res = channel.block_connected(header, height, txn_matched, indexes_of_txn_matched);
 				if let Ok(Some(funding_locked)) = chan_res {
 					pending_msg_events.push(events::MessageSendEvent::SendFundingLocked {
@@ -2429,20 +2439,8 @@ impl ChainListener for ChannelManager {
 					for tx in txn_matched {
 						for inp in tx.input.iter() {
 							if inp.previous_output == funding_txo.into_bitcoin_outpoint() {
-								log_trace!(self, "Detected channel-closing tx {} spending {}:{}, closing channel {}", tx.txid(), inp.previous_output.txid, inp.previous_output.vout, log_bytes!(channel.channel_id()));
-								if let Some(short_id) = channel.get_short_channel_id() {
-									short_to_id.remove(&short_id);
-								}
-								// It looks like our counterparty went on-chain. We go ahead and
-								// broadcast our latest local state as well here, just in case its
-								// some kind of SPV attack, though we expect these to be dropped.
-								failed_channels.push(channel.force_shutdown());
-								if let Ok(update) = self.get_channel_update(&channel) {
-									pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
-										msg: update
-									});
-								}
-								return false;
+								log_trace!(self, "Detected channel-closing tx {} spending {}:{}, waiting until {} to close channel {}", tx.txid(), inp.previous_output.txid, inp.previous_output.vout, height + HTLC_FAIL_ANTI_REORG_DELAY - 1, log_bytes!(channel_id[..]));
+								channel_closings.push((height + HTLC_FAIL_ANTI_REORG_DELAY - 1, *channel_id));
 							}
 						}
 					}
@@ -2465,6 +2463,44 @@ impl ChainListener for ChannelManager {
 				}
 				true
 			});
+			for (block_timer, channel_id) in channel_closings.drain(..) {
+				match channel_state.channel_closing_waiting_threshold_conf.entry(block_timer) {
+					hash_map::Entry::Occupied(mut entry) => {
+						let mut duplicate = false;
+						for id in entry.get().iter() {
+							if *id == channel_id {
+								duplicate = true;
+								break;
+							}
+						}
+						if !duplicate {
+							entry.get_mut().push(channel_id);
+						}
+					}
+					hash_map::Entry::Vacant(entry) => {
+						entry.insert(vec![channel_id]);
+					}
+				}
+			}
+			if let Some(channel_closings) = channel_state.channel_closing_waiting_threshold_conf.remove(&height) {
+				for channel_id in channel_closings {
+					log_trace!(self, "Enough confirmations for a broacast commitment tx, channel {} can be closed", log_bytes!(&channel_id[..]));
+					if let Some(mut channel) = channel_state.by_id.remove(&channel_id) {
+						if let Some(short_id) = channel.get_short_channel_id() {
+							short_to_id.remove(&short_id);
+						}
+						// It looks like our counterparty went on-chain. We go ahead and
+						// broadcast our latest local state as well here, just in case its
+						// some kind of SPV attack, though we expect these to be dropped.
+						failed_channels.push(channel.force_shutdown());
+						if let Ok(update) = self.get_channel_update(&channel) {
+							pending_msg_events.push(events::MessageSendEvent::BroadcastChannelUpdate {
+								msg: update
+							});
+						}
+					}
+				}
+			}
 		}
 		for failure in failed_channels.drain(..) {
 			self.finish_force_close_channel(failure);
@@ -2474,7 +2510,7 @@ impl ChainListener for ChannelManager {
 	}
 
 	/// We force-close the channel without letting our counterparty participate in the shutdown
-	fn block_disconnected(&self, header: &BlockHeader, _: u32) {
+	fn block_disconnected(&self, header: &BlockHeader, height: u32) {
 		let _ = self.total_consistency_lock.read().unwrap();
 		let mut failed_channels = Vec::new();
 		{
@@ -2498,6 +2534,9 @@ impl ChainListener for ChannelManager {
 					true
 				}
 			});
+			if let Some(_) = channel_state.channel_closing_waiting_threshold_conf.remove(&(height + HTLC_FAIL_ANTI_REORG_DELAY - 1)) {
+				// We discard channel_closing there as brooadcast commitment tx has been disconnected, (and may be replaced by a legit closing_signed)
+			}
 		}
 		for failure in failed_channels.drain(..) {
 			self.finish_force_close_channel(failure);
@@ -2936,6 +2975,14 @@ impl Writeable for ChannelManager {
 			}
 		}
 
+		(channel_state.channel_closing_waiting_threshold_conf.len() as u64).write(writer)?;
+		for (confirmation_height, channel_id) in channel_state.channel_closing_waiting_threshold_conf.iter() {
+			confirmation_height.write(writer)?;
+			for id in channel_id {
+				id.write(writer)?;
+			}
+		}
+
 		Ok(())
 	}
 }
@@ -3073,6 +3120,21 @@ impl<'a, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a>> for (S
 			claimable_htlcs.insert(payment_hash, previous_hops);
 		}
 
+		let channel_closing_count: u64 = Readable::read(reader)?;
+		let mut channel_closing_waiting_threshold_conf: HashMap<u32, Vec<[u8; 32]>> = HashMap::with_capacity(cmp::min(channel_closing_count as usize, 32));
+		for _ in 0..channel_closing_count {
+			let confirmation_height: u32 = Readable::read(reader)?;
+			let channel_id: [u8; 32] = Readable::read(reader)?;
+			match channel_closing_waiting_threshold_conf.entry(confirmation_height) {
+				hash_map::Entry::Occupied(mut entry) => {
+					entry.get_mut().push(channel_id);
+				}
+				hash_map::Entry::Vacant(entry) => {
+					entry.insert(vec![channel_id]);
+				}
+			}
+		}
+
 		let channel_manager = ChannelManager {
 			genesis_hash,
 			fee_estimator: args.fee_estimator,
@@ -3091,6 +3153,7 @@ impl<'a, R : ::std::io::Read> ReadableArgs<R, ChannelManagerReadArgs<'a>> for (S
 				forward_htlcs,
 				claimable_htlcs,
 				pending_msg_events: Vec::new(),
+				channel_closing_waiting_threshold_conf,
 			}),
 			our_network_key: args.keys_manager.get_node_secret(),