[DataAvailability] add a check for executors passed by a client

cmd/access/node_builder/access_node_builder.go

@@ -1784,6 +1784,7 @@ func (builder *FlowAccessNodeBuilder) buildExecutionResultInfoProvider() *FlowAc
 			node.Logger,
 			node.State,
 			node.Storage.Receipts,
+			node.Storage.Headers,
 			execNodeSelector,
 			operatorCriteria,
 		)

cmd/observer/node_builder/observer_builder.go

@@ -1141,6 +1141,7 @@ func (builder *ObserverServiceBuilder) buildExecutionResultInfoProvider() *Obser
 			node.Logger,
 			node.State,
 			node.Storage.Receipts,
+			node.Storage.Headers,
 			execNodeSelector,
 			operatorCriteria,
 		)

engine/access/rpc/backend/accounts/accounts.go

@@ -127,6 +127,12 @@ func (a *Accounts) GetAccountAtLatestBlock(ctx context.Context, address flow.Add
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -181,6 +187,12 @@ func (a *Accounts) GetAccountAtBlockHeight(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -231,6 +243,12 @@ func (a *Accounts) GetAccountBalanceAtLatestBlock(ctx context.Context, address f
 			return 0, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return 0, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return 0, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return 0, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return 0, nil, access.NewPreconditionFailedError(err)
 		default:
 			return 0, nil, access.RequireNoError(ctx, err)
 		}
@@ -285,6 +303,12 @@ func (a *Accounts) GetAccountBalanceAtBlockHeight(
 			return 0, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return 0, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return 0, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return 0, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return 0, nil, access.NewPreconditionFailedError(err)
 		default:
 			return 0, nil, access.RequireNoError(ctx, err)
 		}
@@ -339,6 +363,12 @@ func (a *Accounts) GetAccountKeyAtLatestBlock(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -394,6 +424,12 @@ func (a *Accounts) GetAccountKeyAtBlockHeight(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -447,6 +483,12 @@ func (a *Accounts) GetAccountKeysAtLatestBlock(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -501,6 +543,12 @@ func (a *Accounts) GetAccountKeysAtBlockHeight(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}

engine/access/rpc/backend/scripts/scripts.go

@@ -163,6 +163,12 @@ func (b *Scripts) ExecuteScriptAtLatestBlock(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -221,6 +227,12 @@ func (b *Scripts) ExecuteScriptAtBlockID(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}
@@ -284,6 +296,12 @@ func (b *Scripts) ExecuteScriptAtBlockHeight(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}

engine/access/state_stream/backend/backend.go

@@ -223,6 +223,12 @@ func (b *StateStreamBackend) GetRegisterValues(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}

engine/access/state_stream/backend/backend_executiondata.go

@@ -61,6 +61,12 @@ func (b *ExecutionDataBackend) GetExecutionDataByBlockID(
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
 		case common.IsInsufficientExecutionReceipts(err):
 			return nil, nil, access.NewDataNotFoundError("execution data", err)
+		case optimistic_sync.IsRequiredExecutorsCountExceededError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsUnknownRequiredExecutorError(err):
+			return nil, nil, access.NewInvalidRequestError(err)
+		case optimistic_sync.IsCriteriaNotMetError(err):
+			return nil, nil, access.NewPreconditionFailedError(err)
 		default:
 			return nil, nil, access.RequireNoError(ctx, err)
 		}

module/executiondatasync/optimistic_sync/errors.go

@@ -0,0 +1,75 @@
+package optimistic_sync
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+// RequiredExecutorsCountExceededError indicates that the requested number of required executors
+// exceeds the total available execution nodes.
+type RequiredExecutorsCountExceededError struct {
+	requiredExecutorsCount  int
+	availableExecutorsCount int
+}
+
+func NewRequiredExecutorsCountExceededError(requiredExecutorsCount int, availableExecutorsCount int) *RequiredExecutorsCountExceededError {
+	return &RequiredExecutorsCountExceededError{
+		requiredExecutorsCount:  requiredExecutorsCount,
+		availableExecutorsCount: availableExecutorsCount,
+	}
+}
+
+func (e *RequiredExecutorsCountExceededError) Error() string {
+	return fmt.Sprintf("required executors count exceeded: required %d, available %d",
+		e.requiredExecutorsCount, e.availableExecutorsCount,
+	)
+}
+
+func IsRequiredExecutorsCountExceededError(err error) bool {
+	var requiredExecutorsCountExceededError *RequiredExecutorsCountExceededError
+	return errors.As(err, &requiredExecutorsCountExceededError)
+}
+
+// UnknownRequiredExecutorError indicates that a required executor ID is not present
+// in the list of active execution nodes.
+type UnknownRequiredExecutorError struct {
+	executorID flow.Identifier
+}
+
+func NewUnknownRequiredExecutorError(executorID flow.Identifier) *UnknownRequiredExecutorError {
+	return &UnknownRequiredExecutorError{
+		executorID: executorID,
+	}
+}
+
+func (e *UnknownRequiredExecutorError) Error() string {
+	return fmt.Sprintf("unknown required executor ID: %s", e.executorID.String())
+}
+
+func IsUnknownRequiredExecutorError(err error) bool {
+	var unknownRequiredExecutor *UnknownRequiredExecutorError
+	return errors.As(err, &unknownRequiredExecutor)
+}
+
+// CriteriaNotMetError indicates that the execution result criteria could not be
+// satisfied for a given block, when the block is already sealed.
+type CriteriaNotMetError struct {
+	blockID flow.Identifier
+}
+
+func NewCriteriaNotMetError(blockID flow.Identifier) *CriteriaNotMetError {
+	return &CriteriaNotMetError{
+		blockID: blockID,
+	}
+}
+
+func (e *CriteriaNotMetError) Error() string {
+	return fmt.Sprintf("block %s is already sealed but the criteria is still not met,", e.blockID)
+}
+
+func IsCriteriaNotMetError(err error) bool {
+	var criteriaNotMetError *CriteriaNotMetError
+	return errors.As(err, &criteriaNotMetError)
+}

module/executiondatasync/optimistic_sync/execution_result/info_provider.go

@@ -20,6 +20,7 @@ type Provider struct {
 	log zerolog.Logger
 
 	executionReceipts storage.ExecutionReceipts
+	headers           storage.Headers
 	state             protocol.State
 	rootBlockID       flow.Identifier
 	executionNodes    *ExecutionNodeSelector
@@ -33,12 +34,14 @@ func NewExecutionResultInfoProvider(
 	log zerolog.Logger,
 	state protocol.State,
 	executionReceipts storage.ExecutionReceipts,
+	headers storage.Headers,
 	executionNodes *ExecutionNodeSelector,
 	operatorCriteria optimistic_sync.Criteria,
 ) *Provider {
 	return &Provider{
 		log:               log.With().Str("module", "execution_result_info").Logger(),
 		executionReceipts: executionReceipts,
+		headers:           headers,
 		state:             state,
 		executionNodes:    executionNodes,
 		rootBlockID:       state.Params().SporkRootBlock().ID(),
@@ -51,8 +54,9 @@ func NewExecutionResultInfoProvider(
 //
 // Expected errors during normal operations:
 //   - [common.InsufficientExecutionReceipts]: Found insufficient receipts for given block ID.
-//   - [storage.ErrNotFound]: If the request is for the spork root block and the node was bootstrapped
-//     from a newer block.
+//   - [storage.ErrNotFound]: If the data was not found.
+//   - [optimistic_sync.RequiredExecutorsCountExceededError]: Required executor IDs count exceeds available executors.
+//   - [optimistic_sync.UnknownRequiredExecutorError]: A required executor ID is not in the available set.
 func (e *Provider) ExecutionResultInfo(
 	blockID flow.Identifier,
 	criteria optimistic_sync.Criteria,
@@ -62,6 +66,11 @@ func (e *Provider) ExecutionResultInfo(
 		return nil, fmt.Errorf("failed to retrieve execution IDs: %w", err)
 	}
 
+	err = e.validateRequiredExecutors(criteria.RequiredExecutors, executorIdentities)
+	if err != nil {
+		return nil, fmt.Errorf("invalid required executors: %w", err)
+	}
+
 	// if the block ID is the root block, then use the root ExecutionResult and skip the receipt
 	// check since there will not be any.
 	if e.rootBlockID == blockID {
@@ -94,6 +103,19 @@ func (e *Provider) ExecutionResultInfo(
 		return nil, fmt.Errorf("failed to choose execution nodes for block ID %v: %w", blockID, err)
 	}
 
+	sealedHeader, err := e.state.Sealed().Head()
+	if err != nil {
+		return nil, fmt.Errorf("failed to lookup sealed header: %w", err)
+	}
+	header, err := e.headers.ByBlockID(blockID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get header by block ID %v: %w", blockID, err)
+	}
+	// If block is sealed and criteria cannot be met return an error
+	if header.Height <= sealedHeader.Height && len(subsetENs) < len(criteria.RequiredExecutors) {
+		return nil, optimistic_sync.NewCriteriaNotMetError(blockID)
+	}
+
 	if len(subsetENs) == 0 {
 		// this is unexpected, and probably indicates there is a bug.
 		// There are only three ways that SelectExecutionNodes can return an empty list:
@@ -112,6 +134,34 @@ func (e *Provider) ExecutionResultInfo(
 	}, nil
 }
 
+// validateRequiredExecutors verifies that the provided set of execution node
+// identities contains all nodes required for processing and that the requested
+// number of required executors does not exceed the available number.
+//
+// Expected errors during normal operations:
+//   - [common.RequiredExecutorsCountExceededError]: Required executor IDs count exceeds available executors.
+//   - [common.UnknownRequiredExecutorError]: A required executor ID is not in the available set.
+func (e *Provider) validateRequiredExecutors(
+	required flow.IdentifierList,
+	available flow.IdentityList,
+) error {
+	if len(available) < len(required) {
+		return optimistic_sync.NewRequiredExecutorsCountExceededError(
+			len(required),
+			len(available),
+		)
+	}
+
+	lookup := available.Lookup()
+	for _, executorID := range required {
+		if _, ok := lookup[executorID]; !ok {
+			return optimistic_sync.NewUnknownRequiredExecutorError(executorID)
+		}
+	}
+
+	return nil
+}
+
 // findResultAndExecutors returns a query response for a given block ID.
 // The result must match the provided criteria and have at least one acceptable executor. If multiple
 // results are found, then the result with the most executors is returned.