separate state sync notifications to pre-commit and commit (#14889)

consensus/src/counters.rs

@@ -955,6 +955,14 @@ pub static PENDING_STATE_SYNC_NOTIFICATION: Lazy<IntGauge> = Lazy::new(|| {
     .unwrap()
 });
 
+pub static PENDING_COMMIT_NOTIFICATION: Lazy<IntGauge> = Lazy::new(|| {
+    register_int_gauge!(
+        "aptos_consensus_pending_commit_notification",
+        "Count of the pending commit notification"
+    )
+    .unwrap()
+});
+
 /// Count of the pending quorum store commit notification.
 pub static PENDING_QUORUM_STORE_COMMIT_NOTIFICATION: Lazy<IntGauge> = Lazy::new(|| {
     register_int_gauge!(

consensus/src/execution_pipeline.rs

@@ -14,7 +14,7 @@ use aptos_consensus_types::{block::Block, pipeline_execution_result::PipelineExe
 use aptos_crypto::HashValue;
 use aptos_executor_types::{
     state_checkpoint_output::StateCheckpointOutput, BlockExecutorTrait, ExecutorError,
-    ExecutorResult,
+    ExecutorResult, StateComputeResult,
 };
 use aptos_experimental_runtimes::thread_manager::optimal_min_len;
 use aptos_logger::{debug, warn};
@@ -35,6 +35,12 @@ use std::{
 };
 use tokio::sync::{mpsc, oneshot};
 
+pub type PreCommitHook = Box<
+    dyn 'static
+        + FnOnce(&[SignedTransaction], &StateComputeResult) -> BoxFuture<'static, ()>
+        + Send,
+>;
+
 #[allow(clippy::unwrap_used)]
 pub static SIG_VERIFY_POOL: Lazy<Arc<rayon::ThreadPool>> = Lazy::new(|| {
     Arc::new(
@@ -88,6 +94,7 @@ impl ExecutionPipeline {
         parent_block_id: HashValue,
         txn_generator: BlockPreparer,
         block_executor_onchain_config: BlockExecutorConfigFromOnchain,
+        pre_commit_hook: PreCommitHook,
         lifetime_guard: CountedRequest<()>,
     ) -> StateComputeResultFut {
         let (result_tx, result_rx) = oneshot::channel();
@@ -101,6 +108,7 @@ impl ExecutionPipeline {
                 block_preparer: txn_generator,
                 result_tx,
                 command_creation_time: Instant::now(),
+                pre_commit_hook,
                 lifetime_guard,
             })
             .expect("Failed to send block to execution pipeline.");
@@ -127,6 +135,7 @@ impl ExecutionPipeline {
             block_executor_onchain_config,
             parent_block_id,
             block_preparer,
+            pre_commit_hook,
             result_tx,
             command_creation_time,
             lifetime_guard,
@@ -163,6 +172,7 @@ impl ExecutionPipeline {
                     block: (block.id(), sig_verified_txns).into(),
                     parent_block_id,
                     block_executor_onchain_config,
+                    pre_commit_hook,
                     result_tx,
                     command_creation_time: Instant::now(),
                     lifetime_guard,
@@ -196,6 +206,7 @@ impl ExecutionPipeline {
             block,
             parent_block_id,
             block_executor_onchain_config,
+            pre_commit_hook,
             result_tx,
             command_creation_time,
             lifetime_guard,
@@ -232,6 +243,7 @@ impl ExecutionPipeline {
                     block_id,
                     parent_block_id,
                     state_checkpoint_output,
+                    pre_commit_hook,
                     result_tx,
                     command_creation_time: Instant::now(),
                     lifetime_guard,
@@ -252,6 +264,7 @@ impl ExecutionPipeline {
             block_id,
             parent_block_id,
             state_checkpoint_output: execution_result,
+            pre_commit_hook,
             result_tx,
             command_creation_time,
             lifetime_guard,
@@ -274,6 +287,7 @@ impl ExecutionPipeline {
             }
             .await;
             let pipeline_res = res.map(|(output, execution_duration)| {
+                let pre_commit_hook_fut = pre_commit_hook(&input_txns, &output);
                 let pre_commit_fut: BoxFuture<'static, ExecutorResult<()>> =
                     if output.epoch_state().is_some() || !enable_pre_commit {
                         // hack: it causes issue if pre-commit is finished at an epoch ending, and
@@ -285,7 +299,9 @@ impl ExecutionPipeline {
                                 executor.pre_commit_block(block_id, parent_block_id)
                             })
                             .await
-                            .expect("failed to spawn_blocking")
+                            .expect("failed to spawn_blocking")?;
+                            pre_commit_hook_fut.await;
+                            Ok(())
                         })
                     } else {
                         // kick off pre-commit right away
@@ -295,6 +311,7 @@ impl ExecutionPipeline {
                             .send(PreCommitCommand {
                                 block_id,
                                 parent_block_id,
+                                pre_commit_hook_fut,
                                 result_tx: pre_commit_result_tx,
                                 lifetime_guard,
                             })
@@ -322,6 +339,7 @@ impl ExecutionPipeline {
         while let Some(PreCommitCommand {
             block_id,
             parent_block_id,
+            pre_commit_hook_fut,
             result_tx,
             lifetime_guard,
         }) = block_rx.recv().await
@@ -336,7 +354,9 @@ impl ExecutionPipeline {
                     })
                 )
                 .await
-                .expect("Failed to spawn_blocking().")
+                .expect("Failed to spawn_blocking().")?;
+                pre_commit_hook_fut.await;
+                Ok(())
             }
             .await;
             result_tx
@@ -355,6 +375,7 @@ struct PrepareBlockCommand {
     // The parent block id.
     parent_block_id: HashValue,
     block_preparer: BlockPreparer,
+    pre_commit_hook: PreCommitHook,
     result_tx: oneshot::Sender<ExecutorResult<PipelineExecutionResult>>,
     command_creation_time: Instant,
     lifetime_guard: CountedRequest<()>,
@@ -365,6 +386,7 @@ struct ExecuteBlockCommand {
     block: ExecutableBlock,
     parent_block_id: HashValue,
     block_executor_onchain_config: BlockExecutorConfigFromOnchain,
+    pre_commit_hook: PreCommitHook,
     result_tx: oneshot::Sender<ExecutorResult<PipelineExecutionResult>>,
     command_creation_time: Instant,
     lifetime_guard: CountedRequest<()>,
@@ -375,6 +397,7 @@ struct LedgerApplyCommand {
     block_id: HashValue,
     parent_block_id: HashValue,
     state_checkpoint_output: ExecutorResult<(StateCheckpointOutput, Duration)>,
+    pre_commit_hook: PreCommitHook,
     result_tx: oneshot::Sender<ExecutorResult<PipelineExecutionResult>>,
     command_creation_time: Instant,
     lifetime_guard: CountedRequest<()>,
@@ -383,6 +406,7 @@ struct LedgerApplyCommand {
 struct PreCommitCommand {
     block_id: HashValue,
     parent_block_id: HashValue,
+    pre_commit_hook_fut: BoxFuture<'static, ()>,
     result_tx: oneshot::Sender<ExecutorResult<()>>,
     lifetime_guard: CountedRequest<()>,
 }