fix: egress id race condition (#4235)

engine/src/witness/btc.rs

@@ -5,7 +5,7 @@ pub mod btc_source;
 use std::sync::Arc;
 
 use bitcoin::{BlockHash, Transaction};
-use cf_chains::btc::{deposit_address::DepositAddress, CHANGE_ADDRESS_SALT};
+use cf_chains::btc::{self, deposit_address::DepositAddress, BlockNumber, CHANGE_ADDRESS_SALT};
 use cf_primitives::EpochIndex;
 use futures_core::Future;
 use secp256k1::hashes::Hash;
@@ -32,7 +32,7 @@ const SAFETY_MARGIN: usize = 5;
 
 pub async fn process_egress<ProcessCall, ProcessingFut, ExtraInfo, ExtraHistoricInfo>(
 	epoch: Vault<cf_chains::Bitcoin, ExtraInfo, ExtraHistoricInfo>,
-	header: Header<u64, BlockHash, (Vec<Transaction>, Vec<[u8; 32]>)>,
+	header: Header<u64, BlockHash, (Vec<Transaction>, Vec<(btc::Hash, BlockNumber)>)>,
 	process_call: ProcessCall,
 ) where
 	ProcessCall: Fn(state_chain_runtime::RuntimeCall, EpochIndex) -> ProcessingFut
@@ -44,7 +44,9 @@ pub async fn process_egress<ProcessCall, ProcessingFut, ExtraInfo, ExtraHistoric
 {
 	let (txs, monitored_tx_hashes) = header.data;
 
-	for tx_hash in success_witnesses(&monitored_tx_hashes, &txs) {
+	let monitored_tx_hashes = monitored_tx_hashes.iter().map(|(tx_hash, _)| tx_hash);
+
+	for tx_hash in success_witnesses(monitored_tx_hashes, &txs) {
 		process_call(
 			state_chain_runtime::RuntimeCall::BitcoinBroadcaster(
 				pallet_cf_broadcast::Call::transaction_succeeded {
@@ -156,11 +158,16 @@ where
 	Ok(())
 }
 
-fn success_witnesses(monitored_tx_hashes: &[[u8; 32]], txs: &Vec<Transaction>) -> Vec<[u8; 32]> {
+fn success_witnesses<'a>(
+	monitored_tx_hashes: impl Iterator<Item = &'a btc::Hash> + Clone,
+	txs: &Vec<Transaction>,
+) -> Vec<btc::Hash> {
 	let mut successful_witnesses = Vec::new();
+
 	for tx in txs {
+		let mut monitored = monitored_tx_hashes.clone();
 		let tx_hash = tx.txid().as_raw_hash().to_byte_array();
-		if monitored_tx_hashes.contains(&tx_hash) {
+		if monitored.any(|&monitored_hash| monitored_hash == tx_hash) {
 			successful_witnesses.push(tx_hash);
 		}
 	}
@@ -197,19 +204,22 @@ mod tests {
 				value: 232232,
 				script_pubkey: ScriptBuf::from(vec![32, 32, 121, 9]),
 			}]),
+			fake_transaction(vec![TxOut {
+				value: 232232,
+				script_pubkey: ScriptBuf::from(vec![33, 2, 1, 9]),
+			}]),
 		];
 
-		let tx_hashes = txs
-			.iter()
-			.map(|tx| tx.txid().to_raw_hash().to_byte_array())
-			// Only watch for the first 2.
-			.take(2)
-			.collect::<Vec<_>>();
+		let tx_hashes =
+			txs.iter().map(|tx| tx.txid().to_raw_hash().to_byte_array()).collect::<Vec<_>>();
+
+		// we're not monitoring for index 2, and they're out of order.
+		let mut monitored_hashes = vec![tx_hashes[3], tx_hashes[0], tx_hashes[1]];
 
-		let success_witnesses = success_witnesses(&tx_hashes, &txs);
+		let mut success_witnesses = success_witnesses(monitored_hashes.iter(), &txs);
+		success_witnesses.sort();
+		monitored_hashes.sort();
 
-		assert_eq!(success_witnesses.len(), 2);
-		assert_eq!(success_witnesses[0], tx_hashes[0]);
-		assert_eq!(success_witnesses[1], tx_hashes[1]);
+		assert_eq!(success_witnesses, monitored_hashes);
 	}
 }

engine/src/witness/common/chunked_chain_source/chunked_by_vault.rs

@@ -2,6 +2,7 @@ pub mod builder;
 pub mod continuous;
 pub mod deposit_addresses;
 pub mod egress_items;
+pub mod monitored_items;
 
 use cf_chains::Chain;
 use futures_util::StreamExt;