Replace WatchEvent usage with get_outputs_to_watch

Outputs to watch are tracked by ChannelMonitor as of
73dce207dd. Instead of determining new
outputs to watch independently using ChainWatchedUtil, do so by
comparing against outputs already tracked. Thus, ChainWatchedUtil and
WatchEvent are no longer needed.

lightning/src/chain/chaininterface.rs

@@ -14,10 +14,6 @@
 //! disconnections, transaction broadcasting, and feerate information requests.
 
 use bitcoin::blockdata::transaction::Transaction;
-use bitcoin::blockdata::script::Script;
-use bitcoin::hash_types::Txid;
-
-use std::collections::HashSet;
 
 /// An interface to send a transaction to the Bitcoin network.
 pub trait BroadcasterInterface: Sync + Send {
@@ -55,91 +51,3 @@ pub trait FeeEstimator: Sync + Send {
 
 /// Minimum relay fee as required by bitcoin network mempool policy.
 pub const MIN_RELAY_FEE_SAT_PER_1000_WEIGHT: u64 = 4000;
-
-/// Utility for tracking registered txn/outpoints and checking for matches
-#[cfg_attr(test, derive(PartialEq))]
-pub struct ChainWatchedUtil {
-	watch_all: bool,
-
-	// We are more conservative in matching during testing to ensure everything matches *exactly*,
-	// even though during normal runtime we take more optimized match approaches...
-	#[cfg(test)]
-	watched_txn: HashSet<(Txid, Script)>,
-	#[cfg(not(test))]
-	watched_txn: HashSet<Script>,
-
-	watched_outpoints: HashSet<(Txid, u32)>,
-}
-
-impl ChainWatchedUtil {
-	/// Constructs an empty (watches nothing) ChainWatchedUtil
-	pub fn new() -> Self {
-		Self {
-			watch_all: false,
-			watched_txn: HashSet::new(),
-			watched_outpoints: HashSet::new(),
-		}
-	}
-
-	/// Registers a tx for monitoring, returning true if it was a new tx and false if we'd already
-	/// been watching for it.
-	pub fn register_tx(&mut self, txid: &Txid, script_pub_key: &Script) -> bool {
-		if self.watch_all { return false; }
-		#[cfg(test)]
-		{
-			self.watched_txn.insert((txid.clone(), script_pub_key.clone()))
-		}
-		#[cfg(not(test))]
-		{
-			let _tx_unused = txid; // It's used in cfg(test), though
-			self.watched_txn.insert(script_pub_key.clone())
-		}
-	}
-
-	/// Registers an outpoint for monitoring, returning true if it was a new outpoint and false if
-	/// we'd already been watching for it
-	pub fn register_outpoint(&mut self, outpoint: (Txid, u32), _script_pub_key: &Script) -> bool {
-		if self.watch_all { return false; }
-		self.watched_outpoints.insert(outpoint)
-	}
-
-	/// Sets us to match all transactions, returning true if this is a new setting and false if
-	/// we'd already been set to match everything.
-	pub fn watch_all(&mut self) -> bool {
-		if self.watch_all { return false; }
-		self.watch_all = true;
-		true
-	}
-
-	/// Checks if a given transaction matches the current filter.
-	pub fn does_match_tx(&self, tx: &Transaction) -> bool {
-		if self.watch_all {
-			return true;
-		}
-		for out in tx.output.iter() {
-			#[cfg(test)]
-			for &(ref txid, ref script) in self.watched_txn.iter() {
-				if *script == out.script_pubkey {
-					if tx.txid() == *txid {
-						return true;
-					}
-				}
-			}
-			#[cfg(not(test))]
-			for script in self.watched_txn.iter() {
-				if *script == out.script_pubkey {
-					return true;
-				}
-			}
-		}
-		for input in tx.input.iter() {
-			for outpoint in self.watched_outpoints.iter() {
-				let &(outpoint_hash, outpoint_index) = outpoint;
-				if outpoint_hash == input.previous_output.txid && outpoint_index == input.previous_output.vout {
-					return true;
-				}
-			}
-		}
-		false
-	}
-}

lightning/src/chain/mod.rs

@@ -73,8 +73,7 @@ pub trait Watch: Send + Sync {
 	///
 	/// Implementations are responsible for watching the chain for the funding transaction along
 	/// with any spends of outputs returned by [`get_outputs_to_watch`]. In practice, this means
-	/// calling [`block_connected`] and [`block_disconnected`] on the monitor and including all such
+	/// calling [`block_connected`] and [`block_disconnected`] on the monitor.
-	/// transactions that meet this criteria.
 	///
 	/// [`get_outputs_to_watch`]: ../ln/channelmonitor/struct.ChannelMonitor.html#method.get_outputs_to_watch
 	/// [`block_connected`]: ../ln/channelmonitor/struct.ChannelMonitor.html#method.block_connected
@@ -119,9 +118,9 @@ pub trait Watch: Send + Sync {
 pub trait Filter: Send + Sync {
 	/// Registers interest in a transaction with `txid` and having an output with `script_pubkey` as
 	/// a spending condition.
-	fn register_tx(&self, txid: Txid, script_pubkey: Script);
+	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);
+	fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script);
 }

lightning/src/ln/channelmonitor.rs

@@ -44,7 +44,7 @@ use ln::channelmanager::{HTLCSource, PaymentPreimage, PaymentHash};
 use ln::onchaintx::{OnchainTxHandler, InputDescriptors};
 use chain;
 use chain::Filter;
-use chain::chaininterface::{ChainWatchedUtil, BroadcasterInterface, FeeEstimator};
+use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
 use chain::transaction::OutPoint;
 use chain::keysinterface::{SpendableOutputDescriptor, ChannelKeys};
 use util::logger::Logger;
@@ -205,110 +205,12 @@ pub struct ChainMonitor<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L
 {
 	/// The monitors
 	pub monitors: Mutex<HashMap<OutPoint, ChannelMonitor<ChanSigner>>>,
-	watch_events: Mutex<WatchEventCache>,
 	chain_source: Option<C>,
 	broadcaster: T,
 	logger: L,
 	fee_estimator: F
 }
 
-struct WatchEventCache {
-	watched: ChainWatchedUtil,
-	events: Vec<WatchEvent>,
-}
-
-/// An event indicating on-chain activity to watch for pertaining to a channel.
-enum WatchEvent {
-	/// Watch for a transaction with `txid` and having an output with `script_pubkey` as a spending
-	/// condition.
-	WatchTransaction {
-		/// Identifier of the transaction.
-		txid: Txid,
-
-		/// Spending condition for an output of the transaction.
-		script_pubkey: Script,
-	},
-	/// Watch for spends of a transaction output identified by `outpoint` having `script_pubkey` as
-	/// the spending condition.
-	WatchOutput {
-		/// Identifier for the output.
-		outpoint: OutPoint,
-
-		/// Spending condition for the output.
-		script_pubkey: Script,
-	}
-}
-
-impl WatchEventCache {
-	fn new() -> Self {
-		Self {
-			watched: ChainWatchedUtil::new(),
-			events: Vec::new(),
-		}
-	}
-
-	fn watch_tx(&mut self, txid: &Txid, script_pubkey: &Script) {
-		if self.watched.register_tx(txid, script_pubkey) {
-			self.events.push(WatchEvent::WatchTransaction {
-				txid: *txid,
-				script_pubkey: script_pubkey.clone()
-			});
-		}
-	}
-
-	fn watch_output(&mut self, outpoint: (&Txid, usize), script_pubkey: &Script) {
-		let (txid, index) = outpoint;
-		if self.watched.register_outpoint((*txid, index as u32), script_pubkey) {
-			self.events.push(WatchEvent::WatchOutput {
-				outpoint: OutPoint {
-					txid: *txid,
-					index: index as u16,
-				},
-				script_pubkey: script_pubkey.clone(),
-			});
-		}
-	}
-
-	fn flush_events<C: Deref>(&mut self, chain_source: &Option<C>) -> bool where C::Target: chain::Filter {
-		let num_events = self.events.len();
-		match chain_source {
-			&None => self.events.clear(),
-			&Some(ref chain_source) => {
-				for event in self.events.drain(..) {
-					match event {
-						WatchEvent::WatchTransaction { txid, script_pubkey } => {
-							chain_source.register_tx(txid, script_pubkey)
-						},
-						WatchEvent::WatchOutput { outpoint, script_pubkey } => {
-							chain_source.register_output(outpoint, script_pubkey)
-						},
-					}
-				}
-			}
-		}
-		num_events > 0
-	}
-
-	fn filter_block<'a>(&self, txdata: &[(usize, &'a Transaction)]) -> Vec<(usize, &'a Transaction)> {
-		let mut matched_txids = HashSet::new();
-		txdata.iter().filter(|&&(_, tx)| {
-			// A tx matches the filter if it either matches the filter directly (via does_match_tx)
-			// or if it is a descendant of another matched transaction within the same block.
-			let mut matched = self.watched.does_match_tx(tx);
-			for input in tx.input.iter() {
-				if matched || matched_txids.contains(&input.previous_output.txid) {
-					matched = true;
-					break;
-				}
-			}
-			if matched {
-				matched_txids.insert(tx.txid());
-			}
-			matched
-		}).map(|e| *e).collect()
-	}
-}
-
 impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonitor<ChanSigner, C, T, F, L>
 	where C::Target: chain::Filter,
 	      T::Target: BroadcasterInterface,
@@ -329,21 +231,23 @@ impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonit
 	/// [`chain::Watch::release_pending_monitor_events`]: ../../chain/trait.Watch.html#tymethod.release_pending_monitor_events
 	/// [`chain::Filter`]: ../../chain/trait.Filter.html
 	pub fn block_connected(&self, header: &BlockHeader, txdata: &[(usize, &Transaction)], height: u32) -> bool {
-		let mut watch_events = self.watch_events.lock().unwrap();
+		let mut has_new_outputs_to_watch = false;
-		let matched_txn = watch_events.filter_block(txdata);
 		{
 			let mut monitors = self.monitors.lock().unwrap();
 			for monitor in monitors.values_mut() {
-				let txn_outputs = monitor.block_connected(header, &matched_txn, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+				let mut txn_outputs = monitor.block_connected(header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+				has_new_outputs_to_watch |= !txn_outputs.is_empty();
 
-				for (ref txid, ref outputs) in txn_outputs {
+				if let Some(ref chain_source) = self.chain_source {
-					for (idx, output) in outputs.iter().enumerate() {
+					for (txid, outputs) in txn_outputs.drain(..) {
-						watch_events.watch_output((txid, idx), &output.script_pubkey);
+						for (idx, output) in outputs.iter().enumerate() {
+							chain_source.register_output(&OutPoint { txid, index: idx as u16 }, &output.script_pubkey);
+						}
 					}
 				}
 			}
 		}
-		watch_events.flush_events(&self.chain_source)
+		has_new_outputs_to_watch
 	}
 
 	/// Dispatches to per-channel monitors, which are responsible for updating their on-chain view
@@ -377,7 +281,6 @@ impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonit
 	pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F) -> Self {
 		Self {
 			monitors: Mutex::new(HashMap::new()),
-			watch_events: Mutex::new(WatchEventCache::new()),
 			chain_source,
 			broadcaster,
 			logger,
@@ -391,7 +294,6 @@ impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonit
 	///
 	/// [`chain::Filter`]: ../../chain/trait.Filter.html
 	fn add_monitor(&self, outpoint: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
-		let mut watch_events = self.watch_events.lock().unwrap();
 		let mut monitors = self.monitors.lock().unwrap();
 		let entry = match monitors.entry(outpoint) {
 			hash_map::Entry::Occupied(_) => return Err(MonitorUpdateError("Channel monitor for given outpoint is already present")),
@@ -400,16 +302,17 @@ impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonit
 		{
 			let funding_txo = monitor.get_funding_txo();
 			log_trace!(self.logger, "Got new Channel Monitor for channel {}", log_bytes!(funding_txo.0.to_channel_id()[..]));
-			watch_events.watch_tx(&funding_txo.0.txid, &funding_txo.1);
+
-			watch_events.watch_output((&funding_txo.0.txid, funding_txo.0.index as usize), &funding_txo.1);
+			if let Some(ref chain_source) = self.chain_source {
-			for (txid, outputs) in monitor.get_outputs_to_watch().iter() {
+				chain_source.register_tx(&funding_txo.0.txid, &funding_txo.1);
-				for (idx, script) in outputs.iter().enumerate() {
+				for (txid, outputs) in monitor.get_outputs_to_watch().iter() {
-					watch_events.watch_output((txid, idx), script);
+					for (idx, script_pubkey) in outputs.iter().enumerate() {
+						chain_source.register_output(&OutPoint { txid: *txid, index: idx as u16 }, &script_pubkey);
+					}
 				}
 			}
 		}
 		entry.insert(monitor);
-		watch_events.flush_events(&self.chain_source);
 		Ok(())
 	}
 
@@ -1958,16 +1861,17 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 	/// - detect settled outputs for later spending
 	/// - schedule and bump any in-flight claims
 	///
-	/// Returns any transaction outputs from `txn_matched` that spends of should be watched for.
+	/// Returns any new outputs to watch from `txdata`; after called, these are also included in
-	/// After called these are also available via [`get_outputs_to_watch`].
+	/// [`get_outputs_to_watch`].
 	///
 	/// [`get_outputs_to_watch`]: #method.get_outputs_to_watch
-	pub fn block_connected<B: Deref, F: Deref, L: Deref>(&mut self, header: &BlockHeader, txn_matched: &[(usize, &Transaction)], height: u32, broadcaster: B, fee_estimator: F, logger: L)-> Vec<(Txid, Vec<TxOut>)>
+	pub fn block_connected<B: Deref, F: Deref, L: Deref>(&mut self, header: &BlockHeader, txdata: &[(usize, &Transaction)], height: u32, broadcaster: B, fee_estimator: F, logger: L)-> Vec<(Txid, Vec<TxOut>)>
 		where B::Target: BroadcasterInterface,
 		      F::Target: FeeEstimator,
 					L::Target: Logger,
 	{
-		for &(_, tx) in txn_matched {
+		let txn_matched = self.filter_block(txdata);
+		for tx in &txn_matched {
 			let mut output_val = 0;
 			for out in tx.output.iter() {
 				if out.value > 21_000_000_0000_0000 { panic!("Value-overflowing transaction provided to block connected"); }
@@ -1981,7 +1885,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 
 		let mut watch_outputs = Vec::new();
 		let mut claimable_outpoints = Vec::new();
-		for &(_, tx) in txn_matched {
+		for tx in &txn_matched {
 			if tx.input.len() == 1 {
 				// Assuming our keys were not leaked (in which case we're screwed no matter what),
 				// commitment transactions and HTLC transactions will all only ever have one input,
@@ -2056,13 +1960,15 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 			}
 		}
 
-		self.onchain_tx_handler.block_connected(txn_matched, claimable_outpoints, height, &*broadcaster, &*fee_estimator, &*logger);
+		self.onchain_tx_handler.block_connected(&txn_matched, claimable_outpoints, height, &*broadcaster, &*fee_estimator, &*logger);
-
 		self.last_block_hash = block_hash;
-		for &(ref txid, ref output_scripts) in watch_outputs.iter() {
-			self.outputs_to_watch.insert(txid.clone(), output_scripts.iter().map(|o| o.script_pubkey.clone()).collect());
-		}
 
+		// Determine new outputs to watch by comparing against previously known outputs to watch,
+		// updating the latter in the process.
+		watch_outputs.retain(|&(ref txid, ref txouts)| {
+			let output_scripts = txouts.iter().map(|o| o.script_pubkey.clone()).collect();
+			self.outputs_to_watch.insert(txid.clone(), output_scripts).is_none()
+		});
 		watch_outputs
 	}
 
@@ -2087,6 +1993,40 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 		self.last_block_hash = block_hash;
 	}
 
+	/// Filters a block's `txdata` for transactions spending watched outputs or for any child
+	/// transactions thereof.
+	fn filter_block<'a>(&self, txdata: &[(usize, &'a Transaction)]) -> Vec<&'a Transaction> {
+		let mut matched_txn = HashSet::new();
+		txdata.iter().filter(|&&(_, tx)| {
+			let mut matches = self.spends_watched_output(tx);
+			for input in tx.input.iter() {
+				if matches { break; }
+				if matched_txn.contains(&input.previous_output.txid) {
+					matches = true;
+				}
+			}
+			if matches {
+				matched_txn.insert(tx.txid());
+			}
+			matches
+		}).map(|(_, tx)| *tx).collect()
+	}
+
+	/// Checks if a given transaction spends any watched outputs.
+	fn spends_watched_output(&self, tx: &Transaction) -> bool {
+		for input in tx.input.iter() {
+			if let Some(outputs) = self.get_outputs_to_watch().get(&input.previous_output.txid) {
+				for (idx, _script_pubkey) in outputs.iter().enumerate() {
+					if idx == input.previous_output.vout as usize {
+						return true;
+					}
+				}
+			}
+		}
+
+		false
+	}
+
 	fn would_broadcast_at_height<L: Deref>(&self, height: u32, logger: &L) -> bool where L::Target: Logger {
 		// We need to consider all HTLCs which are:
 		//  * in any unrevoked counterparty commitment transaction, as they could broadcast said

lightning/src/ln/onchaintx.rs

@@ -657,7 +657,7 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		None
 	}
 
-	pub(super) fn block_connected<B: Deref, F: Deref, L: Deref>(&mut self, txn_matched: &[(usize, &Transaction)], claimable_outpoints: Vec<ClaimRequest>, height: u32, broadcaster: B, fee_estimator: F, logger: L)
+	pub(crate) fn block_connected<B: Deref, F: Deref, L: Deref>(&mut self, txn_matched: &[&Transaction], claimable_outpoints: Vec<ClaimRequest>, height: u32, broadcaster: B, fee_estimator: F, logger: L)
 		where B::Target: BroadcasterInterface,
 		      F::Target: FeeEstimator,
 					L::Target: Logger,
@@ -706,7 +706,7 @@ impl<ChanSigner: ChannelKeys> OnchainTxHandler<ChanSigner> {
 		}
 
 		let mut bump_candidates = HashMap::new();
-		for &(_, tx) in txn_matched {
+		for tx in txn_matched {
 			// Scan all input to verify is one of the outpoint spent is of interest for us
 			let mut claimed_outputs_material = Vec::new();
 			for inp in &tx.input {

lightning/src/util/test_utils.rs

@@ -423,11 +423,11 @@ impl chain::Access for TestChainSource {
 }
 
 impl chain::Filter for TestChainSource {
-	fn register_tx(&self, txid: Txid, script_pubkey: Script) {
+	fn register_tx(&self, txid: &Txid, script_pubkey: &Script) {
-		self.watched_txn.lock().unwrap().insert((txid, script_pubkey));
+		self.watched_txn.lock().unwrap().insert((*txid, script_pubkey.clone()));
 	}
 
-	fn register_output(&self, outpoint: OutPoint, script_pubkey: Script) {
+	fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script) {
-		self.watched_outputs.lock().unwrap().insert((outpoint, script_pubkey));
+		self.watched_outputs.lock().unwrap().insert((*outpoint, script_pubkey.clone()));
 	}
 }