Replace WatchEventProvider with chain::Notify

WatchEventProvider served as a means for replacing ChainWatchInterface.
However, it requires users to explicitly fetch WatchEvents, even if not
interested in them. Replace WatchEventProvider by chain::Notify, which
is an optional member of ChainMonitor. If set, interesting transactions
and output spends are registered such that blocks containing them can be
retrieved from a chain source in an efficient manner.

This is useful when the chain source is not a full node. For Electrum,
it allows for pre-filtered blocks. For BIP157/158, it serves as a means
to match against compact filters.

Author: jkczyz

ARCH.md

@@ -54,9 +54,9 @@ At a high level, some of the common interfaces fit together as follows:
  |                    -----------------        \       _----------------   /           /
  |                    | chain::Access |         \     / | ChainMonitor |---------------
  |                    -----------------          \   /  ----------------
- |                            |                   \ /
-(as RoutingMessageHandler)    v                    v
-  \                   --------------------     ---------
-   -----------------> | NetGraphMsgHandler |   | Event |
-                      --------------------     ---------
+ |                            |                   \ /          |
+(as RoutingMessageHandler)    v                    v           v
+  \                   --------------------     ---------   -----------------
+   -----------------> | NetGraphMsgHandler |   | Event |   | chain::Notify |
+                      --------------------     ---------   -----------------
 ```

fuzz/src/chanmon_consistency.rs

@@ -75,7 +75,7 @@ impl Writer for VecWriter {
 
 struct TestChainMonitor {
 	pub logger: Arc<dyn Logger>,
-	pub chain_monitor: Arc<channelmonitor::ChainMonitor<OutPoint, EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	pub chain_monitor: Arc<channelmonitor::ChainMonitor<OutPoint, EnforcingChannelKeys, Arc<dyn chain::Notify>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
 	pub update_ret: Mutex<Result<(), channelmonitor::ChannelMonitorUpdateErr>>,
 	// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
 	// logic will automatically force-close our channels for us (as we don't have an up-to-date
@@ -88,7 +88,7 @@ struct TestChainMonitor {
 impl TestChainMonitor {
 	pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>) -> Self {
 		Self {
-			chain_monitor: Arc::new(channelmonitor::ChainMonitor::new(broadcaster, logger.clone(), feeest)),
+			chain_monitor: Arc::new(channelmonitor::ChainMonitor::new(None, broadcaster, logger.clone(), feeest)),
 			logger,
 			update_ret: Mutex::new(Ok(())),
 			latest_monitors: Mutex::new(HashMap::new()),

fuzz/src/full_stack.rs

@@ -137,14 +137,14 @@ impl<'a> std::hash::Hash for Peer<'a> {
 
 type ChannelMan = ChannelManager<
 	EnforcingChannelKeys,
-	Arc<channelmonitor::ChainMonitor<OutPoint, EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	Arc<channelmonitor::ChainMonitor<OutPoint, EnforcingChannelKeys, Arc<dyn chain::Notify>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
 	Arc<TestBroadcaster>, Arc<KeyProvider>, Arc<FuzzEstimator>, Arc<dyn Logger>>;
 type PeerMan<'a> = PeerManager<Peer<'a>, Arc<ChannelMan>, Arc<NetGraphMsgHandler<Arc<dyn chain::Access>, Arc<dyn Logger>>>, Arc<dyn Logger>>;
 
 struct MoneyLossDetector<'a> {
 	manager: Arc<ChannelMan>,
 	monitor: Arc<channelmonitor::ChainMonitor<
-		OutPoint, EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+		OutPoint, EnforcingChannelKeys, Arc<dyn chain::Notify>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
 	handler: PeerMan<'a>,
 
 	peers: &'a RefCell<[bool; 256]>,
@@ -158,7 +158,7 @@ struct MoneyLossDetector<'a> {
 impl<'a> MoneyLossDetector<'a> {
 	pub fn new(peers: &'a RefCell<[bool; 256]>,
 	           manager: Arc<ChannelMan>,
-	           monitor: Arc<channelmonitor::ChainMonitor<OutPoint, EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	           monitor: Arc<channelmonitor::ChainMonitor<OutPoint, EnforcingChannelKeys, Arc<dyn chain::Notify>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
 	           handler: PeerMan<'a>) -> Self {
 		MoneyLossDetector {
 			manager,
@@ -332,7 +332,7 @@ pub fn do_test(data: &[u8], logger: &Arc<dyn Logger>) {
 	};
 
 	let broadcast = Arc::new(TestBroadcaster{});
-	let monitor = Arc::new(channelmonitor::ChainMonitor::new(broadcast.clone(), Arc::clone(&logger), fee_est.clone()));
+	let monitor = Arc::new(channelmonitor::ChainMonitor::new(None, broadcast.clone(), Arc::clone(&logger), fee_est.clone()));
 
 	let keys_manager = Arc::new(KeyProvider { node_secret: our_network_key.clone(), counter: AtomicU64::new(0) });
 	let mut config = UserConfig::default();

lightning-net-tokio/src/lib.rs

@@ -26,7 +26,8 @@
 //! type FeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator;
 //! type Logger = dyn lightning::util::logger::Logger;
 //! type ChainAccess = dyn lightning::chain::Access;
-//! type ChainMonitor = lightning::ln::channelmonitor::ChainMonitor<lightning::chain::transaction::OutPoint, lightning::chain::keysinterface::InMemoryChannelKeys, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>>;
+//! type ChainNotify = dyn lightning::chain::Notify;
+//! type ChainMonitor = lightning::ln::channelmonitor::ChainMonitor<lightning::chain::transaction::OutPoint, lightning::chain::keysinterface::InMemoryChannelKeys, Arc<ChainNotify>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>>;
 //! type ChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>;
 //! type PeerManager = lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger>;
 //!

lightning/src/chain/mod.rs

@@ -79,32 +79,22 @@ pub trait Watch: Send + Sync {
 	fn release_pending_htlc_updates(&self) -> Vec<HTLCUpdate>;
 }
 
-/// An interface for providing [`WatchEvent`]s.
+/// The `Notify` trait defines behavior for indicating chain activity of interest pertaining to
+/// channels.
 ///
-/// [`WatchEvent`]: enum.WatchEvent.html
-pub trait WatchEventProvider {
-	/// Releases events produced since the last call. Subsequent calls must only return new events.
-	fn release_pending_watch_events(&self) -> Vec<WatchEvent>;
-}
-
-/// An event indicating on-chain activity to watch for pertaining to a channel.
-pub 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,
+/// This is useful in order to have a [`Watch`] implementation convey to a chain source which
+/// transactions to be notified of. This may take the form of pre-filtering blocks or, in the case
+/// of [BIP 157]/[BIP 158], only fetching a block if the compact filter matches.
+///
+/// [`Watch`]: trait.Watch.html
+/// [BIP 157]: https://github.com/bitcoin/bips/blob/master/bip-0157.mediawiki
+/// [BIP 158]: https://github.com/bitcoin/bips/blob/master/bip-0158.mediawiki
+pub trait Notify: 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);
 
-		/// Spending condition for the output.
-		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);
 }

lightning/src/ln/channelmonitor.rs

@@ -35,6 +35,7 @@ use ln::chan_utils::{CounterpartyCommitmentSecrets, HTLCOutputInCommitment, Loca
 use ln::channelmanager::{HTLCSource, PaymentPreimage, PaymentHash};
 use ln::onchaintx::{OnchainTxHandler, InputDescriptors};
 use chain;
+use chain::Notify;
 use chain::chaininterface::{ChainWatchedUtil, BroadcasterInterface, FeeEstimator};
 use chain::transaction::OutPoint;
 use chain::keysinterface::{SpendableOutputDescriptor, ChannelKeys};
@@ -167,27 +168,51 @@ impl_writeable!(HTLCUpdate, 0, { payment_hash, payment_preimage, source });
 /// `OutPoint` as the key, which will give you a [`chain::Watch`] implementation.
 ///
 /// [`chain::Watch`]: ../../chain/trait.Watch.html
-pub struct ChainMonitor<Key, ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref>
-	where T::Target: BroadcasterInterface,
+pub struct ChainMonitor<Key, ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref>
+	where C::Target: chain::Notify,
+        T::Target: BroadcasterInterface,
         F::Target: FeeEstimator,
         L::Target: Logger,
 {
 	#[cfg(test)] // Used in ChannelManager tests to manipulate channels directly
 	pub monitors: Mutex<HashMap<Key, ChannelMonitor<ChanSigner>>>,
 	#[cfg(not(test))]
 	monitors: Mutex<HashMap<Key, ChannelMonitor<ChanSigner>>>,
-	watch_events: Mutex<WatchEventQueue>,
+	watch_events: Mutex<WatchEventCache>,
+	chain_source: Option<C>,
 	broadcaster: T,
 	logger: L,
 	fee_estimator: F
 }
 
-struct WatchEventQueue {
+struct WatchEventCache {
 	watched: ChainWatchedUtil,
-	events: Vec<chain::WatchEvent>,
+	events: Vec<WatchEvent>,
 }
 
-impl WatchEventQueue {
+/// 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(),
@@ -197,7 +222,7 @@ impl WatchEventQueue {
 
 	fn watch_tx(&mut self, txid: &Txid, script_pubkey: &Script) {
 		if self.watched.register_tx(txid, script_pubkey) {
-			self.events.push(chain::WatchEvent::WatchTransaction {
+			self.events.push(WatchEvent::WatchTransaction {
 				txid: *txid,
 				script_pubkey: script_pubkey.clone()
 			});
@@ -207,7 +232,7 @@ impl WatchEventQueue {
 	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(chain::WatchEvent::WatchOutput {
+			self.events.push(WatchEvent::WatchOutput {
 				outpoint: OutPoint {
 					txid: *txid,
 					index: index as u16,
@@ -217,20 +242,35 @@ impl WatchEventQueue {
 		}
 	}
 
-	fn dequeue_events(&mut self) -> Vec<chain::WatchEvent> {
-		let mut pending_events = Vec::with_capacity(self.events.len());
-		pending_events.append(&mut self.events);
-		pending_events
+	fn flush_events<C: Deref>(&mut self, chain_source: &Option<C>) -> bool where C::Target: chain::Notify {
+		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
 	}
 }
 
-impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
 	///
-	pub fn block_connected(&self, header: &BlockHeader, txdata: &[(usize, &Transaction)], height: u32) {
+	pub fn block_connected(&self, header: &BlockHeader, txdata: &[(usize, &Transaction)], height: u32) -> bool {
 		let mut watch_events = self.watch_events.lock().unwrap();
 		let matched_txn: Vec<_> = txdata.iter().filter(|&&(_, tx)| watch_events.watched.does_match_tx(tx)).map(|e| *e).collect();
 		{
@@ -245,6 +285,7 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
 				}
 			}
 		}
+		watch_events.flush_events(&self.chain_source)
 	}
 
 	///
@@ -256,17 +297,19 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
 	}
 }
 
-impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
 	/// Creates a new object which can be used to monitor several channels given the chain
 	/// interface with which to register to receive notifications.
-	pub fn new(broadcaster: T, logger: L, feeest: F) -> ChainMonitor<Key, ChanSigner, T, F, L> {
+	pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F) -> ChainMonitor<Key, ChanSigner, C, T, F, L> {
 		Self {
 			monitors: Mutex::new(HashMap::new()),
-			watch_events: Mutex::new(WatchEventQueue::new()),
+			watch_events: Mutex::new(WatchEventCache::new()),
+			chain_source,
 			broadcaster,
 			logger,
 			fee_estimator: feeest,
@@ -293,6 +336,7 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
 			}
 		}
 		entry.insert(monitor);
+		watch_events.flush_events(&self.chain_source);
 		Ok(())
 	}
 
@@ -309,8 +353,9 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
 	}
 }
 
-impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send, L: Deref + Sync + Send> chain::Watch for ChainMonitor<OutPoint, ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<ChanSigner: ChannelKeys, C: Deref + Sync + Send, T: Deref + Sync + Send, F: Deref + Sync + Send, L: Deref + Sync + Send> chain::Watch for ChainMonitor<OutPoint, ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
@@ -339,8 +384,9 @@ impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send, L:
 	}
 }
 
-impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> events::EventsProvider for ChainMonitor<Key, ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> events::EventsProvider for ChainMonitor<Key, ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
@@ -353,16 +399,6 @@ impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref, F: De
 	}
 }
 
-impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> chain::WatchEventProvider for ChainMonitor<Key, ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
-	      F::Target: FeeEstimator,
-	      L::Target: Logger,
-{
-	fn release_pending_watch_events(&self) -> Vec<chain::WatchEvent> {
-		self.watch_events.lock().unwrap().dequeue_events()
-	}
-}
-
 /// If an HTLC expires within this many blocks, don't try to claim it in a shared transaction,
 /// instead claiming it in its own individual transaction.
 pub(crate) const CLTV_SHARED_CLAIM_BUFFER: u32 = 12;

lightning/src/ln/functional_test_utils.rs

@@ -1,7 +1,6 @@
 //! A bunch of useful utilities for building networks of nodes and exchanging messages between
 //! nodes for functional tests.
 
-use chain;
 use chain::Watch;
 use chain::transaction::OutPoint;
 use ln::channelmanager::{ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, PaymentPreimage, PaymentHash, PaymentSecret, PaymentSendFailure};
@@ -73,28 +72,11 @@ pub fn connect_blocks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, depth: u32, he
 }
 
 pub fn connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: &Block, height: u32) {
-	use chain::WatchEventProvider;
-
-	let watch_events = node.chain_monitor.chain_monitor.release_pending_watch_events();
-	process_chain_watch_events(&watch_events);
-
 	let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
-	loop {
-		node.chain_monitor.chain_monitor.block_connected(&block.header, &txdata, height);
-
-		let watch_events = node.chain_monitor.chain_monitor.release_pending_watch_events();
-		process_chain_watch_events(&watch_events);
-
-		if watch_events.is_empty() {
-			break;
-		}
-	}
-
+	while node.chain_monitor.chain_monitor.block_connected(&block.header, &txdata, height) {}
 	node.node.block_connected(&block.header, &txdata, height);
 }
 
-fn process_chain_watch_events(_events: &Vec<chain::WatchEvent>) {}
-
 pub fn disconnect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, header: &BlockHeader, height: u32) {
 	node.chain_monitor.chain_monitor.block_disconnected(header, height);
 	node.node.block_disconnected(header, height);
@@ -207,7 +189,7 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
 				}).unwrap();
 			}
 
-			let channel_monitor = test_utils::TestChainMonitor::new(self.tx_broadcaster.clone(), &self.logger, &feeest);
+			let channel_monitor = test_utils::TestChainMonitor::new(Some(self.chain_source), self.tx_broadcaster.clone(), &self.logger, &feeest);
 			for deserialized_monitor in deserialized_monitors.drain(..) {
 				if let Err(_) = channel_monitor.watch_channel(deserialized_monitor.get_funding_txo().0, deserialized_monitor) {
 					panic!();
@@ -1112,7 +1094,7 @@ pub fn create_node_cfgs<'a>(node_count: usize, chanmon_cfgs: &'a Vec<TestChanMon
 	for i in 0..node_count {
 		let seed = [i as u8; 32];
 		let keys_manager = test_utils::TestKeysInterface::new(&seed, Network::Testnet);
-		let chain_monitor = test_utils::TestChainMonitor::new(&chanmon_cfgs[i].tx_broadcaster, &chanmon_cfgs[i].logger, &chanmon_cfgs[i].fee_estimator);
+		let chain_monitor = test_utils::TestChainMonitor::new(Some(&chanmon_cfgs[i].chain_source), &chanmon_cfgs[i].tx_broadcaster, &chanmon_cfgs[i].logger, &chanmon_cfgs[i].fee_estimator);
 		nodes.push(NodeCfg { chain_source: &chanmon_cfgs[i].chain_source, logger: &chanmon_cfgs[i].logger, tx_broadcaster: &chanmon_cfgs[i].tx_broadcaster, fee_estimator: &chanmon_cfgs[i].fee_estimator, chain_monitor, keys_manager, node_seed: seed });
 	}