Move payload_manager.notify_commit to after commit

consensus/src/pipeline/pipeline_builder.rs

@@ -325,7 +325,6 @@ impl PipelineBuilder {
                 pre_commit_fut.clone(),
                 parent.post_pre_commit_fut.clone(),
                 self.state_sync_notifier.clone(),
-                self.payload_manager.clone(),
                 block.clone(),
             ),
             &mut abort_handles,
@@ -335,6 +334,7 @@ impl PipelineBuilder {
                 pre_commit_fut.clone(),
                 commit_ledger_fut.clone(),
                 parent.post_commit_fut.clone(),
+                self.payload_manager.clone(),
                 block_store_callback,
                 block.clone(),
             ),
@@ -618,15 +618,12 @@ impl PipelineBuilder {
         pre_commit: TaskFuture<PreCommitResult>,
         parent_post_pre_commit: TaskFuture<PostCommitResult>,
         state_sync_notifier: Arc<dyn ConsensusNotificationSender>,
-        payload_manager: Arc<dyn TPayloadManager>,
         block: Arc<Block>,
     ) -> TaskResult<PostPreCommitResult> {
         let compute_result = pre_commit.await?;
         parent_post_pre_commit.await?;
 
         let _tracker = Tracker::new("post_pre_commit", &block);
-        let payload = block.payload().cloned();
-        let timestamp = block.timestamp_usecs();
         let _timer = counters::OP_COUNTERS.timer("pre_commit_notify");
 
         let txns = compute_result.transactions_to_commit().to_vec();
@@ -640,8 +637,6 @@ impl PipelineBuilder {
             error!(error = ?e, "Failed to notify state synchronizer");
         }
 
-        let payload_vec = payload.into_iter().collect();
-        payload_manager.notify_commit(timestamp, payload_vec);
         Ok(())
     }
 
@@ -684,6 +679,7 @@ impl PipelineBuilder {
         pre_commit_fut: TaskFuture<PreCommitResult>,
         commit_ledger_fut: TaskFuture<CommitLedgerResult>,
         parent_post_commit: TaskFuture<PostCommitResult>,
+        payload_manager: Arc<dyn TPayloadManager>,
         block_store_callback: Box<dyn FnOnce(LedgerInfoWithSignatures) + Send + Sync>,
         block: Arc<Block>,
     ) -> TaskResult<PostCommitResult> {
@@ -695,6 +691,11 @@ impl PipelineBuilder {
         update_counters_for_block(&block);
         update_counters_for_compute_result(&compute_result);
 
+        let payload = block.payload().cloned();
+        let timestamp = block.timestamp_usecs();
+        let payload_vec = payload.into_iter().collect();
+        payload_manager.notify_commit(timestamp, payload_vec);
+
         if let Some(ledger_info_with_sigs) = maybe_ledger_info_with_sigs {
             block_store_callback(ledger_info_with_sigs);
         }

consensus/src/state_computer.rs

@@ -131,15 +131,9 @@ impl ExecutionProxy {
         tx
     }
 
-    fn pre_commit_hook(
-        &self,
-        block: &Block,
-        payload_manager: Arc<dyn TPayloadManager>,
-    ) -> PreCommitHook {
+    fn pre_commit_hook(&self) -> PreCommitHook {
         let mut pre_commit_notifier = self.pre_commit_notifier.clone();
         let state_sync_notifier = self.state_sync_notifier.clone();
-        let payload = block.payload().cloned();
-        let timestamp = block.timestamp_usecs();
         Box::new(move |state_compute_result: &StateComputeResult| {
             let state_compute_result = state_compute_result.clone();
             Box::pin(async move {
@@ -158,16 +152,38 @@ impl ExecutionProxy {
                         ) {
                             error!(error = ?e, "Failed to notify state synchronizer");
                         }
-
-                        let payload_vec = payload.into_iter().collect();
-                        payload_manager.notify_commit(timestamp, payload_vec);
                     }))
                     .await
                     .expect("Failed to send pre-commit notification");
             })
         })
     }
 
+    fn commit_hook(
+        &self,
+        blocks: &[Arc<PipelinedBlock>],
+        callback: StateComputerCommitCallBackType,
+        finality_proof: LedgerInfoWithSignatures,
+    ) -> NotificationType {
+        let payload_manager = self
+            .state
+            .read()
+            .as_ref()
+            .expect("must be set within an epoch")
+            .payload_manager
+            .clone();
+        let blocks = blocks.to_vec();
+        Box::pin(async move {
+            for block in blocks.iter() {
+                let payload = block.payload().cloned();
+                let payload_vec = payload.into_iter().collect();
+                let timestamp = block.timestamp_usecs();
+                payload_manager.notify_commit(timestamp, payload_vec);
+            }
+            callback(&blocks, finality_proof);
+        })
+    }
+
     pub fn pipeline_builder(&self, commit_signer: Arc<ValidatorSigner>) -> PipelineBuilder {
         let MutableState {
             validators,
@@ -236,7 +252,7 @@ impl StateComputer for ExecutionProxy {
 
         let txn_notifier = self.txn_notifier.clone();
         let transaction_generator = BlockPreparer::new(
-            payload_manager.clone(),
+            payload_manager,
             self.transaction_filter.clone(),
             transaction_deduper.clone(),
             transaction_shuffler.clone(),
@@ -260,7 +276,7 @@ impl StateComputer for ExecutionProxy {
                 parent_block_id,
                 transaction_generator,
                 block_executor_onchain_config,
-                self.pre_commit_hook(block, payload_manager),
+                self.pre_commit_hook(),
                 lifetime_guard,
             )
             .await;
@@ -343,14 +359,9 @@ impl StateComputer for ExecutionProxy {
         )
         .expect("spawn_blocking failed");
 
-        let blocks = blocks.to_vec();
-        let callback_fut = Box::pin(async move {
-            callback(&blocks, finality_proof);
-        });
-
         self.commit_notifier
             .clone()
-            .send(callback_fut)
+            .send(self.commit_hook(blocks, callback, finality_proof))
             .await
             .expect("Failed to send commit notification");
 

third_party/move/testing-infra/transactional-test-runner/src/vm_test_harness.rs

@@ -302,7 +302,7 @@ impl<'a> MoveTestAdapter<'a> for SimpleVMTestAdapter<'a> {
             Compatibility::new(
                 !extra_args.skip_check_struct_layout,
                 !extra_args.skip_check_friend_linking,
-                false
+                false,
             )
         };
         if vm.vm_config().use_loader_v2 {