WIP

Author: Antoine Riard

src/ln/channelmonitor.rs

@@ -461,9 +461,12 @@ pub struct ChannelMonitor {
 	// for inputs spending this. If height timer (u32) is expired and claim tx hasn't reached enough confirmations
 	// before, use TxMaterial to regenerate a new claim tx with a satoshis-per-1000-weight-units higher than last
 	// one (u64), if timelock expiration (u32) is near, decrease height timer, the in-between bumps delay.
-	// Last field cached (u32) is height of outpoint confirmation, which is needed to flush this tracker
+	// We also cache height of outpoint confirmation (u32), which is needed to flush this tracker
 	// in case of reorgs, given block timer are scaled on timer expiration we can't deduce from it original height.
-	our_claim_txn_waiting_first_conf: HashMap<BitcoinOutPoint, (u32, TxMaterial, u64, u32, u32)>,
+	// Last field cached (Option<Sha256dHash>) is txid of lastest claim tx, it shouldn't be useful for single-input claim
+	// tx but if we claimed outpoints at first in an aggregate one, they must stay claimed in an aggregate bumped
+	// one too to respect RBF rules.
+	our_claim_txn_waiting_first_conf: HashMap<BitcoinOutPoint, (u32, TxMaterial, u64, u32, u32, Vec<BitcoinOutPoint>)>,
 
 	// Used to track onchain events, i.e transactions parts of channels confirmed on chain, on which
 	// we have to take actions once they reach enough confs. Key is a block height timer, i.e we enforce
@@ -1142,6 +1145,10 @@ impl ChannelMonitor {
 			writer.write_all(&byte_utils::be64_to_array(claim_tx_data.2))?;
 			writer.write_all(&byte_utils::be32_to_array(claim_tx_data.3))?;
 			writer.write_all(&byte_utils::be32_to_array(claim_tx_data.4))?;
+			writer.write_all(&byte_utils::be64_to_array(claim_tx_data.5.len() as u64))?;
+			for inp in claim_tx_data.5.iter() {
+				inp.write(writer)?;
+			}
 		}
 
 		writer.write_all(&byte_utils::be64_to_array(self.onchain_events_waiting_threshold_conf.len() as u64))?;
@@ -1382,7 +1389,7 @@ impl ChannelMonitor {
 								log_trace!(self, "Outpoint {}:{} is under claiming process, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", single_htlc_tx.input[0].previous_output.vout, single_htlc_tx.input[0].previous_output.txid, height_timer);
 								match self.our_claim_txn_waiting_first_conf.entry(single_htlc_tx.input[0].previous_output.clone()) {
 									hash_map::Entry::Occupied(_) => {},
-									hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::Revoked { script: redeemscript, pubkey: Some(revocation_pubkey), key: revocation_key, is_htlc: true, amount: htlc.amount_msat / 1000 }, used_feerate, htlc.cltv_expiry, height)); }
+									hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::Revoked { script: redeemscript, pubkey: Some(revocation_pubkey), key: revocation_key, is_htlc: true, amount: htlc.amount_msat / 1000 }, used_feerate, htlc.cltv_expiry, height, Vec::new())); }
 								}
 								txn_to_broadcast.push(single_htlc_tx);
 							}
@@ -1457,13 +1464,17 @@ impl ChannelMonitor {
 
 			let sighash_parts = bip143::SighashComponents::new(&spend_tx);
 
+			let mut buddy_inputs = Vec::new();
+			for inp in spend_tx.input.iter() {
+				buddy_inputs.push(inp.previous_output.clone());
+			}
 			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);
 				let height_timer = Self::get_height_timer(height, info.2);
 				log_trace!(self, "Outpoint {}:{} is under claiming process, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", input.previous_output.vout, input.previous_output.txid, height_timer);
 				match self.our_claim_txn_waiting_first_conf.entry(input.previous_output.clone()) {
 					hash_map::Entry::Occupied(_) => {},
-					hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::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 }, used_feerate, if !info.0.is_some() { height + info.2 } else { info.2 }, height)); }
+					hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::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 }, used_feerate, if !info.0.is_some() { height + info.2 } else { info.2 }, height, buddy_inputs.clone())); }
 				}
 			}
 			assert!(predicted_weight >= spend_tx.get_weight());
@@ -1653,7 +1664,7 @@ impl ChannelMonitor {
 										log_trace!(self, "Outpoint {}:{} is under claiming process, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", single_htlc_tx.input[0].previous_output.vout, single_htlc_tx.input[0].previous_output.txid, height_timer);
 										match self.our_claim_txn_waiting_first_conf.entry(single_htlc_tx.input[0].previous_output.clone()) {
 											hash_map::Entry::Occupied(_) => {},
-											hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::RemoteHTLC { script: redeemscript, key: htlc_key, preimage: Some(*payment_preimage), amount: htlc.amount_msat / 1000 }, used_feerate, htlc.cltv_expiry, height)); }
+											hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::RemoteHTLC { script: redeemscript, key: htlc_key, preimage: Some(*payment_preimage), amount: htlc.amount_msat / 1000 }, used_feerate, htlc.cltv_expiry, height, Vec::new())); }
 										}
 										txn_to_broadcast.push(single_htlc_tx);
 									}
@@ -1691,7 +1702,7 @@ impl ChannelMonitor {
 									log_trace!(self, "Outpoint {}:{} is under claiming process, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", timeout_tx.input[0].previous_output.vout, timeout_tx.input[0].previous_output.txid, height_timer);
 									match self.our_claim_txn_waiting_first_conf.entry(timeout_tx.input[0].previous_output.clone()) {
 										hash_map::Entry::Occupied(_) => {},
-										hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::RemoteHTLC { script : redeemscript, key: htlc_key, preimage: None, amount: htlc.amount_msat / 1000 }, used_feerate, htlc.cltv_expiry, height)); }
+										hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::RemoteHTLC { script : redeemscript, key: htlc_key, preimage: None, amount: htlc.amount_msat / 1000 }, used_feerate, htlc.cltv_expiry, height, Vec::new())); }
 									}
 								}
 								txn_to_broadcast.push(timeout_tx);
@@ -1721,13 +1732,17 @@ impl ChannelMonitor {
 
 					let sighash_parts = bip143::SighashComponents::new(&spend_tx);
 
+					let mut buddy_inputs = Vec::new();
+					for inp in spend_tx.input.iter() {
+						buddy_inputs.push(inp.previous_output.clone());
+					}
 					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());
 						let height_timer = Self::get_height_timer(height, info.2);
 						log_trace!(self, "Outpoint {}:{} is under claiming process, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", input.previous_output.vout, input.previous_output.txid, height_timer);
 						match self.our_claim_txn_waiting_first_conf.entry(input.previous_output.clone()) {
 							hash_map::Entry::Occupied(_) => {},
-							hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::RemoteHTLC { script: redeemscript, key: htlc_key, preimage: Some(*(info.0)), amount: info.1}, used_feerate, info.2, height)); }
+							hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::RemoteHTLC { script: redeemscript, key: htlc_key, preimage: Some(*(info.0)), amount: info.1}, used_feerate, info.2, height, buddy_inputs.clone())); }
 						}
 					}
 					assert!(predicted_weight >= spend_tx.get_weight());
@@ -1835,13 +1850,13 @@ impl ChannelMonitor {
 			log_trace!(self, "Outpoint {}:{} is under claiming process, if it doesn't succeed, a bumped claiming txn is going to be broadcast at height {}", spend_tx.input[0].previous_output.vout, spend_tx.input[0].previous_output.txid, height_timer);
 			match self.our_claim_txn_waiting_first_conf.entry(spend_tx.input[0].previous_output.clone()) {
 				hash_map::Entry::Occupied(_) => {},
-				hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::Revoked { script: redeemscript, pubkey: None, key: revocation_key, is_htlc: false, amount: tx.output[0].value }, used_feerate, height + self.our_to_self_delay as u32, height)); }
+				hash_map::Entry::Vacant(entry) => { entry.insert((height_timer, TxMaterial::Revoked { script: redeemscript, pubkey: None, key: revocation_key, is_htlc: false, amount: tx.output[0].value }, used_feerate, height + self.our_to_self_delay as u32, height, Vec::new())); }
 			}
 			(Some(spend_tx), Some(SpendableOutputDescriptor::StaticOutput { outpoint, output }))
 		} else { (None, None) }
 	}
 
-	fn broadcast_by_local_state(&self, local_tx: &LocalSignedTx, per_commitment_point: &Option<PublicKey>, delayed_payment_base_key: &Option<SecretKey>, height: u32) -> (Vec<Transaction>, Vec<SpendableOutputDescriptor>, Vec<TxOut>, Vec<(BitcoinOutPoint, (u32, TxMaterial, u64, u32, u32))>) {
+	fn broadcast_by_local_state(&self, local_tx: &LocalSignedTx, per_commitment_point: &Option<PublicKey>, delayed_payment_base_key: &Option<SecretKey>, height: u32) -> (Vec<Transaction>, Vec<SpendableOutputDescriptor>, Vec<TxOut>, Vec<(BitcoinOutPoint, (u32, TxMaterial, u64, u32, u32, Vec<BitcoinOutPoint>))>) {
 		let mut res = Vec::with_capacity(local_tx.htlc_outputs.len());
 		let mut spendable_outputs = Vec::with_capacity(local_tx.htlc_outputs.len());
 		let mut watch_outputs = Vec::with_capacity(local_tx.htlc_outputs.len());
@@ -1895,7 +1910,7 @@ impl ChannelMonitor {
 
 						add_dynamic_output!(htlc_timeout_tx, 0);
 						let height_timer = Self::get_height_timer(height, htlc.cltv_expiry);
-						pending_claims.push((htlc_timeout_tx.input[0].previous_output.clone(), (height_timer, TxMaterial::LocalHTLC { script: htlc_script, sigs: (*their_sig, *our_sig), preimage: None, amount: htlc.amount_msat / 1000}, 0, htlc.cltv_expiry, height)));
+						pending_claims.push((htlc_timeout_tx.input[0].previous_output.clone(), (height_timer, TxMaterial::LocalHTLC { script: htlc_script, sigs: (*their_sig, *our_sig), preimage: None, amount: htlc.amount_msat / 1000}, 0, htlc.cltv_expiry, height, Vec::new())));
 						res.push(htlc_timeout_tx);
 					} else {
 						if let Some(payment_preimage) = self.payment_preimages.get(&htlc.payment_hash) {
@@ -1915,7 +1930,7 @@ impl ChannelMonitor {
 
 							add_dynamic_output!(htlc_success_tx, 0);
 							let height_timer = Self::get_height_timer(height, htlc.cltv_expiry);
-							pending_claims.push((htlc_success_tx.input[0].previous_output.clone(), (height_timer, TxMaterial::LocalHTLC { script: htlc_script, sigs: (*their_sig, *our_sig), preimage: Some(*payment_preimage), amount: htlc.amount_msat / 1000}, 0, htlc.cltv_expiry, height)));
+							pending_claims.push((htlc_success_tx.input[0].previous_output.clone(), (height_timer, TxMaterial::LocalHTLC { script: htlc_script, sigs: (*their_sig, *our_sig), preimage: Some(*payment_preimage), amount: htlc.amount_msat / 1000}, 0, htlc.cltv_expiry, height, Vec::new())));
 							res.push(htlc_success_tx);
 						}
 					}
@@ -2210,9 +2225,10 @@ impl ChannelMonitor {
 					bump_candidates.push((claimed_outpoint, claim_tx_data));
 				}
 			}
-			for candidate in bump_candidates {
-				if let Some((new_timer, bumped_tx, feerate)) = self.bump_claim_tx(candidate.0, (candidate.1).0, &(candidate.1).1, (candidate.1).2, fee_estimator) {
-					pending_claims.push((*candidate.0, (new_timer, (candidate.1).1.clone() , feerate, (candidate.1).3, (candidate.1).4)));
+			// TODO: retrieve and sort aggregate claim outpoints based on last_txid
+			for (outpoint, (height_timer, tx_material, old_feerate, timelock_expiration, seen_height, buddy_inputs)) in bump_candidates.drain(..) {
+				if let Some((new_timer, bumped_tx, feerate)) = self.bump_claim_tx(outpoint, *height_timer, &tx_material, *old_feerate, fee_estimator) {
+					pending_claims.push((*outpoint, (new_timer, *tx_material, feerate, *timelock_expiration, *seen_height, *buddy_inputs)));
 					bumped_txn.append(&mut vec![bumped_tx]);
 				}
 			}
@@ -2799,7 +2815,12 @@ impl<R: ::std::io::Read> ReadableArgs<R, Arc<Logger>> for (Sha256dHash, ChannelM
 			let last_fee = Readable::read(reader)?;
 			let timelock_expiration = Readable::read(reader)?;
 			let height = Readable::read(reader)?;
-			our_claim_txn_waiting_first_conf.insert(outpoint, (height_target, tx_material, last_fee, timelock_expiration, height));
+			let buddy_len: u64 = Readable::read(reader)?;
+			let mut buddy_inputs = Vec::new();
+			for _ in 0..buddy_len {
+				buddy_inputs.push(Readable::read(reader)?);
+			}
+			our_claim_txn_waiting_first_conf.insert(outpoint, (height_target, tx_material, last_fee, timelock_expiration, height, buddy_inputs));
 		}
 
 		let waiting_threshold_conf_len: u64 = Readable::read(reader)?;