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<EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	pub chain_monitor: Arc<channelmonitor::ChainMonitor<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,13 +137,13 @@ impl<'a> std::hash::Hash for Peer<'a> {
 
 type ChannelMan = ChannelManager<
 	EnforcingChannelKeys,
-	Arc<channelmonitor::ChainMonitor<EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	Arc<channelmonitor::ChainMonitor<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<EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	monitor: Arc<channelmonitor::ChainMonitor<EnforcingChannelKeys, Arc<dyn chain::Notify>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
 	handler: PeerMan<'a>,
 
 	peers: &'a RefCell<[bool; 256]>,
@@ -157,7 +157,7 @@ struct MoneyLossDetector<'a> {
 impl<'a> MoneyLossDetector<'a> {
 	pub fn new(peers: &'a RefCell<[bool; 256]>,
 	           manager: Arc<ChannelMan>,
-	           monitor: Arc<channelmonitor::ChainMonitor<EnforcingChannelKeys, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
+	           monitor: Arc<channelmonitor::ChainMonitor<EnforcingChannelKeys, Arc<dyn chain::Notify>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>>>,
 	           handler: PeerMan<'a>) -> Self {
 		MoneyLossDetector {
 			manager,
@@ -331,7 +331,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::keysinterface::InMemoryChannelKeys, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>>;
+//! type ChainNotify = dyn lightning::chain::Notify;
+//! type ChainMonitor = lightning::ln::channelmonitor::ChainMonitor<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};
@@ -160,28 +161,51 @@ impl_writeable!(HTLCUpdate, 0, { payment_hash, payment_preimage, source });
 /// independently to monitor channels remotely.
 ///
 /// [`chain::Watch`]: ../../chain/trait.Watch.html
-/// [`ChannelManager`]: ../channelmanager/struct.ChannelManager.html
-pub struct ChainMonitor<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref>
-	where T::Target: BroadcasterInterface,
+pub struct ChainMonitor<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<OutPoint, ChannelMonitor<ChanSigner>>>,
 	#[cfg(not(test))]
 	monitors: Mutex<HashMap<OutPoint, 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(),
@@ -191,7 +215,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()
 			});
@@ -201,7 +225,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,
@@ -211,24 +235,43 @@ 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<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonitor<ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
 	/// Delegates to [`ChannelMonitor::block_connected`] for each watched channel. Any HTLCs that
 	/// were resolved on chain will be retuned by [`chain::Watch::release_pending_htlc_updates`].
 	///
+	/// Calls back to [`chain::Notify`] if any monitor indicated new outputs to watch, returning
+	/// `true` if so.
+	///
 	/// [`ChannelMonitor::block_connected`]: struct.ChannelMonitor.html#method.block_connected
 	/// [`chain::Watch::release_pending_htlc_updates`]: ../../chain/trait.Watch.html#tymethod.release_pending_htlc_updates
-	pub fn block_connected(&self, header: &BlockHeader, txdata: &[(usize, &Transaction)], height: u32) {
+	/// [`chain::Notify`]: ../../chain/trait.Notify.html
+	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();
 		{
@@ -243,6 +286,7 @@ impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, Ch
 				}
 			}
 		}
+		watch_events.flush_events(&self.chain_source)
 	}
 
 	/// Delegates to [`ChannelMonitor::block_disconnected`] for each watched channel.
@@ -256,24 +300,30 @@ impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<Key, Ch
 	}
 }
 
-impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> ChainMonitor<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) -> Self {
+	pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F) -> Self {
 		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,
 		}
 	}
 
 	/// Adds or updates the monitor which monitors the channel referred to by the given outpoint.
+	///
+	/// Calls back to [`chain::Notify`] with the funding transaction and outputs to watch.
+	///
+	/// [`chain::Notify`]: ../../chain/trait.Notify.html
 	pub 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();
@@ -293,6 +343,7 @@ impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<ChanSig
 			}
 		}
 		entry.insert(monitor);
+		watch_events.flush_events(&self.chain_source);
 		Ok(())
 	}
 
@@ -309,8 +360,9 @@ impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> ChainMonitor<ChanSig
 	}
 }
 
-impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send, L: Deref + Sync + Send> chain::Watch for ChainMonitor<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<ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
@@ -339,8 +391,9 @@ impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send, L:
 	}
 }
 
-impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> events::EventsProvider for ChainMonitor<ChanSigner, T, F, L>
-	where T::Target: BroadcasterInterface,
+impl<ChanSigner: ChannelKeys, C: Deref, T: Deref, F: Deref, L: Deref> events::EventsProvider for ChainMonitor<ChanSigner, C, T, F, L>
+	where C::Target: chain::Notify,
+	      T::Target: BroadcasterInterface,
 	      F::Target: FeeEstimator,
 	      L::Target: Logger,
 {
@@ -353,16 +406,6 @@ impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> events::EventsProvid
 	}
 }
 
-impl<ChanSigner: ChannelKeys, T: Deref, F: Deref, L: Deref> chain::WatchEventProvider for ChainMonitor<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;