apollo_protobuf: move executed_transaction_count into ProposalFin (#11492)

Author: asmaastarkware

crates/apollo_consensus_orchestrator/src/build_proposal.rs

@@ -288,18 +288,10 @@ async fn get_proposal_content(
                     }
                 }
 
-                let final_n_executed_txs_u64 = final_n_executed_txs
+                let executed_transaction_count: u64 = final_n_executed_txs
                     .try_into()
                     .expect("Number of executed transactions should fit in u64");
-                args.stream_sender
-                    .send(ProposalPart::ExecutedTransactionCount(final_n_executed_txs_u64))
-                    .await
-                    .map_err(|e| {
-                        BuildProposalError::SendError(format!(
-                            "Failed to send executed transaction count: {e:?}"
-                        ))
-                    })?;
-                let fin = ProposalFin { proposal_commitment };
+                let fin = ProposalFin { proposal_commitment, executed_transaction_count };
                 info!("Sending fin={fin:?}");
                 args.stream_sender.send(ProposalPart::Fin(fin)).await.map_err(|e| {
                     BuildProposalError::SendError(format!("Failed to send proposal fin: {e:?}"))

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context.rs

@@ -972,12 +972,10 @@ async fn send_reproposal(
         stream_sender.send(ProposalPart::Transactions(TransactionBatch { transactions })).await?;
         n_executed_txs += batch.len();
     }
-    stream_sender
-        .send(ProposalPart::ExecutedTransactionCount(
-            n_executed_txs.try_into().expect("Number of executed transactions should fit in u64"),
-        ))
-        .await?;
-    stream_sender.send(ProposalPart::Fin(ProposalFin { proposal_commitment: id })).await?;
+    let executed_transaction_count: u64 =
+        n_executed_txs.try_into().expect("Number of executed transactions should fit in u64");
+    let fin = ProposalFin { proposal_commitment: id, executed_transaction_count };
+    stream_sender.send(ProposalPart::Fin(fin)).await?;
 
     Ok(())
 }

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context_test.rs

@@ -139,12 +139,9 @@ async fn validate_then_repropose(#[case] execute_all_txs: bool) {
     let transactions =
         ProposalPart::Transactions(TransactionBatch { transactions: TX_BATCH.to_vec() });
     content_sender.send(transactions.clone()).await.unwrap();
-    content_sender
-        .send(ProposalPart::ExecutedTransactionCount(n_executed_txs_count.try_into().unwrap()))
-        .await
-        .unwrap();
     let fin = ProposalPart::Fin(ProposalFin {
         proposal_commitment: ProposalCommitment(STATE_DIFF_COMMITMENT.0.0),
+        executed_transaction_count: n_executed_txs_count.try_into().unwrap(),
     });
     content_sender.send(fin.clone()).await.unwrap();
     let fin_receiver =
@@ -161,10 +158,6 @@ async fn validate_then_repropose(#[case] execute_all_txs: bool) {
         receiver.next().await.unwrap(),
         ProposalPart::Transactions(TransactionBatch { transactions: executed_transactions })
     );
-    assert_eq!(
-        receiver.next().await.unwrap(),
-        ProposalPart::ExecutedTransactionCount(n_executed_txs_count.try_into().unwrap())
-    );
     assert_eq!(receiver.next().await.unwrap(), fin);
     assert!(receiver.next().await.is_none());
 }
@@ -181,18 +174,16 @@ async fn proposals_from_different_rounds() {
     // Proposal parts sent in the proposals.
     let prop_part_txs =
         ProposalPart::Transactions(TransactionBatch { transactions: TX_BATCH.to_vec() });
-    let prop_part_executed_count =
-        ProposalPart::ExecutedTransactionCount(INTERNAL_TX_BATCH.len().try_into().unwrap());
     let prop_part_fin = ProposalPart::Fin(ProposalFin {
         proposal_commitment: ProposalCommitment(STATE_DIFF_COMMITMENT.0.0),
+        executed_transaction_count: INTERNAL_TX_BATCH.len().try_into().unwrap(),
     });
 
     // The proposal from the past round is ignored.
     let (mut content_sender, content_receiver) =
         mpsc::channel(context.config.static_config.proposal_buffer_size);
     content_sender.send(ProposalPart::BlockInfo(block_info(BlockNumber(0)))).await.unwrap();
     content_sender.send(prop_part_txs.clone()).await.unwrap();
-    content_sender.send(prop_part_executed_count.clone()).await.unwrap();
 
     let mut init = ProposalInit { round: 0, ..Default::default() };
     let fin_receiver_past_round = context.validate_proposal(init, TIMEOUT, content_receiver).await;
@@ -204,7 +195,6 @@ async fn proposals_from_different_rounds() {
         mpsc::channel(context.config.static_config.proposal_buffer_size);
     content_sender.send(ProposalPart::BlockInfo(block_info(BlockNumber(0)))).await.unwrap();
     content_sender.send(prop_part_txs.clone()).await.unwrap();
-    content_sender.send(prop_part_executed_count.clone()).await.unwrap();
     content_sender.send(prop_part_fin.clone()).await.unwrap();
     init.round = 1;
     let fin_receiver_curr_round = context.validate_proposal(init, TIMEOUT, content_receiver).await;
@@ -215,7 +205,6 @@ async fn proposals_from_different_rounds() {
         mpsc::channel(context.config.static_config.proposal_buffer_size);
     content_sender.send(ProposalPart::BlockInfo(block_info(BlockNumber(0)))).await.unwrap();
     content_sender.send(prop_part_txs.clone()).await.unwrap();
-    content_sender.send(prop_part_executed_count.clone()).await.unwrap();
     content_sender.send(prop_part_fin.clone()).await.unwrap();
     let fin_receiver_future_round = context
         .validate_proposal(
@@ -284,14 +273,11 @@ async fn build_proposal() {
         receiver.next().await.unwrap(),
         ProposalPart::Transactions(TransactionBatch { transactions: TX_BATCH.to_vec() })
     );
-    assert_eq!(
-        receiver.next().await.unwrap(),
-        ProposalPart::ExecutedTransactionCount(INTERNAL_TX_BATCH.len().try_into().unwrap())
-    );
     assert_eq!(
         receiver.next().await.unwrap(),
         ProposalPart::Fin(ProposalFin {
             proposal_commitment: ProposalCommitment(STATE_DIFF_COMMITMENT.0.0),
+            executed_transaction_count: INTERNAL_TX_BATCH.len().try_into().unwrap(),
         })
     );
     assert!(receiver.next().await.is_none());
@@ -494,8 +480,6 @@ async fn propose_then_repropose(#[case] execute_all_txs: bool) {
     let _init = receiver.next().await.unwrap();
     let block_info = receiver.next().await.unwrap();
     let _txs = receiver.next().await.unwrap();
-    let final_n_executed_txs = receiver.next().await.unwrap();
-    assert!(matches!(final_n_executed_txs, ProposalPart::ExecutedTransactionCount(_)));
     let fin = receiver.next().await.unwrap();
     assert_eq!(fin_receiver.await.unwrap().0, STATE_DIFF_COMMITMENT.0.0);
 
@@ -514,7 +498,6 @@ async fn propose_then_repropose(#[case] execute_all_txs: bool) {
     let reproposed_txs = ProposalPart::Transactions(TransactionBatch { transactions });
     assert_eq!(receiver.next().await.unwrap(), reproposed_txs);
 
-    assert_eq!(receiver.next().await.unwrap(), final_n_executed_txs);
     assert_eq!(receiver.next().await.unwrap(), fin);
     assert!(receiver.next().await.is_none());
 }
@@ -711,14 +694,11 @@ async fn oracle_fails_on_startup(#[case] l1_oracle_failure: bool) {
         receiver.next().await.unwrap(),
         ProposalPart::Transactions(TransactionBatch { transactions: TX_BATCH.to_vec() })
     );
-    assert_eq!(
-        receiver.next().await.unwrap(),
-        ProposalPart::ExecutedTransactionCount(INTERNAL_TX_BATCH.len().try_into().unwrap())
-    );
     assert_eq!(
         receiver.next().await.unwrap(),
         ProposalPart::Fin(ProposalFin {
             proposal_commitment: ProposalCommitment(STATE_DIFF_COMMITMENT.0.0),
+            executed_transaction_count: INTERNAL_TX_BATCH.len().try_into().unwrap(),
         })
     );
     assert!(receiver.next().await.is_none());
@@ -837,14 +817,11 @@ async fn oracle_fails_on_second_block(#[case] l1_oracle_failure: bool) {
         receiver.next().await.unwrap(),
         ProposalPart::Transactions(TransactionBatch { transactions: TX_BATCH.to_vec() })
     );
-    assert_eq!(
-        receiver.next().await.unwrap(),
-        ProposalPart::ExecutedTransactionCount(INTERNAL_TX_BATCH.len().try_into().unwrap())
-    );
     assert_eq!(
         receiver.next().await.unwrap(),
         ProposalPart::Fin(ProposalFin {
             proposal_commitment: ProposalCommitment(STATE_DIFF_COMMITMENT.0.0),
+            executed_transaction_count: INTERNAL_TX_BATCH.len().try_into().unwrap(),
         })
     );
     assert!(receiver.next().await.is_none());

crates/apollo_consensus_orchestrator/src/test_utils.rs

@@ -382,13 +382,10 @@ pub(crate) async fn send_proposal_to_validator_context(
         .send(ProposalPart::Transactions(TransactionBatch { transactions: TX_BATCH.to_vec() }))
         .await
         .unwrap();
-    content_sender
-        .send(ProposalPart::ExecutedTransactionCount(INTERNAL_TX_BATCH.len().try_into().unwrap()))
-        .await
-        .unwrap();
     content_sender
         .send(ProposalPart::Fin(ProposalFin {
             proposal_commitment: ProtoProposalCommitment(STATE_DIFF_COMMITMENT.0.0),
+            executed_transaction_count: INTERNAL_TX_BATCH.len().try_into().unwrap(),
         }))
         .await
         .unwrap();

crates/apollo_consensus_orchestrator/src/validate_proposal.rs

@@ -133,7 +133,6 @@ pub(crate) async fn validate_proposal(
     mut args: ProposalValidateArguments,
 ) -> ValidateProposalResult<ProposalCommitment> {
     let mut content = Vec::new();
-    let mut final_n_executed_txs: Option<usize> = None;
     let now = args.deps.clock.now();
 
     let Some(deadline) = now.checked_add_signed(chrono::TimeDelta::from_std(args.timeout).unwrap())
@@ -150,7 +149,10 @@ pub(crate) async fn validate_proposal(
     .await?
     {
         SecondProposalPart::BlockInfo(block_info) => block_info,
-        SecondProposalPart::Fin(ProposalFin { proposal_commitment }) => {
+        SecondProposalPart::Fin(ProposalFin {
+            proposal_commitment,
+            executed_transaction_count: _,
+        }) => {
             return Ok(proposal_commitment);
         }
     };
@@ -196,7 +198,6 @@ pub(crate) async fn validate_proposal(
                     args.deps.batcher.as_ref(),
                     proposal_part.clone(),
                     &mut content,
-                    &mut final_n_executed_txs,
                     args.deps.transaction_converter.clone(),
                 ).await {
                     HandledProposalPart::Finished(built_block, received_fin) => {
@@ -373,9 +374,9 @@ async fn await_second_proposal_part(
                 Some(ProposalPart::BlockInfo(block_info)) => {
                     Ok(SecondProposalPart::BlockInfo(block_info))
                 }
-                Some(ProposalPart::Fin(ProposalFin { proposal_commitment })) => {
+                Some(ProposalPart::Fin(ProposalFin { proposal_commitment, executed_transaction_count })) => {
                     warn!("Received an empty proposal.");
-                    Ok(SecondProposalPart::Fin(ProposalFin { proposal_commitment }))
+                    Ok(SecondProposalPart::Fin(ProposalFin { proposal_commitment, executed_transaction_count }))
                 }
                 x => {
                     Err(ValidateProposalError::InvalidSecondProposalPart(x
@@ -428,7 +429,6 @@ async fn handle_proposal_part(
     batcher: &dyn BatcherClient,
     proposal_part: Option<ProposalPart>,
     content: &mut Vec<Vec<InternalConsensusTransaction>>,
-    final_n_executed_txs: &mut Option<usize>,
     transaction_converter: Arc<dyn TransactionConverterTrait>,
 ) -> HandledProposalPart {
     match proposal_part {
@@ -444,15 +444,18 @@ async fn handle_proposal_part(
         }
         Some(ProposalPart::Fin(fin)) => {
             info!("Received fin={fin:?}");
-            let Some(final_n_executed_txs_nonopt) = *final_n_executed_txs else {
+            let Ok(executed_txs_count) = fin.executed_transaction_count.try_into() else {
                 return HandledProposalPart::Failed(
-                    "Received Fin without executed transaction count".to_string(),
+                    "Number of executed transactions should fit in usize".to_string(),
                 );
             };
+
+            *content = truncate_to_executed_txs(content, executed_txs_count);
+
             // Output this along with the ID from batcher, to compare them.
             let input = SendProposalContentInput {
                 proposal_id,
-                content: SendProposalContent::Finish(final_n_executed_txs_nonopt),
+                content: SendProposalContent::Finish(executed_txs_count),
             };
             let response = match batcher.send_proposal_content(input).await {
                 Ok(response) => response,
@@ -474,21 +477,16 @@ async fn handle_proposal_part(
             info!(
                 network_block_id = ?fin.proposal_commitment,
                 ?batcher_block_id,
-                final_n_executed_txs_nonopt,
+                executed_txs_count,
                 "Finished validating proposal."
             );
-            if final_n_executed_txs_nonopt == 0 {
+            if executed_txs_count == 0 {
                 warn!("Validated an empty proposal.");
             }
             HandledProposalPart::Finished(batcher_block_id, fin)
         }
         Some(ProposalPart::Transactions(TransactionBatch { transactions: txs })) => {
             debug!("Received transaction batch with {} txs", txs.len());
-            if final_n_executed_txs.is_some() {
-                return HandledProposalPart::Failed(
-                    "Received transactions after executed transaction count".to_string(),
-                );
-            }
             let txs =
                 futures::future::join_all(txs.into_iter().map(|tx| {
                     transaction_converter.convert_consensus_tx_to_internal_consensus_tx(tx)
@@ -529,24 +527,6 @@ async fn handle_proposal_part(
                 }
             }
         }
-        Some(ProposalPart::ExecutedTransactionCount(executed_txs_count)) => {
-            debug!("Received executed transaction count: {executed_txs_count}");
-            if final_n_executed_txs.is_some() {
-                return HandledProposalPart::Failed(
-                    "Received executed transaction count more than once".to_string(),
-                );
-            }
-            let executed_txs_count_usize_res: Result<usize, _> = executed_txs_count.try_into();
-            let Ok(executed_txs_count_usize) = executed_txs_count_usize_res else {
-                return HandledProposalPart::Failed(
-                    "Number of executed transactions should fit in usize".to_string(),
-                );
-            };
-            *final_n_executed_txs = Some(executed_txs_count_usize);
-            *content = truncate_to_executed_txs(content, executed_txs_count_usize);
-
-            HandledProposalPart::Continue
-        }
         _ => HandledProposalPart::Failed("Invalid proposal part".to_string()),
     }
 }