X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fchain%2Fonchaintx.rs;h=ac01cacaa8649d6676050e3b6eb4d173026250ed;hb=28c9b56113ff1ebb1b505a2c979c55c1626aa06b;hp=04de721f901a3a9e67ea01bca8ce2c6e3df3f362;hpb=12461fcba179a1b6726f5c2557d12cb1f1b33b25;p=rust-lightning diff --git a/lightning/src/chain/onchaintx.rs b/lightning/src/chain/onchaintx.rs index 04de721f..ac01caca 100644 --- a/lightning/src/chain/onchaintx.rs +++ b/lightning/src/chain/onchaintx.rs @@ -18,22 +18,23 @@ use bitcoin::blockdata::script::Script; use bitcoin::hash_types::Txid; -use bitcoin::secp256k1::{Secp256k1, Signature}; +use bitcoin::secp256k1::{Secp256k1, ecdsa::Signature}; use bitcoin::secp256k1; use ln::msgs::DecodeError; use ln::PaymentPreimage; use ln::chan_utils::{ChannelTransactionParameters, HolderCommitmentTransaction}; -use chain::chaininterface::{FeeEstimator, BroadcasterInterface}; +use chain::chaininterface::{FeeEstimator, BroadcasterInterface, LowerBoundedFeeEstimator}; use chain::channelmonitor::{ANTI_REORG_DELAY, CLTV_SHARED_CLAIM_BUFFER}; use chain::keysinterface::{Sign, KeysInterface}; use chain::package::PackageTemplate; use util::logger::Logger; -use util::ser::{Readable, ReadableArgs, Writer, Writeable, VecWriter}; +use util::ser::{Readable, ReadableArgs, MaybeReadable, Writer, Writeable, VecWriter}; use util::byte_utils; +use io; use prelude::*; -use std::collections::HashMap; +use alloc::collections::BTreeMap; use core::cmp; use core::ops::Deref; use core::mem::replace; @@ -78,8 +79,46 @@ enum OnchainEvent { } } +impl Writeable for OnchainEventEntry { + fn write(&self, writer: &mut W) -> Result<(), io::Error> { + write_tlv_fields!(writer, { + (0, self.txid, required), + (2, self.height, required), + (4, self.event, required), + }); + Ok(()) + } +} + +impl MaybeReadable for OnchainEventEntry { + fn read(reader: &mut R) -> Result, DecodeError> { + let mut txid = Default::default(); + let mut height = 0; + let mut event = None; + read_tlv_fields!(reader, { + (0, txid, required), + (2, height, required), + (4, event, ignorable), + }); + if let Some(ev) = event { + Ok(Some(Self { txid, height, event: ev })) + } else { + Ok(None) + } + } +} + +impl_writeable_tlv_based_enum_upgradable!(OnchainEvent, + (0, Claim) => { + (0, claim_request, required), + }, + (1, ContentiousOutpoint) => { + (0, package, required), + }, +); + impl Readable for Option>> { - fn read(reader: &mut R) -> Result { + fn read(reader: &mut R) -> Result { match Readable::read(reader)? { 0u8 => Ok(None), 1u8 => { @@ -100,7 +139,7 @@ impl Readable for Option>> { } impl Writeable for Option>> { - fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + fn write(&self, writer: &mut W) -> Result<(), io::Error> { match self { &Some(ref vec) => { 1u8.write(writer)?; @@ -165,9 +204,9 @@ pub struct OnchainTxHandler { #[cfg(not(test))] claimable_outpoints: HashMap, - onchain_events_awaiting_threshold_conf: Vec, + locktimed_packages: BTreeMap>, - latest_height: u32, + onchain_events_awaiting_threshold_conf: Vec, pub(super) secp_ctx: Secp256k1, } @@ -176,7 +215,7 @@ const SERIALIZATION_VERSION: u8 = 1; const MIN_SERIALIZATION_VERSION: u8 = 1; impl OnchainTxHandler { - pub(crate) fn write(&self, writer: &mut W) -> Result<(), ::std::io::Error> { + pub(crate) fn write(&self, writer: &mut W) -> Result<(), io::Error> { write_ver_prefix!(writer, SERIALIZATION_VERSION, MIN_SERIALIZATION_VERSION); self.destination_script.write(writer)?; @@ -207,30 +246,27 @@ impl OnchainTxHandler { 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)?; - writer.write_all(&byte_utils::be32_to_array(entry.height))?; - match entry.event { - OnchainEvent::Claim { ref claim_request } => { - writer.write_all(&[0; 1])?; - claim_request.write(writer)?; - }, - OnchainEvent::ContentiousOutpoint { ref package } => { - writer.write_all(&[1; 1])?; - package.write(writer)?; - } - } + entry.write(writer)?; } - self.latest_height.write(writer)?; - write_tlv_fields!(writer, {}, {}); + write_tlv_fields!(writer, {}); Ok(()) } } impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler { - fn read(reader: &mut R, keys_manager: &'a K) -> Result { + fn read(reader: &mut R, keys_manager: &'a K) -> Result { let _ver = read_ver_prefix!(reader, SERIALIZATION_VERSION); let destination_script = Readable::read(reader)?; @@ -267,31 +303,28 @@ impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler { 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 / core::mem::size_of::())); + 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 { - let txid = Readable::read(reader)?; - let height = Readable::read(reader)?; - let event = match ::read(reader)? { - 0 => { - let claim_request = Readable::read(reader)?; - OnchainEvent::Claim { - claim_request - } - }, - 1 => { - let package = Readable::read(reader)?; - OnchainEvent::ContentiousOutpoint { - package - } - } - _ => return Err(DecodeError::InvalidValue), - }; - onchain_events_awaiting_threshold_conf.push(OnchainEventEntry { txid, height, event }); + if let Some(val) = MaybeReadable::read(reader)? { + onchain_events_awaiting_threshold_conf.push(val); + } } - let latest_height = Readable::read(reader)?; - read_tlv_fields!(reader, {}, {}); + read_tlv_fields!(reader, {}); let mut secp_ctx = Secp256k1::new(); secp_ctx.seeded_randomize(&keys_manager.get_secure_random_bytes()); @@ -305,9 +338,9 @@ impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler { signer, channel_transaction_parameters: channel_parameters, claimable_outpoints, + locktimed_packages, pending_claim_requests, onchain_events_awaiting_threshold_conf, - latest_height, secp_ctx, }) } @@ -325,18 +358,26 @@ impl OnchainTxHandler { 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(), - latest_height: 0, secp_ctx, } } + pub(crate) fn get_prev_holder_commitment_to_self_value(&self) -> Option { + self.prev_holder_commitment.as_ref().map(|commitment| commitment.to_broadcaster_value_sat()) + } + + pub(crate) fn get_cur_holder_commitment_to_self_value(&self) -> u64 { + self.holder_commitment.to_broadcaster_value_sat() + } + /// Lightning security model (i.e being able to redeem/timeout HTLC or penalize coutnerparty onchain) lays on the assumption of claim transactions getting confirmed before timelock expiration /// (CSV or CLTV following cases). In case of high-fee spikes, claim tx may stuck in the mempool, so you need to bump its feerate quickly using Replace-By-Fee or Child-Pay-For-Parent. /// Panics if there are signing errors, because signing operations in reaction to on-chain events /// are not expected to fail, and if they do, we may lose funds. - fn generate_claim_tx(&mut self, height: u32, cached_request: &PackageTemplate, fee_estimator: &F, logger: &L) -> Option<(Option, u64, Transaction)> + fn generate_claim_tx(&mut self, cur_height: u32, cached_request: &PackageTemplate, fee_estimator: &LowerBoundedFeeEstimator, logger: &L) -> Option<(Option, u64, Transaction)> where F::Target: FeeEstimator, L::Target: Logger, { @@ -344,24 +385,23 @@ impl OnchainTxHandler { // Compute new height timer to decide when we need to regenerate a new bumped version of the claim tx (if we // didn't receive confirmation of it before, or not enough reorg-safe depth on top of it). - let new_timer = Some(cached_request.get_height_timer(height)); - let amt = cached_request.package_amount(); + let new_timer = Some(cached_request.get_height_timer(cur_height)); if cached_request.is_malleable() { - let predicted_weight = cached_request.package_weight(&self.destination_script); - if let Some((output_value, new_feerate)) = cached_request.compute_package_output(predicted_weight, amt, fee_estimator, logger) { + let predicted_weight = cached_request.package_weight(&self.destination_script, self.channel_transaction_parameters.opt_anchors.is_some()); + if let Some((output_value, new_feerate)) = + cached_request.compute_package_output(predicted_weight, self.destination_script.dust_value().as_sat(), fee_estimator, logger) { assert!(new_feerate != 0); let transaction = cached_request.finalize_package(self, output_value, self.destination_script.clone(), logger).unwrap(); log_trace!(logger, "...with timer {} and feerate {}", new_timer.unwrap(), new_feerate); - assert!(predicted_weight >= transaction.get_weight()); + assert!(predicted_weight >= transaction.weight()); return Some((new_timer, new_feerate, transaction)) } } else { // Note: Currently, amounts of holder outputs spending witnesses aren't used // as we can't malleate spending package to increase their feerate. This // should change with the remaining anchor output patchset. - debug_assert!(amt == 0); - if let Some(transaction) = cached_request.finalize_package(self, amt, self.destination_script.clone(), logger) { + if let Some(transaction) = cached_request.finalize_package(self, 0, self.destination_script.clone(), logger) { return Some((None, 0, transaction)); } } @@ -372,16 +412,15 @@ impl OnchainTxHandler { /// for this channel, provide new relevant on-chain transactions and/or new claim requests. /// Formerly this was named `block_connected`, but it is now also used for claiming an HTLC output /// if we receive a preimage after force-close. - pub(crate) fn update_claims_view(&mut self, txn_matched: &[&Transaction], requests: Vec, latest_height: Option, broadcaster: &B, fee_estimator: &F, logger: &L) + /// `conf_height` represents the height at which the transactions in `txn_matched` were + /// confirmed. This does not need to equal the current blockchain tip height, which should be + /// provided via `cur_height`, however it must never be higher than `cur_height`. + pub(crate) fn update_claims_view(&mut self, txn_matched: &[&Transaction], requests: Vec, conf_height: u32, cur_height: u32, broadcaster: &B, fee_estimator: &LowerBoundedFeeEstimator, logger: &L) where B::Target: BroadcasterInterface, F::Target: FeeEstimator, L::Target: Logger, { - let height = match latest_height { - Some(h) => h, - None => self.latest_height, - }; - log_trace!(logger, "Updating claims view at height {} with {} matched transactions and {} claim requests", height, txn_matched.len(), requests.len()); + log_debug!(logger, "Updating claims view at height {} with {} matched transactions in block {} and {} claim requests", cur_height, txn_matched.len(), conf_height, requests.len()); let mut preprocessed_requests = Vec::with_capacity(requests.len()); let mut aggregated_request = None; @@ -389,9 +428,28 @@ impl OnchainTxHandler { // <= CLTV_SHARED_CLAIM_BUFFER) and they don't require an immediate nLockTime (aggregable). 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 { - 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() { + if let Some(_) = self.claimable_outpoints.get(req.outpoints()[0]) { + log_info!(logger, "Ignoring second claim for outpoint {}:{}, already registered its claiming request", req.outpoints()[0].txid, req.outpoints()[0].vout); + } else { + let timelocked_equivalent_package = self.locktimed_packages.iter().map(|v| v.1.iter()).flatten() + .find(|locked_package| locked_package.outpoints() == req.outpoints()); + if let Some(package) = timelocked_equivalent_package { + log_info!(logger, "Ignoring second claim for outpoint {}:{}, we already have one which we're waiting on a timelock at {} for.", + req.outpoints()[0].txid, req.outpoints()[0].vout, package.package_timelock()); + continue; + } + + if req.package_timelock() > cur_height + 1 { + log_info!(logger, "Delaying claim of package until its timelock at {} (current height {}), the following outpoints are spent:", req.package_timelock(), cur_height); + for outpoint in req.outpoints() { + log_info!(logger, " Outpoint {}", outpoint); + } + self.locktimed_packages.entry(req.package_timelock()).or_insert(Vec::new()).push(req); + continue; + } + + log_trace!(logger, "Test if outpoint can be aggregated with expiration {} against {}", req.timelock(), cur_height + CLTV_SHARED_CLAIM_BUFFER); + if req.timelock() <= cur_height + CLTV_SHARED_CLAIM_BUFFER || !req.aggregable() { // Don't aggregate if outpoint package timelock is soon or marked as non-aggregable preprocessed_requests.push(req); } else if aggregated_request.is_none() { @@ -405,19 +463,27 @@ impl OnchainTxHandler { preprocessed_requests.push(req); } + // Claim everything up to and including cur_height + 1 + let remaining_locked_packages = self.locktimed_packages.split_off(&(cur_height + 2)); + for (pop_height, mut entry) in self.locktimed_packages.iter_mut() { + log_trace!(logger, "Restoring delayed claim of package(s) at their timelock at {}.", pop_height); + preprocessed_requests.append(&mut entry); + } + self.locktimed_packages = remaining_locked_packages; + // 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 let Some((new_timer, new_feerate, tx)) = self.generate_claim_tx(cur_height, &req, &*fee_estimator, &*logger) { req.set_timer(new_timer); req.set_feerate(new_feerate); let txid = tx.txid(); for k in req.outpoints() { - log_trace!(logger, "Registering claiming request for {}:{}", k.txid, k.vout); - self.claimable_outpoints.insert(k.clone(), (txid, height)); + log_info!(logger, "Registering claiming request for {}:{}", k.txid, k.vout); + self.claimable_outpoints.insert(k.clone(), (txid, conf_height)); } self.pending_claim_requests.insert(txid, req); - log_trace!(logger, "Broadcasting onchain {}", log_tx!(tx)); + log_info!(logger, "Broadcasting onchain {}", log_tx!(tx)); broadcaster.broadcast_transaction(&tx); } } @@ -448,7 +514,7 @@ impl OnchainTxHandler { () => { let entry = OnchainEventEntry { txid: tx.txid(), - height, + height: conf_height, event: OnchainEvent::Claim { claim_request: first_claim_txid_height.0.clone() } }; if !self.onchain_events_awaiting_threshold_conf.contains(&entry) { @@ -488,7 +554,7 @@ impl OnchainTxHandler { for package in claimed_outputs_material.drain(..) { let entry = OnchainEventEntry { txid: tx.txid(), - height, + height: conf_height, event: OnchainEvent::ContentiousOutpoint { package }, }; if !self.onchain_events_awaiting_threshold_conf.contains(&entry) { @@ -501,18 +567,21 @@ impl OnchainTxHandler { let onchain_events_awaiting_threshold_conf = self.onchain_events_awaiting_threshold_conf.drain(..).collect::>(); for entry in onchain_events_awaiting_threshold_conf { - if entry.has_reached_confirmation_threshold(height) { + if entry.has_reached_confirmation_threshold(cur_height) { match entry.event { OnchainEvent::Claim { claim_request } => { // We may remove a whole set of claim outpoints here, as these one may have // been aggregated in a single tx and claimed so atomically if let Some(request) = self.pending_claim_requests.remove(&claim_request) { for outpoint in request.outpoints() { + log_debug!(logger, "Removing claim tracking for {} due to maturation of claim tx {}.", outpoint, claim_request); self.claimable_outpoints.remove(&outpoint); } } }, OnchainEvent::ContentiousOutpoint { package } => { + log_debug!(logger, "Removing claim tracking due to maturation of claim tx for outpoints:"); + log_debug!(logger, " {:?}", package.outpoints()); self.claimable_outpoints.remove(&package.outpoints()[0]); } } @@ -524,7 +593,7 @@ impl OnchainTxHandler { // Check if any pending claim request must be rescheduled for (first_claim_txid, ref request) in self.pending_claim_requests.iter() { if let Some(h) = request.timer() { - if height >= h { + if cur_height >= h { bump_candidates.insert(*first_claim_txid, (*request).clone()); } } @@ -533,8 +602,8 @@ impl OnchainTxHandler { // Build, bump and rebroadcast tx accordingly log_trace!(logger, "Bumping {} candidates", bump_candidates.len()); for (first_claim_txid, request) in bump_candidates.iter() { - if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &request, &*fee_estimator, &*logger) { - log_trace!(logger, "Broadcasting onchain {}", log_tx!(bump_tx)); + if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(cur_height, &request, &*fee_estimator, &*logger) { + log_info!(logger, "Broadcasting RBF-bumped onchain {}", log_tx!(bump_tx)); broadcaster.broadcast_transaction(&bump_tx); if let Some(request) = self.pending_claim_requests.get_mut(first_claim_txid) { request.set_timer(new_timer); @@ -548,7 +617,7 @@ impl OnchainTxHandler { &mut self, txid: &Txid, broadcaster: B, - fee_estimator: F, + fee_estimator: &LowerBoundedFeeEstimator, logger: L, ) where B::Target: BroadcasterInterface, @@ -568,7 +637,7 @@ impl OnchainTxHandler { } } - pub(crate) fn block_disconnected(&mut self, height: u32, broadcaster: B, fee_estimator: F, logger: L) + pub(crate) fn block_disconnected(&mut self, height: u32, broadcaster: B, fee_estimator: &LowerBoundedFeeEstimator, logger: L) where B::Target: BroadcasterInterface, F::Target: FeeEstimator, L::Target: Logger, @@ -598,7 +667,7 @@ impl OnchainTxHandler { } } for (_, request) in bump_candidates.iter_mut() { - if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &request, &&*fee_estimator, &&*logger) { + if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &request, fee_estimator, &&*logger) { request.set_timer(new_timer); request.set_feerate(new_feerate); log_info!(logger, "Broadcasting onchain {}", log_tx!(bump_tx)); @@ -717,6 +786,10 @@ impl OnchainTxHandler { htlc_tx } + pub(crate) fn opt_anchors(&self) -> bool { + self.channel_transaction_parameters.opt_anchors.is_some() + } + #[cfg(any(test,feature = "unsafe_revoked_tx_signing"))] pub(crate) fn unsafe_get_fully_signed_htlc_tx(&mut self, outp: &::bitcoin::OutPoint, preimage: &Option) -> Option { let latest_had_sigs = self.holder_htlc_sigs.is_some();