Merge PR 'Allow filtered block rescans at the current tip' (#4847)

### CONTRIBUTING.md
@@ -195,6 +195,7 @@ welcomed.
   * `TRANSACTIONS_DUPLICATIVELY_FIRST_SKIPPING_BLOCKS`
   * `HIGHLY_REDUNDANT_TRANSACTIONS_FIRST_SKIPPING_BLOCKS`
   * `TRANSACTIONS_FIRST_REORGS_ONLY_TIP`
+  * `REPLAYED_FULL_BLOCK_VIA_LISTEN`
   * `FULL_BLOCK_VIA_LISTEN`
   * `FULL_BLOCK_DISCONNECTIONS_SKIPPING_VIA_LISTEN`
 

### lightning/src/ln/channelmanager.rs
@@ -16636,16 +16636,23 @@ impl<
 	> chain::Listen for ChannelManager<M, T, ES, NS, SP, F, R, MR, L>
 {
 	fn filtered_block_connected(&self, header: &Header, txdata: &TransactionData, height: u32) {
-		{
+		let is_rescan = {
 			let best_block = self.best_block.read().unwrap();
-			assert_eq!(best_block.block_hash, header.prev_blockhash,
-				"Blocks must be connected in chain-order - the connected header must build on the last connected header");
-			assert_eq!(best_block.height, height - 1,
-				"Blocks must be connected in chain-order - the connected block height must be one greater than the previous height");
-		}
+			let is_rescan =
+				best_block.block_hash == header.block_hash() && best_block.height == height;
+			if !is_rescan {
+				assert_eq!(best_block.block_hash, header.prev_blockhash,
+					"Blocks must be connected in chain-order - the connected header must build on the last connected header");
+				assert_eq!(best_block.height, height - 1,
+					"Blocks must be connected in chain-order - the connected block height must be one greater than the previous height");
+			}
+			is_rescan
+		};
 
 		self.transactions_confirmed(header, txdata, height);
-		self.best_block_updated(header, height);
+		if !is_rescan {
+			self.best_block_updated(header, height);
+		}
 	}
 
 	fn blocks_disconnected(&self, fork_point: BlockLocator) {

### lightning/src/ln/functional_test_utils.rs
@@ -204,6 +204,9 @@ pub enum ConnectStyle {
 	/// Provides the full block via the `chain::Listen` interface. In the current code this is
 	/// equivalent to `TransactionsFirst` with some additional assertions.
 	FullBlockViaListen,
+	/// Provides the full block via the `chain::Listen` interface, but replays it a second time
+	/// similar to what a filtering client might do.
+	ReplayedFullBlockViaListen,
 	/// Provides the full block via the `chain::Listen` interface, condensing multiple block
 	/// disconnections into a single `blocks_disconnected` call.
 	FullBlockDisconnectionsSkippingViaListen,
@@ -221,6 +224,7 @@ impl ConnectStyle {
 			ConnectStyle::HighlyRedundantTransactionsFirstSkippingBlocks => true,
 			ConnectStyle::TransactionsFirstReorgsOnlyTip => true,
 			ConnectStyle::FullBlockViaListen => false,
+			ConnectStyle::ReplayedFullBlockViaListen => false,
 			ConnectStyle::FullBlockDisconnectionsSkippingViaListen => false,
 		}
 	}
@@ -236,6 +240,7 @@ impl ConnectStyle {
 			ConnectStyle::HighlyRedundantTransactionsFirstSkippingBlocks => false,
 			ConnectStyle::TransactionsFirstReorgsOnlyTip => false,
 			ConnectStyle::FullBlockViaListen => false,
+			ConnectStyle::ReplayedFullBlockViaListen => true,
 			ConnectStyle::FullBlockDisconnectionsSkippingViaListen => false,
 		}
 	}
@@ -244,7 +249,7 @@ impl ConnectStyle {
 		use core::hash::{BuildHasher, Hasher};
 		// Get a random value using the only std API to do so - the DefaultHasher
 		let rand_val = std::collections::hash_map::RandomState::new().build_hasher().finish();
-		let res = match rand_val % 10 {
+		let res = match rand_val % 11 {
 			0 => ConnectStyle::BestBlockFirst,
 			1 => ConnectStyle::BestBlockFirstSkippingBlocks,
 			2 => ConnectStyle::BestBlockFirstReorgsOnlyTip,
@@ -254,7 +259,8 @@ impl ConnectStyle {
 			6 => ConnectStyle::HighlyRedundantTransactionsFirstSkippingBlocks,
 			7 => ConnectStyle::TransactionsFirstReorgsOnlyTip,
 			8 => ConnectStyle::FullBlockViaListen,
-			9 => ConnectStyle::FullBlockDisconnectionsSkippingViaListen,
+			9 => ConnectStyle::ReplayedFullBlockViaListen,
+			10 => ConnectStyle::FullBlockDisconnectionsSkippingViaListen,
 			_ => unreachable!(),
 		};
 		eprintln!("Using Block Connection Style: {:?}", res);
@@ -316,11 +322,25 @@ fn do_connect_block_with_consistency_checks<'a, 'b, 'c, 'd>(
 fn do_connect_block_without_consistency_checks<'a, 'b, 'c, 'd>(
 	node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool,
 ) {
-	let height = node.best_block_info().1 + 1;
 	eprintln!("Connecting block using Block Connection Style: {:?}", *node.connect_style.borrow());
-	// Update the block internally before handing it over to LDK, to ensure our assertions regarding
-	// transaction broadcast are correct.
-	node.blocks.lock().unwrap().push((block.clone(), height));
+	let (new_block, height) = {
+		let mut blocks = node.blocks.lock().unwrap();
+		let existing =
+			blocks.iter().rev().find(|(candidate, _)| candidate == &block).map(|(_, h)| *h);
+		if let Some(height) = existing {
+			// We're being handed a block we've already connected, i.e. this is a redundant rescan
+			// rather than a new block. Reuse the height it was originally connected at rather than
+			// extending the chain.
+			(false, height)
+		} else {
+			let height = blocks.last().unwrap().1 + 1;
+			// Update the block internally before handing it over to LDK, to ensure our assertions
+			// regarding transaction broadcast are correct.
+			blocks.push((block.clone(), height));
+			(true, height)
+		}
+	};
+
 	if !skip_intermediaries {
 		let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
 		match *node.connect_style.borrow() {
@@ -390,17 +410,26 @@ fn do_connect_block_without_consistency_checks<'a, 'b, 'c, 'd>(
 				node.chain_monitor.chain_monitor.block_connected(&block, height);
 				node.node.block_connected(&block, height);
 			},
+			ConnectStyle::ReplayedFullBlockViaListen => {
+				let header = &block.header;
+				node.chain_monitor.chain_monitor.filtered_block_connected(header, &[], height);
+				node.node.filtered_block_connected(header, &[], height);
+				node.chain_monitor.chain_monitor.block_connected(&block, height);
+				node.node.block_connected(&block, height);
+			},
 		}
 	}
 
-	for tx in &block.txdata {
-		for input in &tx.input {
-			node.wallet_source.remove_utxo(input.previous_output);
-		}
-		let wallet_script = node.wallet_source.get_change_script().unwrap();
-		for (idx, output) in tx.output.iter().enumerate() {
-			if output.script_pubkey == wallet_script {
-				node.wallet_source.add_utxo(tx.clone(), idx as u32);
+	if new_block {
+		for tx in &block.txdata {
+			for input in &tx.input {
+				node.wallet_source.remove_utxo(input.previous_output);
+			}
+			let wallet_script = node.wallet_source.get_change_script().unwrap();
+			for (idx, output) in tx.output.iter().enumerate() {
+				if output.script_pubkey == wallet_script {
+					node.wallet_source.add_utxo(tx.clone(), idx as u32);
+				}
 			}
 		}
 	}
@@ -447,7 +476,7 @@ pub fn disconnect_blocks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, count: u32)
 		let prev = node.blocks.lock().unwrap().last().unwrap().clone();
 
 		match *node.connect_style.borrow() {
-			ConnectStyle::FullBlockViaListen => {
+			ConnectStyle::FullBlockViaListen | ConnectStyle::ReplayedFullBlockViaListen => {
 				let best_block = BlockLocator::new(orig.0.header.prev_blockhash, orig.1 - 1);
 				node.chain_monitor.chain_monitor.blocks_disconnected(best_block);
 				Listen::blocks_disconnected(node.node, best_block);
@@ -4795,6 +4824,7 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(
 			},
 			"TRANSACTIONS_FIRST_REORGS_ONLY_TIP" => ConnectStyle::TransactionsFirstReorgsOnlyTip,
 			"FULL_BLOCK_VIA_LISTEN" => ConnectStyle::FullBlockViaListen,
+			"REPLAYED_FULL_BLOCK_VIA_LISTEN" => ConnectStyle::ReplayedFullBlockViaListen,
 			"FULL_BLOCK_DISCONNECTIONS_SKIPPING_VIA_LISTEN" => {
 				ConnectStyle::FullBlockDisconnectionsSkippingViaListen
 			},

### lightning/src/ln/functional_tests.rs
@@ -2362,16 +2362,6 @@ pub fn test_htlc_ignore_latest_remote_commitment() {
 	let node_a_id = nodes[0].node.get_our_node_id();
 	let node_b_id = nodes[1].node.get_our_node_id();
 
-	match *nodes[1].connect_style.borrow() {
-		ConnectStyle::FullBlockViaListen
-		| ConnectStyle::FullBlockDisconnectionsSkippingViaListen => {
-			// We rely on the ability to connect a block redundantly, which isn't allowed via
-			// `chain::Listen`, so we never run the test if we randomly get assigned that
-			// connect_style.
-			return;
-		},
-		_ => {},
-	}
 	let funding_tx = create_announced_chan_between_nodes(&nodes, 0, 1).3;
 	let message = "Channel force-closed".to_owned();
 	route_payment(&nodes[0], &[&nodes[1]], 10000000);
@@ -3005,6 +2995,28 @@ pub fn test_drop_messages_peer_disconnect_b() {
 	do_test_drop_messages_peer_disconnect(6, false);
 }
 
+#[xtest(feature = "_externalize_tests")]
+pub fn test_filtered_block_connected_allows_same_block_rescan() {
+	let chanmon_cfgs = create_chanmon_cfgs(1);
+	let node_cfgs = create_node_cfgs(1, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(1, &node_cfgs, &[None]);
+	let nodes = create_network(1, &node_cfgs, &node_chanmgrs);
+
+	let best_block = nodes[0].node.current_best_block();
+	let height = best_block.height + 1;
+	let header = create_dummy_header(best_block.block_hash, height);
+	nodes[0].node.filtered_block_connected(&header, &[], height);
+	nodes[0].node.filtered_block_connected(&header, &[], height);
+
+	let current_best_block = nodes[0].node.current_best_block();
+	assert_eq!(current_best_block.block_hash, header.block_hash());
+	assert_eq!(current_best_block.height, height);
+	assert_eq!(
+		current_best_block.get_hash_at_height(best_block.height),
+		Some(best_block.block_hash)
+	);
+}
+
 #[xtest(feature = "_externalize_tests")]
 pub fn test_channel_ready_without_best_block_updated() {
 	// Previously, if we were offline when a funding transaction was locked in, and then we came

### lightning/src/ln/monitor_tests.rs
@@ -2859,17 +2859,19 @@ fn do_test_anchors_aggregated_revoked_htlc_tx(p2a_anchor: bool) {
 	let mut events = nodes[1].chain_monitor.chain_monitor.get_and_clear_pending_events();
 	// Certain block `ConnectStyle`s cause an extra `ChannelClose` event to be emitted since the
 	// best block is updated before the confirmed transactions are notified.
-	match *nodes[1].connect_style.borrow() {
-		ConnectStyle::BestBlockFirst|ConnectStyle::BestBlockFirstReorgsOnlyTip|ConnectStyle::BestBlockFirstSkippingBlocks => {
-			assert_eq!(events.len(), 4);
-			if let Event::BumpTransaction(BumpTransactionEvent::ChannelClose { .. }) = events.remove(0) {}
-			else { panic!("unexpected event"); }
-			if let Event::BumpTransaction(BumpTransactionEvent::ChannelClose { .. }) = events.remove(1) {}
-			else { panic!("unexpected event"); }
-
-		},
-		_ => assert_eq!(events.len(), 2),
-	};
+	if nodes[1].connect_style.borrow().updates_best_block_first() {
+		assert_eq!(events.len(), 4);
+		if let Event::BumpTransaction(BumpTransactionEvent::ChannelClose { .. }) = events.remove(0) {
+		} else {
+			panic!("unexpected event");
+		}
+		if let Event::BumpTransaction(BumpTransactionEvent::ChannelClose { .. }) = events.remove(1) {
+		} else {
+			panic!("unexpected event");
+		}
+	} else {
+		assert_eq!(events.len(), 2);
+	}
 	let htlc_tx = {
 		let secret_key = SecretKey::from_slice(&[1; 32]).unwrap();
 		let public_key = PublicKey::new(secret_key.public_key(&secp));

### lightning/src/util/sweep.rs
@@ -763,13 +763,19 @@ where
 		&self, header: &Header, txdata: &chain::transaction::TransactionData, height: u32,
 	) {
 		let mut state_lock = self.sweeper_state.lock().unwrap();
-		assert_eq!(state_lock.best_block.block_hash, header.prev_blockhash,
-			"Blocks must be connected in chain-order - the connected header must build on the last connected header");
-		assert_eq!(state_lock.best_block.height, height - 1,
-			"Blocks must be connected in chain-order - the connected block height must be one greater than the previous height");
+		let is_rescan = state_lock.best_block.block_hash == header.block_hash()
+			&& state_lock.best_block.height == height;
+		if !is_rescan {
+			assert_eq!(state_lock.best_block.block_hash, header.prev_blockhash,
+				"Blocks must be connected in chain-order - the connected header must build on the last connected header");
+			assert_eq!(state_lock.best_block.height, height - 1,
+				"Blocks must be connected in chain-order - the connected block height must be one greater than the previous height");
+		}
 
 		self.transactions_confirmed_internal(&mut state_lock, header, txdata, height);
-		self.best_block_updated_internal(&mut state_lock, header, height);
+		if !is_rescan {
+			self.best_block_updated_internal(&mut state_lock, header, height);
+		}
 	}
 
 	fn blocks_disconnected(&self, fork_point: BlockLocator) {
@@ -1329,4 +1335,75 @@ mod tests {
 			"pending_sweep flag was not reset when the future was dropped",
 		);
 	}
+
+	#[test]
+	fn filtered_block_connected_allows_same_block_rescan() {
+		let best_block = BlockLocator::new(BlockHash::all_zeros(), 0);
+		let sweeper: OutputSweeper<
+			DummyBroadcaster,
+			Box<DummyChangeDestSource>,
+			DummyFeeEstimator,
+			DummyFilter,
+			PendingKVStore,
+			DummyLogger,
+			DummyOutputSpender,
+		> = OutputSweeper::new(
+			best_block.clone(),
+			DummyBroadcaster,
+			DummyFeeEstimator,
+			None,
+			DummyOutputSpender,
+			Box::new(DummyChangeDestSource),
+			PendingKVStore,
+			DummyLogger,
+		);
+
+		let header = Header {
+			version: bitcoin::block::Version::NO_SOFT_FORK_SIGNALLING,
+			prev_blockhash: best_block.block_hash,
+			merkle_root: bitcoin::hash_types::TxMerkleNode::all_zeros(),
+			time: 1,
+			bits: bitcoin::pow::CompactTarget::from_consensus(42),
+			nonce: 42,
+		};
+		let tracked_outpoint = bitcoin::OutPoint { txid: Txid::all_zeros(), vout: 0 };
+		let spending_tx = Transaction {
+			version: Version::TWO,
+			lock_time: LockTime::ZERO,
+			input: vec![bitcoin::TxIn { previous_output: tracked_outpoint, ..Default::default() }],
+			output: Vec::new(),
+		};
+		let descriptor = SpendableOutputDescriptor::StaticOutput {
+			outpoint: OutPoint { txid: tracked_outpoint.txid, index: tracked_outpoint.vout as u16 },
+			output: TxOut { value: Amount::from_sat(100_000), script_pubkey: ScriptBuf::new() },
+			channel_keys_id: None,
+		};
+		sweeper.sweeper_state.lock().unwrap().outputs.push(TrackedSpendableOutput {
+			descriptor,
+			channel_id: None,
+			counterparty_node_id: None,
+			status: OutputSpendStatus::PendingFirstConfirmation {
+				first_broadcast_hash: best_block.block_hash,
+				latest_broadcast_height: best_block.height,
+				latest_spending_tx: spending_tx.clone(),
+			},
+		});
+
+		sweeper.filtered_block_connected(&header, &[], 1);
+		let txdata = [(0, &spending_tx)];
+		sweeper.filtered_block_connected(&header, &txdata, 1);
+
+		let current_best_block = sweeper.current_best_block();
+		assert_eq!(current_best_block.block_hash, header.block_hash());
+		assert_eq!(current_best_block.height, 1);
+		assert_eq!(current_best_block.get_hash_at_height(0), Some(best_block.block_hash));
+		assert!(matches!(
+			sweeper.tracked_spendable_outputs()[0].status,
+			OutputSpendStatus::PendingThresholdConfirmations {
+				confirmation_height: 1,
+				confirmation_hash,
+				..
+			} if confirmation_hash == header.block_hash()
+		));
+	}
 }