diff --git a/crates/starfish/core/src/authority_service.rs b/crates/starfish/core/src/authority_service.rs index 2de5ea7f1b3a..295f62b7887d 100644 --- a/crates/starfish/core/src/authority_service.rs +++ b/crates/starfish/core/src/authority_service.rs @@ -48,7 +48,7 @@ use crate::{ shard_reconstructor::TransactionMessage, stake_aggregator::{QuorumThreshold, StakeAggregator}, storage::Store, - transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI as _}, + transaction_ref::{GenericTransactionRef, TransactionRef}, transactions_synchronizer::TransactionsSynchronizerHandle, }; @@ -1372,10 +1372,14 @@ impl NetworkService for AuthorityService { .handle_fetch_commits(peer, commit_range, CommitSyncType::Fast) .await?; - let transaction_refs: Vec = commits + // The `BlockRef` arm exists only for `CommitV1`, which is no longer + // produced and never enters the per-epoch store these commits are read + // from. + let transaction_refs: Vec = commits .iter() .flat_map(|commit| commit.committed_transactions()) - .collect(); + .map(GenericTransactionRef::expect_transaction_ref) + .collect::>()?; let serialized_transactions = self .handle_fetch_transactions(peer, transaction_refs, TransactionFetchMode::FastCommitSync) @@ -1444,7 +1448,7 @@ impl NetworkService for AuthorityService { async fn handle_fetch_transactions( &self, peer: AuthorityIndex, - mut committed_transactions_refs: Vec, + mut committed_transactions_refs: Vec, fetch_mode: TransactionFetchMode, ) -> ConsensusResult> { fail_point_async!("consensus-rpc-response"); @@ -1490,12 +1494,14 @@ impl NetworkService for AuthorityService { let (below_gc, above_gc): (Vec<_>, Vec<_>) = committed_transactions_refs .iter() .cloned() - .partition(|gen_tx_ref| gen_tx_ref.round() < gc_round); + .partition(|tx_ref| tx_ref.round < gc_round); // Fetch transactions below GC from store let store_transactions = if !below_gc.is_empty() { + let refs: Vec = + below_gc.iter().copied().map(Into::into).collect(); self.store - .read_serialized_transactions(&below_gc)? + .read_serialized_transactions(&refs)? .into_iter() .zip(below_gc) .collect::>() @@ -1505,9 +1511,11 @@ impl NetworkService for AuthorityService { // Fetch transactions at-or-above GC from dag_state let dag_transactions = if !above_gc.is_empty() { + let refs: Vec = + above_gc.iter().copied().map(Into::into).collect(); self.dag_state .read() - .get_serialized_transactions(&above_gc) + .get_serialized_transactions(&refs) .into_iter() .zip(above_gc) .collect::>() @@ -1517,9 +1525,10 @@ impl NetworkService for AuthorityService { // Combine and serialize the results let mut result = Vec::new(); - for (opt_serialized_tx, gen_ref) in store_transactions.into_iter().chain(dag_transactions) { + for (opt_serialized_tx, transaction_ref) in + store_transactions.into_iter().chain(dag_transactions) + { if let Some(serialized_tx) = opt_serialized_tx { - let transaction_ref = gen_ref.expect_transaction_ref()?; let serialized = bcs::to_bytes(&SerializedTransactionsV2 { transaction_ref, serialized_transactions: serialized_tx, @@ -1807,7 +1816,7 @@ mod tests { storage::{Store, WriteBatch, mem_store::MemStore}, test_dag_builder::DagBuilder, transaction::TransactionConsumer, - transaction_ref::GenericTransactionRef, + transaction_ref::{GenericTransactionRef, TransactionRef}, transactions_synchronizer::TransactionsSynchronizer, }; @@ -1866,7 +1875,7 @@ mod tests { async fn fetch_transactions( &self, _peer: AuthorityIndex, - _block_refs: Vec, + _transaction_refs: Vec, _timeout: Duration, ) -> ConsensusResult> { unimplemented!("Unimplemented") @@ -4053,20 +4062,19 @@ mod tests { all_block_headers.push(dag_builder.block_headers(round..=round)); } - let mut block_refs_to_request_first_batch: Vec = (1..=rounds) + let mut tx_refs_to_request_first_batch: Vec = (1..=rounds) .flat_map(|round| { all_block_headers[round as usize] .iter() - .map(|bh| GenericTransactionRef::TransactionRef(bh.transaction_ref())) + .map(|bh| bh.transaction_ref()) }) .collect(); - let mut block_refs_to_request_second_batch: Vec = (rounds + 1 - ..=2 * rounds) + let mut tx_refs_to_request_second_batch: Vec = (rounds + 1..=2 * rounds) .flat_map(|round| { all_block_headers[round as usize] .iter() - .map(|bh| GenericTransactionRef::TransactionRef(bh.transaction_ref())) + .map(|bh| bh.transaction_ref()) }) .collect(); @@ -4074,13 +4082,13 @@ mod tests { let serialized_transactions = authority_service .handle_fetch_transactions( peer, - block_refs_to_request_first_batch.clone(), + tx_refs_to_request_first_batch.clone(), TransactionFetchMode::TransactionSync, ) .await .expect("We should expect a correct return of serialized transactions"); - block_refs_to_request_first_batch.truncate( + tx_refs_to_request_first_batch.truncate( context .parameters .max_transactions_per_transaction_sync_fetch, @@ -4088,9 +4096,9 @@ mod tests { // Verify that we received the correct number of requested transactions assert_eq!( serialized_transactions.len(), - block_refs_to_request_first_batch.len(), - "Should receive {} block transactions", - block_refs_to_request_first_batch.len() + tx_refs_to_request_first_batch.len(), + "Should receive {} transactions", + tx_refs_to_request_first_batch.len() ); // Check the correctness of the received transactions @@ -4101,10 +4109,7 @@ mod tests { let transaction_ref = deserialized.transaction_ref; // Verify it matches the expected ref - assert_eq!( - GenericTransactionRef::TransactionRef(transaction_ref), - block_refs_to_request_first_batch[i] - ); + assert_eq!(transaction_ref, tx_refs_to_request_first_batch[i]); let serialized_transactions = deserialized.serialized_transactions; // Verify the transaction commitment matches @@ -4119,7 +4124,7 @@ mod tests { ); } - block_refs_to_request_second_batch.truncate( + tx_refs_to_request_second_batch.truncate( context .parameters .max_transactions_per_transaction_sync_fetch, @@ -4128,7 +4133,7 @@ mod tests { let serialized_transactions = authority_service .handle_fetch_transactions( peer, - block_refs_to_request_second_batch.clone(), + tx_refs_to_request_second_batch.clone(), TransactionFetchMode::TransactionSync, ) .await diff --git a/crates/starfish/core/src/commit_syncer/fast.rs b/crates/starfish/core/src/commit_syncer/fast.rs index 1b8d2218c5f9..45aaf61b3ca9 100644 --- a/crates/starfish/core/src/commit_syncer/fast.rs +++ b/crates/starfish/core/src/commit_syncer/fast.rs @@ -619,14 +619,20 @@ impl FastCommitSyncer { .await .expect("Spawn blocking should not fail")?; - // 3. Collect all committed transaction refs from commits. Commits passing + // 3. Collect the committed transaction refs of each commit. Commits passing // verify_commits are V2/V3, which only carry `TransactionRef`s, so the // legacy `BlockRef` variant is an error. - let mut committed_tx_refs: BTreeSet = commits + let mut commits_tx_refs: Vec> = commits .iter() - .flat_map(|c| c.committed_transactions()) - .map(GenericTransactionRef::expect_transaction_ref) + .map(|c| { + c.committed_transactions() + .into_iter() + .map(GenericTransactionRef::expect_transaction_ref) + .collect() + }) .collect::>()?; + let mut committed_tx_refs: BTreeSet = + commits_tx_refs.iter().flatten().copied().collect(); // 4. Process fetched transactions. Each serialized_transaction is a // SerializedTransactionsV2 containing both the TransactionRef and the actual @@ -647,6 +653,7 @@ impl FastCommitSyncer { truncate_to_fully_fetched_prefix( target_authority, &mut commits, + &mut commits_tx_refs, &mut fetched_transactions, )?; info!( @@ -693,7 +700,7 @@ impl FastCommitSyncer { // consumers merge-max against their last-seen, so repeating absolute // totals across the batch is a no-op after the first. let misbehavior_counts = inner.dag_state.read().misbehavior_store().snapshot_totals(); - for commit in &commits { + for (commit, commit_tx_refs) in commits.iter().zip(&commits_tx_refs) { // Get block headers from the commit let committed_header_refs = commit.block_headers().to_vec(); @@ -701,8 +708,7 @@ impl FastCommitSyncer { let reputation_scores = commit.reputation_scores().to_vec(); // Collect transactions for this commit - let commit_transactions: Vec = commit - .committed_transactions() + let commit_transactions: Vec = commit_tx_refs .iter() .filter_map(|tx_ref| transactions_map.remove(tx_ref)) .collect(); @@ -956,7 +962,7 @@ fn process_serialized_transactions( peer: AuthorityIndex, serialized_transactions: Vec, committed_tx_refs: &mut BTreeSet, -) -> ConsensusResult> { +) -> ConsensusResult> { let mut fetched_transactions = BTreeMap::new(); for serialized_transaction in serialized_transactions { let tx_v2: SerializedTransactionsV2 = bcs::from_bytes(&serialized_transaction) @@ -965,19 +971,17 @@ fn process_serialized_transactions( if !committed_tx_refs.contains(&transaction_ref) { return Err(ConsensusError::UnexpectedTransactionForCommit { peer, - received: GenericTransactionRef::TransactionRef(transaction_ref), + received: transaction_ref, }); } - fetched_transactions.insert( - GenericTransactionRef::TransactionRef(transaction_ref), - tx_v2.serialized_transactions, - ); + fetched_transactions.insert(transaction_ref, tx_v2.serialized_transactions); committed_tx_refs.remove(&transaction_ref); } Ok(fetched_transactions) } -/// Truncates verified `commits` to the longest prefix whose committed +/// Truncates verified `commits` (and their aligned per-commit transaction +/// refs in `commits_tx_refs`) to the longest prefix whose committed /// transactions are all present in `fetched_transactions`, and drops fetched /// transactions not referenced by that prefix. Commits verified by /// `verify_commits` are chained by digest up to a vote-certified last commit, @@ -988,13 +992,13 @@ fn process_serialized_transactions( fn truncate_to_fully_fetched_prefix( peer: AuthorityIndex, commits: &mut Vec, - fetched_transactions: &mut BTreeMap, + commits_tx_refs: &mut Vec>, + fetched_transactions: &mut BTreeMap, ) -> ConsensusResult<()> { - let prefix_len = commits + let prefix_len = commits_tx_refs .iter() - .take_while(|commit| { - commit - .committed_transactions() + .take_while(|commit_tx_refs| { + commit_tx_refs .iter() .all(|tx_ref| fetched_transactions.contains_key(tx_ref)) }) @@ -1002,19 +1006,17 @@ fn truncate_to_fully_fetched_prefix( if prefix_len == 0 { return Err(ConsensusError::FetchedTransactionsMismatch { peer, - expected: commits + expected: commits_tx_refs .iter() - .map(|commit| commit.committed_transactions().len()) + .map(|commit_tx_refs| commit_tx_refs.len()) .sum(), received: fetched_transactions.len(), }); } if prefix_len < commits.len() { commits.truncate(prefix_len); - let prefix_tx_refs: BTreeSet = commits - .iter() - .flat_map(|commit| commit.committed_transactions()) - .collect(); + commits_tx_refs.truncate(prefix_len); + let prefix_tx_refs: BTreeSet<&TransactionRef> = commits_tx_refs.iter().flatten().collect(); fetched_transactions.retain(|tx_ref, _| prefix_tx_refs.contains(tx_ref)); } Ok(()) @@ -1533,17 +1535,13 @@ mod tests { context::Context, error::ConsensusError, network::SerializedTransactionsV2, - transaction_ref::{GenericTransactionRef, TransactionRef}, + transaction_ref::TransactionRef, }; - fn transaction_ref(round: Round) -> GenericTransactionRef { - GenericTransactionRef::TransactionRef(plain_ref(round)) - } - fn commit( context: &Arc, index: u32, - transactions: &[GenericTransactionRef], + transactions: &[TransactionRef], ) -> TrustedCommit { let leader = BlockRef::new( index, @@ -1557,11 +1555,11 @@ mod tests { 0, leader, vec![leader], - transactions.to_vec(), + transactions.iter().copied().map(Into::into).collect(), ) } - fn fetched(refs: &[GenericTransactionRef]) -> BTreeMap { + fn fetched(refs: &[TransactionRef]) -> BTreeMap { refs.iter().map(|r| (*r, Bytes::new())).collect() } @@ -1569,16 +1567,18 @@ mod tests { async fn keeps_all_commits_when_all_transactions_fetched() { let (context, _) = Context::new_for_test(4); let context = Arc::new(context); - let (tx_a, tx_b, tx_c) = (transaction_ref(1), transaction_ref(2), transaction_ref(3)); + let (tx_a, tx_b, tx_c) = (plain_ref(1), plain_ref(2), plain_ref(3)); let mut commits = vec![ commit(&context, 1, &[tx_a]), commit(&context, 2, &[tx_b, tx_c]), ]; + let mut commits_tx_refs = vec![vec![tx_a], vec![tx_b, tx_c]]; let mut transactions = fetched(&[tx_a, tx_b, tx_c]); truncate_to_fully_fetched_prefix( AuthorityIndex::new_for_test(1), &mut commits, + &mut commits_tx_refs, &mut transactions, ) .unwrap(); @@ -1591,12 +1591,7 @@ mod tests { async fn truncates_to_prefix_and_drops_unreferenced_transactions() { let (context, _) = Context::new_for_test(4); let context = Arc::new(context); - let (tx_a, tx_b, tx_c, tx_d) = ( - transaction_ref(1), - transaction_ref(2), - transaction_ref(3), - transaction_ref(4), - ); + let (tx_a, tx_b, tx_c, tx_d) = (plain_ref(1), plain_ref(2), plain_ref(3), plain_ref(4)); let first = commit(&context, 1, &[tx_a]); let mut commits = vec![ first.clone(), @@ -1604,16 +1599,19 @@ mod tests { commit(&context, 2, &[tx_b, tx_c]), commit(&context, 3, &[tx_d]), ]; + let mut commits_tx_refs = vec![vec![tx_a], vec![tx_b, tx_c], vec![tx_d]]; let mut transactions = fetched(&[tx_a, tx_c, tx_d]); truncate_to_fully_fetched_prefix( AuthorityIndex::new_for_test(1), &mut commits, + &mut commits_tx_refs, &mut transactions, ) .unwrap(); assert_eq!(commits, vec![first]); + assert_eq!(commits_tx_refs, vec![vec![tx_a]]); assert_eq!(transactions.into_keys().collect::>(), vec![tx_a]); } @@ -1621,12 +1619,18 @@ mod tests { async fn errors_when_first_commit_transactions_missing() { let (context, _) = Context::new_for_test(4); let context = Arc::new(context); - let (tx_a, tx_b) = (transaction_ref(1), transaction_ref(2)); + let (tx_a, tx_b) = (plain_ref(1), plain_ref(2)); let peer = AuthorityIndex::new_for_test(1); let mut commits = vec![commit(&context, 1, &[tx_a]), commit(&context, 2, &[tx_b])]; + let mut commits_tx_refs = vec![vec![tx_a], vec![tx_b]]; let mut transactions = fetched(&[tx_b]); - let result = truncate_to_fully_fetched_prefix(peer, &mut commits, &mut transactions); + let result = truncate_to_fully_fetched_prefix( + peer, + &mut commits, + &mut commits_tx_refs, + &mut transactions, + ); assert!(matches!( result, @@ -1642,14 +1646,16 @@ mod tests { async fn commit_without_transactions_counts_toward_prefix() { let (context, _) = Context::new_for_test(4); let context = Arc::new(context); - let tx_a = transaction_ref(1); + let tx_a = plain_ref(1); let empty = commit(&context, 1, &[]); let mut commits = vec![empty.clone(), commit(&context, 2, &[tx_a])]; + let mut commits_tx_refs = vec![vec![], vec![tx_a]]; let mut transactions = fetched(&[]); truncate_to_fully_fetched_prefix( AuthorityIndex::new_for_test(1), &mut commits, + &mut commits_tx_refs, &mut transactions, ) .unwrap(); @@ -1721,7 +1727,7 @@ mod tests { result, Err(ConsensusError::UnexpectedTransactionForCommit { peer: error_peer, - received: GenericTransactionRef::TransactionRef(received), + received, }) if error_peer == peer && received == tx_b )); } diff --git a/crates/starfish/core/src/commit_syncer/mod.rs b/crates/starfish/core/src/commit_syncer/mod.rs index dd34cb3696b8..e9097e1c80df 100644 --- a/crates/starfish/core/src/commit_syncer/mod.rs +++ b/crates/starfish/core/src/commit_syncer/mod.rs @@ -73,7 +73,7 @@ use crate::{ misbehavior_store::MisbehaviorStore, network::NetworkClient, stake_aggregator::{QuorumThreshold, StakeAggregator}, - transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI}, + transaction_ref::{GenericTransactionRefAPI, TransactionRef}, }; /// Allowed multiplicity of commit vote headers per authority in a @@ -428,13 +428,12 @@ pub(crate) fn verify_commits( pub(crate) fn verify_transactions_with_transactions_refs( context: &Arc, peer: AuthorityIndex, - serialized_transactions: BTreeMap, -) -> ConsensusResult> { + serialized_transactions: BTreeMap, +) -> ConsensusResult> { let mut verified_transactions_map = BTreeMap::new(); let mut encoder = create_encoder(context); let size_limit = serialized_transactions_size_limit(context); - for (committed_transactions_ref, inner_serialized_transactions) in serialized_transactions { - let transaction_ref = committed_transactions_ref.expect_transaction_ref()?; + for (transaction_ref, inner_serialized_transactions) in serialized_transactions { // Range-check the peer-supplied author and round before any consumer // indexes the committee by author. if !context.committee.is_valid_index(transaction_ref.author) { @@ -481,10 +480,7 @@ pub(crate) fn verify_transactions_with_transactions_refs( inner_serialized_transactions, ); - verified_transactions_map.insert( - GenericTransactionRef::TransactionRef(transaction_ref), - verified_transactions, - ); + verified_transactions_map.insert(transaction_ref, verified_transactions); } Ok(verified_transactions_map) @@ -961,7 +957,7 @@ pub(crate) mod tests { async fn fetch_transactions( &self, _peer: AuthorityIndex, - _block_refs: Vec, + _transaction_refs: Vec, _timeout: Duration, ) -> ConsensusResult> { unimplemented!("Unimplemented") @@ -1113,10 +1109,8 @@ pub(crate) mod tests { author: AuthorityIndex::new_for_test(0), transactions_commitment: TransactionsCommitment::MIN, }; - let serialized_transactions = BTreeMap::from([( - GenericTransactionRef::TransactionRef(transaction_ref), - Bytes::from(vec![0u8; size_limit + 1]), - )]); + let serialized_transactions = + BTreeMap::from([(transaction_ref, Bytes::from(vec![0u8; size_limit + 1]))]); let result = verify_transactions_with_transactions_refs(&context, peer, serialized_transactions); @@ -1148,10 +1142,8 @@ pub(crate) mod tests { author: out_of_range_author, transactions_commitment, }; - let serialized_transactions = BTreeMap::from([( - GenericTransactionRef::TransactionRef(transaction_ref), - inner_serialized_transactions, - )]); + let serialized_transactions = + BTreeMap::from([(transaction_ref, inner_serialized_transactions)]); let result = verify_transactions_with_transactions_refs(&context, peer, serialized_transactions); @@ -1185,10 +1177,8 @@ pub(crate) mod tests { author: AuthorityIndex::new_for_test(0), transactions_commitment, }; - let serialized_transactions = BTreeMap::from([( - GenericTransactionRef::TransactionRef(transaction_ref), - inner_serialized_transactions, - )]); + let serialized_transactions = + BTreeMap::from([(transaction_ref, inner_serialized_transactions)]); let result = verify_transactions_with_transactions_refs(&context, peer, serialized_transactions); diff --git a/crates/starfish/core/src/commit_syncer/regular.rs b/crates/starfish/core/src/commit_syncer/regular.rs index 240e1aaf923c..0788a5cda20f 100644 --- a/crates/starfish/core/src/commit_syncer/regular.rs +++ b/crates/starfish/core/src/commit_syncer/regular.rs @@ -41,7 +41,7 @@ use crate::{ header_synchronizer::HeaderSynchronizerHandle, misbehavior_store::MisbehaviorStore, network::{NetworkClient, SerializedTransactionsV2}, - transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI as _}, + transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI as _, TransactionRef}, }; pub(crate) struct RegularCommitSyncer { @@ -550,11 +550,20 @@ impl RegularCommitSyncer { .cloned() .collect(); - // 3a. Collect all committed transaction block refs from commits - let committed_tx_refs: Vec = commits + // 3a. Collect the committed transaction refs of each commit. Commits + // passing verify_commits are V2/V3, which only carry + // `TransactionRef`s, so the legacy `BlockRef` variant is an error. + let commits_tx_refs: Vec> = commits .iter() - .flat_map(|c| c.committed_transactions()) - .collect(); + .map(|c| { + c.committed_transactions() + .into_iter() + .map(GenericTransactionRef::expect_transaction_ref) + .collect() + }) + .collect::>()?; + let committed_tx_refs: Vec = + commits_tx_refs.iter().flatten().copied().collect(); let num_chunks = block_refs .len() @@ -612,7 +621,7 @@ impl RegularCommitSyncer { .max_transactions_per_commit_sync_fetch, ) .enumerate() - .map(|(i, request_block_refs)| { + .map(|(i, request_tx_refs)| { let inner = inner.clone(); async move { // 9. Send out pipelined fetch requests to avoid overloading the target @@ -622,11 +631,7 @@ impl RegularCommitSyncer { sleep(individual_delay * i as u32 + individual_delay / 2).await; let serialized_transactions = inner .network_client - .fetch_transactions( - target_authority, - request_block_refs.to_vec(), - timeout, - ) + .fetch_transactions(target_authority, request_tx_refs.to_vec(), timeout) .await?; // 10. Verify that the number of returned transactions is not greater than @@ -635,14 +640,15 @@ impl RegularCommitSyncer { // headers. We don't want to fail the whole fetch in this case. // TransactionSynchronizer will take care of fetching missing // transactions later. - if request_block_refs.len() < serialized_transactions.len() { + if request_tx_refs.len() < serialized_transactions.len() { return Err(ConsensusError::TooManyFetchedTransactionsReturned( target_authority, )); } - let requested_block_refs_set: BTreeSet<_> = - request_block_refs.iter().cloned().collect(); - // Deserialize to extract BlockRef and build a map directly + let requested_tx_refs_set: BTreeSet<_> = + request_tx_refs.iter().cloned().collect(); + // Deserialize to extract the TransactionRef and build a map + // directly let mut result = BTreeMap::new(); for serialized_bytes in serialized_transactions { let serialized_tx: SerializedTransactionsV2 = @@ -651,10 +657,8 @@ impl RegularCommitSyncer { // 11. Verify the returned transactions match the requested transaction // refs. - let committed_transaction_ref = GenericTransactionRef::TransactionRef( - serialized_tx.transaction_ref, - ); - if !requested_block_refs_set.contains(&committed_transaction_ref) { + let committed_transaction_ref = serialized_tx.transaction_ref; + if !requested_tx_refs_set.contains(&committed_transaction_ref) { return Err(ConsensusError::UnexpectedTransactionForCommit { peer: target_authority, received: committed_transaction_ref, @@ -667,7 +671,7 @@ impl RegularCommitSyncer { ); } - Ok::, ConsensusError>(result) + Ok::, ConsensusError>(result) } }) .collect() @@ -716,7 +720,7 @@ impl RegularCommitSyncer { // 14. Now create the Certified commits by assigning the block headers and // transactions to each commit and retaining the commit votes history. let mut certified_commits = Vec::new(); - for commit in &commits { + for (commit, commit_tx_refs) in commits.iter().zip(&commits_tx_refs) { let block_headers = commit .block_headers() .iter() @@ -730,8 +734,7 @@ impl RegularCommitSyncer { .collect::>(); // Collect transactions for this commit - let commit_transactions = commit - .committed_transactions() + let commit_transactions = commit_tx_refs .iter() .filter_map(|tx_ref| transactions_map.remove(tx_ref)) .collect::>(); diff --git a/crates/starfish/core/src/error.rs b/crates/starfish/core/src/error.rs index b78a95a3f094..e80fb5d5110f 100644 --- a/crates/starfish/core/src/error.rs +++ b/crates/starfish/core/src/error.rs @@ -11,7 +11,7 @@ use typed_store::TypedStoreError; use crate::{ block_header::{BlockRef, GENESIS_ROUND, Round}, commit::{Commit, CommitIndex}, - transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI as _, TransactionRef}, + transaction_ref::TransactionRef, }; /// Errors that can occur when processing blocks, reading from storage, or @@ -286,7 +286,7 @@ pub(crate) enum ConsensusError { #[error("Received unexpected transaction from peer {peer}: {received:?}")] UnexpectedTransactionForCommit { peer: AuthorityIndex, - received: GenericTransactionRef, + received: TransactionRef, }, #[error( @@ -455,19 +455,19 @@ impl ConsensusError { } pub fn quick_validation_requested_tx_refs( - gen_tx_refs: &[GenericTransactionRef], + tx_refs: &[TransactionRef], peer: AuthorityIndex, committee: &Committee, ) -> ConsensusResult<()> { - for gen_tx_ref in gen_tx_refs { - if !committee.is_valid_index(gen_tx_ref.author()) { + for tx_ref in tx_refs { + if !committee.is_valid_index(tx_ref.author) { return Err(ConsensusError::InvalidAuthorityIndexRequested { - index: gen_tx_ref.author(), + index: tx_ref.author, max: committee.size(), peer, }); } - if gen_tx_ref.round() == GENESIS_ROUND { + if tx_ref.round == GENESIS_ROUND { return Err(ConsensusError::UnexpectedGenesisRequested { peer }); } } diff --git a/crates/starfish/core/src/header_synchronizer.rs b/crates/starfish/core/src/header_synchronizer.rs index d82ab89c73c8..24e79d303dfb 100644 --- a/crates/starfish/core/src/header_synchronizer.rs +++ b/crates/starfish/core/src/header_synchronizer.rs @@ -1868,7 +1868,7 @@ mod tests { misbehavior_store::MisbehaviorStore, network::{BlockBundleStream, NetworkClient}, storage::mem_store::MemStore, - transaction_ref::GenericTransactionRef, + transaction_ref::{GenericTransactionRef, TransactionRef}, transactions_synchronizer::TransactionsSynchronizer, }; @@ -1940,7 +1940,7 @@ mod tests { async fn fetch_transactions( &self, _peer: AuthorityIndex, - _block_refs: Vec, + _transaction_refs: Vec, _timeout: Duration, ) -> ConsensusResult> { unimplemented!("Unimplemented") diff --git a/crates/starfish/core/src/leader_timeout.rs b/crates/starfish/core/src/leader_timeout.rs index 0abfbc364d9d..6c143d01f3d7 100644 --- a/crates/starfish/core/src/leader_timeout.rs +++ b/crates/starfish/core/src/leader_timeout.rs @@ -231,7 +231,7 @@ mod tests { leader_timeout::LeaderTimeoutTask, network::{BlockBundleStream, NetworkClient}, storage::mem_store::MemStore, - transaction_ref::GenericTransactionRef, + transaction_ref::TransactionRef, transactions_synchronizer::TransactionsSynchronizer, }; @@ -252,7 +252,7 @@ mod tests { async fn fetch_transactions( &self, _peer: AuthorityIndex, - _block_refs: Vec, + _transaction_refs: Vec, _timeout: Duration, ) -> ConsensusResult> { unimplemented!("Unimplemented") diff --git a/crates/starfish/core/src/network/mod.rs b/crates/starfish/core/src/network/mod.rs index cbc14f87a37d..1d3817316f6d 100644 --- a/crates/starfish/core/src/network/mod.rs +++ b/crates/starfish/core/src/network/mod.rs @@ -57,9 +57,7 @@ pub mod tonic_network; mod tonic_tls; use crate::{ - commit_syncer::CommitSyncType, - encoder::ShardEncoder, - transaction_ref::{GenericTransactionRef, TransactionRef}, + commit_syncer::CommitSyncType, encoder::ShardEncoder, transaction_ref::TransactionRef, }; /// Controls transaction fetching truncation behavior for different sync modes @@ -95,11 +93,11 @@ pub(crate) trait NetworkClient: Send + Sync + Sized + 'static { timeout: Duration, ) -> ConsensusResult; - /// Fetches transactions for the given block references from a peer. + /// Fetches transactions for the given transaction references from a peer. async fn fetch_transactions( &self, peer: AuthorityIndex, - transactions_refs: Vec, + transactions_refs: Vec, timeout: Duration, ) -> ConsensusResult>; @@ -217,7 +215,7 @@ pub(crate) trait NetworkService: Send + Sync + 'static { async fn handle_fetch_transactions( &self, peer: AuthorityIndex, - block_refs: Vec, + transactions_refs: Vec, fetch_mode: TransactionFetchMode, ) -> ConsensusResult>; } diff --git a/crates/starfish/core/src/network/test_network.rs b/crates/starfish/core/src/network/test_network.rs index 7420f66c9d2c..49642ec55aed 100644 --- a/crates/starfish/core/src/network/test_network.rs +++ b/crates/starfish/core/src/network/test_network.rs @@ -16,7 +16,7 @@ use crate::{ encoder::ShardEncoder, error::ConsensusResult, network::{BlockBundleStream, NetworkService, SerializedBlockBundle}, - transaction_ref::GenericTransactionRef, + transaction_ref::TransactionRef, }; pub(crate) struct TestService { @@ -119,7 +119,7 @@ impl NetworkService for Mutex { async fn handle_fetch_transactions( &self, _peer: AuthorityIndex, - _block_refs: Vec, + _transaction_refs: Vec, _fetch_mode: crate::network::TransactionFetchMode, ) -> ConsensusResult> { unimplemented!("Unimplemented") diff --git a/crates/starfish/core/src/network/tonic_network.rs b/crates/starfish/core/src/network/tonic_network.rs index 4648351f2441..76ff98470379 100644 --- a/crates/starfish/core/src/network/tonic_network.rs +++ b/crates/starfish/core/src/network/tonic_network.rs @@ -48,7 +48,7 @@ use crate::{ tonic_gen::consensus_service_server::ConsensusServiceServer, tonic_tls::certificate_server_name, }, - transaction_ref::{GenericTransactionRef, TransactionRef}, + transaction_ref::TransactionRef, }; // Maximum bytes size in a single fetch_blocks()response. @@ -344,28 +344,19 @@ impl NetworkClient for TonicClient { async fn fetch_transactions( &self, peer: AuthorityIndex, - transactions_refs: Vec, + transactions_refs: Vec, timeout: Duration, ) -> ConsensusResult> { let mut client = self.get_client(peer, timeout).await?; let mut request = Request::new(FetchTransactionsRequest { - block_refs: transactions_refs + transaction_refs: transactions_refs .iter() - .filter_map(|r| match r { - GenericTransactionRef::BlockRef(block_ref) => match bcs::to_bytes(block_ref) { - Ok(serialized) => Some(serialized), - Err(e) => { - debug!("Failed to serialize BlockRef {:?}: {e:?}", block_ref); - None - } - }, - GenericTransactionRef::TransactionRef(tx_ref) => match bcs::to_bytes(tx_ref) { - Ok(serialized) => Some(serialized), - Err(e) => { - debug!("Failed to serialize TransactionRef {:?}: {e:?}", tx_ref); - None - } - }, + .filter_map(|tx_ref| match bcs::to_bytes(tx_ref) { + Ok(serialized) => Some(serialized), + Err(e) => { + debug!("Failed to serialize TransactionRef {:?}: {e:?}", tx_ref); + None + } }) .collect(), }); @@ -1050,11 +1041,11 @@ impl ConsensusService for TonicServiceProxy { let permit = self.admit(RpcGroup::TransactionFetch, peer_index)?; let request = request.into_inner(); - let committed_transactions_refs: Vec = request - .block_refs + let committed_transactions_refs: Vec = request + .transaction_refs .iter() .filter_map(|r| match bcs::from_bytes::(r) { - Ok(transaction_ref) => Some(GenericTransactionRef::TransactionRef(transaction_ref)), + Ok(transaction_ref) => Some(transaction_ref), Err(e) => { debug!("Failed to deserialize transaction ref: {e:?}"); None @@ -1626,8 +1617,9 @@ pub(crate) struct GetLatestRoundsResponse { #[derive(Clone, prost::Message)] pub(crate) struct FetchTransactionsRequest { + // BCS-serialized `TransactionRef`s. #[prost(bytes = "vec", repeated, tag = "1")] - block_refs: Vec>, + transaction_refs: Vec>, } #[derive(Clone, prost::Message)] diff --git a/crates/starfish/core/src/subscriber.rs b/crates/starfish/core/src/subscriber.rs index b363ef6586cb..34585a9c5ae7 100644 --- a/crates/starfish/core/src/subscriber.rs +++ b/crates/starfish/core/src/subscriber.rs @@ -276,7 +276,7 @@ mod test { error::ConsensusResult, network::{BlockBundleStream, SerializedBlockBundle, test_network::TestService}, storage::mem_store::MemStore, - transaction_ref::GenericTransactionRef, + transaction_ref::TransactionRef, }; struct SubscriberTestClient {} @@ -309,7 +309,7 @@ mod test { async fn fetch_transactions( &self, _peer: AuthorityIndex, - _block_refs: Vec, + _transaction_refs: Vec, _timeout: Duration, ) -> ConsensusResult> { unimplemented!("Unimplemented") diff --git a/crates/starfish/core/src/transactions_synchronizer.rs b/crates/starfish/core/src/transactions_synchronizer.rs index 3dfcba9156ed..ab152dc889e2 100644 --- a/crates/starfish/core/src/transactions_synchronizer.rs +++ b/crates/starfish/core/src/transactions_synchronizer.rs @@ -36,7 +36,7 @@ use crate::{ error::{ConsensusError, ConsensusResult}, misbehavior_store::MisbehaviorStore, network::{NetworkClient, SerializedTransactionsV2}, - transaction_ref::{GenericTransactionRef, GenericTransactionRefAPI as _}, + transaction_ref::{GenericTransactionRef, TransactionRef}, }; /// The number of concurrent live transaction fetch requests @@ -179,7 +179,7 @@ impl Drop for ActiveRequestGuard { struct TransactionsGuard { map: Arc, - transactions_refs: BTreeSet, + transactions_refs: BTreeSet, peer: AuthorityIndex, } @@ -191,12 +191,13 @@ impl Drop for TransactionsGuard { } // Keeps a mapping between the missing transactions that have been instructed to -// be fetched and the authorities that are currently fetching them. For a block -// ref there is a maximum number of authorities that can concurrently fetch it. -// The authority ids that are currently fetching a transaction are set on the -// corresponding `BTreeSet` and basically they act as "locks". +// be fetched and the authorities that are currently fetching them. For a +// transaction ref there is a maximum number of authorities that can +// concurrently fetch it. The authority ids that are currently fetching a +// transaction are set on the corresponding `BTreeSet` and basically they act +// as "locks". struct InflightTransactionsMap { - inner: Mutex>>, + inner: Mutex>>, } impl InflightTransactionsMap { @@ -209,13 +210,13 @@ impl InflightTransactionsMap { /// Locks the transactions to be fetched for the assigned `peer`. We /// want to avoid re-fetching the missing transactions from too many /// authorities at the same time, thus we limit the concurrency per - /// transaction by attempting to lock per block_ref. In addition, we check - /// whether a given `peer` has many concurrent requests. If so, we will - /// not lock transactions. The method return optionally two guards. One for - /// the fetched transactions and one for active fetch request. + /// transaction by attempting to lock per transaction ref. In addition, we + /// check whether a given `peer` has many concurrent requests. If so, we + /// will not lock transactions. The method return optionally two guards. + /// One for the fetched transactions and one for active fetch request. fn lock_transactions_and_active_request( self: &Arc, - missing_block_refs: BTreeSet, + missing_transaction_refs: BTreeSet, peer: AuthorityIndex, max_number_transactions_per_fetch: usize, sync_method: SyncMethod, @@ -237,18 +238,18 @@ impl InflightTransactionsMap { // Now try to lock transactions let mut selected_transactions_to_fetch = BTreeSet::new(); - let mut selected_block_refs_num = 0; + let mut selected_transaction_refs_num = 0; - for block_ref in missing_block_refs { - let authorities = transaction_map.entry(block_ref).or_default(); + for tx_ref in missing_transaction_refs { + let authorities = transaction_map.entry(tx_ref).or_default(); if authorities.len() < MAX_AUTHORITIES_TO_FETCH_PER_TRANSACTION && authorities.insert(peer) { - selected_transactions_to_fetch.insert(block_ref); - selected_block_refs_num += 1; + selected_transactions_to_fetch.insert(tx_ref); + selected_transaction_refs_num += 1; - if selected_block_refs_num >= max_number_transactions_per_fetch { + if selected_transaction_refs_num >= max_number_transactions_per_fetch { break; } } @@ -281,13 +282,13 @@ impl InflightTransactionsMap { Some((transactions_guard, active_request_guard)) } - /// Unlocks the provided block references for the given `peer`. The + /// Unlocks the provided transaction references for the given `peer`. The /// unlocking is strict, meaning that if this method is called for a - /// specific block ref and peer more times than the corresponding lock - /// has been called, it will panic. + /// specific transaction ref and peer more times than the corresponding + /// lock has been called, it will panic. fn unlock_transactions( self: &Arc, - tx_refs: &BTreeSet, + tx_refs: &BTreeSet, peer: AuthorityIndex, ) { // Now mark all the transactions as fetched from the map @@ -312,7 +313,7 @@ impl InflightTransactionsMap { enum Command { FetchTransactions { - missing_transaction_refs: BTreeMap>, + missing_transaction_refs: BTreeMap>, result: oneshot::Sender>, }, KickOffScheduler, @@ -325,16 +326,22 @@ pub(crate) struct TransactionsSynchronizerHandle { impl TransactionsSynchronizerHandle { /// Explicitly asks from the transactions synchronizer to fetch the - /// transactions - provided the block_refs set - from the peer - /// authority. + /// transactions - provided the refs set - from the peer authority. pub(crate) async fn fetch_transactions( &self, - missing_block_refs: BTreeMap>, + missing_transaction_refs: BTreeMap>, ) -> ConsensusResult<()> { + // The `BlockRef` arm exists only for `CommitV1`, which no longer reaches + // the transaction-sync paths, so the synchronizer works with + // `TransactionRef` directly past this point. + let missing_transaction_refs = missing_transaction_refs + .into_iter() + .map(|(tx_ref, authorities)| Ok((tx_ref.expect_transaction_ref()?, authorities))) + .collect::>>()?; let (sender, receiver) = oneshot::channel(); self.commands_sender .send(Command::FetchTransactions { - missing_transaction_refs: missing_block_refs, + missing_transaction_refs, result: sender, }) .await @@ -381,7 +388,7 @@ impl TransactionsSynchronizerHandle { pub(crate) struct TransactionsSynchronizer { context: Arc, commands_receiver: Receiver, - live_fetch_requests: Sender>>, + live_fetch_requests: Sender>>, core_dispatcher: Arc, dag_state: Arc>, active_requests: Arc, @@ -543,7 +550,7 @@ impl TransactionsSynchronizer { network_client: Arc, context: Arc, core_dispatcher: Arc, - mut receiver: Receiver>>, + mut receiver: Receiver>>, inflight_transactions_map: Arc, last_failure_by_peer: Arc, block_verifier: Arc, @@ -560,7 +567,7 @@ impl TransactionsSynchronizer { .expect("We expect semaphore to be valid"); match receiver.recv().await { - Some(missing_transactions_block_refs) => { + Some(missing_transactions) => { let context = context.clone(); let active_requests = active_requests.clone(); let inflight_transactions_map = inflight_transactions_map.clone(); @@ -575,7 +582,7 @@ impl TransactionsSynchronizer { active_requests, inflight_transactions_map, network_client, - missing_transactions_block_refs, + missing_transactions, core_dispatcher, last_failure_by_peer, SyncMethod::Live, @@ -605,6 +612,13 @@ impl TransactionsSynchronizer { .get_missing_transaction_data() .await .map_err(|_err| ConsensusError::Shutdown)?; + // The `BlockRef` arm exists only for `CommitV1`, which no longer reaches + // the transaction-sync paths, so the synchronizer works with + // `TransactionRef` directly past this point. + let missing_transactions = missing_transactions + .into_iter() + .map(|(tx_ref, authorities)| Ok((tx_ref.expect_transaction_ref()?, authorities))) + .collect::>>()?; let dag_state = self.dag_state.clone(); @@ -615,7 +629,7 @@ impl TransactionsSynchronizer { let accepted_round = dag_state.read().highest_accepted_round(); let earliest_unavailable_transaction_round = missing_transactions .first_key_value() - .map(|(block_ref, _)| block_ref.round()) + .map(|(tx_ref, _)| tx_ref.round) .unwrap_or(accepted_round); let gap_to_unavailable_transactions = accepted_round.saturating_sub(earliest_unavailable_transaction_round); @@ -634,8 +648,8 @@ impl TransactionsSynchronizer { // Update metrics for missing transactions per authority before fetching let mut missing_transactions_per_authority = vec![0; context.committee.size()]; - for block_ref in missing_transactions.keys() { - missing_transactions_per_authority[block_ref.author()] += 1; + for tx_ref in missing_transactions.keys() { + missing_transactions_per_authority[tx_ref.author] += 1; } for (missing, (_, authority)) in missing_transactions_per_authority .into_iter() @@ -705,20 +719,21 @@ impl TransactionsSynchronizer { active_requests: Arc, inflight_transactions_map: Arc, network_client: Arc, - missing_transactions: BTreeMap>, + missing_transactions: BTreeMap>, core_dispatcher: Arc, last_failure_by_peer: Arc, sync_method: SyncMethod, block_verifier: Arc, misbehavior_store: Arc, ) { - // Build a mapping from authority -> set of BlockRefs it has acknowledged - let mut blocks_by_authority: BTreeMap> = + // Build a mapping from authority -> set of transaction refs it has + // acknowledged + let mut transaction_refs_by_authority: BTreeMap> = BTreeMap::new(); for (tx_ref, authorities) in &missing_transactions { for authority in authorities { if *authority != context.own_index { - blocks_by_authority + transaction_refs_by_authority .entry(*authority) .or_default() .insert(*tx_ref); @@ -746,53 +761,54 @@ impl TransactionsSynchronizer { // Randomness for ordering the authorities below. let mut rng = StdRng::from_rng(thread_rng()).expect("thread_rng should be available"); - // Create an iterator over authorities with their corresponding block refs. + // Create an iterator over authorities with their corresponding + // transaction refs. // When ranking is enabled, responsive acknowledgers are tried earlier // and a peer whose last fetch failed is ordered behind the healthy // candidates rather than removed from the set. When ranking is // disabled: a stable order under test, otherwise the previous // selection — a uniform shuffle that excludes the most recently failed // peers (up to less than f+1 by stake). - let iter_authorities: Box< - dyn Iterator)>, - > = if context.parameters.enable_peer_responsiveness_ranking { - let mut order: Vec = blocks_by_authority.keys().copied().collect(); - context.peer_responsiveness.prioritize( - DataSource::TransactionSynchronizer, - &mut order, - &mut rng, - ); - Box::new(order.into_iter().map(move |authority| { - let block_refs = blocks_by_authority - .remove(&authority) - .expect("prioritized order is a permutation of the candidate set"); - (authority, block_refs) - })) - } else if cfg!(test) { - // Stable order for tests. - Box::new(blocks_by_authority.into_iter()) - } else { - let excluded_authorities = last_failure_by_peer.get_excluded_authorities_by_stake(); - let mut vec: Vec<_> = blocks_by_authority - .into_iter() - .filter(|(authority, _)| !excluded_authorities.contains(authority)) - .collect(); - vec.shuffle(&mut rng); - Box::new(vec.into_iter()) - }; + let iter_authorities: Box)>> = + if context.parameters.enable_peer_responsiveness_ranking { + let mut order: Vec = + transaction_refs_by_authority.keys().copied().collect(); + context.peer_responsiveness.prioritize( + DataSource::TransactionSynchronizer, + &mut order, + &mut rng, + ); + Box::new(order.into_iter().map(move |authority| { + let transaction_refs = transaction_refs_by_authority + .remove(&authority) + .expect("prioritized order is a permutation of the candidate set"); + (authority, transaction_refs) + })) + } else if cfg!(test) { + // Stable order for tests. + Box::new(transaction_refs_by_authority.into_iter()) + } else { + let excluded_authorities = last_failure_by_peer.get_excluded_authorities_by_stake(); + let mut vec: Vec<_> = transaction_refs_by_authority + .into_iter() + .filter(|(authority, _)| !excluded_authorities.contains(authority)) + .collect(); + vec.shuffle(&mut rng); + Box::new(vec.into_iter()) + }; let mut request_futures = FuturesUnordered::new(); let mut assigned_authorities_for_transaction_fetch = 0; - for (authority, authority_block_refs) in iter_authorities { + for (authority, authority_transaction_refs) in iter_authorities { // * If transactions are successfully locked, and we didn't make too many to // this authority, then send a request to the network client to fetch the // transactions from the authority. If the fetch is successful, then process // the transactions and send them to the core for processing. if let Some((transactions_guard, active_request_guard)) = inflight_transactions_map .lock_transactions_and_active_request( - authority_block_refs.clone(), + authority_transaction_refs.clone(), authority, context .parameters @@ -962,7 +978,7 @@ impl TransactionsSynchronizer { .transactions_refs .iter() .cloned() - .collect::>(); + .collect::>(); let peer_hostname = &context.committee.authority(peer).hostname; let start_time = Instant::now(); @@ -1075,11 +1091,7 @@ impl TransactionsSynchronizer { // inside verify_transactions let transactions = match Handle::current() .spawn_blocking({ - for tx_ref in &requested_transactions_guard.transactions_refs { - tx_ref.expect_transaction_ref()?; - } - - let mut serialized_transactions_map: BTreeMap = + let mut serialized_transactions_map: BTreeMap = BTreeMap::new(); for serialized_transaction_bytes in &serialized_transactions_vec { let serialized_transactions: SerializedTransactionsV2 = @@ -1092,9 +1104,7 @@ impl TransactionsSynchronizer { ) }) .map_err(ConsensusError::MalformedTransactions)?; - let committed_transaction_ref = GenericTransactionRef::TransactionRef( - serialized_transactions.transaction_ref, - ); + let committed_transaction_ref = serialized_transactions.transaction_ref; // The commitment check below only proves each payload matches // its own claimed ref; it does not tie the ref to anything we // asked for. Reject a ref outside the requested set so a peer @@ -1245,8 +1255,8 @@ mod tests { use crate::{ Round, TestBlockHeader, Transaction, block_header::{ - BlockHeaderDigest, BlockRef, TransactionsCommitment, VerifiedBlock, - VerifiedBlockHeader, VerifiedOwnShard, VerifiedTransactions, + BlockRef, TransactionsCommitment, VerifiedBlock, VerifiedBlockHeader, VerifiedOwnShard, + VerifiedTransactions, }, block_verifier::{NoopBlockVerifier, SignedBlockVerifier, test::TxnSizeVerifier}, commit::{CertifiedCommits, CommitRange}, @@ -1603,7 +1613,7 @@ mod tests { verified_transactions.push(verified_transaction); } - // Create a map of block refs to authorities that have them + // Create a map of transaction refs to authorities that have them let mut missing_transactions = Vec::new(); for (index, header) in block_headers.iter().enumerate() { let mut authorities = BTreeSet::new(); @@ -1611,7 +1621,10 @@ mod tests { authorities.insert(from_whom); network_client.set_timeout_peer(from_whom).await; let mut missing_txs = BTreeMap::new(); - missing_txs.insert(GenericTransactionRef::from(header.reference()), authorities); + missing_txs.insert( + GenericTransactionRef::from(header.transaction_ref()), + authorities, + ); missing_transactions.push(missing_txs) } @@ -2436,14 +2449,16 @@ mod tests { block_headers.push(header); } - // Create a map of block refs to authorities that have them + // Create a map of transaction refs to authorities that have them let mut missing_transactions = BTreeMap::new(); for header in &block_headers { let mut authorities = BTreeSet::new(); authorities.insert(AuthorityIndex::new_for_test(1)); // This peer will timeout authorities.insert(AuthorityIndex::new_for_test(2)); // This peer will return an error - missing_transactions - .insert(GenericTransactionRef::from(header.reference()), authorities); + missing_transactions.insert( + GenericTransactionRef::from(header.transaction_ref()), + authorities, + ); } // Set peer 1 to timeout @@ -2599,18 +2614,20 @@ mod tests { let active_requests = InflightActiveRequests::new(); let sync_method = SyncMethod::Periodic; - let some_block_refs = [ - BlockRef::new(1, AuthorityIndex::new_for_test(0), BlockHeaderDigest::MIN), - BlockRef::new(10, AuthorityIndex::new_for_test(0), BlockHeaderDigest::MIN), - BlockRef::new(12, AuthorityIndex::new_for_test(3), BlockHeaderDigest::MIN), - BlockRef::new(15, AuthorityIndex::new_for_test(2), BlockHeaderDigest::MIN), - ]; let context = Context::new_for_test(10).0; - let missing_block_refs = some_block_refs.iter().cloned().collect::>(); - let missing_transactions_refs = missing_block_refs - .iter() - .map(|&br| GenericTransactionRef::from(br)) - .collect::>(); + let missing_transactions_refs = [ + (1, AuthorityIndex::new_for_test(0)), + (10, AuthorityIndex::new_for_test(0)), + (12, AuthorityIndex::new_for_test(3)), + (15, AuthorityIndex::new_for_test(2)), + ] + .into_iter() + .map(|(round, author)| TransactionRef { + round, + author, + transactions_commitment: TransactionsCommitment::MIN, + }) + .collect::>(); // We keep both guards so that drops happen at the end let mut all_guards: Vec<(TransactionsGuard, ActiveRequestGuard)> = Vec::new(); @@ -2747,7 +2764,7 @@ mod tests { } struct MockNetworkClient { - transactions: Arc>>, + transactions: Arc>>, error_peers: Arc>>, timeout_peers: Arc>>, empty_peers: Arc>>, @@ -2783,9 +2800,8 @@ mod tests { transaction_ref, serialized_transactions: transaction.serialized().clone(), }; - let tx_ref = GenericTransactionRef::TransactionRef(transaction_ref); let serialized = bcs::to_bytes(&serialized_transactions).unwrap(); - transactions_map.insert((peer, tx_ref), serialized.into()); + transactions_map.insert((peer, transaction_ref), serialized.into()); } } @@ -3019,7 +3035,7 @@ mod tests { async fn fetch_transactions( &self, peer: AuthorityIndex, - block_refs: Vec, + transaction_refs: Vec, _timeout: Duration, ) -> ConsensusResult> { // Check if this peer is set to timeout @@ -3048,7 +3064,7 @@ mod tests { if corrupted_peers.contains(&peer) { // Return corrupted data (invalid bytes that can't be deserialized) let mut result = Vec::new(); - for _ in 0..block_refs.len() { + for _ in 0..transaction_refs.len() { result.push(Bytes::from(vec![0, 1, 2, 3])); // Invalid serialized data } return Ok(result); @@ -3063,8 +3079,8 @@ mod tests { // Normal case - return transactions from the map let transactions_map = self.transactions.lock().await; let mut result = Vec::new(); - for block_ref in block_refs { - if let Some(serialized) = transactions_map.get(&(peer, block_ref)) { + for transaction_ref in transaction_refs { + if let Some(serialized) = transactions_map.get(&(peer, transaction_ref)) { result.push(serialized.clone()); } }