Track and react to remote partial-claiming of pending claim request

A pending claim request may contain a set of multiple outpoints.
If one or multiple of them get claimed by remote party, our in-flight
claiming transactions aren't valid anymore so we need to react
quickly and regenerate claiming transaction with accurate set.

However, a claimed outpoint may be disconnected and we need to resurrect
back outpoint among set of orignal pending claim request.

To guarantee consistency of contentious claimed outpoint we cache it
as OnchainEvent::ContentionsOutpoint and only delete it after
ANTI_REORG_DELAY.

Fix test broken by change, partial claiming on revoked txn
force us to regenerate txn

Author: Antoine Riard

fuzz/fuzz_targets/full_stack_target.rs

@@ -212,7 +212,7 @@ impl<'a, Key : Send + cmp::Eq + hash::Hash> ChainListener for SimpleManyChannelM
 		let block_hash = header.bitcoin_hash();
 		let mut monitors = self.monitors.lock().unwrap();
 		for monitor in monitors.values_mut() {
-			monitor.block_disconnected(disconnected_height, &block_hash);
+			monitor.block_disconnected(disconnected_height, &block_hash, &*self.broadcaster, &*self.fee_estimator);
 		}
 	}
 }
@@ -502,6 +502,13 @@ enum OnchainEvent {
 	HTLCUpdate {
 		htlc_update: (HTLCSource, PaymentHash),
 	},
+	/// Claim tx aggregate multiple claimable outpoints. One of the outpoint may be claimed by a remote party tx.
+	/// In this case, we need to drop the outpoint and regenerate a new claim tx. By safety, we keep tracking
+	/// the outpoint to be sure to resurect it back to the claim tx if reorgs happen.
+	ContentiousOutpoint {
+		outpoint: BitcoinOutPoint,
+		input_material: InputMaterial,
+	}
 }
 
 /// Higher-level cache structure needed to re-generate bumped claim txn if needed
@@ -1321,6 +1328,11 @@ impl ChannelMonitor {
 						writer.write_all(&[1; 1])?;
 						htlc_update.0.write(writer)?;
 						htlc_update.1.write(writer)?;
+					},
+					OnchainEvent::ContentiousOutpoint { ref outpoint, ref input_material } => {
+						writer.write_all(&[2; 1])?;
+						outpoint.write(writer)?;
+						input_material.write(writer)?;
 					}
 				}
 			}
@@ -1545,6 +1557,10 @@ impl ChannelMonitor {
 								log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", single_htlc_tx.input[0].previous_output.txid, single_htlc_tx.input[0].previous_output.vout, height_timer);
 								let mut per_input_material = HashMap::with_capacity(1);
 								per_input_material.insert(single_htlc_tx.input[0].previous_output, InputMaterial::Revoked { script: redeemscript, pubkey: Some(revocation_pubkey), key: revocation_key, is_htlc: true, amount: htlc.amount_msat / 1000 });
+								match self.claimable_outpoints.entry(single_htlc_tx.input[0].previous_output) {
+									hash_map::Entry::Occupied(_) => {},
+									hash_map::Entry::Vacant(entry) => { entry.insert((single_htlc_tx.txid(), height)); }
+								}
 								match self.pending_claim_requests.entry(single_htlc_tx.txid()) {
 									hash_map::Entry::Occupied(_) => {},
 									hash_map::Entry::Vacant(entry) => { entry.insert(ClaimTxBumpMaterial { height_timer, feerate_previous: used_feerate, soonest_timelock: htlc.cltv_expiry, per_input_material }); }
@@ -1630,15 +1646,17 @@ impl ChannelMonitor {
 				}
 			}
 			let height_timer = Self::get_height_timer(height, soonest_timelock);
+			let spend_txid = spend_tx.txid();
 			for (input, info) in spend_tx.input.iter_mut().zip(inputs_info.iter()) {
 				let (redeemscript, revocation_key) = sign_input!(sighash_parts, input, info.0, info.1);
 				log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", input.previous_output.txid, input.previous_output.vout, height_timer);
 				per_input_material.insert(input.previous_output, InputMaterial::Revoked { script: redeemscript, pubkey: if info.0.is_some() { Some(revocation_pubkey) } else { None }, key: revocation_key, is_htlc: if info.0.is_some() { true } else { false }, amount: info.1 });
-				if info.2 < soonest_timelock {
-					soonest_timelock = info.2;
+				match self.claimable_outpoints.entry(input.previous_output) {
+					hash_map::Entry::Occupied(_) => {},
+					hash_map::Entry::Vacant(entry) => { entry.insert((spend_txid, height)); }
 				}
 			}
-			match self.pending_claim_requests.entry(spend_tx.txid()) {
+			match self.pending_claim_requests.entry(spend_txid) {
 				hash_map::Entry::Occupied(_) => {},
 				hash_map::Entry::Vacant(entry) => { entry.insert(ClaimTxBumpMaterial { height_timer, feerate_previous: used_feerate, soonest_timelock, per_input_material }); }
 			}
@@ -1831,6 +1849,10 @@ impl ChannelMonitor {
 											log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", single_htlc_tx.input[0].previous_output.txid, single_htlc_tx.input[0].previous_output.vout, height_timer);
 											let mut per_input_material = HashMap::with_capacity(1);
 											per_input_material.insert(single_htlc_tx.input[0].previous_output, InputMaterial::RemoteHTLC { script: redeemscript, key: htlc_key, preimage: Some(*payment_preimage), amount: htlc.amount_msat / 1000, locktime: 0 });
+											match self.claimable_outpoints.entry(single_htlc_tx.input[0].previous_output) {
+												hash_map::Entry::Occupied(_) => {},
+												hash_map::Entry::Vacant(entry) => { entry.insert((single_htlc_tx.txid(), height)); }
+											}
 											match self.pending_claim_requests.entry(single_htlc_tx.txid()) {
 												hash_map::Entry::Occupied(_) => {},
 												hash_map::Entry::Vacant(entry) => { entry.insert(ClaimTxBumpMaterial { height_timer, feerate_previous: used_feerate, soonest_timelock: htlc.cltv_expiry, per_input_material}); }
@@ -1872,6 +1894,10 @@ impl ChannelMonitor {
 									log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", timeout_tx.input[0].previous_output.txid, timeout_tx.input[0].previous_output.vout, height_timer);
 									let mut per_input_material = HashMap::with_capacity(1);
 									per_input_material.insert(timeout_tx.input[0].previous_output, InputMaterial::RemoteHTLC { script : redeemscript, key: htlc_key, preimage: None, amount: htlc.amount_msat / 1000, locktime: htlc.cltv_expiry });
+									match self.claimable_outpoints.entry(timeout_tx.input[0].previous_output) {
+										hash_map::Entry::Occupied(_) => {},
+										hash_map::Entry::Vacant(entry) => { entry.insert((timeout_tx.txid(), height)); }
+									}
 									match self.pending_claim_requests.entry(timeout_tx.txid()) {
 										hash_map::Entry::Occupied(_) => {},
 										hash_map::Entry::Vacant(entry) => { entry.insert(ClaimTxBumpMaterial { height_timer, feerate_previous: used_feerate, soonest_timelock: htlc.cltv_expiry, per_input_material }); }
@@ -1912,12 +1938,17 @@ impl ChannelMonitor {
 						}
 					}
 					let height_timer = Self::get_height_timer(height, soonest_timelock);
+					let spend_txid = spend_tx.txid();
 					for (input, info) in spend_tx.input.iter_mut().zip(inputs_info.iter()) {
 						let (redeemscript, htlc_key) = sign_input!(sighash_parts, input, info.1, (info.0).0.to_vec());
 						log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", input.previous_output.txid, input.previous_output.vout, height_timer);
 						per_input_material.insert(input.previous_output, InputMaterial::RemoteHTLC { script: redeemscript, key: htlc_key, preimage: Some(*(info.0)), amount: info.1, locktime: 0});
+						match self.claimable_outpoints.entry(input.previous_output) {
+							hash_map::Entry::Occupied(_) => {},
+							hash_map::Entry::Vacant(entry) => { entry.insert((spend_txid, height)); }
+						}
 					}
-					match self.pending_claim_requests.entry(spend_tx.txid()) {
+					match self.pending_claim_requests.entry(spend_txid) {
 						hash_map::Entry::Occupied(_) => {},
 						hash_map::Entry::Vacant(entry) => { entry.insert(ClaimTxBumpMaterial { height_timer, feerate_previous: used_feerate, soonest_timelock, per_input_material }); }
 					}
@@ -2036,6 +2067,10 @@ impl ChannelMonitor {
 			log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", spend_tx.input[0].previous_output.txid, spend_tx.input[0].previous_output.vout, height_timer);
 			let mut per_input_material = HashMap::with_capacity(1);
 			per_input_material.insert(spend_tx.input[0].previous_output, InputMaterial::Revoked { script: redeemscript, pubkey: None, key: revocation_key, is_htlc: false, amount: tx.output[0].value });
+			match self.claimable_outpoints.entry(spend_tx.input[0].previous_output) {
+				hash_map::Entry::Occupied(_) => {},
+				hash_map::Entry::Vacant(entry) => { entry.insert((spend_tx.txid(), height)); }
+			}
 			match self.pending_claim_requests.entry(spend_tx.txid()) {
 				hash_map::Entry::Occupied(_) => {},
 				hash_map::Entry::Vacant(entry) => { entry.insert(ClaimTxBumpMaterial { height_timer, feerate_previous: used_feerate, soonest_timelock: height + self.our_to_self_delay as u32, per_input_material }); }
@@ -2099,6 +2134,7 @@ impl ChannelMonitor {
 						let height_timer = Self::get_height_timer(height, htlc.cltv_expiry);
 						let mut per_input_material = HashMap::with_capacity(1);
 						per_input_material.insert(htlc_timeout_tx.input[0].previous_output, InputMaterial::LocalHTLC { script: htlc_script, sigs: (*their_sig, *our_sig), preimage: None, amount: htlc.amount_msat / 1000});
+						//TODO: with option_simplified_commitment track outpoint too
 						log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", htlc_timeout_tx.input[0].previous_output.vout, htlc_timeout_tx.input[0].previous_output.txid, height_timer);
 						pending_claims.push((htlc_timeout_tx.txid(), ClaimTxBumpMaterial { height_timer, feerate_previous: 0, soonest_timelock: htlc.cltv_expiry, per_input_material }));
 						res.push(htlc_timeout_tx);
@@ -2122,6 +2158,7 @@ impl ChannelMonitor {
 							let height_timer = Self::get_height_timer(height, htlc.cltv_expiry);
 							let mut per_input_material = HashMap::with_capacity(1);
 							per_input_material.insert(htlc_success_tx.input[0].previous_output, InputMaterial::LocalHTLC { script: htlc_script, sigs: (*their_sig, *our_sig), preimage: Some(*payment_preimage), amount: htlc.amount_msat / 1000});
+							//TODO: with option_simplified_commitment track outpoint too
 							log_trace!(self, "Outpoint {}:{} is being being claimed, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", htlc_success_tx.input[0].previous_output.vout, htlc_success_tx.input[0].previous_output.txid, height_timer);
 							pending_claims.push((htlc_success_tx.txid(), ClaimTxBumpMaterial { height_timer, feerate_previous: 0, soonest_timelock: htlc.cltv_expiry, per_input_material }));
 							res.push(htlc_success_tx);
@@ -2356,17 +2393,18 @@ impl ChannelMonitor {
 			}
 
 			// Scan all input to verify is one of the outpoint spent is of interest for us
+			let mut claimed_outpoints = Vec::new();
+			let mut claimed_input_material = Vec::new();
 			for inp in &tx.input {
 				if let Some(ancestor_claimable_txid) = self.claimable_outpoints.get(&inp.previous_output) {
 					// If outpoint has claim request pending on it...
 					if let Some(claim_material) = self.pending_claim_requests.get_mut(&ancestor_claimable_txid.0) {
 						//... we need to verify equality between transaction outpoints and claim request
 						// outpoints to know if transaction is the original claim or a bumped one issued
 						// by us.
-						let mut claimed_outpoints = Vec::new();
-						for (claim_inp, tx_inp) in claim_material.per_input_material.keys().zip(tx.input.iter()) {
-							if *claim_inp != tx_inp.previous_output {
-								claimed_outpoints.push(tx_inp.previous_output.clone());
+						for claim_inp in claim_material.per_input_material.keys() {
+							if *claim_inp == inp.previous_output {
+								claimed_outpoints.push(inp.previous_output.clone());
 							}
 						}
 						if claimed_outpoints.len() == 0 && claim_material.per_input_material.len() == tx.input.len() { // If true, register claim request to be removed after reaching a block security height
@@ -2377,19 +2415,30 @@ impl ChannelMonitor {
 								}
 							}
 						} else { // If false, generate new claim request with update outpoint set
-							for already_claimed in claimed_outpoints {
-								claim_material.per_input_material.remove(&already_claimed);
+							for already_claimed in claimed_outpoints.iter() {
+								if let Some(input_material) = claim_material.per_input_material.remove(&already_claimed) {
+									claimed_input_material.push(input_material);
+								}
 							}
 							// Avoid bump engine using inaccurate feerate due to new transaction size
 							claim_material.feerate_previous = 0;
 							//TODO: recompute soonest_timelock to avoid wasting a bit on fees
 							bump_candidates.push((ancestor_claimable_txid.0.clone(), claim_material.clone()));
 						}
+						break; //No need to iterate further, either tx is our or their
 					} else {
 						panic!("Inconsistencies between pending_claim_requests map and claimable_outpoints map");
 					}
 				}
 			}
+			for (outpoint, input_material) in claimed_outpoints.iter().zip(claimed_input_material.drain(..)) {
+				match self.onchain_events_waiting_threshold_conf.entry(height + ANTI_REORG_DELAY - 1) {
+					hash_map::Entry::Occupied(_) => {},
+					hash_map::Entry::Vacant(entry) => {
+						entry.insert(vec![OnchainEvent::ContentiousOutpoint { outpoint: *outpoint, input_material: input_material }]);
+					}
+				}
+			}
 		}
 		if let Some(ref cur_local_tx) = self.current_local_signed_commitment_tx {
 			if self.would_broadcast_at_height(height) {
@@ -2432,6 +2481,9 @@ impl ChannelMonitor {
 						log_trace!(self, "HTLC {} failure update has got enough confirmations to be passed upstream", log_bytes!((htlc_update.1).0));
 						htlc_updated.push((htlc_update.0, None, htlc_update.1));
 					},
+					OnchainEvent::ContentiousOutpoint { outpoint, .. } => {
+						self.claimable_outpoints.remove(&outpoint);
+					}
 				}
 			}
 		}
@@ -2454,13 +2506,52 @@ impl ChannelMonitor {
 		(watch_outputs, spendable_outputs, htlc_updated)
 	}
 
-	fn block_disconnected(&mut self, height: u32, block_hash: &Sha256dHash) {
-		if let Some(_) = self.onchain_events_waiting_threshold_conf.remove(&(height + ANTI_REORG_DELAY - 1)) {
+	fn block_disconnected(&mut self, height: u32, block_hash: &Sha256dHash, broadcaster: &BroadcasterInterface, fee_estimator: &FeeEstimator) {
+		let mut bump_candidates = HashMap::new();
+		if let Some(events) = self.onchain_events_waiting_threshold_conf.remove(&(height + ANTI_REORG_DELAY - 1)) {
 			//We may discard:
 			//- htlc update there as failure-trigger tx (revoked commitment tx, non-revoked commitment tx, HTLC-timeout tx) has been disconnected
 			//- our claim tx on a commitment tx output
+			//- resurect outpoint back in its claimable set and regenerate tx
+			for ev in events {
+				match ev {
+					OnchainEvent::ContentiousOutpoint { outpoint, input_material } => {
+						if let Some(ancestor_claimable_txid) = self.claimable_outpoints.get(&outpoint) {
+							if let Some(claim_material) = self.pending_claim_requests.get_mut(&ancestor_claimable_txid.0) {
+								// Avoid bump engine using inaccurate feerate due to new transaction size
+								claim_material.feerate_previous = 0;
+								claim_material.per_input_material.insert(outpoint, input_material);
+								// Using a HashMap guarantee us than if we have multiple outpoints getting
+								// resurrected only one bump claim tx is going to be broadcast
+								bump_candidates.insert(ancestor_claimable_txid.clone(), claim_material.clone());
+							}
+						}
+					},
+					_ => {},
+				}
+			}
+		}
+		for (_, claim_material) in bump_candidates.iter_mut() {
+			if let Some((new_timer, new_feerate, bump_tx)) = self.bump_claim_tx(height, &claim_material, fee_estimator) {
+				claim_material.height_timer = new_timer;
+				claim_material.feerate_previous = new_feerate;
+				broadcaster.broadcast_transaction(&bump_tx);
+			}
+		}
+		for (ancestor_claim_txid, claim_material) in bump_candidates.drain() {
+			self.pending_claim_requests.insert(ancestor_claim_txid.0, claim_material);
+		}
+		//TODO: if we implement cross-block aggregated claim transaction we need to refresh set of outpoints and regenerate tx but
+		// right now if one of the outpoint get disconnected, just erase whole pending claim request.
+		let mut remove_request = Vec::new();
+		self.claimable_outpoints.retain(|_, ref v|
+			if v.1 == height {
+			remove_request.push(v.0.clone());
+			false
+			} else { true });
+		for req in remove_request {
+			self.pending_claim_requests.remove(&req);
 		}
-		self.claimable_outpoints.retain(|_, ref v| if v.1 == height { false } else { true });
 		self.last_block_hash = block_hash.clone();
 	}
 
@@ -3060,6 +3151,14 @@ impl<R: ::std::io::Read> ReadableArgs<R, Arc<Logger>> for (Sha256dHash, ChannelM
 							htlc_update: (htlc_source, hash)
 						}
 					},
+					2 => {
+						let outpoint = Readable::read(reader)?;
+						let input_material = Readable::read(reader)?;
+						OnchainEvent::ContentiousOutpoint {
+							outpoint,
+							input_material
+						}
+					}
 					_ => return Err(DecodeError::InvalidValue),
 				};
 				events.push(ev);

lightning/src/ln/channelmonitor.rs

@@ -920,15 +920,21 @@ pub fn test_txn_broadcast(node: &Node, chan: &(msgs::ChannelUpdate, msgs::Channe
 
 /// Tests that the given node has broadcast a claim transaction against the provided revoked
 /// HTLC transaction.
-pub fn test_revoked_htlc_claim_txn_broadcast(node: &Node, revoked_tx: Transaction) {
+pub fn test_revoked_htlc_claim_txn_broadcast(node: &Node, revoked_tx: Transaction, commitment_revoked_tx: Transaction) {
 	let mut node_txn = node.tx_broadcaster.txn_broadcasted.lock().unwrap();
-	assert_eq!(node_txn.len(), 1);
+	// We should issue a 2nd transaction if one htlc is dropped from initial claiming tx
+	// but sometimes not as feerate is too-low
+	if node_txn.len() != 1 && node_txn.len() != 2 { assert!(false); }
 	node_txn.retain(|tx| {
 		if tx.input.len() == 1 && tx.input[0].previous_output.txid == revoked_tx.txid() {
-			check_spends!(tx, revoked_tx.clone());
+			check_spends!(tx, revoked_tx);
 			false
 		} else { true }
 	});
+	node_txn.retain(|tx| {
+		check_spends!(tx, commitment_revoked_tx);
+		false
+	});
 	assert!(node_txn.is_empty());
 }