Merge pull request #840 from jkczyz/2021-03-rescan-logic
authorMatt Corallo <649246+TheBlueMatt@users.noreply.github.com>
Sun, 28 Mar 2021 19:47:09 +0000 (19:47 +0000)
committerGitHub <noreply@github.com>
Sun, 28 Mar 2021 19:47:09 +0000 (19:47 +0000)
Rescan dependent transactions in ChainMonitor

lightning/src/chain/chainmonitor.rs
lightning/src/chain/channelmonitor.rs
lightning/src/chain/mod.rs
lightning/src/ln/functional_test_utils.rs
lightning/src/util/test_utils.rs

index eb17b469a0365df2de4464ef313cc243441b6767..0fd088019247e4e8e9adbea39adb99798e274dd4 100644 (file)
@@ -26,7 +26,7 @@
 use bitcoin::blockdata::block::{Block, BlockHeader};
 
 use chain;
-use chain::Filter;
+use chain::{Filter, WatchedOutput};
 use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use chain::channelmonitor;
 use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent, Persist};
@@ -82,18 +82,40 @@ where C::Target: chain::Filter,
        /// descendants of such transactions. It is not necessary to re-fetch the block to obtain
        /// updated `txdata`.
        pub fn block_connected(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
+               let mut dependent_txdata = Vec::new();
                let monitors = self.monitors.read().unwrap();
                for monitor in monitors.values() {
                        let mut txn_outputs = monitor.block_connected(header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
 
+                       // Register any new outputs with the chain source for filtering, storing any dependent
+                       // transactions from within the block that previously had not been included in txdata.
                        if let Some(ref chain_source) = self.chain_source {
+                               let block_hash = header.block_hash();
                                for (txid, outputs) in txn_outputs.drain(..) {
                                        for (idx, output) in outputs.iter() {
-                                               chain_source.register_output(&OutPoint { txid, index: *idx as u16 }, &output.script_pubkey);
+                                               // Register any new outputs with the chain source for filtering and recurse
+                                               // if it indicates that there are dependent transactions within the block
+                                               // that had not been previously included in txdata.
+                                               let output = WatchedOutput {
+                                                       block_hash: Some(block_hash),
+                                                       outpoint: OutPoint { txid, index: *idx as u16 },
+                                                       script_pubkey: output.script_pubkey.clone(),
+                                               };
+                                               if let Some(tx) = chain_source.register_output(output) {
+                                                       dependent_txdata.push(tx);
+                                               }
                                        }
                                }
                        }
                }
+
+               // Recursively call for any dependent transactions that were identified by the chain source.
+               if !dependent_txdata.is_empty() {
+                       dependent_txdata.sort_unstable_by_key(|(index, _tx)| *index);
+                       dependent_txdata.dedup_by_key(|(index, _tx)| *index);
+                       let txdata: Vec<_> = dependent_txdata.iter().map(|(index, tx)| (*index, tx)).collect();
+                       self.block_connected(header, &txdata, height);
+               }
        }
 
        /// Dispatches to per-channel monitors, which are responsible for updating their on-chain view
@@ -245,3 +267,56 @@ impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> even
                pending_events
        }
 }
+
+#[cfg(test)]
+mod tests {
+       use ::{check_added_monitors, get_local_commitment_txn};
+       use ln::features::InitFeatures;
+       use ln::functional_test_utils::*;
+       use util::events::EventsProvider;
+       use util::events::MessageSendEventsProvider;
+       use util::test_utils::{OnRegisterOutput, TxOutReference};
+
+       /// Tests that in-block dependent transactions are processed by `block_connected` when not
+       /// included in `txdata` but returned by [`chain::Filter::register_output`]. For instance,
+       /// a (non-anchor) commitment transaction's HTLC output may be spent in the same block as the
+       /// commitment transaction itself. An Electrum client may filter the commitment transaction but
+       /// needs to return the HTLC transaction so it can be processed.
+       #[test]
+       fn connect_block_checks_dependent_transactions() {
+               let chanmon_cfgs = create_chanmon_cfgs(2);
+               let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+               let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+               let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+               let channel = create_announced_chan_between_nodes(
+                       &nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
+
+               // Send a payment, saving nodes[0]'s revoked commitment and HTLC-Timeout transactions.
+               let (commitment_tx, htlc_tx) = {
+                       let payment_preimage = route_payment(&nodes[0], &vec!(&nodes[1])[..], 5_000_000).0;
+                       let mut txn = get_local_commitment_txn!(nodes[0], channel.2);
+                       claim_payment(&nodes[0], &vec!(&nodes[1])[..], payment_preimage, 5_000_000);
+
+                       assert_eq!(txn.len(), 2);
+                       (txn.remove(0), txn.remove(0))
+               };
+
+               // Set expectations on nodes[1]'s chain source to return dependent transactions.
+               let htlc_output = TxOutReference(commitment_tx.clone(), 0);
+               let to_local_output = TxOutReference(commitment_tx.clone(), 1);
+               let htlc_timeout_output = TxOutReference(htlc_tx.clone(), 0);
+               nodes[1].chain_source
+                       .expect(OnRegisterOutput { with: htlc_output, returns: Some((1, htlc_tx)) })
+                       .expect(OnRegisterOutput { with: to_local_output, returns: None })
+                       .expect(OnRegisterOutput { with: htlc_timeout_output, returns: None });
+
+               // Notify nodes[1] that nodes[0]'s revoked commitment transaction was mined. The chain
+               // source should return the dependent HTLC transaction when the HTLC output is registered.
+               mine_transaction(&nodes[1], &commitment_tx);
+
+               // Clean up so uninteresting assertions don't fail.
+               check_added_monitors!(nodes[1], 1);
+               nodes[1].node.get_and_clear_pending_msg_events();
+               nodes[1].node.get_and_clear_pending_events();
+       }
+}
index 28b6da4722a7b4e3820541b30c89047d1e8e5cfe..939337d7b42d572d64c7ab5b1a6224479f6c1ce9 100644 (file)
@@ -40,6 +40,7 @@ use ln::chan_utils::{CounterpartyCommitmentSecrets, HTLCOutputInCommitment, HTLC
 use ln::channelmanager::{HTLCSource, PaymentPreimage, PaymentHash};
 use ln::onchaintx::{OnchainTxHandler, InputDescriptors};
 use chain;
+use chain::WatchedOutput;
 use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use chain::transaction::{OutPoint, TransactionData};
 use chain::keysinterface::{SpendableOutputDescriptor, StaticPaymentOutputDescriptor, DelayedPaymentOutputDescriptor, Sign, KeysInterface};
@@ -1174,7 +1175,11 @@ impl<Signer: Sign> ChannelMonitor<Signer> {
                for (txid, outputs) in lock.get_outputs_to_watch().iter() {
                        for (index, script_pubkey) in outputs.iter() {
                                assert!(*index <= u16::max_value() as u32);
-                               filter.register_output(&OutPoint { txid: *txid, index: *index as u16 }, script_pubkey);
+                               filter.register_output(WatchedOutput {
+                                       block_hash: None,
+                                       outpoint: OutPoint { txid: *txid, index: *index as u16 },
+                                       script_pubkey: script_pubkey.clone(),
+                               });
                        }
                }
        }
index 7d410b9b71dd116cbe22d2858e9664ec6bc14381..18c7fd55d7b6643bbbc140a5208df42e39e42ea3 100644 (file)
@@ -11,7 +11,7 @@
 
 use bitcoin::blockdata::block::{Block, BlockHeader};
 use bitcoin::blockdata::script::Script;
-use bitcoin::blockdata::transaction::TxOut;
+use bitcoin::blockdata::transaction::{Transaction, TxOut};
 use bitcoin::hash_types::{BlockHash, Txid};
 
 use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent};
@@ -129,9 +129,38 @@ pub trait Filter: Send + Sync {
        /// a spending condition.
        fn register_tx(&self, txid: &Txid, script_pubkey: &Script);
 
-       /// Registers interest in spends of a transaction output identified by `outpoint` having
-       /// `script_pubkey` as the spending condition.
-       fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script);
+       /// Registers interest in spends of a transaction output.
+       ///
+       /// Optionally, when `output.block_hash` is set, should return any transaction spending the
+       /// output that is found in the corresponding block along with its index.
+       ///
+       /// This return value is useful for Electrum clients in order to supply in-block descendant
+       /// transactions which otherwise were not included. This is not necessary for other clients if
+       /// such descendant transactions were already included (e.g., when a BIP 157 client provides the
+       /// full block).
+       fn register_output(&self, output: WatchedOutput) -> Option<(usize, Transaction)>;
+}
+
+/// A transaction output watched by a [`ChannelMonitor`] for spends on-chain.
+///
+/// Used to convey to a [`Filter`] such an output with a given spending condition. Any transaction
+/// spending the output must be given to [`ChannelMonitor::block_connected`] either directly or via
+/// the return value of [`Filter::register_output`].
+///
+/// If `block_hash` is `Some`, this indicates the output was created in the corresponding block and
+/// may have been spent there. See [`Filter::register_output`] for details.
+///
+/// [`ChannelMonitor`]: channelmonitor::ChannelMonitor
+/// [`ChannelMonitor::block_connected`]: channelmonitor::ChannelMonitor::block_connected
+pub struct WatchedOutput {
+       /// First block where the transaction output may have been spent.
+       pub block_hash: Option<BlockHash>,
+
+       /// Outpoint identifying the transaction output.
+       pub outpoint: OutPoint,
+
+       /// Spending condition of the transaction output.
+       pub script_pubkey: Script,
 }
 
 impl<T: Listen> Listen for std::ops::Deref<Target = T> {
index e7f61c6a3dfeefb23f845b7a47d6ab0efa5abf60..36ca7d64e5d4a2a16e3cb409214b446369ea1843 100644 (file)
@@ -341,7 +341,8 @@ macro_rules! get_feerate {
        }
 }
 
-#[cfg(test)]
+/// Returns any local commitment transactions for the channel.
+#[macro_export]
 macro_rules! get_local_commitment_txn {
        ($node: expr, $channel_id: expr) => {
                {
index a0ccf4f816ea991e31a019351018d43a95633324..2d3f930a0e425776a7bd0a24f5c595353ed86130 100644 (file)
@@ -8,6 +8,7 @@
 // licenses.
 
 use chain;
+use chain::WatchedOutput;
 use chain::chaininterface;
 use chain::chaininterface::ConfirmationTarget;
 use chain::chainmonitor;
@@ -38,7 +39,7 @@ use std::time::Duration;
 use std::sync::{Mutex, Arc};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::{cmp, mem};
-use std::collections::{HashMap, HashSet};
+use std::collections::{HashMap, HashSet, VecDeque};
 use chain::keysinterface::InMemorySigner;
 
 pub struct TestVecWriter(pub Vec<u8>);
@@ -517,6 +518,7 @@ pub struct TestChainSource {
        pub utxo_ret: Mutex<Result<TxOut, chain::AccessError>>,
        pub watched_txn: Mutex<HashSet<(Txid, Script)>>,
        pub watched_outputs: Mutex<HashSet<(OutPoint, Script)>>,
+       expectations: Mutex<Option<VecDeque<OnRegisterOutput>>>,
 }
 
 impl TestChainSource {
@@ -527,8 +529,17 @@ impl TestChainSource {
                        utxo_ret: Mutex::new(Ok(TxOut { value: u64::max_value(), script_pubkey })),
                        watched_txn: Mutex::new(HashSet::new()),
                        watched_outputs: Mutex::new(HashSet::new()),
+                       expectations: Mutex::new(None),
                }
        }
+
+       /// Sets an expectation that [`chain::Filter::register_output`] is called.
+       pub fn expect(&self, expectation: OnRegisterOutput) -> &Self {
+               self.expectations.lock().unwrap()
+                       .get_or_insert_with(|| VecDeque::new())
+                       .push_back(expectation);
+               self
+       }
 }
 
 impl chain::Access for TestChainSource {
@@ -546,7 +557,72 @@ impl chain::Filter for TestChainSource {
                self.watched_txn.lock().unwrap().insert((*txid, script_pubkey.clone()));
        }
 
-       fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script) {
-               self.watched_outputs.lock().unwrap().insert((*outpoint, script_pubkey.clone()));
+       fn register_output(&self, output: WatchedOutput) -> Option<(usize, Transaction)> {
+               let dependent_tx = match &mut *self.expectations.lock().unwrap() {
+                       None => None,
+                       Some(expectations) => match expectations.pop_front() {
+                               None => {
+                                       panic!("Unexpected register_output: {:?}",
+                                               (output.outpoint, output.script_pubkey));
+                               },
+                               Some(expectation) => {
+                                       assert_eq!(output.outpoint, expectation.outpoint());
+                                       assert_eq!(&output.script_pubkey, expectation.script_pubkey());
+                                       expectation.returns
+                               },
+                       },
+               };
+
+               self.watched_outputs.lock().unwrap().insert((output.outpoint, output.script_pubkey));
+               dependent_tx
+       }
+}
+
+impl Drop for TestChainSource {
+       fn drop(&mut self) {
+               if std::thread::panicking() {
+                       return;
+               }
+
+               if let Some(expectations) = &*self.expectations.lock().unwrap() {
+                       if !expectations.is_empty() {
+                               panic!("Unsatisfied expectations: {:?}", expectations);
+                       }
+               }
+       }
+}
+
+/// An expectation that [`chain::Filter::register_output`] was called with a transaction output and
+/// returns an optional dependent transaction that spends the output in the same block.
+pub struct OnRegisterOutput {
+       /// The transaction output to register.
+       pub with: TxOutReference,
+
+       /// A dependent transaction spending the output along with its position in the block.
+       pub returns: Option<(usize, Transaction)>,
+}
+
+/// A transaction output as identified by an index into a transaction's output list.
+pub struct TxOutReference(pub Transaction, pub usize);
+
+impl OnRegisterOutput {
+       fn outpoint(&self) -> OutPoint {
+               let txid = self.with.0.txid();
+               let index = self.with.1 as u16;
+               OutPoint { txid, index }
+       }
+
+       fn script_pubkey(&self) -> &Script {
+               let index = self.with.1;
+               &self.with.0.output[index].script_pubkey
+       }
+}
+
+impl std::fmt::Debug for OnRegisterOutput {
+       fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+               f.debug_struct("OnRegisterOutput")
+                       .field("outpoint", &self.outpoint())
+                       .field("script_pubkey", self.script_pubkey())
+                       .finish()
        }
 }