Merge pull request #8134 from onflow/andron/7655-registers-endpoints-error-handling

Author: AndriiDiachuk

engine/access/state_stream/backend/backend.go

@@ -7,9 +7,8 @@ import (
 	"time"
 
 	"github.com/rs/zerolog"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 
+	"github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/engine/access/index"
 	"github.com/onflow/flow-go/engine/access/rpc/backend/common"
 	"github.com/onflow/flow-go/engine/access/state_stream"
@@ -188,61 +187,76 @@ func (b *StateStreamBackend) getExecutionData(ctx context.Context, height uint64
 }
 
 // GetRegisterValues returns the register values for the given register IDs at the given block height.
+//
+// CAUTION: this layer SIMPLIFIES the ERROR HANDLING convention
+//   - All errors returned are guaranteed to be benign. The node can continue normal operations after such errors.
+//   - To prevent delivering incorrect results to clients in case of an error, all other return values should be discarded.
+//
+// Expected sentinel errors providing details to clients about failed requests:
+//   - [access.InvalidRequestError]: If the request had invalid arguments.
+//   - [access.DataNotFoundError]: When data required to process the request is not available.
+//   - [access.OutOfRangeError]: If the data for the requested height is outside the node's available range.
+//   - [access.PreconditionFailedError]: When the register's database isn't initialized yet.
 func (b *StateStreamBackend) GetRegisterValues(
+	ctx context.Context,
 	ids flow.RegisterIDs,
 	height uint64,
 	criteria optimistic_sync.Criteria,
 ) ([]flow.RegisterValue, *accessmodel.ExecutorMetadata, error) {
 	if len(ids) > b.registerRequestLimit {
-		return nil, nil, status.Errorf(codes.InvalidArgument, "number of register IDs exceeds limit of %d", b.registerRequestLimit)
+		return nil, nil, access.NewInvalidRequestError(
+			fmt.Errorf("number of register IDs exceeds limit of %d", b.registerRequestLimit))
 	}
 
 	header, err := b.headers.ByHeight(height)
 	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) {
-			return nil, nil, status.Errorf(codes.NotFound, "no finalized block is known at the given height %d", height)
-		}
-		return nil, nil, err
+		err = access.RequireErrorIs(ctx, err, storage.ErrNotFound)
+		err = fmt.Errorf("failed to find header by height %d: %w", height, err)
+		return nil, nil, access.NewDataNotFoundError("header", err)
 	}
 
 	execResultInfo, err := b.executionResultProvider.ExecutionResultInfo(header.ID(), criteria)
 	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) || common.IsInsufficientExecutionReceipts(err) {
-			return nil, nil, status.Errorf(codes.NotFound, "failed to get execution result info for block")
+		err = fmt.Errorf("failed to get execution result info for block: %w", err)
+		switch {
+		case errors.Is(err, storage.ErrNotFound):
+			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case common.IsInsufficientExecutionReceipts(err):
+			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		default:
+			return nil, nil, access.RequireNoError(ctx, err)
 		}
-		return nil, nil, err
 	}
 
 	executionResultID := execResultInfo.ExecutionResultID
 	snapshot, err := b.executionStateCache.Snapshot(executionResultID)
 	if err != nil {
-		if errors.Is(err, storage.ErrNotFound) {
-			return nil, nil, status.Errorf(codes.NotFound, "register values for block %d not found", height)
-		}
-		return nil, nil, err
+		err = access.RequireErrorIs(ctx, err, storage.ErrNotFound)
+		err = fmt.Errorf("failed to find snapshot by execution result ID %s: %w", executionResultID.String(), err)
+		return nil, nil, access.NewDataNotFoundError("snapshot", err)
 	}
 
 	registers, err := snapshot.Registers()
 	if err != nil {
-		if errors.Is(err, indexer.ErrIndexNotInitialized) {
-			return nil, nil, status.Errorf(codes.FailedPrecondition, "failed to get lowest indexed height: %v", err)
-		}
-
-		return nil, nil, err
+		err = access.RequireErrorIs(ctx, err, indexer.ErrIndexNotInitialized)
+		err = fmt.Errorf("failed to get registers storage from snapshot: %w", err)
+		return nil, nil, access.NewPreconditionFailedError(err)
 	}
 
 	result := make([]flow.RegisterValue, len(ids))
 	for i, regID := range ids {
 		val, err := registers.Get(regID, height)
 		if err != nil {
-			if errors.Is(err, storage.ErrHeightNotIndexed) {
-				return nil, nil, status.Errorf(codes.OutOfRange, "register values for block %d is not available", height)
-			}
-			if errors.Is(err, storage.ErrNotFound) {
-				return nil, nil, status.Errorf(codes.NotFound, "register values for block %d not found", height)
+			err = fmt.Errorf("failed to get register by the register ID at a given block height: %w", err)
+			switch {
+			case errors.Is(err, storage.ErrNotFound):
+				return nil, nil, access.NewDataNotFoundError("registers", err)
+			case errors.Is(err, storage.ErrHeightNotIndexed):
+				return nil, nil, access.NewOutOfRangeError(err)
+			default:
+				return nil, nil, access.RequireNoError(ctx, err)
 			}
 
-			return nil, nil, err
 		}
 		result[i] = val
 	}

engine/access/state_stream/backend/backend_executiondata_test.go

@@ -406,13 +406,13 @@ func (s *BackendExecutionDataSuite) TestGetExecutionDataByBlockID() {
 	})
 
 	s.Run("execution result info returns unexpected error", func() {
-		expectedErr := fmt.Errorf("failed to get execution result info for block: %w", storage.ErrDataMismatch)
+		exception := fmt.Errorf("failed to get execution result info for block: %w", storage.ErrDataMismatch)
 		s.executionResultProvider.
 			On("ExecutionResultInfo", block.ID(), mock.Anything).
 			Return(nil, storage.ErrDataMismatch).
 			Once()
 
-		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, expectedErr)
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
 		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
 
 		execDataRes, metadata, err := s.backend.GetExecutionDataByBlockID(ctxIrr, block.ID(), s.criteria)
@@ -451,13 +451,13 @@ func (s *BackendExecutionDataSuite) TestGetExecutionDataByBlockID() {
 			}, nil).
 			Once()
 
-		expectedError := fmt.Errorf("unexpected error")
+		exception := fmt.Errorf("unexpected error")
 		s.executionStateCache.
 			On("Snapshot", result.ID()).
-			Return(nil, expectedError).
+			Return(nil, exception).
 			Once()
 
-		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, expectedError)
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
 		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
 
 		execDataRes, metadata, err := s.backend.GetExecutionDataByBlockID(ctxIrr, block.ID(), s.criteria)
@@ -522,8 +522,8 @@ func (s *BackendExecutionDataSuite) TestGetExecutionDataByBlockID() {
 			Return(nil, storage.ErrDataMismatch).
 			Once()
 
-		expectedError := fmt.Errorf("unexpected error getting execution data: %w", storage.ErrDataMismatch)
-		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, expectedError)
+		exception := fmt.Errorf("unexpected error getting execution data: %w", storage.ErrDataMismatch)
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
 		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
 
 		execDataRes, metadata, err := s.backend.GetExecutionDataByBlockID(ctxIrr, block.ID(), s.criteria)
@@ -897,6 +897,9 @@ func (s *BackendExecutionDataSuite) TestSubscribeExecutionDataHandlesErrors() {
 // TestGetRegisterValues tests that GetRegisterValues correctly returns register data
 // in normal conditions and propagates appropriate errors for all failure scenarios.
 func (s *BackendExecutionDataSuite) TestGetRegisterValues() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	block := s.blocks[0]
 	seal := s.sealMap[block.ID()]
 	result := s.resultMap[seal.ResultID]
@@ -928,41 +931,60 @@ func (s *BackendExecutionDataSuite) TestGetRegisterValues() {
 		expectedRegister := flow.RegisterValue("value0")
 		s.registers.On("Get", s.registerID, block.Height).Return(expectedRegister, nil).Once()
 
-		res, resMetadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, resMetadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 
 		require.Equal(s.T(), []flow.RegisterValue{expectedRegister}, res)
 		require.Equal(s.T(), metadata, resMetadata)
 		require.NoError(s.T(), err)
 	})
 
 	s.Run("returns error if too many registers are requested", func() {
-		res, metadata, err := s.backend.GetRegisterValues(make(flow.RegisterIDs, s.backend.registerRequestLimit+1), block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, make(flow.RegisterIDs, s.backend.registerRequestLimit+1), block.Height, s.criteria)
 
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.InvalidArgument, status.Code(err))
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsInvalidRequestError(err))
 	})
 
 	s.Run("returns error if failed to get execution result info for block - insufficient receipts", func() {
 		s.executionResultProvider.
 			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
 			Return(nil, common.NewInsufficientExecutionReceipts(block.ID(), 0)).Once()
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.NotFound, status.Code(err))
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsDataNotFoundError(err))
 	})
 
 	s.Run("returns error if failed to get execution result info for block - not found", func() {
 		s.executionResultProvider.
 			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
 			Return(nil, storage.ErrNotFound).Once()
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		require.Nil(s.T(), res)
+		require.Nil(s.T(), metadata)
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsDataNotFoundError(err))
+	})
+
+	s.Run("execution result info returns unexpected error", func() {
+		exception := fmt.Errorf("failed to get execution result info for block: %w", storage.ErrDataMismatch)
+
+		s.executionResultProvider.
+			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
+			Return(nil, storage.ErrDataMismatch).Once()
+
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
+		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
+
+		res, metadata, err := s.backend.GetRegisterValues(ctxIrr, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.NotFound, status.Code(err))
+		require.Error(s.T(), err)
 	})
 
 	s.Run("returns error when snapshot is not found", func() {
@@ -978,10 +1000,34 @@ func (s *BackendExecutionDataSuite) TestGetRegisterValues() {
 			Return(nil, storage.ErrNotFound).
 			Once()
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.NotFound, status.Code(err))
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsDataNotFoundError(err))
+	})
+
+	s.Run("snapshot returns unexpected error", func() {
+		s.executionResultProvider.
+			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
+			Return(&optimistic_sync.ExecutionResultInfo{
+				ExecutionResultID: result.ID(),
+				ExecutionNodes:    executionNodes.ToSkeleton(),
+			}, nil).Once()
+
+		exception := fmt.Errorf("unexpected error")
+		s.executionStateCache.
+			On("Snapshot", result.ID()).
+			Return(nil, exception).
+			Once()
+
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
+		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
+
+		execDataRes, metadata, err := s.backend.GetExecutionDataByBlockID(ctxIrr, block.ID(), s.criteria)
+		assert.Nil(s.T(), execDataRes)
+		assert.Nil(s.T(), metadata)
+		assert.Error(s.T(), err)
 	})
 
 	s.Run("returns error if the storage is still bootstrapping.", func() {
@@ -999,10 +1045,36 @@ func (s *BackendExecutionDataSuite) TestGetRegisterValues() {
 
 		s.executionDataSnapshot.On("Registers").Return(nil, indexer.ErrIndexNotInitialized).Once()
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		require.Nil(s.T(), res)
+		require.Nil(s.T(), metadata)
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsPreconditionFailedError(err))
+	})
+
+	s.Run("registers returns unexpected error", func() {
+		s.executionResultProvider.
+			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
+			Return(&optimistic_sync.ExecutionResultInfo{
+				ExecutionResultID: result.ID(),
+				ExecutionNodes:    executionNodes.ToSkeleton(),
+			}, nil).Once()
+
+		s.executionStateCache.
+			On("Snapshot", result.ID()).
+			Return(s.executionDataSnapshot, nil).
+			Once()
+
+		exception := fmt.Errorf("unexpected error")
+		s.executionDataSnapshot.On("Registers").Return(nil, exception).Once()
+
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
+		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
+
+		res, metadata, err := s.backend.GetRegisterValues(ctxIrr, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.FailedPrecondition, status.Code(err))
+		require.Error(s.T(), err)
 	})
 
 	s.Run("returns error if the requested height is outside the range of indexed blocks", func() {
@@ -1022,10 +1094,34 @@ func (s *BackendExecutionDataSuite) TestGetRegisterValues() {
 
 		s.registers.On("Get", s.registerID, block.Height).Return(nil, storage.ErrHeightNotIndexed).Once()
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		require.Nil(s.T(), res)
+		require.Nil(s.T(), metadata)
+		require.Error(s.T(), err)
+		require.ErrorIs(s.T(), err, storage.ErrHeightNotIndexed)
+	})
+
+	s.Run("returns error if block or registerSnapshot value at height was not found", func() {
+		s.executionResultProvider.
+			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
+			Return(&optimistic_sync.ExecutionResultInfo{
+				ExecutionResultID: result.ID(),
+				ExecutionNodes:    executionNodes.ToSkeleton(),
+			}, nil).Once()
+
+		s.executionStateCache.
+			On("Snapshot", result.ID()).
+			Return(s.executionDataSnapshot, nil).
+			Once()
+
+		s.executionDataSnapshot.On("Registers").Return(s.registers, nil).Once()
+		s.registers.On("Get", s.registerID, block.Height).Return(nil, storage.ErrNotFound).Once()
+
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.OutOfRange, status.Code(err))
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsDataNotFoundError(err))
 	})
 
 	s.Run("returns error if block or registerSnapshot value at height was not found", func() {
@@ -1044,19 +1140,48 @@ func (s *BackendExecutionDataSuite) TestGetRegisterValues() {
 		s.executionDataSnapshot.On("Registers").Return(s.registers, nil).Once()
 		s.registers.On("Get", s.registerID, block.Height).Return(nil, storage.ErrNotFound).Once()
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		require.Nil(s.T(), res)
+		require.Nil(s.T(), metadata)
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsDataNotFoundError(err))
+	})
+
+	s.Run("registers Get returns unexpected error", func() {
+		s.executionResultProvider.
+			On("ExecutionResultInfo", block.ToHeader().ID(), mock.Anything).
+			Return(&optimistic_sync.ExecutionResultInfo{
+				ExecutionResultID: result.ID(),
+				ExecutionNodes:    executionNodes.ToSkeleton(),
+			}, nil).Once()
+
+		s.executionStateCache.
+			On("Snapshot", result.ID()).
+			Return(s.executionDataSnapshot, nil).
+			Once()
+
+		s.executionDataSnapshot.On("Registers").Return(s.registers, nil).Once()
+
+		exception := fmt.Errorf("failed to get register by the register ID at a given block height: %w", storage.ErrDataMismatch)
+		ctxSignaler := irrecoverable.NewMockSignalerContextExpectError(s.T(), ctx, exception)
+		ctxIrr := irrecoverable.WithSignalerContext(ctx, ctxSignaler)
+
+		s.registers.On("Get", s.registerID, block.Height).Return(nil, storage.ErrDataMismatch).Once()
+
+		res, metadata, err := s.backend.GetRegisterValues(ctxIrr, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.NotFound, status.Code(err))
+		require.Error(s.T(), err)
 	})
 
 	s.Run("returns error if no finalized block is known at the given height", func() {
 		s.headers.On("ByHeight", block.Height).Unset()
 		s.headers.On("ByHeight", block.Height).Return(nil, storage.ErrNotFound)
 
-		res, metadata, err := s.backend.GetRegisterValues(flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
+		res, metadata, err := s.backend.GetRegisterValues(ctx, flow.RegisterIDs{s.registerID}, block.Height, s.criteria)
 		require.Nil(s.T(), res)
 		require.Nil(s.T(), metadata)
-		require.Equal(s.T(), codes.NotFound, status.Code(err))
+		require.Error(s.T(), err)
+		require.True(s.T(), access.IsDataNotFoundError(err))
 	})
 }