Support serializing TLV fields which may or may not be present
[rust-lightning] / lightning / src / ln / onchaintx.rs
index f3d8b2e9489d5078dc79dbc42be835fa54fe56f5..16bb646ed1837c61297d3aa17693875f0ea1c14f 100644 (file)
@@ -22,7 +22,7 @@ use bitcoin::secp256k1::{Secp256k1, Signature};
 use bitcoin::secp256k1;
 
 use ln::msgs::DecodeError;
-use ln::channelmanager::PaymentPreimage;
+use ln::PaymentPreimage;
 use ln::chan_utils;
 use ln::chan_utils::{TxCreationKeys, ChannelTransactionParameters, HolderCommitmentTransaction};
 use chain::chaininterface::{FeeEstimator, BroadcasterInterface, ConfirmationTarget, MIN_RELAY_FEE_SAT_PER_1000_WEIGHT};
@@ -32,16 +32,37 @@ use util::logger::Logger;
 use util::ser::{Readable, ReadableArgs, Writer, Writeable, VecWriter};
 use util::byte_utils;
 
-use std::collections::{HashMap, hash_map};
-use std::cmp;
-use std::ops::Deref;
-use std::mem::replace;
+use std::collections::HashMap;
+use core::cmp;
+use core::ops::Deref;
+use core::mem::replace;
 
 const MAX_ALLOC_SIZE: usize = 64*1024;
 
+/// An entry for an [`OnchainEvent`], stating the block height when the event was observed and the
+/// transaction causing it.
+///
+/// Used to determine when the on-chain event can be considered safe from a chain reorganization.
+#[derive(PartialEq)]
+struct OnchainEventEntry {
+       txid: Txid,
+       height: u32,
+       event: OnchainEvent,
+}
+
+impl OnchainEventEntry {
+       fn confirmation_threshold(&self) -> u32 {
+               self.height + ANTI_REORG_DELAY - 1
+       }
+
+       fn has_reached_confirmation_threshold(&self, height: u32) -> bool {
+               height >= self.confirmation_threshold()
+       }
+}
+
 /// Upon discovering of some classes of onchain tx by ChannelMonitor, we may have to take actions on it
 /// once they mature to enough confirmations (ANTI_REORG_DELAY)
-#[derive(Clone, PartialEq)]
+#[derive(PartialEq)]
 enum OnchainEvent {
        /// Outpoint under claim process by our own tx, once this one get enough confirmations, we remove it from
        /// bump-txn candidate buffer.
@@ -199,7 +220,7 @@ impl Readable for Option<Vec<Option<(usize, Signature)>>> {
                        0u8 => Ok(None),
                        1u8 => {
                                let vlen: u64 = Readable::read(reader)?;
-                               let mut ret = Vec::with_capacity(cmp::min(vlen as usize, MAX_ALLOC_SIZE / ::std::mem::size_of::<Option<(usize, Signature)>>()));
+                               let mut ret = Vec::with_capacity(cmp::min(vlen as usize, MAX_ALLOC_SIZE / ::core::mem::size_of::<Option<(usize, Signature)>>()));
                                for _ in 0..vlen {
                                        ret.push(match Readable::read(reader)? {
                                                0u8 => None,
@@ -280,15 +301,20 @@ pub struct OnchainTxHandler<ChannelSigner: Sign> {
        #[cfg(not(test))]
        claimable_outpoints: HashMap<BitcoinOutPoint, (Txid, u32)>,
 
-       onchain_events_waiting_threshold_conf: HashMap<u32, Vec<OnchainEvent>>,
+       onchain_events_awaiting_threshold_conf: Vec<OnchainEventEntry>,
 
        latest_height: u32,
 
        secp_ctx: Secp256k1<secp256k1::All>,
 }
 
+const SERIALIZATION_VERSION: u8 = 1;
+const MIN_SERIALIZATION_VERSION: u8 = 1;
+
 impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
        pub(crate) fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ::std::io::Error> {
+               write_ver_prefix!(writer, SERIALIZATION_VERSION, MIN_SERIALIZATION_VERSION);
+
                self.destination_script.write(writer)?;
                self.holder_commitment.write(writer)?;
                self.holder_htlc_sigs.write(writer)?;
@@ -299,8 +325,8 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
 
                let mut key_data = VecWriter(Vec::new());
                self.signer.write(&mut key_data)?;
-               assert!(key_data.0.len() < std::usize::MAX);
-               assert!(key_data.0.len() < std::u32::MAX as usize);
+               assert!(key_data.0.len() < core::usize::MAX);
+               assert!(key_data.0.len() < core::u32::MAX as usize);
                (key_data.0.len() as u32).write(writer)?;
                writer.write_all(&key_data.0[..])?;
 
@@ -317,31 +343,33 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                        claim_and_height.1.write(writer)?;
                }
 
-               writer.write_all(&byte_utils::be64_to_array(self.onchain_events_waiting_threshold_conf.len() as u64))?;
-               for (ref target, ref events) in self.onchain_events_waiting_threshold_conf.iter() {
-                       writer.write_all(&byte_utils::be32_to_array(**target))?;
-                       writer.write_all(&byte_utils::be64_to_array(events.len() as u64))?;
-                       for ev in events.iter() {
-                               match *ev {
-                                       OnchainEvent::Claim { ref claim_request } => {
-                                               writer.write_all(&[0; 1])?;
-                                               claim_request.write(writer)?;
-                                       },
-                                       OnchainEvent::ContentiousOutpoint { ref outpoint, ref input_material } => {
-                                               writer.write_all(&[1; 1])?;
-                                               outpoint.write(writer)?;
-                                               input_material.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 outpoint, ref input_material } => {
+                                       writer.write_all(&[1; 1])?;
+                                       outpoint.write(writer)?;
+                                       input_material.write(writer)?;
                                }
                        }
                }
                self.latest_height.write(writer)?;
+
+               write_tlv_fields!(writer, {}, {});
                Ok(())
        }
 }
 
 impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler<K::Signer> {
        fn read<R: ::std::io::Read>(reader: &mut R, keys_manager: &'a K) -> Result<Self, DecodeError> {
+               let _ver = read_ver_prefix!(reader, SERIALIZATION_VERSION);
+
                let destination_script = Readable::read(reader)?;
 
                let holder_commitment = Readable::read(reader)?;
@@ -377,35 +405,33 @@ impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler<K::Signer> {
                        claimable_outpoints.insert(outpoint, (ancestor_claim_txid, height));
                }
                let waiting_threshold_conf_len: u64 = Readable::read(reader)?;
-               let mut onchain_events_waiting_threshold_conf = HashMap::with_capacity(cmp::min(waiting_threshold_conf_len as usize, MAX_ALLOC_SIZE / 128));
+               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 height_target = Readable::read(reader)?;
-                       let events_len: u64 = Readable::read(reader)?;
-                       let mut events = Vec::with_capacity(cmp::min(events_len as usize, MAX_ALLOC_SIZE / 128));
-                       for _ in 0..events_len {
-                               let ev = match <u8 as Readable>::read(reader)? {
-                                       0 => {
-                                               let claim_request = Readable::read(reader)?;
-                                               OnchainEvent::Claim {
-                                                       claim_request
-                                               }
-                                       },
-                                       1 => {
-                                               let outpoint = Readable::read(reader)?;
-                                               let input_material = Readable::read(reader)?;
-                                               OnchainEvent::ContentiousOutpoint {
-                                                       outpoint,
-                                                       input_material
-                                               }
+                       let txid = Readable::read(reader)?;
+                       let height = Readable::read(reader)?;
+                       let event = match <u8 as Readable>::read(reader)? {
+                               0 => {
+                                       let claim_request = Readable::read(reader)?;
+                                       OnchainEvent::Claim {
+                                               claim_request
                                        }
-                                       _ => return Err(DecodeError::InvalidValue),
-                               };
-                               events.push(ev);
-                       }
-                       onchain_events_waiting_threshold_conf.insert(height_target, events);
+                               },
+                               1 => {
+                                       let outpoint = Readable::read(reader)?;
+                                       let input_material = Readable::read(reader)?;
+                                       OnchainEvent::ContentiousOutpoint {
+                                               outpoint,
+                                               input_material
+                                       }
+                               }
+                               _ => return Err(DecodeError::InvalidValue),
+                       };
+                       onchain_events_awaiting_threshold_conf.push(OnchainEventEntry { txid, height, event });
                }
                let latest_height = Readable::read(reader)?;
 
+               read_tlv_fields!(reader, {}, {});
+
                let mut secp_ctx = Secp256k1::new();
                secp_ctx.seeded_randomize(&keys_manager.get_secure_random_bytes());
 
@@ -419,7 +445,7 @@ impl<'a, K: KeysInterface> ReadableArgs<&'a K> for OnchainTxHandler<K::Signer> {
                        channel_transaction_parameters: channel_parameters,
                        claimable_outpoints,
                        pending_claim_requests,
-                       onchain_events_waiting_threshold_conf,
+                       onchain_events_awaiting_threshold_conf,
                        latest_height,
                        secp_ctx,
                })
@@ -438,7 +464,7 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                        channel_transaction_parameters: channel_parameters,
                        pending_claim_requests: HashMap::new(),
                        claimable_outpoints: HashMap::new(),
-                       onchain_events_waiting_threshold_conf: HashMap::new(),
+                       onchain_events_awaiting_threshold_conf: Vec::new(),
                        latest_height: 0,
 
                        secp_ctx,
@@ -613,7 +639,11 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                                                                chan_utils::get_revokeable_redeemscript(&tx_keys.revocation_key, *on_counterparty_tx_csv, &tx_keys.broadcaster_delayed_payment_key)
                                                        };
 
-                                                       let sig = self.signer.sign_justice_transaction(&bumped_tx, i, *amount, &per_commitment_key, htlc, &self.secp_ctx).expect("sign justice tx");
+                                                       let sig = if let Some(ref htlc) = *htlc {
+                                                               self.signer.sign_justice_revoked_htlc(&bumped_tx, i, *amount, &per_commitment_key, &htlc, &self.secp_ctx).expect("sign justice tx")
+                                                       } else {
+                                                               self.signer.sign_justice_revoked_output(&bumped_tx, i, *amount, &per_commitment_key, &self.secp_ctx).expect("sign justice tx")
+                                                       };
                                                        bumped_tx.input[i].witness.push(sig.serialize_der().to_vec());
                                                        bumped_tx.input[i].witness[0].push(SigHashType::All as u8);
                                                        if htlc.is_some() {
@@ -692,7 +722,7 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                log_trace!(logger, "Updating claims view at height {} with {} matched transactions and {} claim requests", height, txn_matched.len(), claimable_outpoints.len());
                let mut new_claims = Vec::new();
                let mut aggregated_claim = HashMap::new();
-               let mut aggregated_soonest = ::std::u32::MAX;
+               let mut aggregated_soonest = ::core::u32::MAX;
 
                // Try to aggregate outputs if their timelock expiration isn't imminent (absolute_timelock
                // <= CLTV_SHARED_CLAIM_BUFFER) and they don't require an immediate nLockTime (aggregable).
@@ -727,7 +757,7 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                                        self.claimable_outpoints.insert(k.clone(), (txid, height));
                                }
                                self.pending_claim_requests.insert(txid, claim_material);
-                               log_trace!(logger, "Broadcast onchain {}", log_tx!(tx));
+                               log_info!(logger, "Broadcasting onchain {}", log_tx!(tx));
                                broadcaster.broadcast_transaction(&tx);
                        }
                }
@@ -756,16 +786,13 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
 
                                                macro_rules! clean_claim_request_after_safety_delay {
                                                        () => {
-                                                               let new_event = OnchainEvent::Claim { claim_request: first_claim_txid_height.0.clone() };
-                                                               match self.onchain_events_waiting_threshold_conf.entry(height + ANTI_REORG_DELAY - 1) {
-                                                                       hash_map::Entry::Occupied(mut entry) => {
-                                                                               if !entry.get().contains(&new_event) {
-                                                                                       entry.get_mut().push(new_event);
-                                                                               }
-                                                                       },
-                                                                       hash_map::Entry::Vacant(entry) => {
-                                                                               entry.insert(vec![new_event]);
-                                                                       }
+                                                               let entry = OnchainEventEntry {
+                                                                       txid: tx.txid(),
+                                                                       height,
+                                                                       event: OnchainEvent::Claim { claim_request: first_claim_txid_height.0.clone() }
+                                                               };
+                                                               if !self.onchain_events_awaiting_threshold_conf.contains(&entry) {
+                                                                       self.onchain_events_awaiting_threshold_conf.push(entry);
                                                                }
                                                        }
                                                }
@@ -799,24 +826,23 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                                }
                        }
                        for (outpoint, input_material) in claimed_outputs_material.drain(..) {
-                               let new_event = OnchainEvent::ContentiousOutpoint { outpoint, input_material };
-                               match self.onchain_events_waiting_threshold_conf.entry(height + ANTI_REORG_DELAY - 1) {
-                                       hash_map::Entry::Occupied(mut entry) => {
-                                               if !entry.get().contains(&new_event) {
-                                                       entry.get_mut().push(new_event);
-                                               }
-                                       },
-                                       hash_map::Entry::Vacant(entry) => {
-                                               entry.insert(vec![new_event]);
-                                       }
+                               let entry = OnchainEventEntry {
+                                       txid: tx.txid(),
+                                       height,
+                                       event: OnchainEvent::ContentiousOutpoint { outpoint, input_material },
+                               };
+                               if !self.onchain_events_awaiting_threshold_conf.contains(&entry) {
+                                       self.onchain_events_awaiting_threshold_conf.push(entry);
                                }
                        }
                }
 
                // After security delay, either our claim tx got enough confs or outpoint is definetely out of reach
-               if let Some(events) = self.onchain_events_waiting_threshold_conf.remove(&height) {
-                       for ev in events {
-                               match ev {
+               let onchain_events_awaiting_threshold_conf =
+                       self.onchain_events_awaiting_threshold_conf.drain(..).collect::<Vec<_>>();
+               for entry in onchain_events_awaiting_threshold_conf {
+                       if entry.has_reached_confirmation_threshold(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
@@ -830,13 +856,15 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                                                self.claimable_outpoints.remove(&outpoint);
                                        }
                                }
+                       } else {
+                               self.onchain_events_awaiting_threshold_conf.push(entry);
                        }
                }
 
                // Check if any pending claim request must be rescheduled
                for (first_claim_txid, ref claim_data) in self.pending_claim_requests.iter() {
-                       if let Some(h) = claim_data.height_timer {
-                               if h == height {
+                       if let Some(height_timer) = claim_data.height_timer {
+                               if height >= height_timer {
                                        bump_candidates.insert(*first_claim_txid, (*claim_data).clone());
                                }
                        }
@@ -846,7 +874,7 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                log_trace!(logger, "Bumping {} candidates", bump_candidates.len());
                for (first_claim_txid, claim_material) in bump_candidates.iter() {
                        if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &claim_material, &*fee_estimator, &*logger) {
-                               log_trace!(logger, "Broadcast onchain {}", log_tx!(bump_tx));
+                               log_info!(logger, "Broadcasting onchain {}", log_tx!(bump_tx));
                                broadcaster.broadcast_transaction(&bump_tx);
                                if let Some(claim_material) = self.pending_claim_requests.get_mut(first_claim_txid) {
                                        claim_material.height_timer = new_timer;
@@ -856,17 +884,43 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                }
        }
 
+       pub(crate) fn transaction_unconfirmed<B: Deref, F: Deref, L: Deref>(
+               &mut self,
+               txid: &Txid,
+               broadcaster: B,
+               fee_estimator: F,
+               logger: L,
+       ) where
+               B::Target: BroadcasterInterface,
+               F::Target: FeeEstimator,
+               L::Target: Logger,
+       {
+               let mut height = None;
+               for entry in self.onchain_events_awaiting_threshold_conf.iter() {
+                       if entry.txid == *txid {
+                               height = Some(entry.height);
+                               break;
+                       }
+               }
+
+               if let Some(height) = height {
+                       self.block_disconnected(height, broadcaster, fee_estimator, logger);
+               }
+       }
+
        pub(crate) fn block_disconnected<B: Deref, F: Deref, L: Deref>(&mut self, height: u32, broadcaster: B, fee_estimator: F, logger: L)
                where B::Target: BroadcasterInterface,
                      F::Target: FeeEstimator,
                                        L::Target: Logger,
        {
                let mut bump_candidates = HashMap::new();
-               if let Some(events) = self.onchain_events_waiting_threshold_conf.remove(&(height + ANTI_REORG_DELAY - 1)) {
-                       //- our claim tx on a commitment tx output
-                       //- resurect outpoint back in its claimable set and regenerate tx
-                       for ev in events {
-                               match ev {
+               let onchain_events_awaiting_threshold_conf =
+                       self.onchain_events_awaiting_threshold_conf.drain(..).collect::<Vec<_>>();
+               for entry in onchain_events_awaiting_threshold_conf {
+                       if entry.height >= height {
+                               //- our claim tx on a commitment tx output
+                               //- resurect outpoint back in its claimable set and regenerate tx
+                               match entry.event {
                                        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) {
@@ -879,12 +933,15 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                                        },
                                        _ => {},
                                }
+                       } else {
+                               self.onchain_events_awaiting_threshold_conf.push(entry);
                        }
                }
                for (_, claim_material) in bump_candidates.iter_mut() {
                        if let Some((new_timer, new_feerate, bump_tx)) = self.generate_claim_tx(height, &claim_material, &&*fee_estimator, &&*logger) {
                                claim_material.height_timer = new_timer;
                                claim_material.feerate_previous = new_feerate;
+                               log_info!(logger, "Broadcasting onchain {}", log_tx!(bump_tx));
                                broadcaster.broadcast_transaction(&bump_tx);
                        }
                }
@@ -895,7 +952,7 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                // 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 {
+                       if v.1 >= height {
                        remove_request.push(v.0.clone());
                        false
                        } else { true });
@@ -904,6 +961,16 @@ impl<ChannelSigner: Sign> OnchainTxHandler<ChannelSigner> {
                }
        }
 
+       pub(crate) fn get_relevant_txids(&self) -> Vec<Txid> {
+               let mut txids: Vec<Txid> = self.onchain_events_awaiting_threshold_conf
+                       .iter()
+                       .map(|entry| entry.txid)
+                       .collect();
+               txids.sort_unstable();
+               txids.dedup();
+               txids
+       }
+
        pub(crate) fn provide_latest_holder_tx(&mut self, tx: HolderCommitmentTransaction) {
                self.prev_holder_commitment = Some(replace(&mut self.holder_commitment, tx));
                self.holder_htlc_sigs = None;