Delay broadcast of PackageTemplate packages until their locktime

This stores transaction templates temporarily until their locktime
is reached, avoiding broadcasting (or RBF bumping) transactions
prior to their locktime. For those broadcasting transactions
(potentially indirectly) via Bitcoin Core RPC, this ensures no
automated rebroadcast of transactions on the client side is
required to get transactions confirmed.

Author: TheBlueMatt

lightning/src/chain/onchaintx.rs

@@ -32,7 +32,7 @@ use util::logger::Logger;
 use util::ser::{Readable, ReadableArgs, Writer, Writeable, VecWriter};
 use util::byte_utils;
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use core::cmp;
 use core::ops::Deref;
 use core::mem::replace;
@@ -164,8 +164,9 @@ pub struct OnchainTxHandler<ChannelSigner: Sign> {
 	#[cfg(not(test))]
 	claimable_outpoints: HashMap<BitcoinOutPoint, (Txid, u32)>,
 
-	onchain_events_awaiting_threshold_conf: Vec<OnchainEventEntry>,
+	locktimed_packages: BTreeMap<u32, Vec<PackageTemplate>>,
 
+	onchain_events_awaiting_threshold_conf: Vec<OnchainEventEntry>,
 
 	pub(super) secp_ctx: Secp256k1<secp256k1::All>,
 }
@@ -205,6 +206,15 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
 			claim_and_height.1.write(writer)?;
 		}
 
+		writer.write_all(&byte_utils::be64_to_array(self.locktimed_packages.len() as u64))?;
+		for (ref locktime, ref packages) in self.locktimed_packages.iter() {
+			locktime.write(writer)?;
+			writer.write_all(&byte_utils::be64_to_array(packages.len() as u64))?;
+			for ref package in packages.iter() {
+				package.write(writer)?;
+			}
+		}
+
 		writer.write_all(&byte_utils::be64_to_array(self.onchain_events_awaiting_threshold_conf.len() as u64))?;
 		for ref entry in self.onchain_events_awaiting_threshold_conf.iter() {
 			entry.txid.write(writer)?;
@@ -264,6 +274,19 @@ impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler<K::Signer> {
 			let height = Readable::read(reader)?;
 			claimable_outpoints.insert(outpoint, (ancestor_claim_txid, height));
 		}
+
+		let locktimed_packages_len: u64 = Readable::read(reader)?;
+		let mut locktimed_packages = BTreeMap::new();
+		for _ in 0..locktimed_packages_len {
+			let locktime = Readable::read(reader)?;
+			let packages_len: u64 = Readable::read(reader)?;
+			let mut packages = Vec::with_capacity(cmp::min(packages_len as usize, MAX_ALLOC_SIZE / std::mem::size_of::<PackageTemplate>()));
+			for _ in 0..packages_len {
+				packages.push(Readable::read(reader)?);
+			}
+			locktimed_packages.insert(locktime, packages);
+		}
+
 		let waiting_threshold_conf_len: u64 = Readable::read(reader)?;
 		let mut onchain_events_awaiting_threshold_conf = Vec::with_capacity(cmp::min(waiting_threshold_conf_len as usize, MAX_ALLOC_SIZE / 128));
 		for _ in 0..waiting_threshold_conf_len {
@@ -301,6 +324,7 @@ impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler<K::Signer> {
 			signer,
 			channel_transaction_parameters: channel_parameters,
 			claimable_outpoints,
+			locktimed_packages,
 			pending_claim_requests,
 			onchain_events_awaiting_threshold_conf,
 			secp_ctx,
@@ -320,6 +344,7 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
 			channel_transaction_parameters: channel_parameters,
 			pending_claim_requests: HashMap::new(),
 			claimable_outpoints: HashMap::new(),
+			locktimed_packages: BTreeMap::new(),
 			onchain_events_awaiting_threshold_conf: Vec::new(),
 
 			secp_ctx,
@@ -375,9 +400,17 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
 
 		// Try to aggregate outputs if their timelock expiration isn't imminent (package timelock
 		// <= CLTV_SHARED_CLAIM_BUFFER) and they don't require an immediate nLockTime (aggregable).
-		for req in requests {
+		'req_walk: for req in requests {
 			// Don't claim a outpoint twice that would be bad for privacy and may uselessly lock a CPFP input for a while
-			if let Some(_) = self.claimable_outpoints.get(req.outpoints()[0]) { log_trace!(logger, "Bouncing off outpoint {}:{}, already registered its claiming request", req.outpoints()[0].txid, req.outpoints()[0].vout); } else {
+			if let Some(_) = self.claimable_outpoints.get(req.outpoints()[0]) {
+				log_trace!(logger, "Bouncing off outpoint {}:{}, already registered its claiming request", req.outpoints()[0].txid, req.outpoints()[0].vout);
+			} else {
+				for locked_package in self.locktimed_packages.iter().map(|v| v.1.iter()).flatten() {
+					if locked_package.outpoints() == req.outpoints() {
+						continue 'req_walk;
+					}
+				}
+
 				log_trace!(logger, "Test if outpoint can be aggregated with expiration {} against {}", req.timelock(), height + CLTV_SHARED_CLAIM_BUFFER);
 				if req.timelock() <= height + CLTV_SHARED_CLAIM_BUFFER || !req.aggregable() {
 					// Don't aggregate if outpoint package timelock is soon or marked as non-aggregable
@@ -393,10 +426,26 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
 			preprocessed_requests.push(req);
 		}
 
+		loop {
+			let pop_height = if let Some(first_entry) = self.locktimed_packages.iter().next() {
+				if *first_entry.0 <= height + 1 {
+					*first_entry.0
+				} else { break; }
+			} else { break; };
+			log_trace!(logger, "Restoring delayed claim of package(s) at their timelock at {}.", pop_height);
+			preprocessed_requests.append(&mut self.locktimed_packages.remove(&pop_height).unwrap());
+		}
+
 		// Generate claim transactions and track them to bump if necessary at
 		// height timer expiration (i.e in how many blocks we're going to take action).
 		for mut req in preprocessed_requests {
-			if let Some((new_timer, new_feerate, tx)) = self.generate_claim_tx(height, &req, &*fee_estimator, &*logger) {
+			if req.package_timelock() > height + 1 {
+				log_debug!(logger, "Delaying claim of package until its timelock at {} (current height {}), the following outpoints are spent:", req.package_timelock(), height);
+				for outpoint in req.outpoints() {
+					log_debug!(logger, "  Outpoint {}", outpoint);
+				}
+				self.locktimed_packages.entry(req.package_timelock()).or_insert(Vec::new()).push(req);
+			} else if let Some((new_timer, new_feerate, tx)) = self.generate_claim_tx(height, &req, &*fee_estimator, &*logger) {
 				req.set_timer(new_timer);
 				req.set_feerate(new_feerate);
 				let txid = tx.txid();