Merge pull request #1663 from tnull/2022-08-drop-register-output-return

TheBlueMatt · web-flow · commit 4005724ae671 · 2022-08-15T23:10:32.000Z
Drop return value from `Filter::register_output`
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.
@@ -746,50 +731,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;
@@ -151,15 +151,15 @@ pub trait Confirm {
 	/// in the event of a chain reorganization, it must not be called with a `header` that is no
 	/// longer in the chain as of the last call to [`best_block_updated`].
 	///
-	/// [chain order]: Confirm#Order
+	/// [chain order]: Confirm#order
 	/// [`best_block_updated`]: Self::best_block_updated
 	fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32);
 
 	/// Processes a transaction that is no longer confirmed as result of a chain reorganization.
 	///
 	/// Should be called for any transaction returned by [`get_relevant_txids`] if it has been
-	/// reorganized out of the best chain. Once called, the given transaction should not be returned
-	/// by [`get_relevant_txids`] unless it has been reconfirmed via [`transactions_confirmed`].
+	/// reorganized out of the best chain. Once called, the given transaction will not be returned
+	/// by [`get_relevant_txids`], unless it has been reconfirmed via [`transactions_confirmed`].
 	///
 	/// [`get_relevant_txids`]: Self::get_relevant_txids
 	/// [`transactions_confirmed`]: Self::transactions_confirmed
@@ -173,9 +173,9 @@ pub trait Confirm {
 
 	/// Returns transactions that should be monitored for reorganization out of the chain.
 	///
-	/// Should include any transactions passed to [`transactions_confirmed`] that have insufficient
-	/// confirmations to be safe from a chain reorganization. Should not include any transactions
-	/// passed to [`transaction_unconfirmed`] unless later reconfirmed.
+	/// Will include any transactions passed to [`transactions_confirmed`] that have insufficient
+	/// confirmations to be safe from a chain reorganization. Will not include any transactions
+	/// passed to [`transaction_unconfirmed`], unless later reconfirmed.
 	///
 	/// May be called to determine the subset of transactions that must still be monitored for
 	/// reorganization. Will be idempotent between calls but may change as a result of calls to the
@@ -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
@@ -730,7 +730,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 {
@@ -741,17 +740,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 {
@@ -769,24 +759,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
 	}
 }
 
@@ -795,47 +769,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()
 	}
 }
 
