Merge pull request #5109 from onflow/leo/fix-get-highest

[Storehouse] Fix getting highest executed block ID when storehouse is enabled

cmd/execution_builder.go

@@ -540,15 +540,16 @@ func (exeNode *ExecutionNode) LoadProviderEngine(
 
 	// Get latest executed block and a view at that block
 	ctx := context.Background()
-	_, blockID, err := exeNode.executionState.GetHighestExecutedBlockID(ctx)
+	height, blockID, err := exeNode.executionState.GetHighestExecutedBlockID(ctx)
 	if err != nil {
 		return nil, fmt.Errorf(
-			"cannot get the latest executed block id: %w",
-			err)
+			"cannot get the latest executed block id at height %v: %w",
+			height, err)
 	}
 	blockSnapshot, _, err := exeNode.executionState.CreateStorageSnapshot(blockID)
 	if err != nil {
-		return nil, fmt.Errorf("cannot create a storage snapshot at block %v: %w", blockID, err)
+		return nil, fmt.Errorf("cannot create a storage snapshot at block %v at height %v: %w", blockID,
+			height, err)
 	}
 
 	// Get the epoch counter from the smart contract at the last executed block.
@@ -558,7 +559,8 @@ func (exeNode *ExecutionNode) LoadProviderEngine(
 		blockSnapshot)
 	// Failing to fetch the epoch counter from the smart contract is a fatal error.
 	if err != nil {
-		return nil, fmt.Errorf("cannot get epoch counter from the smart contract at block %s: %w", blockID.String(), err)
+		return nil, fmt.Errorf("cannot get epoch counter from the smart contract at block %s at height %v: %w",
+			blockID.String(), height, err)
 	}
 
 	// Get the epoch counter form the protocol state, at the same block.
@@ -577,6 +579,7 @@ func (exeNode *ExecutionNode) LoadProviderEngine(
 		Uint64("contractEpochCounter", contractEpochCounter).
 		Uint64("protocolStateEpochCounter", protocolStateEpochCounter).
 		Str("blockID", blockID.String()).
+		Uint64("height", height).
 		Logger()
 
 	if contractEpochCounter != protocolStateEpochCounter {

engine/execution/execution_test.go

@@ -233,7 +233,13 @@ func TestExecutionFlow(t *testing.T) {
 	}, time.Second*10, time.Millisecond*500)
 
 	// check that the block has been executed.
-	exeNode.AssertHighestExecutedBlock(t, block.Header)
+	exeNode.AssertBlockIsExecuted(t, block.Header)
+
+	if exeNode.StorehouseEnabled {
+		exeNode.AssertHighestExecutedBlock(t, genesis)
+	} else {
+		exeNode.AssertHighestExecutedBlock(t, block.Header)
+	}
 
 	myReceipt, err := exeNode.MyExecutionReceipts.MyReceipt(block.ID())
 	require.NoError(t, err)
@@ -435,7 +441,15 @@ func TestFailedTxWillNotChangeStateCommitment(t *testing.T) {
 	hub.DeliverAllEventually(t, func() bool {
 		return receiptsReceived.Load() == 1
 	})
-	exe1Node.AssertHighestExecutedBlock(t, block1.Header)
+
+	if exe1Node.StorehouseEnabled {
+		exe1Node.AssertHighestExecutedBlock(t, genesis)
+	} else {
+		exe1Node.AssertHighestExecutedBlock(t, block1.Header)
+	}
+
+	exe1Node.AssertBlockIsExecuted(t, block1.Header)
+	exe1Node.AssertBlockNotExecuted(t, block2.Header)
 
 	scExe1Genesis, err := exe1Node.ExecutionState.StateCommitmentByBlockID(genesis.ID())
 	assert.NoError(t, err)
@@ -457,8 +471,8 @@ func TestFailedTxWillNotChangeStateCommitment(t *testing.T) {
 	})
 
 	// ensure state has been synced across both nodes
-	exe1Node.AssertHighestExecutedBlock(t, block3.Header)
-	// exe2Node.AssertHighestExecutedBlock(t, block3.Header)
+	exe1Node.AssertBlockIsExecuted(t, block2.Header)
+	exe1Node.AssertBlockIsExecuted(t, block3.Header)
 
 	// verify state commitment of block 2 is the same as block 1, since tx failed on seq number verification
 	scExe1Block2, err := exe1Node.ExecutionState.StateCommitmentByBlockID(block2.ID())

engine/execution/state/state.go

@@ -482,7 +482,19 @@ func (s *state) UpdateHighestExecutedBlockIfHigher(ctx context.Context, header *
 	return operation.RetryOnConflict(s.db.Update, procedure.UpdateHighestExecutedBlockIfHigher(header))
 }
 
+// deprecated by storehouse's GetHighestFinalizedExecuted
 func (s *state) GetHighestExecutedBlockID(ctx context.Context) (uint64, flow.Identifier, error) {
+	if s.enableRegisterStore {
+		// when storehouse is enabled, the highest executed block is consisted as
+		// the highest finalized and executed block
+		height := s.GetHighestFinalizedExecuted()
+		header, err := s.headers.ByHeight(height)
+		if err != nil {
+			return 0, flow.ZeroID, fmt.Errorf("could not get header by height %v: %w", height, err)
+		}
+		return height, header.ID(), nil
+	}
+
 	var blockID flow.Identifier
 	var height uint64
 	err := s.db.View(procedure.GetHighestExecutedBlock(&height, &blockID))

engine/testutil/mock/nodes.go

@@ -203,6 +203,7 @@ type ExecutionNode struct {
 	Collections         storage.Collections
 	Finalizer           *consensus.Finalizer
 	MyExecutionReceipts storage.MyExecutionReceipts
+	StorehouseEnabled   bool
 }
 
 func (en ExecutionNode) Ready(ctx context.Context) {
@@ -247,14 +248,25 @@ func (en ExecutionNode) Done(cancelFunc context.CancelFunc) {
 }
 
 func (en ExecutionNode) AssertHighestExecutedBlock(t *testing.T, header *flow.Header) {
-
 	height, blockID, err := en.ExecutionState.GetHighestExecutedBlockID(context.Background())
 	require.NoError(t, err)
 
 	require.Equal(t, header.ID(), blockID)
 	require.Equal(t, header.Height, height)
 }
 
+func (en ExecutionNode) AssertBlockIsExecuted(t *testing.T, header *flow.Header) {
+	executed, err := en.ExecutionState.IsBlockExecuted(header.Height, header.ID())
+	require.NoError(t, err)
+	require.True(t, executed)
+}
+
+func (en ExecutionNode) AssertBlockNotExecuted(t *testing.T, header *flow.Header) {
+	executed, err := en.ExecutionState.IsBlockExecuted(header.Height, header.ID())
+	require.NoError(t, err)
+	require.False(t, executed)
+}
+
 // VerificationNode implements an in-process verification node for tests.
 type VerificationNode struct {
 	*GenericNode

engine/testutil/nodes.go

@@ -650,11 +650,12 @@ func ExecutionNode(t *testing.T, hub *stub.Hub, identity *flow.Identity, identit
 		node.Log)
 	require.NoError(t, err)
 
+	storehouseEnabled := true
 	execState := executionState.NewExecutionState(
 		ls, commitsStorage, node.Blocks, node.Headers, collectionsStorage, chunkDataPackStorage, results, myReceipts, eventsStorage, serviceEventsStorage, txResultStorage, node.PublicDB, node.Tracer,
 		// TODO: test with register store
 		registerStore,
-		false,
+		storehouseEnabled,
 	)
 
 	requestEngine, err := requester.New(
@@ -852,6 +853,7 @@ func ExecutionNode(t *testing.T, hub *stub.Hub, identity *flow.Identity, identit
 		Finalizer:           finalizer,
 		MyExecutionReceipts: myReceipts,
 		Compactor:           compactor,
+		StorehouseEnabled:   storehouseEnabled,
 	}
 }