chaininterface: add BlockNotifier struct

Adding this struct will allow us to remove the circular reference
between ChainListeners and the ChainWatchInterface, because it
separates out the responsibility of notifying listeners about new
blocks from the responsibility of storing and retrieving watched
transactions.

Author: valentinewallace

src/chain/chaininterface.rs

@@ -193,6 +193,75 @@ impl ChainWatchedUtil {
 	}
 }
 
+/// Utility for notifying listeners about new blocks, and handling block rescans if new watch
+/// data is registered.
+pub struct BlockNotifier<'a, 'b: 'a> {
+	listeners: Mutex<Vec<Weak<ChainListener + 'a>>>, //TODO(vmw): try removing Weak
+	chain_monitor: Arc<ChainWatchInterface<'b>>,
+}
+
+impl<'a, 'b> BlockNotifier<'a, 'b> {
+	/// Constructs a new BlockNotifier without any listeners.
+	pub fn new(chain_monitor: Arc<ChainWatchInterface>) -> BlockNotifier {
+		BlockNotifier {
+			listeners: Mutex::new(Vec::new()),
+			chain_monitor,
+		}
+	}
+
+	/// Register the given listener to receive events. Only a weak pointer is provided and
+	/// the registration should be freed once that pointer expires.
+	// TODO: unregister
+	pub fn register_listener(&self, listener: Weak<ChainListener + 'a>) {
+		let mut vec = self.listeners.lock().unwrap();
+		vec.push(listener);
+	}
+
+	/// Notify listeners that a block was connected given a full, unfiltered block.
+	///
+	/// Handles re-scanning the block and calling block_connected again if listeners register new
+	/// watch data during the callbacks for you (see ChainListener::block_connected for more info).
+	pub fn block_connected(&self, block: &'b Block, height: u32) {
+		let mut reentered = true;
+		while reentered {
+			let (matched, matched_index) = self.chain_monitor.filter_block(block);
+			reentered = self.block_connected_checked(&block.header, height, matched.as_slice(), matched_index.as_slice());
+		}
+	}
+
+	/// Notify listeners that a block was connected, given pre-filtered list of transactions in the
+	/// block which matched the filter (probably using does_match_tx).
+	///
+	/// Returns true if notified listeners registered additional watch data (implying that the
+	/// block must be re-scanned and this function called again prior to further block_connected
+	/// calls, see ChainListener::block_connected for more info).
+	pub fn block_connected_checked(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) -> bool {
+		let last_seen = self.chain_monitor.reentered();
+
+		let listeners = self.listeners.lock().unwrap().clone();
+		for listener in listeners.iter() {
+			match listener.upgrade() {
+				Some(arc) => arc.block_connected(header, height, txn_matched, indexes_of_txn_matched),
+				None => ()
+			}
+		}
+		return last_seen != self.chain_monitor.reentered();
+	}
+
+
+	/// Notify listeners that a block was disconnected.
+	pub fn block_disconnected(&self, header: &BlockHeader, disconnected_height: u32) {
+		let listeners = self.listeners.lock().unwrap().clone();
+		for listener in listeners.iter() {
+			match listener.upgrade() {
+				Some(arc) => arc.block_disconnected(&header, disconnected_height),
+				None => ()
+			}
+		}
+	}
+
+}
+
 /// Utility to capture some common parts of ChainWatchInterface implementors.
 ///
 /// Keeping a local copy of this in a ChainWatchInterface implementor is likely useful.
@@ -240,63 +309,11 @@ impl ChainWatchInterfaceUtil {
 		ChainWatchInterfaceUtil {
 			network: network,
 			watched: Mutex::new(ChainWatchedUtil::new()),
-			listeners: Mutex::new(Vec::new()),
 			reentered: AtomicUsize::new(1),
 			logger: logger,
 		}
 	}
 
-	/// Notify listeners that a block was connected given a full, unfiltered block.
-	///
-	/// Handles re-scanning the block and calling block_connected again if listeners register new
-	/// watch data during the callbacks for you (see ChainListener::block_connected for more info).
-	pub fn block_connected_with_filtering(&self, block: &Block, height: u32) {
-		let mut reentered = true;
-		while reentered {
-			let mut matched = Vec::new();
-			let mut matched_index = Vec::new();
-			{
-				let watched = self.watched.lock().unwrap();
-				for (index, transaction) in block.txdata.iter().enumerate() {
-					if self.does_match_tx_unguarded(transaction, &watched) {
-						matched.push(transaction);
-						matched_index.push(index as u32);
-					}
-				}
-			}
-			reentered = self.block_connected_checked(&block.header, height, matched.as_slice(), matched_index.as_slice());
-		}
-	}
-
-	/// Notify listeners that a block was disconnected.
-	pub fn block_disconnected(&self, header: &BlockHeader, disconnected_height: u32) {
-		let listeners = self.listeners.lock().unwrap().clone();
-		for listener in listeners.iter() {
-			match listener.upgrade() {
-				Some(arc) => arc.block_disconnected(&header, disconnected_height),
-				None => ()
-			}
-		}
-	}
-
-	/// Notify listeners that a block was connected, given pre-filtered list of transactions in the
-	/// block which matched the filter (probably using does_match_tx).
-	///
-	/// Returns true if notified listeners registered additional watch data (implying that the
-	/// block must be re-scanned and this function called again prior to further block_connected
-	/// calls, see ChainListener::block_connected for more info).
-	pub fn block_connected_checked(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) -> bool {
-		let last_seen = self.reentered.load(Ordering::Relaxed);
-
-		let listeners = self.listeners.lock().unwrap().clone();
-		for listener in listeners.iter() {
-			match listener.upgrade() {
-				Some(arc) => arc.block_connected(header, height, txn_matched, indexes_of_txn_matched),
-				None => ()
-			}
-		}
-		return last_seen != self.reentered.load(Ordering::Relaxed);
-	}
 
 	/// Checks if a given transaction matches the current filter.
 	pub fn does_match_tx(&self, tx: &Transaction) -> bool {

src/ln/chanmon_update_fail_tests.rs

@@ -1627,11 +1627,11 @@ fn do_during_funding_monitor_fail(fail_on_generate: bool, restore_between_fails:
 	};
 
 	if confirm_a_first {
-		confirm_transaction(&nodes[0].chain_monitor, &funding_tx, funding_tx.version);
+		confirm_transaction(&nodes[0].block_notifier, &nodes[0].chain_monitor, &funding_tx, funding_tx.version);
 		nodes[1].node.handle_funding_locked(&nodes[0].node.get_our_node_id(), &get_event_msg!(nodes[0], MessageSendEvent::SendFundingLocked, nodes[1].node.get_our_node_id())).unwrap();
 	} else {
 		assert!(!restore_b_before_conf);
-		confirm_transaction(&nodes[1].chain_monitor, &funding_tx, funding_tx.version);
+		confirm_transaction(&nodes[1].block_notifier, &nodes[1].chain_monitor, &funding_tx, funding_tx.version);
 		assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 	}
 
@@ -1643,7 +1643,7 @@ fn do_during_funding_monitor_fail(fail_on_generate: bool, restore_between_fails:
 	assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 
 	if !restore_b_before_conf {
-		confirm_transaction(&nodes[1].chain_monitor, &funding_tx, funding_tx.version);
+		confirm_transaction(&nodes[1].block_notifier, &nodes[1].chain_monitor, &funding_tx, funding_tx.version);
 		assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
 		assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
 	}
@@ -1655,12 +1655,12 @@ fn do_during_funding_monitor_fail(fail_on_generate: bool, restore_between_fails:
 	let (channel_id, (announcement, as_update, bs_update)) = if !confirm_a_first {
 		nodes[0].node.handle_funding_locked(&nodes[1].node.get_our_node_id(), &get_event_msg!(nodes[1], MessageSendEvent::SendFundingLocked, nodes[0].node.get_our_node_id())).unwrap();
 
-		confirm_transaction(&nodes[0].chain_monitor, &funding_tx, funding_tx.version);
+		confirm_transaction(&nodes[0].block_notifier, &nodes[0].chain_monitor, &funding_tx, funding_tx.version);
 		let (funding_locked, channel_id) = create_chan_between_nodes_with_value_confirm_second(&nodes[1], &nodes[0]);
 		(channel_id, create_chan_between_nodes_with_value_b(&nodes[0], &nodes[1], &funding_locked))
 	} else {
 		if restore_b_before_conf {
-			confirm_transaction(&nodes[1].chain_monitor, &funding_tx, funding_tx.version);
+			confirm_transaction(&nodes[1].block_notifier, &nodes[1].chain_monitor, &funding_tx, funding_tx.version);
 		}
 		let (funding_locked, channel_id) = create_chan_between_nodes_with_value_confirm_second(&nodes[0], &nodes[1]);
 		(channel_id, create_chan_between_nodes_with_value_b(&nodes[1], &nodes[0], &funding_locked))

src/ln/functional_test_utils.rs

@@ -34,27 +34,28 @@ use std::sync::{Arc, Mutex};
 use std::mem;
 
 pub const CHAN_CONFIRM_DEPTH: u32 = 100;
-pub fn confirm_transaction(chain: &chaininterface::ChainWatchInterfaceUtil, tx: &Transaction, chan_id: u32) {
+pub fn confirm_transaction(notifier: &chaininterface::BlockNotifier, chain: &chaininterface::ChainWatchInterfaceUtil, tx: &Transaction, chan_id: u32) {
 	assert!(chain.does_match_tx(tx));
 	let mut header = BlockHeader { version: 0x20000000, prev_blockhash: Default::default(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
-	chain.block_connected_checked(&header, 1, &[tx; 1], &[chan_id; 1]);
+	notifier.block_connected_checked(&header, 1, &[tx; 1], &[chan_id; 1]);
 	for i in 2..CHAN_CONFIRM_DEPTH {
 		header = BlockHeader { version: 0x20000000, prev_blockhash: header.bitcoin_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
-		chain.block_connected_checked(&header, i, &[tx; 0], &[0; 0]);
+		notifier.block_connected_checked(&header, i, &vec![], &[0; 0]);
 	}
 }
 
-pub fn connect_blocks(chain: &chaininterface::ChainWatchInterfaceUtil, depth: u32, height: u32, parent: bool, prev_blockhash: Sha256d) -> Sha256d {
+pub fn connect_blocks(notifier: &chaininterface::BlockNotifier, depth: u32, height: u32, parent: bool, prev_blockhash: Sha256d) -> Sha256d {
 	let mut header = BlockHeader { version: 0x2000000, prev_blockhash: if parent { prev_blockhash } else { Default::default() }, merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
-	chain.block_connected_checked(&header, height + 1, &Vec::new(), &Vec::new());
+	notifier.block_connected_checked(&header, height + 1, &Vec::new(), &Vec::new());
 	for i in 2..depth + 1 {
 		header = BlockHeader { version: 0x20000000, prev_blockhash: header.bitcoin_hash(), merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 };
-		chain.block_connected_checked(&header, height + i, &Vec::new(), &Vec::new());
+		notifier.block_connected_checked(&header, height + i, &Vec::new(), &Vec::new());
 	}
 	header.bitcoin_hash()
 }
 
 pub struct Node {
+	pub block_notifier: Arc<chaininterface::BlockNotifier<'a, 'b>>,
 	pub chain_monitor: Arc<chaininterface::ChainWatchInterfaceUtil>,
 	pub tx_broadcaster: Arc<test_utils::TestBroadcaster>,
 	pub chan_monitor: Arc<test_utils::TestChannelMonitor>,
@@ -220,7 +221,7 @@ pub fn create_chan_between_nodes_with_value_init(node_a: &Node, node_b: &Node, c
 }
 
 pub fn create_chan_between_nodes_with_value_confirm_first(node_recv: &Node, node_conf: &Node, tx: &Transaction) {
-	confirm_transaction(&node_conf.chain_monitor, &tx, tx.version);
+	confirm_transaction(&node_conf.block_notifier, &node_conf.chain_monitor, &tx, tx.version);
 	node_recv.node.handle_funding_locked(&node_conf.node.get_our_node_id(), &get_event_msg!(node_conf, MessageSendEvent::SendFundingLocked, node_recv.node.get_our_node_id())).unwrap();
 }
 
@@ -246,7 +247,7 @@ pub fn create_chan_between_nodes_with_value_confirm_second(node_recv: &Node, nod
 
 pub fn create_chan_between_nodes_with_value_confirm(node_a: &Node, node_b: &Node, tx: &Transaction) -> ((msgs::FundingLocked, msgs::AnnouncementSignatures), [u8; 32]) {
 	create_chan_between_nodes_with_value_confirm_first(node_a, node_b, tx);
-	confirm_transaction(&node_a.chain_monitor, &tx, tx.version);
+	confirm_transaction(&node_a.block_notifier, &node_a.chain_monitor, &tx, tx.version);
 	create_chan_between_nodes_with_value_confirm_second(node_b, node_a)
 }
 
@@ -836,19 +837,25 @@ pub fn create_network(node_count: usize, node_config: &[Option<UserConfig>]) ->
 		let logger: Arc<Logger> = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
 		let feeest = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: 253 });
 		let chain_monitor = Arc::new(chaininterface::ChainWatchInterfaceUtil::new(Network::Testnet, Arc::clone(&logger)));
+		let block_notifier = Arc::new(chaininterface::BlockNotifier::new(chain_monitor.clone()));
 		let tx_broadcaster = Arc::new(test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new())});
 		let mut seed = [0; 32];
 		rng.fill_bytes(&mut seed);
 		let keys_manager = Arc::new(test_utils::TestKeysInterface::new(&seed, Network::Testnet, Arc::clone(&logger)));
 		let chan_monitor = Arc::new(test_utils::TestChannelMonitor::new(chain_monitor.clone(), tx_broadcaster.clone(), logger.clone(), feeest.clone()));
+		let weak_res = Arc::downgrade(&chan_monitor.simple_monitor);
+		block_notifier.register_listener(weak_res);
 		let mut default_config = UserConfig::new();
 		default_config.channel_options.announced_channel = true;
 		default_config.peer_channel_config_limits.force_announced_channel_preference = false;
 		let node = ChannelManager::new(Network::Testnet, feeest.clone(), chan_monitor.clone(), tx_broadcaster.clone(), Arc::clone(&logger), keys_manager.clone(), if node_config[i].is_some() { node_config[i].clone().unwrap() } else { default_config }).unwrap();
+		let weak_res = Arc::downgrade(&node);
+		block_notifier.register_listener(weak_res);
 		let router = Router::new(PublicKey::from_secret_key(&secp_ctx, &keys_manager.get_node_secret()), chain_monitor.clone(), Arc::clone(&logger));
 		nodes.push(Node { chain_monitor, tx_broadcaster, chan_monitor, node, router, keys_manager, node_seed: seed,
 			network_payment_count: payment_count.clone(),
 			network_chan_count: chan_count.clone(),
+			block_notifier,
 		});
 	}