Merge pull request #840 from jkczyz/2021-03-rescan-logic

Rescan dependent transactions in ChainMonitor

Author: TheBlueMatt

lightning/src/chain/chainmonitor.rs

@@ -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();
+	}
+}

lightning/src/chain/channelmonitor.rs

@@ -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(),
+				});
 			}
 		}
 	}

lightning/src/chain/mod.rs

@@ -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> {

lightning/src/ln/functional_test_utils.rs

@@ -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) => {
 		{

lightning/src/util/test_utils.rs

@@ -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()
 	}
 }