Drop return value from Filter::register_output

tnull · tnull · commit 2f4457fd7e18 · 2022-08-15T20:47:33.000+02:00
This commit removes the return value from `Filter::register_output` as
creating a suitable value almost always entails blocking operations
(e.g., lookups via network request), which however conflicts with the
requirement that user calls should avoid blocking calls at all cost.

Removing the return value also rendered quite a bit of test code for
dependent transaction handling superfluous, which is therefore also
removed with this commit.
diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs
@@ -263,82 +263,67 @@ where C::Target: chain::Filter,
 	where
 		FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>
 	{
-		let mut dependent_txdata = Vec::new();
-		{
-			let monitor_states = self.monitors.write().unwrap();
-			if let Some(height) = best_height {
-				// If the best block height is being updated, update highest_chain_height under the
-				// monitors write lock.
-				let old_height = self.highest_chain_height.load(Ordering::Acquire);
-				let new_height = height as usize;
-				if new_height > old_height {
-					self.highest_chain_height.store(new_height, Ordering::Release);
-				}
+		let monitor_states = self.monitors.write().unwrap();
+		if let Some(height) = best_height {
+			// If the best block height is being updated, update highest_chain_height under the
+			// monitors write lock.
+			let old_height = self.highest_chain_height.load(Ordering::Acquire);
+			let new_height = height as usize;
+			if new_height > old_height {
+				self.highest_chain_height.store(new_height, Ordering::Release);
 			}
+		}
 
-			for (funding_outpoint, monitor_state) in monitor_states.iter() {
-				let monitor = &monitor_state.monitor;
-				let mut txn_outputs;
-				{
-					txn_outputs = process(monitor, txdata);
-					let update_id = MonitorUpdateId {
-						contents: UpdateOrigin::ChainSync(self.sync_persistence_id.get_increment()),
-					};
-					let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
-					if let Some(height) = best_height {
-						if !monitor_state.has_pending_chainsync_updates(&pending_monitor_updates) {
-							// If there are not ChainSync persists awaiting completion, go ahead and
-							// set last_chain_persist_height here - we wouldn't want the first
-							// TemporaryFailure to always immediately be considered "overly delayed".
-							monitor_state.last_chain_persist_height.store(height as usize, Ordering::Release);
-						}
+		for (funding_outpoint, monitor_state) in monitor_states.iter() {
+			let monitor = &monitor_state.monitor;
+			let mut txn_outputs;
+			{
+				txn_outputs = process(monitor, txdata);
+				let update_id = MonitorUpdateId {
+					contents: UpdateOrigin::ChainSync(self.sync_persistence_id.get_increment()),
+				};
+				let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
+				if let Some(height) = best_height {
+					if !monitor_state.has_pending_chainsync_updates(&pending_monitor_updates) {
+						// If there are not ChainSync persists awaiting completion, go ahead and
+						// set last_chain_persist_height here - we wouldn't want the first
+						// TemporaryFailure to always immediately be considered "overly delayed".
+						monitor_state.last_chain_persist_height.store(height as usize, Ordering::Release);
 					}
+				}
 
-					log_trace!(self.logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor));
-					match self.persister.update_persisted_channel(*funding_outpoint, &None, monitor, update_id) {
-						Ok(()) =>
-							log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
-						Err(ChannelMonitorUpdateErr::PermanentFailure) => {
-							monitor_state.channel_perm_failed.store(true, Ordering::Release);
-							self.pending_monitor_events.lock().unwrap().push((*funding_outpoint, vec![MonitorEvent::UpdateFailed(*funding_outpoint)], monitor.get_counterparty_node_id()));
-						},
-						Err(ChannelMonitorUpdateErr::TemporaryFailure) => {
-							log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
-							pending_monitor_updates.push(update_id);
-						},
-					}
+				log_trace!(self.logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor));
+				match self.persister.update_persisted_channel(*funding_outpoint, &None, monitor, update_id) {
+					Ok(()) =>
+						log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
+					Err(ChannelMonitorUpdateErr::PermanentFailure) => {
+						monitor_state.channel_perm_failed.store(true, Ordering::Release);
+						self.pending_monitor_events.lock().unwrap().push((*funding_outpoint, vec![MonitorEvent::UpdateFailed(*funding_outpoint)], monitor.get_counterparty_node_id()));
+					},
+					Err(ChannelMonitorUpdateErr::TemporaryFailure) => {
+						log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
+						pending_monitor_updates.push(update_id);
+					},
 				}
+			}
 
-				// 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, mut outputs) in txn_outputs.drain(..) {
-						for (idx, output) in outputs.drain(..) {
-							// 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,
-							};
-							if let Some(tx) = chain_source.register_output(output) {
-								dependent_txdata.push(tx);
-							}
-						}
+			// 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, mut outputs) in txn_outputs.drain(..) {
+					for (idx, output) in outputs.drain(..) {
+						// Register any new outputs with the chain source for filtering
+						let output = WatchedOutput {
+							block_hash: Some(block_hash),
+							outpoint: OutPoint { txid, index: idx as u16 },
+							script_pubkey: output.script_pubkey,
+						};
+						chain_source.register_output(output)
 					}
 				}
 			}
 		}
-
-		// 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.process_chain_data(header, None, &txdata, process); // We skip the best height the second go-around
-		}
 	}
 
 	/// Creates a new `ChainMonitor` used to watch on-chain activity pertaining to channels.
@@ -745,50 +730,6 @@ mod tests {
 	use ln::msgs::ChannelMessageHandler;
 	use util::errors::APIError;
 	use util::events::{ClosureReason, MessageSendEvent, 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);
-
-			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();
-	}
 
 	#[test]
 	fn test_async_ooo_offchain_updates() {
diff --git a/lightning/src/chain/mod.rs b/lightning/src/chain/mod.rs
@@ -12,7 +12,7 @@
 use bitcoin::blockdata::block::{Block, BlockHeader};
 use bitcoin::blockdata::constants::genesis_block;
 use bitcoin::blockdata::script::Script;
-use bitcoin::blockdata::transaction::{Transaction, TxOut};
+use bitcoin::blockdata::transaction::TxOut;
 use bitcoin::hash_types::{BlockHash, Txid};
 use bitcoin::network::constants::Network;
 use bitcoin::secp256k1::PublicKey;
@@ -333,21 +333,18 @@ pub trait Filter {
 
 	/// 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)>;
+	/// Note that this method might be called during processing of a new block. You therefore need
+	/// to ensure that also dependent output spents within an already connected block are correctly
+	/// handled, e.g., by re-scanning the block in question whenever new outputs have been
+	/// registered mid-processing.
+	fn register_output(&self, output: WatchedOutput);
 }
 
 /// 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`].
+/// [`Confirm::transactions_confirmed`].
 ///
 /// 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.
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
@@ -728,7 +728,6 @@ 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 {
@@ -739,17 +738,8 @@ 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 {
@@ -767,24 +757,8 @@ impl chain::Filter for TestChainSource {
 		self.watched_txn.lock().unwrap().insert((*txid, 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
-				},
-			},
-		};
-
+	fn register_output(&self, output: WatchedOutput) {
 		self.watched_outputs.lock().unwrap().insert((output.outpoint, output.script_pubkey));
-		dependent_tx
 	}
 }
 
@@ -793,47 +767,6 @@ impl Drop for TestChainSource {
 		if 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 core::fmt::Debug for OnRegisterOutput {
-	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
-		f.debug_struct("OnRegisterOutput")
-			.field("outpoint", &self.outpoint())
-			.field("script_pubkey", self.script_pubkey())
-			.finish()
 	}
 }
 
