extract ExecutionOutput, DoGetExecutionOutput

1. put ExecutionOutput under executor-types; next is to make
   StateComputeResult a combination of ExecutionOutput + StateCheckpointOutput
   + LedgerUpdateOutput
2. leave the algorithm in the executor crate, creating
   DoGetExecutionOutput

execution/executor-benchmark/src/native_executor.rs

@@ -6,9 +6,8 @@ use crate::{
     metrics::TIMER,
 };
 use anyhow::Result;
-use aptos_executor::{
-    block_executor::TransactionBlockExecutor, components::chunk_output::ChunkOutput,
-};
+use aptos_executor::block_executor::TransactionBlockExecutor;
+use aptos_executor_types::execution_output::ExecutionOutput;
 use aptos_storage_interface::cached_state_view::CachedStateView;
 use aptos_types::{
     account_address::AccountAddress,
@@ -351,7 +350,7 @@ impl TransactionBlockExecutor for NativeExecutor {
         transactions: ExecutableTransactions,
         state_view: CachedStateView,
         _onchain_config: BlockExecutorConfigFromOnchain,
-    ) -> Result<ChunkOutput> {
+    ) -> Result<ExecutionOutput> {
         let transactions = match transactions {
             ExecutableTransactions::Unsharded(txns) => txns,
             _ => todo!("sharded execution not yet supported"),
@@ -422,7 +421,7 @@ impl TransactionBlockExecutor for NativeExecutor {
                 })
                 .collect::<Result<Vec<_>>>()
         })?;
-        Ok(ChunkOutput {
+        Ok(ExecutionOutput {
             transactions: transactions.into_iter().map(|t| t.into_inner()).collect(),
             transaction_outputs,
             state_cache: state_view.into_state_cache(),

execution/executor-types/src/execution_output.rs

@@ -0,0 +1,21 @@
+// Copyright © Aptos Foundation
+// Parts of the project are originally copyright © Meta Platforms, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+#![forbid(unsafe_code)]
+
+use aptos_storage_interface::cached_state_view::StateCache;
+use aptos_types::transaction::{block_epilogue::BlockEndInfo, Transaction, TransactionOutput};
+
+pub struct ExecutionOutput {
+    /// Input transactions.
+    pub transactions: Vec<Transaction>,
+    /// Raw VM output.
+    pub transaction_outputs: Vec<TransactionOutput>,
+    /// Carries the frozen base state view, so all in-mem nodes involved won't drop before the
+    /// execution result is processed; as well as all the accounts touched during execution, together
+    /// with their proofs.
+    pub state_cache: StateCache,
+    /// Optional StateCheckpoint payload
+    pub block_end_info: Option<BlockEndInfo>,
+}

execution/executor-types/src/lib.rs

@@ -36,6 +36,7 @@ use std::{
 };
 
 mod error;
+pub mod execution_output;
 mod ledger_update_output;
 pub mod parsed_transaction_output;
 pub mod state_checkpoint_output;

execution/executor/src/block_executor.rs

@@ -6,7 +6,8 @@
 
 use crate::{
     components::{
-        block_tree::BlockTree, chunk_output::ChunkOutput, do_ledger_update::DoLedgerUpdate,
+        apply_chunk_output::ApplyChunkOutput, block_tree::BlockTree,
+        do_get_execution_output::DoGetExecutionOutput, do_ledger_update::DoLedgerUpdate,
         partial_state_compute_result::PartialStateComputeResult,
     },
     logging::{LogEntry, LogSchema},
@@ -18,8 +19,8 @@ use crate::{
 use anyhow::Result;
 use aptos_crypto::HashValue;
 use aptos_executor_types::{
-    state_checkpoint_output::StateCheckpointOutput, state_compute_result::StateComputeResult,
-    BlockExecutorTrait, ExecutorError, ExecutorResult,
+    execution_output::ExecutionOutput, state_checkpoint_output::StateCheckpointOutput,
+    state_compute_result::StateComputeResult, BlockExecutorTrait, ExecutorError, ExecutorResult,
 };
 use aptos_experimental_runtimes::thread_manager::THREAD_MANAGER;
 use aptos_infallible::RwLock;
@@ -46,16 +47,20 @@ pub trait TransactionBlockExecutor: Send + Sync {
         transactions: ExecutableTransactions,
         state_view: CachedStateView,
         onchain_config: BlockExecutorConfigFromOnchain,
-    ) -> Result<ChunkOutput>;
+    ) -> Result<ExecutionOutput>;
 }
 
 impl TransactionBlockExecutor for AptosVM {
     fn execute_transaction_block(
         transactions: ExecutableTransactions,
         state_view: CachedStateView,
         onchain_config: BlockExecutorConfigFromOnchain,
-    ) -> Result<ChunkOutput> {
-        ChunkOutput::by_transaction_execution::<AptosVM>(transactions, state_view, onchain_config)
+    ) -> Result<ExecutionOutput> {
+        DoGetExecutionOutput::by_transaction_execution::<AptosVM>(
+            transactions,
+            state_view,
+            onchain_config,
+        )
     }
 }
 
@@ -267,7 +272,17 @@ where
                 let _timer = OTHER_TIMERS.timer_with(&["state_checkpoint"]);
 
                 THREAD_MANAGER.get_exe_cpu_pool().install(|| {
-                    chunk_output.into_state_checkpoint_output(parent_output.state(), block_id)
+                    fail_point!("executor::block_state_checkpoint", |_| {
+                        Err(anyhow::anyhow!("Injected error in block state checkpoint."))
+                    });
+
+                    ApplyChunkOutput::calculate_state_checkpoint(
+                        chunk_output,
+                        parent_output.state(),
+                        Some(block_id),
+                        None,
+                        /*is_block=*/ true,
+                    )
                 })?
             };
 

execution/executor/src/chunk_executor.rs

@@ -8,8 +8,8 @@ use crate::{
     components::{
         apply_chunk_output::ApplyChunkOutput,
         chunk_commit_queue::{ChunkCommitQueue, ChunkToUpdateLedger},
-        chunk_output::ChunkOutput,
         chunk_result_verifier::{ChunkResultVerifier, ReplayChunkVerifier, StateSyncChunkVerifier},
+        do_get_execution_output::DoGetExecutionOutput,
         do_ledger_update::DoLedgerUpdate,
         executed_chunk::ExecutedChunk,
         partial_state_compute_result::PartialStateComputeResult,
@@ -597,7 +597,7 @@ impl<V: VMExecutor> ChunkExecutorInner<V> {
             .collect::<Vec<SignatureVerifiedTransaction>>();
 
         // State sync executor shouldn't have block gas limit.
-        let chunk_output = ChunkOutput::by_transaction_execution::<V>(
+        let chunk_output = DoGetExecutionOutput::by_transaction_execution::<V>(
             txns.into(),
             state_view,
             BlockExecutorConfigFromOnchain::new_no_block_limit(),

execution/executor/src/components/apply_chunk_output.rs

@@ -6,15 +6,15 @@
 
 use crate::{
     components::{
-        chunk_output::ChunkOutput, do_ledger_update::DoLedgerUpdate, executed_chunk::ExecutedChunk,
-        in_memory_state_calculator_v2::InMemoryStateCalculatorV2,
+        do_ledger_update::DoLedgerUpdate, in_memory_state_calculator_v2::InMemoryStateCalculatorV2,
         partial_state_compute_result::PartialStateComputeResult,
     },
     metrics::{EXECUTOR_ERRORS, OTHER_TIMERS},
 };
 use anyhow::Result;
 use aptos_crypto::HashValue;
 use aptos_executor_types::{
+    execution_output::ExecutionOutput,
     parsed_transaction_output::TransactionsWithParsedOutput,
     state_checkpoint_output::{StateCheckpointOutput, TransactionsByStatus},
     ParsedTransactionOutput,
@@ -37,13 +37,13 @@ pub struct ApplyChunkOutput;
 
 impl ApplyChunkOutput {
     pub fn calculate_state_checkpoint(
-        chunk_output: ChunkOutput,
+        chunk_output: ExecutionOutput,
         parent_state: &StateDelta,
         append_state_checkpoint_to_block: Option<HashValue>,
         known_state_checkpoints: Option<Vec<Option<HashValue>>>,
         is_block: bool,
     ) -> Result<(Arc<StateDelta>, Option<EpochState>, StateCheckpointOutput)> {
-        let ChunkOutput {
+        let ExecutionOutput {
             state_cache,
             transactions,
             transaction_outputs,
@@ -109,10 +109,14 @@ impl ApplyChunkOutput {
     }
 
     pub fn apply_chunk(
-        chunk_output: ChunkOutput,
+        chunk_output: ExecutionOutput,
         base_view: &ExecutedTrees,
         known_state_checkpoint_hashes: Option<Vec<Option<HashValue>>>,
-    ) -> Result<(ExecutedChunk, Vec<Transaction>, Vec<Transaction>)> {
+    ) -> Result<(
+        PartialStateComputeResult,
+        Vec<Transaction>,
+        Vec<Transaction>,
+    )> {
         let (result_state, next_epoch_state, state_checkpoint_output) =
             Self::calculate_state_checkpoint(
                 chunk_output,
@@ -130,14 +134,7 @@ impl ApplyChunkOutput {
         );
         output.set_ledger_update_output(ledger_update_output);
 
-        Ok((
-            ExecutedChunk {
-                output,
-                ledger_info_opt: None,
-            },
-            to_discard,
-            to_retry,
-        ))
+        Ok((output, to_discard, to_retry))
     }
 
     fn sort_transactions_with_state_checkpoint(