Track HTLC-failure trigger tx until anti-reorg delay reached

Broadcasting a commitment tx means that we have to fail
inbound HTLC in backward channel. Doing it prematurely would
put us at risk in case of reorg. So we delay passing failure
update upstream until solving tx mature to HTLC_FAIL_ANTI_
REORG_DELAY.
Requirements differ if HTLC is a revoked/non-revoked dust/
non-revoked non-dust one.

Add connect_blocks in test_utils to fix broken tests due to
anti-reorg delay enforcement

Remove anti-duplicate htlc update stuff in ManySimpleChannelMonitor

Author: Antoine Riard

src/ln/channelmonitor.rs

@@ -178,10 +178,6 @@ impl<Key : Send + cmp::Eq + hash::Hash> ChainListener for SimpleManyChannelMonit
 						// In case of reorg we may have htlc outputs solved in a different way so
 						// we prefer to keep claims but don't store duplicate updates for a given
 						// (payment_hash, HTLCSource) pair.
-						// TODO: Note that we currently don't really use this as ChannelManager
-						// will fail/claim backwards after the first block. We really should delay
-						// a few blocks before failing backwards (but can claim backwards
-						// immediately) as long as we have a few blocks of headroom.
 						let mut existing_claim = false;
 						e.get_mut().retain(|htlc_data| {
 							if htlc.0 == htlc_data.0 {
@@ -306,7 +302,6 @@ pub(crate) const HTLC_FAIL_TIMEOUT_BLOCKS: u32 = 3;
 /// Number of blocks we wait on seeing a confirmed HTLC-Timeout or previous revoked commitment
 /// transaction before we fail corresponding inbound HTLCs. This prevents us from failing backwards
 /// and then getting a reorg resulting in us losing money.
-//TODO: We currently don't actually use this...we should
 pub(crate) const HTLC_FAIL_ANTI_REORG_DELAY: u32 = 6;
 
 #[derive(Clone, PartialEq)]
@@ -401,6 +396,8 @@ pub struct ChannelMonitor {
 
 	destination_script: Script,
 
+	htlc_updated_waiting_threshold_conf: HashMap<u32, Vec<(HTLCSource, Option<PaymentPreimage>, PaymentHash)>>,
+
 	// We simply modify last_block_hash in Channel's block_connected so that serialization is
 	// consistent but hopefully the users' copy handles block_connected in a consistent way.
 	// (we do *not*, however, update them in insert_combine to ensure any local user copies keep
@@ -430,7 +427,8 @@ impl PartialEq for ChannelMonitor {
 			self.current_remote_commitment_number != other.current_remote_commitment_number ||
 			self.current_local_signed_commitment_tx != other.current_local_signed_commitment_tx ||
 			self.payment_preimages != other.payment_preimages ||
-			self.destination_script != other.destination_script
+			self.destination_script != other.destination_script ||
+			self.htlc_updated_waiting_threshold_conf != other.htlc_updated_waiting_threshold_conf
 		{
 			false
 		} else {
@@ -480,6 +478,8 @@ impl ChannelMonitor {
 			payment_preimages: HashMap::new(),
 			destination_script: destination_script,
 
+			htlc_updated_waiting_threshold_conf: HashMap::new(),
+
 			last_block_hash: Default::default(),
 			secp_ctx: Secp256k1::new(),
 			logger,
@@ -987,6 +987,17 @@ impl ChannelMonitor {
 		self.last_block_hash.write(writer)?;
 		self.destination_script.write(writer)?;
 
+		writer.write_all(&byte_utils::be64_to_array(self.htlc_updated_waiting_threshold_conf.len() as u64))?;
+		for (ref target, ref updates) in self.htlc_updated_waiting_threshold_conf.iter() {
+			writer.write_all(&byte_utils::be32_to_array(**target))?;
+			writer.write_all(&byte_utils::be64_to_array(updates.len() as u64))?;
+			for ref update in updates.iter() {
+				update.0.write(writer)?;
+				update.1.write(writer)?;
+				update.2.write(writer)?;
+			}
+		}
+
 		Ok(())
 	}
 
@@ -1050,13 +1061,12 @@ impl ChannelMonitor {
 	/// HTLC-Success/HTLC-Timeout transactions.
 	/// Return updates for HTLC pending in the channel and failed automatically by the broadcast of
 	/// revoked remote commitment tx
-	fn check_spend_remote_transaction(&mut self, tx: &Transaction, height: u32, fee_estimator: &FeeEstimator) -> (Vec<Transaction>, (Sha256dHash, Vec<TxOut>), Vec<SpendableOutputDescriptor>, Vec<(HTLCSource, Option<PaymentPreimage>, PaymentHash)>)  {
+	fn check_spend_remote_transaction(&mut self, tx: &Transaction, height: u32, fee_estimator: &FeeEstimator) -> (Vec<Transaction>, (Sha256dHash, Vec<TxOut>), Vec<SpendableOutputDescriptor>) {
 		// Most secp and related errors trying to create keys means we have no hope of constructing
 		// a spend transaction...so we return no transactions to broadcast
 		let mut txn_to_broadcast = Vec::new();
 		let mut watch_outputs = Vec::new();
 		let mut spendable_outputs = Vec::new();
-		let mut htlc_updated = Vec::new();
 
 		let commitment_txid = tx.txid(); //TODO: This is gonna be a performance bottleneck for watchtowers!
 		let per_commitment_option = self.remote_claimable_outpoints.get(&commitment_txid);
@@ -1065,7 +1075,7 @@ impl ChannelMonitor {
 			( $thing : expr ) => {
 				match $thing {
 					Ok(a) => a,
-					Err(_) => return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated)
+					Err(_) => return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs)
 				}
 			};
 		}
@@ -1090,7 +1100,7 @@ impl ChannelMonitor {
 			};
 			let delayed_key = ignore_error!(chan_utils::derive_public_key(&self.secp_ctx, &PublicKey::from_secret_key(&self.secp_ctx, &per_commitment_key), &self.their_delayed_payment_base_key.unwrap()));
 			let a_htlc_key = match self.their_htlc_base_key {
-				None => return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated),
+				None => return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs),
 				Some(their_htlc_base_key) => ignore_error!(chan_utils::derive_public_key(&self.secp_ctx, &PublicKey::from_secret_key(&self.secp_ctx, &per_commitment_key), &their_htlc_base_key)),
 			};
 
@@ -1172,7 +1182,7 @@ impl ChannelMonitor {
 						if transaction_output_index as usize >= tx.output.len() ||
 								tx.output[transaction_output_index as usize].value != htlc.amount_msat / 1000 ||
 								tx.output[transaction_output_index as usize].script_pubkey != expected_script.to_v0_p2wsh() {
-							return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated); // Corrupted per_commitment_data, fuck this user
+							return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs); // Corrupted per_commitment_data, fuck this user
 						}
 						let input = TxIn {
 							previous_output: BitcoinOutPoint {
@@ -1216,16 +1226,22 @@ impl ChannelMonitor {
 				watch_outputs.append(&mut tx.output.clone());
 				self.remote_commitment_txn_on_chain.insert(commitment_txid, (commitment_number, tx.output.iter().map(|output| { output.script_pubkey.clone() }).collect()));
 
-				// TODO: We really should only fail backwards after our revocation claims have been
-				// confirmed, but we also need to do more other tracking of in-flight pre-confirm
-				// on-chain claims, so we can do that at the same time.
 				macro_rules! check_htlc_fails {
 					($txid: expr, $commitment_tx: expr) => {
 						if let Some(ref outpoints) = self.remote_claimable_outpoints.get($txid) {
 							for &(ref htlc, ref source_option) in outpoints.iter() {
 								if let &Some(ref source) = source_option {
-									log_trace!(self, "Failing HTLC with payment_hash {} from {} remote commitment tx due to broadcast of revoked remote commitment transaction", log_bytes!(htlc.payment_hash.0), $commitment_tx);
-									htlc_updated.push(((**source).clone(), None, htlc.payment_hash.clone()));
+									log_trace!(self, "Failing HTLC with payment_hash {} from {} remote commitment tx due to broadcast of revoked remote commitment transaction, waiting confirmation until {} height", log_bytes!(htlc.payment_hash.0), $commitment_tx, height + HTLC_FAIL_ANTI_REORG_DELAY - 1);
+									match self.htlc_updated_waiting_threshold_conf.entry(height + HTLC_FAIL_ANTI_REORG_DELAY - 1) {
+										hash_map::Entry::Occupied(mut entry) => {
+											let e = entry.get_mut();
+											e.retain(|ref update| update.0 != **source);
+											e.push(((**source).clone(), None, htlc.payment_hash.clone()));
+										}
+										hash_map::Entry::Vacant(entry) => {
+											entry.insert(vec![((**source).clone(), None, htlc.payment_hash.clone())]);
+										}
+									}
 								}
 							}
 						}
@@ -1241,7 +1257,7 @@ impl ChannelMonitor {
 				}
 				// No need to check local commitment txn, symmetric HTLCSource must be present as per-htlc data on remote commitment tx
 			}
-			if inputs.is_empty() { return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated); } // Nothing to be done...probably a false positive/local tx
+			if inputs.is_empty() { return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs); } // Nothing to be done...probably a false positive/local tx
 
 			let outputs = vec!(TxOut {
 				script_pubkey: self.destination_script.clone(),
@@ -1283,9 +1299,6 @@ impl ChannelMonitor {
 
 			log_trace!(self, "Got broadcast of non-revoked remote commitment transaction {}", commitment_txid);
 
-			// TODO: We really should only fail backwards after our revocation claims have been
-			// confirmed, but we also need to do more other tracking of in-flight pre-confirm
-			// on-chain claims, so we can do that at the same time.
 			macro_rules! check_htlc_fails {
 				($txid: expr, $commitment_tx: expr, $id: tt) => {
 					if let Some(ref latest_outpoints) = self.remote_claimable_outpoints.get($txid) {
@@ -1306,7 +1319,16 @@ impl ChannelMonitor {
 									}
 								}
 								log_trace!(self, "Failing HTLC with payment_hash {} from {} remote commitment tx due to broadcast of remote commitment transaction", log_bytes!(htlc.payment_hash.0), $commitment_tx);
-								htlc_updated.push(((**source).clone(), None, htlc.payment_hash.clone()));
+								match self.htlc_updated_waiting_threshold_conf.entry(height + HTLC_FAIL_ANTI_REORG_DELAY - 1) {
+									hash_map::Entry::Occupied(mut entry) => {
+										let e = entry.get_mut();
+										e.retain(|ref update| update.0 != **source);
+										e.push(((**source).clone(), None, htlc.payment_hash.clone()));
+									}
+									hash_map::Entry::Vacant(entry) => {
+										entry.insert(vec![((**source).clone(), None, htlc.payment_hash.clone())]);
+									}
+								}
 							}
 						}
 					}
@@ -1339,7 +1361,7 @@ impl ChannelMonitor {
 						},
 					};
 					let a_htlc_key = match self.their_htlc_base_key {
-						None => return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated),
+						None => return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs),
 						Some(their_htlc_base_key) => ignore_error!(chan_utils::derive_public_key(&self.secp_ctx, revocation_point, &their_htlc_base_key)),
 					};
 
@@ -1395,7 +1417,7 @@ impl ChannelMonitor {
 							if transaction_output_index as usize >= tx.output.len() ||
 									tx.output[transaction_output_index as usize].value != htlc.amount_msat / 1000 ||
 									tx.output[transaction_output_index as usize].script_pubkey != expected_script.to_v0_p2wsh() {
-								return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated); // Corrupted per_commitment_data, fuck this user
+								return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs); // Corrupted per_commitment_data, fuck this user
 							}
 							if let Some(payment_preimage) = self.payment_preimages.get(&htlc.payment_hash) {
 								let input = TxIn {
@@ -1462,7 +1484,7 @@ impl ChannelMonitor {
 						}
 					}
 
-					if inputs.is_empty() { return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated); } // Nothing to be done...probably a false positive/local tx
+					if inputs.is_empty() { return (txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs); } // Nothing to be done...probably a false positive/local tx
 
 					let outputs = vec!(TxOut {
 						script_pubkey: self.destination_script.clone(),
@@ -1495,7 +1517,7 @@ impl ChannelMonitor {
 			}
 		}
 
-		(txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs, htlc_updated)
+		(txn_to_broadcast, (commitment_txid, watch_outputs), spendable_outputs)
 	}
 
 	/// Attempts to claim a remote HTLC-Success/HTLC-Timeout's outputs using the revocation key
@@ -1770,7 +1792,7 @@ impl ChannelMonitor {
 					}
 				};
 				if funding_txo.is_none() || (prevout.txid == funding_txo.as_ref().unwrap().0.txid && prevout.vout == funding_txo.as_ref().unwrap().0.index as u32) {
-					let (remote_txn, new_outputs, mut spendable_output, mut updated) = self.check_spend_remote_transaction(tx, height, fee_estimator);
+					let (remote_txn, new_outputs, mut spendable_output) = self.check_spend_remote_transaction(tx, height, fee_estimator);
 					txn = remote_txn;
 					spendable_outputs.append(&mut spendable_output);
 					if !new_outputs.1.is_empty() {
@@ -1789,9 +1811,6 @@ impl ChannelMonitor {
 							spendable_outputs.push(spendable_output);
 						}
 					}
-					if updated.len() > 0 {
-						htlc_updated.append(&mut updated);
-					}
 				} else {
 					if let Some(&(commitment_number, _)) = self.remote_commitment_txn_on_chain.get(&prevout.txid) {
 						let (tx, spendable_output) = self.check_spend_remote_htlc(tx, commitment_number, fee_estimator);
@@ -1842,6 +1861,12 @@ impl ChannelMonitor {
 				}
 			}
 		}
+		if let Some(updates) = self.htlc_updated_waiting_threshold_conf.remove(&height) {
+			for update in updates {
+				log_trace!(self, "HTLC {} failure update has get enough confirmation to be pass upstream", log_bytes!((update.2).0));
+				htlc_updated.push(update);
+			}
+		}
 		self.last_block_hash = block_hash.clone();
 		(watch_outputs, spendable_outputs, htlc_updated)
 	}
@@ -2242,6 +2267,21 @@ impl<R: ::std::io::Read> ReadableArgs<R, Arc<Logger>> for (Sha256dHash, ChannelM
 		let last_block_hash: Sha256dHash = Readable::read(reader)?;
 		let destination_script = Readable::read(reader)?;
 
+		let waiting_threshold_conf_len: u64 = Readable::read(reader)?;
+		let mut htlc_updated_waiting_threshold_conf = HashMap::with_capacity(cmp::min(waiting_threshold_conf_len as usize, MAX_ALLOC_SIZE / 128));
+		for _ in 0..waiting_threshold_conf_len {
+			let height_target = Readable::read(reader)?;
+			let updates_len: u64 = Readable::read(reader)?;
+			let mut updates = Vec::with_capacity(cmp::min(updates_len as usize, MAX_ALLOC_SIZE / 128));
+			for _ in 0..updates_len {
+				let htlc_source = Readable::read(reader)?;
+				let preimage = Readable::read(reader)?;
+				let hash = Readable::read(reader)?;
+				updates.push((htlc_source, preimage, hash));
+			}
+			htlc_updated_waiting_threshold_conf.insert(height_target, updates);
+		}
+
 		Ok((last_block_hash.clone(), ChannelMonitor {
 			commitment_transaction_number_obscure_factor,
 
@@ -2265,6 +2305,9 @@ impl<R: ::std::io::Read> ReadableArgs<R, Arc<Logger>> for (Sha256dHash, ChannelM
 			payment_preimages,
 
 			destination_script,
+
+			htlc_updated_waiting_threshold_conf,
+
 			last_block_hash,
 			secp_ctx,
 			logger,

src/ln/functional_test_utils.rs

@@ -20,6 +20,7 @@ use bitcoin::blockdata::transaction::{Transaction, TxOut};
 use bitcoin::network::constants::Network;
 
 use bitcoin_hashes::sha256::Hash as Sha256;
+use bitcoin_hashes::sha256d::Hash as Sha256d;
 use bitcoin_hashes::Hash;
 
 use secp256k1::Secp256k1;
@@ -46,6 +47,15 @@ pub fn confirm_transaction(chain: &chaininterface::ChainWatchInterfaceUtil, tx:
 	}
 }
 
+pub fn connect_blocks(chain: &chaininterface::ChainWatchInterfaceUtil, depth: u32, height: u32, parent: bool, prev_blockhash: Sha256d) {
+	let mut header = BlockHeader { version: 0x2000000, prev_blockhash: if parent { prev_blockhash } else { Default::default() }, merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
+	chain.block_connected_checked(&header, height + 1, &Vec::new(), &Vec::new());
+	for i in 2..depth {
+		header = BlockHeader { version: 0x20000000, prev_blockhash: header.bitcoin_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
+		chain.block_connected_checked(&header, height + i, &Vec::new(), &Vec::new());
+	}
+}
+
 pub struct Node {
 	pub chain_monitor: Arc<chaininterface::ChainWatchInterfaceUtil>,
 	pub tx_broadcaster: Arc<test_utils::TestBroadcaster>,