From 20b98a3e8a1fcd4929ca5bd672e3354cbd6fff81 Mon Sep 17 00:00:00 2001
From: Peter Argue <89119817+peterargue@users.noreply.github.com>
Date: Thu, 29 Aug 2024 17:30:42 +0000
Subject: [PATCH 1/5] Merge pull request #6239 from
 The-K-R-O-K/AndriiSlisarchuk/6128-metrics-for-tx-validator

[Access] Added metrics for transaction validator
---
 access/validator.go                           | 140 +++++++++---------
 access/validator_test.go                      |  10 +-
 .../node_builder/access_node_builder.go       |  86 ++++++-----
 engine/access/rpc/backend/backend.go          |  11 +-
 engine/collection/ingest/engine.go            |   1 +
 go.mod                                        |   2 +-
 go.sum                                        |   6 +-
 insecure/go.mod                               |   2 +-
 insecure/go.sum                               |   3 -
 integration/go.mod                            |   6 +-
 integration/go.sum                            |   8 +-
 module/metrics.go                             |  11 ++
 module/metrics/access.go                      |   7 +
 module/metrics/collection.go                  |   7 +-
 module/metrics/labels.go                      |  14 ++
 module/metrics/namespaces.go                  |   1 +
 module/metrics/noop.go                        |   4 +
 module/metrics/transaction_validation.go      |  59 ++++++++
 module/mock/access_metrics.go                 |  15 ++
 module/mock/collection_metrics.go             |  15 ++
 module/mock/transaction_validation_metrics.go |  39 +++++
 21 files changed, 318 insertions(+), 129 deletions(-)
 create mode 100644 module/metrics/transaction_validation.go
 create mode 100644 module/mock/transaction_validation_metrics.go

diff --git a/access/validator.go b/access/validator.go
index 61081897b2d..00d902d0c10 100644
--- a/access/validator.go
+++ b/access/validator.go
@@ -5,18 +5,21 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/rs/zerolog/log"
+
 	"github.com/onflow/cadence"
 	jsoncdc "github.com/onflow/cadence/encoding/json"
 	"github.com/onflow/cadence/runtime/parser"
 	"github.com/onflow/crypto"
 	"github.com/onflow/flow-core-contracts/lib/go/templates"
-	"github.com/rs/zerolog/log"
 
 	cadenceutils "github.com/onflow/flow-go/access/utils"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/fvm/systemcontracts"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/execution"
+	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/state"
 	"github.com/onflow/flow-go/state/protocol"
 )
@@ -87,19 +90,28 @@ type TransactionValidationOptions struct {
 	CheckPayerBalance            bool
 }
 
+type ValidationStep struct {
+	check      func(*flow.TransactionBody) error
+	failReason string
+}
+
 type TransactionValidator struct {
-	blocks                   Blocks     // for looking up blocks to check transaction expiry
-	chain                    flow.Chain // for checking validity of addresses
-	options                  TransactionValidationOptions
-	serviceAccountAddress    flow.Address
-	limiter                  RateLimiter
-	scriptExecutor           execution.ScriptExecutor
-	verifyPayerBalanceScript []byte
+	blocks                       Blocks     // for looking up blocks to check transaction expiry
+	chain                        flow.Chain // for checking validity of addresses
+	options                      TransactionValidationOptions
+	serviceAccountAddress        flow.Address
+	limiter                      RateLimiter
+	scriptExecutor               execution.ScriptExecutor
+	verifyPayerBalanceScript     []byte
+	transactionValidationMetrics module.TransactionValidationMetrics
+
+	validationSteps []ValidationStep
 }
 
 func NewTransactionValidator(
 	blocks Blocks,
 	chain flow.Chain,
+	transactionValidationMetrics module.TransactionValidationMetrics,
 	options TransactionValidationOptions,
 	executor execution.ScriptExecutor,
 ) (*TransactionValidator, error) {
@@ -109,80 +121,68 @@ func NewTransactionValidator(
 
 	env := systemcontracts.SystemContractsForChain(chain.ChainID()).AsTemplateEnv()
 
-	return &TransactionValidator{
-		blocks:                   blocks,
-		chain:                    chain,
-		options:                  options,
-		serviceAccountAddress:    chain.ServiceAddress(),
-		limiter:                  NewNoopLimiter(),
-		scriptExecutor:           executor,
-		verifyPayerBalanceScript: templates.GenerateVerifyPayerBalanceForTxExecution(env),
-	}, nil
+	txValidator := &TransactionValidator{
+		blocks:                       blocks,
+		chain:                        chain,
+		options:                      options,
+		serviceAccountAddress:        chain.ServiceAddress(),
+		limiter:                      NewNoopLimiter(),
+		scriptExecutor:               executor,
+		verifyPayerBalanceScript:     templates.GenerateVerifyPayerBalanceForTxExecution(env),
+		transactionValidationMetrics: transactionValidationMetrics,
+	}
+
+	txValidator.initValidationSteps()
+
+	return txValidator, nil
 }
 
 func NewTransactionValidatorWithLimiter(
 	blocks Blocks,
 	chain flow.Chain,
 	options TransactionValidationOptions,
+	transactionValidationMetrics module.TransactionValidationMetrics,
 	rateLimiter RateLimiter,
 ) *TransactionValidator {
-	return &TransactionValidator{
-		blocks:                blocks,
-		chain:                 chain,
-		options:               options,
-		serviceAccountAddress: chain.ServiceAddress(),
-		limiter:               rateLimiter,
-	}
-}
-
-func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.TransactionBody) (err error) {
-	// rate limit transactions for specific payers.
-	// a short term solution to prevent attacks that send too many failed transactions
-	// if a transaction is from a payer that should be rate limited, all the following
-	// checks will be skipped
-	err = v.checkRateLimitPayer(tx)
-	if err != nil {
-		return err
-	}
-
-	err = v.checkTxSizeLimit(tx)
-	if err != nil {
-		return err
-	}
-
-	err = v.checkMissingFields(tx)
-	if err != nil {
-		return err
+	txValidator := &TransactionValidator{
+		blocks:                       blocks,
+		chain:                        chain,
+		options:                      options,
+		serviceAccountAddress:        chain.ServiceAddress(),
+		limiter:                      rateLimiter,
+		transactionValidationMetrics: transactionValidationMetrics,
 	}
 
-	err = v.checkGasLimit(tx)
-	if err != nil {
-		return err
-	}
-
-	err = v.checkExpiry(tx)
-	if err != nil {
-		return err
-	}
+	txValidator.initValidationSteps()
 
-	err = v.checkCanBeParsed(tx)
-	if err != nil {
-		return err
-	}
+	return txValidator
+}
 
-	err = v.checkAddresses(tx)
-	if err != nil {
-		return err
+func (v *TransactionValidator) initValidationSteps() {
+	v.validationSteps = []ValidationStep{
+		// rate limit transactions for specific payers.
+		// a short term solution to prevent attacks that send too many failed transactions
+		// if a transaction is from a payer that should be rate limited, all the following
+		// checks will be skipped
+		{v.checkRateLimitPayer, metrics.InvalidTransactionRateLimit},
+		{v.checkTxSizeLimit, metrics.InvalidTransactionByteSize},
+		{v.checkMissingFields, metrics.IncompleteTransaction},
+		{v.checkGasLimit, metrics.InvalidGasLimit},
+		{v.checkExpiry, metrics.ExpiredTransaction},
+		{v.checkCanBeParsed, metrics.InvalidScript},
+		{v.checkAddresses, metrics.InvalidAddresses},
+		{v.checkSignatureFormat, metrics.InvalidSignature},
+		{v.checkSignatureDuplications, metrics.DuplicatedSignature},
 	}
+}
 
-	err = v.checkSignatureFormat(tx)
-	if err != nil {
-		return err
-	}
+func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.TransactionBody) (err error) {
 
-	err = v.checkSignatureDuplications(tx)
-	if err != nil {
-		return err
+	for _, step := range v.validationSteps {
+		if err = step.check(tx); err != nil {
+			v.transactionValidationMetrics.TransactionValidationFailed(step.failReason)
+			return err
+		}
 	}
 
 	err = v.checkSufficientBalanceToPayForTransaction(ctx, tx)
@@ -192,15 +192,19 @@ func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.Transactio
 		// are 'internal' and related to script execution process. they shouldn't
 		// prevent the transaction from proceeding.
 		if IsInsufficientBalanceError(err) {
+			v.transactionValidationMetrics.TransactionValidationFailed(metrics.InsufficientBalance)
 			return err
 		}
 
 		// log and ignore all other errors
+		v.transactionValidationMetrics.TransactionValidationSkipped()
 		log.Info().Err(err).Msg("check payer validation skipped due to error")
 	}
 
 	// TODO replace checkSignatureFormat by verifying the account/payer signatures
 
+	v.transactionValidationMetrics.TransactionValidated()
+
 	return nil
 }
 
@@ -337,7 +341,6 @@ func (v *TransactionValidator) checkCanBeParsed(tx *flow.TransactionBody) (err e
 }
 
 func (v *TransactionValidator) checkAddresses(tx *flow.TransactionBody) error {
-
 	for _, address := range append(tx.Authorizers, tx.Payer) {
 		// we check whether this is a valid output of the address generator
 		if !v.chain.IsValid(address) {
@@ -365,7 +368,6 @@ func (v *TransactionValidator) checkSignatureDuplications(tx *flow.TransactionBo
 }
 
 func (v *TransactionValidator) checkSignatureFormat(tx *flow.TransactionBody) error {
-
 	for _, signature := range append(tx.PayloadSignatures, tx.EnvelopeSignatures...) {
 		// check the format of the signature is valid.
 		// a valid signature is an ECDSA signature of either P-256 or secp256k1 curve.
diff --git a/access/validator_test.go b/access/validator_test.go
index 1554a91cfcb..87b7ade62a9 100644
--- a/access/validator_test.go
+++ b/access/validator_test.go
@@ -17,7 +17,9 @@ import (
 	accessmock "github.com/onflow/flow-go/access/mock"
 	"github.com/onflow/flow-go/fvm"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module"
 	execmock "github.com/onflow/flow-go/module/execution/mock"
+	"github.com/onflow/flow-go/module/metrics"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -31,9 +33,11 @@ type TransactionValidatorSuite struct {
 	header           *flow.Header
 	chain            flow.Chain
 	validatorOptions access.TransactionValidationOptions
+	metrics          module.TransactionValidationMetrics
 }
 
 func (s *TransactionValidatorSuite) SetupTest() {
+	s.metrics = metrics.NewNoopCollector()
 	s.blocks = accessmock.NewBlocks(s.T())
 	assert.NotNil(s.T(), s.blocks)
 
@@ -89,7 +93,7 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_ScriptExecutorInter
 		Return(nil, errors.New("script executor internal error")).
 		Once()
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.validatorOptions, scriptExecutor)
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), validator)
 
@@ -116,7 +120,7 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_SufficientBalance()
 		Return(actualResponse, nil).
 		Once()
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.validatorOptions, scriptExecutor)
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), validator)
 
@@ -147,7 +151,7 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_InsufficientBalance
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), actualAccountResponse)
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.validatorOptions, scriptExecutor)
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), validator)
 
diff --git a/cmd/access/node_builder/access_node_builder.go b/cmd/access/node_builder/access_node_builder.go
index b31ea749060..469c1f0462f 100644
--- a/cmd/access/node_builder/access_node_builder.go
+++ b/cmd/access/node_builder/access_node_builder.go
@@ -287,43 +287,44 @@ type FlowAccessNodeBuilder struct {
 	*AccessNodeConfig
 
 	// components
-	FollowerState              protocol.FollowerState
-	SyncCore                   *chainsync.Core
-	RpcEng                     *rpc.Engine
-	FollowerDistributor        *consensuspubsub.FollowerDistributor
-	CollectionRPC              access.AccessAPIClient
-	TransactionTimings         *stdmap.TransactionTimings
-	CollectionsToMarkFinalized *stdmap.Times
-	CollectionsToMarkExecuted  *stdmap.Times
-	BlocksToMarkExecuted       *stdmap.Times
-	TransactionMetrics         *metrics.TransactionCollector
-	RestMetrics                *metrics.RestCollector
-	AccessMetrics              module.AccessMetrics
-	PingMetrics                module.PingMetrics
-	Committee                  hotstuff.DynamicCommittee
-	Finalized                  *flow.Header // latest finalized block that the node knows of at startup time
-	Pending                    []*flow.Header
-	FollowerCore               module.HotStuffFollower
-	Validator                  hotstuff.Validator
-	ExecutionDataDownloader    execution_data.Downloader
-	PublicBlobService          network.BlobService
-	ExecutionDataRequester     state_synchronization.ExecutionDataRequester
-	ExecutionDataStore         execution_data.ExecutionDataStore
-	ExecutionDataBlobstore     blobs.Blobstore
-	ExecutionDataCache         *execdatacache.ExecutionDataCache
-	ExecutionIndexer           *indexer.Indexer
-	ExecutionIndexerCore       *indexer.IndexerCore
-	ScriptExecutor             *backend.ScriptExecutor
-	RegistersAsyncStore        *execution.RegistersAsyncStore
-	Reporter                   *index.Reporter
-	EventsIndex                *index.EventsIndex
-	TxResultsIndex             *index.TransactionResultsIndex
-	IndexerDependencies        *cmd.DependencyList
-	collectionExecutedMetric   module.CollectionExecutedMetric
-	ExecutionDataPruner        *pruner.Pruner
-	ExecutionDatastoreManager  edstorage.DatastoreManager
-	ExecutionDataTracker       tracker.Storage
-	versionControl             *version.VersionControl
+	FollowerState                protocol.FollowerState
+	SyncCore                     *chainsync.Core
+	RpcEng                       *rpc.Engine
+	FollowerDistributor          *consensuspubsub.FollowerDistributor
+	CollectionRPC                access.AccessAPIClient
+	TransactionTimings           *stdmap.TransactionTimings
+	CollectionsToMarkFinalized   *stdmap.Times
+	CollectionsToMarkExecuted    *stdmap.Times
+	BlocksToMarkExecuted         *stdmap.Times
+	TransactionMetrics           *metrics.TransactionCollector
+	TransactionValidationMetrics *metrics.TransactionValidationCollector
+	RestMetrics                  *metrics.RestCollector
+	AccessMetrics                module.AccessMetrics
+	PingMetrics                  module.PingMetrics
+	Committee                    hotstuff.DynamicCommittee
+	Finalized                    *flow.Header // latest finalized block that the node knows of at startup time
+	Pending                      []*flow.Header
+	FollowerCore                 module.HotStuffFollower
+	Validator                    hotstuff.Validator
+	ExecutionDataDownloader      execution_data.Downloader
+	PublicBlobService            network.BlobService
+	ExecutionDataRequester       state_synchronization.ExecutionDataRequester
+	ExecutionDataStore           execution_data.ExecutionDataStore
+	ExecutionDataBlobstore       blobs.Blobstore
+	ExecutionDataCache           *execdatacache.ExecutionDataCache
+	ExecutionIndexer             *indexer.Indexer
+	ExecutionIndexerCore         *indexer.IndexerCore
+	ScriptExecutor               *backend.ScriptExecutor
+	RegistersAsyncStore          *execution.RegistersAsyncStore
+	Reporter                     *index.Reporter
+	EventsIndex                  *index.EventsIndex
+	TxResultsIndex               *index.TransactionResultsIndex
+	IndexerDependencies          *cmd.DependencyList
+	collectionExecutedMetric     module.CollectionExecutedMetric
+	ExecutionDataPruner          *pruner.Pruner
+	ExecutionDatastoreManager    edstorage.DatastoreManager
+	ExecutionDataTracker         tracker.Storage
+	VersionControl               *version.VersionControl
 
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
@@ -978,7 +979,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					builder.programCacheSize > 0,
 				)
 
-				err = builder.ScriptExecutor.Initialize(builder.ExecutionIndexer, scripts, builder.versionControl)
+				err = builder.ScriptExecutor.Initialize(builder.ExecutionIndexer, scripts, builder.VersionControl)
 				if err != nil {
 					return nil, err
 				}
@@ -1665,6 +1666,10 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			)
 			return nil
 		}).
+		Module("transaction validation metrics", func(node *cmd.NodeConfig) error {
+			builder.TransactionValidationMetrics = metrics.NewTransactionValidationCollector()
+			return nil
+		}).
 		Module("rest metrics", func(node *cmd.NodeConfig) error {
 			m, err := metrics.NewRestCollector(routes.URLToRoute, node.MetricsRegisterer)
 			if err != nil {
@@ -1676,6 +1681,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		Module("access metrics", func(node *cmd.NodeConfig) error {
 			builder.AccessMetrics = metrics.NewAccessCollector(
 				metrics.WithTransactionMetrics(builder.TransactionMetrics),
+				metrics.WithTransactionValidationMetrics(builder.TransactionValidationMetrics),
 				metrics.WithBackendScriptsMetrics(builder.TransactionMetrics),
 				metrics.WithRestMetrics(builder.RestMetrics),
 			)
@@ -1813,8 +1819,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			// VersionControl needs to consume BlockFinalized events.
 			node.ProtocolEvents.AddConsumer(versionControl)
 
-			builder.versionControl = versionControl
-			versionControlDependable.Init(builder.versionControl)
+			builder.VersionControl = versionControl
+			versionControlDependable.Init(builder.VersionControl)
 
 			return versionControl, nil
 		}).
diff --git a/engine/access/rpc/backend/backend.go b/engine/access/rpc/backend/backend.go
index 3af46045697..d250b078d6b 100644
--- a/engine/access/rpc/backend/backend.go
+++ b/engine/access/rpc/backend/backend.go
@@ -246,7 +246,7 @@ func New(params Params) (*Backend, error) {
 		nodeInfo:          nodeInfo,
 	}
 
-	txValidator, err := configureTransactionValidator(params.State, params.ChainID, params.ScriptExecutor, params.CheckPayerBalance)
+	txValidator, err := configureTransactionValidator(params.State, params.ChainID, params.AccessMetrics, params.ScriptExecutor, params.CheckPayerBalance)
 	if err != nil {
 		return nil, fmt.Errorf("could not create transaction validator: %w", err)
 	}
@@ -310,10 +310,17 @@ func identifierList(ids []string) (flow.IdentifierList, error) {
 	return idList, nil
 }
 
-func configureTransactionValidator(state protocol.State, chainID flow.ChainID, executor execution.ScriptExecutor, checkPayerBalance bool) (*access.TransactionValidator, error) {
+func configureTransactionValidator(
+	state protocol.State,
+	chainID flow.ChainID,
+	transactionMetrics module.TransactionValidationMetrics,
+	executor execution.ScriptExecutor,
+	checkPayerBalance bool,
+) (*access.TransactionValidator, error) {
 	return access.NewTransactionValidator(
 		access.NewProtocolStateBlocks(state),
 		chainID.Chain(),
+		transactionMetrics,
 		access.TransactionValidationOptions{
 			Expiry:                       flow.DefaultTransactionExpiry,
 			ExpiryBuffer:                 flow.DefaultTransactionExpiryBuffer,
diff --git a/engine/collection/ingest/engine.go b/engine/collection/ingest/engine.go
index a8adcedaded..438e6bea88b 100644
--- a/engine/collection/ingest/engine.go
+++ b/engine/collection/ingest/engine.go
@@ -71,6 +71,7 @@ func New(
 			MaxTransactionByteSize: config.MaxTransactionByteSize,
 			MaxCollectionByteSize:  config.MaxCollectionByteSize,
 		},
+		colMetrics,
 		limiter,
 	)
 
diff --git a/go.mod b/go.mod
index 0db9f4ca1c4..670a32ff780 100644
--- a/go.mod
+++ b/go.mod
@@ -255,13 +255,13 @@ require (
 	github.com/multiformats/go-multicodec v0.9.0 // indirect
 	github.com/multiformats/go-multistream v0.5.0 // indirect
 	github.com/multiformats/go-varint v0.0.7 // indirect
-	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/onflow/flow-ft/lib/go/contracts v1.0.0 // indirect
 	github.com/onflow/flow-ft/lib/go/templates v1.0.0 // indirect
 	github.com/onflow/flow-nft/lib/go/contracts v1.2.1 // indirect
 	github.com/onflow/flow-nft/lib/go/templates v1.2.0 // indirect
 	github.com/onflow/sdks v0.6.0-preview.1 // indirect
+	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/onsi/ginkgo/v2 v2.13.2 // indirect
 	github.com/opencontainers/runtime-spec v1.1.0 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
diff --git a/go.sum b/go.sum
index 8a68e9ac0a1..2086a0aac3a 100644
--- a/go.sum
+++ b/go.sum
@@ -1480,6 +1480,7 @@ github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG
 github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/go-stack/stack v1.8.1/go.mod h1:dcoOX6HbPZSZptuspn9bctJ+N/CnF5gGygcUP3XYfe4=
+github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls=
 github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
@@ -2209,8 +2210,9 @@ github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+W
 github.com/onsi/ginkgo v1.10.3/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
 github.com/onsi/ginkgo v1.13.0/go.mod h1:+REjRxOmWfHCjfv9TTWB1jD1Frx4XydAD3zm1lskyM0=
-github.com/onsi/ginkgo v1.14.0 h1:2mOpI4JVVPBN+WQRa0WKH2eXR+Ey+uK4n7Zj0aYpIQA=
 github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
+github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
+github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/ginkgo/v2 v2.13.2 h1:Bi2gGVkfn6gQcjNjZJVO8Gf0FHzMPf2phUei9tejVMs=
 github.com/onsi/ginkgo/v2 v2.13.2/go.mod h1:XStQ8QcGwLyF4HdfcZB8SFOS/MWCgDuXMSBe6zrvLgM=
 github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
@@ -2991,6 +2993,7 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210104204734-6f8348627aad/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210220050731-9a76102bfb43/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -3181,6 +3184,7 @@ golang.org/x/tools v0.0.0-20201110124207-079ba7bd75cd/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.0.0-20201124115921-2c860bdd6e78/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201201161351-ac6f37ff4c2a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201208233053-a543418bbed2/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
diff --git a/insecure/go.mod b/insecure/go.mod
index 32271beb2d9..3e9c3314617 100644
--- a/insecure/go.mod
+++ b/insecure/go.mod
@@ -200,6 +200,7 @@ require (
 	github.com/multiformats/go-multihash v0.2.3 // indirect
 	github.com/multiformats/go-multistream v0.5.0 // indirect
 	github.com/multiformats/go-varint v0.0.7 // indirect
+	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/onflow/atree v0.8.0-rc.6 // indirect
 	github.com/onflow/cadence v1.0.1 // indirect
@@ -214,7 +215,6 @@ require (
 	github.com/onflow/go-ethereum v1.14.7 // indirect
 	github.com/onflow/sdks v0.6.0-preview.1 // indirect
 	github.com/onflow/wal v1.0.2 // indirect
-	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/onsi/ginkgo/v2 v2.13.2 // indirect
 	github.com/opencontainers/runtime-spec v1.1.0 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
diff --git a/insecure/go.sum b/insecure/go.sum
index c2cd45ea6c8..a1e8406ede6 100644
--- a/insecure/go.sum
+++ b/insecure/go.sum
@@ -1475,7 +1475,6 @@ github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG
 github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/go-stack/stack v1.8.1/go.mod h1:dcoOX6HbPZSZptuspn9bctJ+N/CnF5gGygcUP3XYfe4=
-github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI=
 github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls=
 github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
@@ -2977,7 +2976,6 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210104204734-6f8348627aad/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210220050731-9a76102bfb43/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -3168,7 +3166,6 @@ golang.org/x/tools v0.0.0-20201110124207-079ba7bd75cd/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.0.0-20201124115921-2c860bdd6e78/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201201161351-ac6f37ff4c2a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20201208233053-a543418bbed2/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
-golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
diff --git a/integration/go.mod b/integration/go.mod
index adb7e341e90..4b956e824c3 100644
--- a/integration/go.mod
+++ b/integration/go.mod
@@ -24,11 +24,11 @@ require (
 	github.com/onflow/crypto v0.25.2
 	github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1
 	github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1
-	github.com/onflow/flow-emulator v1.0.0-preview.36.0.20240729195722-d4eb1c30eb9f
-	github.com/onflow/flow-go v0.36.8-0.20240729193633-433a32eeb0cd
+	github.com/onflow/flow-emulator v1.0.0-preview.41.0.20240829134601-0be55d6970b5
+	github.com/onflow/flow-go v0.37.7-0.20240826193109-e211841b59f5
 	github.com/onflow/flow-go-sdk v1.0.0-preview.54
 	github.com/onflow/flow-go/insecure v0.0.0-00010101000000-000000000000
-	github.com/onflow/flow/protobuf/go/flow v0.4.5
+	github.com/onflow/flow/protobuf/go/flow v0.4.6
 	github.com/onflow/go-ethereum v1.14.7
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/client_model v0.5.0
diff --git a/integration/go.sum b/integration/go.sum
index a67854cbe5b..df6b01ed2a5 100644
--- a/integration/go.sum
+++ b/integration/go.sum
@@ -2151,8 +2151,8 @@ github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1 h1:q9tXLIALwQ76bO4
 github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1/go.mod h1:u/mkP/B+PbV33tEG3qfkhhBlydSvAKxfLZSfB4lsJHg=
 github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1 h1:FfhMBAb78p6VAWkJ+iqdKLErGQVQgxk5w6DP5ZruWX8=
 github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1/go.mod h1:NgbMOYnMh0GN48VsNKZuiwK7uyk38Wyo8jN9+C9QE30=
-github.com/onflow/flow-emulator v1.0.0-preview.36.0.20240729195722-d4eb1c30eb9f h1:2Ejpmm2Vrl/XLaE6lniE1vNfi6WYhzqHiCk6oomGoFE=
-github.com/onflow/flow-emulator v1.0.0-preview.36.0.20240729195722-d4eb1c30eb9f/go.mod h1:0rqp896zEcjNqqDiQNBUlpS/7nzS4+E+yG/4s0P13bQ=
+github.com/onflow/flow-emulator v1.0.0-preview.41.0.20240829134601-0be55d6970b5 h1:Z5PC3Sqvl2UemY27uwUwzkLb8EXUf+m/aEimxFzOXuc=
+github.com/onflow/flow-emulator v1.0.0-preview.41.0.20240829134601-0be55d6970b5/go.mod h1:gFafyGD4Qxs+XT2BRSIjQJ86ILSmgm1VYUoCr1nVxVI=
 github.com/onflow/flow-ft/lib/go/contracts v1.0.0 h1:mToacZ5NWqtlWwk/7RgIl/jeKB/Sy/tIXdw90yKHcV0=
 github.com/onflow/flow-ft/lib/go/contracts v1.0.0/go.mod h1:PwsL8fC81cjnUnTfmyL/HOIyHnyaw/JA474Wfj2tl6A=
 github.com/onflow/flow-ft/lib/go/templates v1.0.0 h1:6cMS/lUJJ17HjKBfMO/eh0GGvnpElPgBXx7h5aoWJhs=
@@ -2165,8 +2165,8 @@ github.com/onflow/flow-nft/lib/go/contracts v1.2.1/go.mod h1:2gpbza+uzs1k7x31hkp
 github.com/onflow/flow-nft/lib/go/templates v1.2.0 h1:JSQyh9rg0RC+D1930BiRXN8lrtMs+ubVMK6aQPon6Yc=
 github.com/onflow/flow-nft/lib/go/templates v1.2.0/go.mod h1:p+2hRvtjLUR3MW1NsoJe5Gqgr2eeH49QB6+s6ze00w0=
 github.com/onflow/flow/protobuf/go/flow v0.3.2-0.20231121210617-52ee94b830c2/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
-github.com/onflow/flow/protobuf/go/flow v0.4.5 h1:6o+pgYGqwXdEhqSJxu2BdnDXkOQVOkfGAb6IiXB+NPM=
-github.com/onflow/flow/protobuf/go/flow v0.4.5/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
+github.com/onflow/flow/protobuf/go/flow v0.4.6 h1:KE/CsRVfyG5lGBtm1aNcjojMciQyS5GfPF3ixOWRfi0=
+github.com/onflow/flow/protobuf/go/flow v0.4.6/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c h1:T0jDCm7k7uqDo26JiiujQ5oryl30itPnlmZQywTu9ng=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c/go.mod h1:XYnWtulwJvHVOr2B0WVA/UC3dvRgFevjp8Pn9a3E1xo=
 github.com/onflow/go-ethereum v1.14.7 h1:gg3awYqI02e3AypRdpJKEvNTJ6kz/OhAqRti0h54Wlc=
diff --git a/module/metrics.go b/module/metrics.go
index cfad9bb11a0..f43c8b9325e 100644
--- a/module/metrics.go
+++ b/module/metrics.go
@@ -607,6 +607,7 @@ type CruiseCtlMetrics interface {
 }
 
 type CollectionMetrics interface {
+	TransactionValidationMetrics
 	// TransactionIngested is called when a new transaction is ingested by the
 	// node. It increments the total count of ingested transactions and starts
 	// a tx->col span for the transaction.
@@ -907,6 +908,7 @@ type AccessMetrics interface {
 	RestMetrics
 	GRPCConnectionPoolMetrics
 	TransactionMetrics
+	TransactionValidationMetrics
 	BackendScriptsMetrics
 
 	// UpdateExecutionReceiptMaxHeight is called whenever we store an execution receipt from a block from a newer height
@@ -1085,6 +1087,15 @@ type TransactionMetrics interface {
 	TransactionSubmissionFailed()
 }
 
+type TransactionValidationMetrics interface {
+	// TransactionValidated tracks number of successfully validated transactions
+	TransactionValidated()
+	// TransactionValidationFailed tracks number of validation failed transactions with reason
+	TransactionValidationFailed(reason string)
+	// TransactionValidationSkipped tracks number of skipped transaction validations
+	TransactionValidationSkipped()
+}
+
 type PingMetrics interface {
 	// NodeReachable tracks the round trip time in milliseconds taken to ping a node
 	// The nodeInfo provides additional information about the node such as the name of the node operator
diff --git a/module/metrics/access.go b/module/metrics/access.go
index 1116f87f433..aacfe316c76 100644
--- a/module/metrics/access.go
+++ b/module/metrics/access.go
@@ -10,6 +10,12 @@ import (
 
 type AccessCollectorOpts func(*AccessCollector)
 
+func WithTransactionValidationMetrics(m module.TransactionValidationMetrics) AccessCollectorOpts {
+	return func(ac *AccessCollector) {
+		ac.TransactionValidationMetrics = m
+	}
+}
+
 func WithTransactionMetrics(m module.TransactionMetrics) AccessCollectorOpts {
 	return func(ac *AccessCollector) {
 		ac.TransactionMetrics = m
@@ -31,6 +37,7 @@ func WithRestMetrics(m module.RestMetrics) AccessCollectorOpts {
 type AccessCollector struct {
 	module.RestMetrics
 	module.TransactionMetrics
+	module.TransactionValidationMetrics
 	module.BackendScriptsMetrics
 
 	connectionReused      prometheus.Counter
diff --git a/module/metrics/collection.go b/module/metrics/collection.go
index fa97a28b679..557eab48b77 100644
--- a/module/metrics/collection.go
+++ b/module/metrics/collection.go
@@ -10,6 +10,7 @@ import (
 )
 
 type CollectionCollector struct {
+	module.TransactionValidationMetrics
 	tracer               module.Tracer
 	transactionsIngested prometheus.Counter       // tracks the number of ingested transactions
 	finalizedHeight      *prometheus.GaugeVec     // tracks the finalized height
@@ -17,11 +18,13 @@ type CollectionCollector struct {
 	guarantees           *prometheus.HistogramVec // counts the number/size of FINALIZED collections
 }
 
+var _ module.CollectionMetrics = (*CollectionCollector)(nil)
+
 func NewCollectionCollector(tracer module.Tracer) *CollectionCollector {
 
 	cc := &CollectionCollector{
-		tracer: tracer,
-
+		TransactionValidationMetrics: NewTransactionValidationCollector(),
+		tracer:                       tracer,
 		transactionsIngested: promauto.NewCounter(prometheus.CounterOpts{
 			Namespace: namespaceCollection,
 			Name:      "ingested_transactions_total",
diff --git a/module/metrics/labels.go b/module/metrics/labels.go
index 916f3e2bc07..82260ca3c5d 100644
--- a/module/metrics/labels.go
+++ b/module/metrics/labels.go
@@ -155,6 +155,20 @@ const (
 	MessageEntityResponse      = "entity_response"
 )
 
+// transaction validation labels
+const (
+	InvalidTransactionRateLimit = "payer_exceeded_rate_limit"
+	InvalidTransactionByteSize  = "transaction_exceeded_size_limit"
+	IncompleteTransaction       = "missing_fields"
+	InvalidGasLimit             = "invalid_gas_limit"
+	ExpiredTransaction          = "transaction_expired"
+	InvalidScript               = "invalid_script"
+	InvalidAddresses            = "invalid_address"
+	InvalidSignature            = "invalid_signature"
+	DuplicatedSignature         = "duplicate_signature"
+	InsufficientBalance         = "payer_insufficient_balance"
+)
+
 const ExecutionDataRequestRetryable = "retryable"
 
 const LabelViolationReason = "reason"
diff --git a/module/metrics/namespaces.go b/module/metrics/namespaces.go
index a35885db7dc..c0f99af3fcf 100644
--- a/module/metrics/namespaces.go
+++ b/module/metrics/namespaces.go
@@ -44,6 +44,7 @@ const (
 const (
 	subsystemTransactionTiming     = "transaction_timing"
 	subsystemTransactionSubmission = "transaction_submission"
+	subsystemTransactionValidation = "transaction_validation"
 	subsystemConnectionPool        = "connection_pool"
 	subsystemHTTP                  = "http"
 )
diff --git a/module/metrics/noop.go b/module/metrics/noop.go
index fb5ceeeed81..17460bf460a 100644
--- a/module/metrics/noop.go
+++ b/module/metrics/noop.go
@@ -37,6 +37,7 @@ func (nc *NoopCollector) BlockProposalDuration(duration time.Duration)
 // interface check
 var _ module.BackendScriptsMetrics = (*NoopCollector)(nil)
 var _ module.TransactionMetrics = (*NoopCollector)(nil)
+var _ module.TransactionValidationMetrics = (*NoopCollector)(nil)
 var _ module.HotstuffMetrics = (*NoopCollector)(nil)
 var _ module.EngineMetrics = (*NoopCollector)(nil)
 var _ module.HeroCacheMetrics = (*NoopCollector)(nil)
@@ -215,6 +216,9 @@ func (nc *NoopCollector) TransactionReceived(txID flow.Identifier, when time.Tim
 func (nc *NoopCollector) TransactionFinalized(txID flow.Identifier, when time.Time)             {}
 func (nc *NoopCollector) TransactionExecuted(txID flow.Identifier, when time.Time)              {}
 func (nc *NoopCollector) TransactionExpired(txID flow.Identifier)                               {}
+func (nc *NoopCollector) TransactionValidated()                                                 {}
+func (nc *NoopCollector) TransactionValidationFailed(reason string)                             {}
+func (nc *NoopCollector) TransactionValidationSkipped()                                         {}
 func (nc *NoopCollector) TransactionSubmissionFailed()                                          {}
 func (nc *NoopCollector) UpdateExecutionReceiptMaxHeight(height uint64)                         {}
 func (nc *NoopCollector) UpdateLastFullBlockHeight(height uint64)                               {}
diff --git a/module/metrics/transaction_validation.go b/module/metrics/transaction_validation.go
new file mode 100644
index 00000000000..70cfe2d7a6a
--- /dev/null
+++ b/module/metrics/transaction_validation.go
@@ -0,0 +1,59 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/onflow/flow-go/module"
+)
+
+type NamespaceType string
+
+// TransactionValidationCollector the metrics for transaction validation functionality
+type TransactionValidationCollector struct {
+	transactionValidated         prometheus.Counter
+	transactionValidationSkipped prometheus.Counter
+	transactionValidationFailed  *prometheus.CounterVec
+}
+
+// interface check
+var _ module.TransactionValidationMetrics = (*TransactionValidationCollector)(nil)
+
+// NewTransactionValidationCollector creates new instance of TransactionValidationCollector
+func NewTransactionValidationCollector() *TransactionValidationCollector {
+	return &TransactionValidationCollector{
+		transactionValidated: promauto.NewCounter(prometheus.CounterOpts{
+			Name:      "transaction_validation_successes_total",
+			Namespace: namespaceAccess,
+			Subsystem: subsystemTransactionValidation,
+			Help:      "counter for the validated transactions",
+		}),
+		transactionValidationSkipped: promauto.NewCounter(prometheus.CounterOpts{
+			Name:      "transaction_validation_skipped_total",
+			Namespace: namespaceAccess,
+			Subsystem: subsystemTransactionValidation,
+			Help:      "counter for the skipped transaction validations",
+		}),
+		transactionValidationFailed: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name:      "transaction_validation_failed_total",
+			Namespace: namespaceAccess,
+			Subsystem: subsystemTransactionValidation,
+			Help:      "counter for the failed transactions validation",
+		}, []string{"reason"}),
+	}
+}
+
+// TransactionValidated tracks number of successfully validated transactions
+func (tc *TransactionValidationCollector) TransactionValidated() {
+	tc.transactionValidated.Inc()
+}
+
+// TransactionValidationFailed tracks number of validation failed transactions with reason
+func (tc *TransactionValidationCollector) TransactionValidationFailed(reason string) {
+	tc.transactionValidationFailed.WithLabelValues(reason).Inc()
+}
+
+// TransactionValidationSkipped tracks number of skipped transaction validations
+func (tc *TransactionValidationCollector) TransactionValidationSkipped() {
+	tc.transactionValidationSkipped.Inc()
+}
diff --git a/module/mock/access_metrics.go b/module/mock/access_metrics.go
index 9866ad90e02..21ecc03740f 100644
--- a/module/mock/access_metrics.go
+++ b/module/mock/access_metrics.go
@@ -143,6 +143,21 @@ func (_m *AccessMetrics) TransactionSubmissionFailed() {
 	_m.Called()
 }
 
+// TransactionValidated provides a mock function with given fields:
+func (_m *AccessMetrics) TransactionValidated() {
+	_m.Called()
+}
+
+// TransactionValidationFailed provides a mock function with given fields: reason
+func (_m *AccessMetrics) TransactionValidationFailed(reason string) {
+	_m.Called(reason)
+}
+
+// TransactionValidationSkipped provides a mock function with given fields:
+func (_m *AccessMetrics) TransactionValidationSkipped() {
+	_m.Called()
+}
+
 // UpdateExecutionReceiptMaxHeight provides a mock function with given fields: height
 func (_m *AccessMetrics) UpdateExecutionReceiptMaxHeight(height uint64) {
 	_m.Called(height)
diff --git a/module/mock/collection_metrics.go b/module/mock/collection_metrics.go
index bde071e902e..34c5e7efee5 100644
--- a/module/mock/collection_metrics.go
+++ b/module/mock/collection_metrics.go
@@ -29,6 +29,21 @@ func (_m *CollectionMetrics) TransactionIngested(txID flow.Identifier) {
 	_m.Called(txID)
 }
 
+// TransactionValidated provides a mock function with given fields:
+func (_m *CollectionMetrics) TransactionValidated() {
+	_m.Called()
+}
+
+// TransactionValidationFailed provides a mock function with given fields: reason
+func (_m *CollectionMetrics) TransactionValidationFailed(reason string) {
+	_m.Called(reason)
+}
+
+// TransactionValidationSkipped provides a mock function with given fields:
+func (_m *CollectionMetrics) TransactionValidationSkipped() {
+	_m.Called()
+}
+
 // NewCollectionMetrics creates a new instance of CollectionMetrics. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewCollectionMetrics(t interface {
diff --git a/module/mock/transaction_validation_metrics.go b/module/mock/transaction_validation_metrics.go
new file mode 100644
index 00000000000..0003f6762f7
--- /dev/null
+++ b/module/mock/transaction_validation_metrics.go
@@ -0,0 +1,39 @@
+// Code generated by mockery v2.43.2. DO NOT EDIT.
+
+package mock
+
+import mock "github.com/stretchr/testify/mock"
+
+// TransactionValidationMetrics is an autogenerated mock type for the TransactionValidationMetrics type
+type TransactionValidationMetrics struct {
+	mock.Mock
+}
+
+// TransactionValidated provides a mock function with given fields:
+func (_m *TransactionValidationMetrics) TransactionValidated() {
+	_m.Called()
+}
+
+// TransactionValidationFailed provides a mock function with given fields: reason
+func (_m *TransactionValidationMetrics) TransactionValidationFailed(reason string) {
+	_m.Called(reason)
+}
+
+// TransactionValidationSkipped provides a mock function with given fields:
+func (_m *TransactionValidationMetrics) TransactionValidationSkipped() {
+	_m.Called()
+}
+
+// NewTransactionValidationMetrics creates a new instance of TransactionValidationMetrics. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewTransactionValidationMetrics(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *TransactionValidationMetrics {
+	mock := &TransactionValidationMetrics{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

From f1c9a948e83e0a63900dc863d539cb402c45016d Mon Sep 17 00:00:00 2001
From: Peter Argue <89119817+peterargue@users.noreply.github.com>
Date: Tue, 1 Oct 2024 14:04:29 +0000
Subject: [PATCH 2/5] Merge pull request #6297 from
 The-K-R-O-K/AndriiSlisarchuk/6139-fail-or-warn-tx-balance

[Access] Implemented options to fail or warn when payer is not able to pay
---
 access/validator.go                           | 65 +++++++++++++++++--
 access/validator_test.go                      | 37 +++++++----
 .../node_builder/access_node_builder.go       | 24 ++++---
 engine/access/rpc/backend/backend.go          | 14 ++--
 integration/go.mod                            |  8 +--
 integration/go.sum                            | 12 ++--
 6 files changed, 120 insertions(+), 40 deletions(-)

diff --git a/access/validator.go b/access/validator.go
index 00d902d0c10..61a58069c88 100644
--- a/access/validator.go
+++ b/access/validator.go
@@ -78,6 +78,59 @@ func (l *NoopLimiter) IsRateLimited(address flow.Address) bool {
 	return false
 }
 
+// PayerBalanceMode represents the mode for checking the payer's balance
+// when validating transactions. It controls whether and how the balance
+// check is performed during transaction validation.
+//
+// There are few modes available:
+//
+//   - `Disabled` - Balance checking is completely disabled. No checks are
+//     performed to verify if the payer has sufficient balance to cover the
+//     transaction fees.
+//   - `WarnCheck` - Balance is checked, and a warning is logged if the payer
+//     does not have enough balance. The transaction is still accepted and
+//     processed regardless of the check result.
+//   - `EnforceCheck` - Balance is checked, and the transaction is rejected if
+//     the payer does not have sufficient balance to cover the transaction fees.
+type PayerBalanceMode int
+
+const (
+	// Disabled indicates that payer balance checking is turned off.
+	Disabled PayerBalanceMode = iota
+
+	// WarnCheck logs a warning if the payer's balance is insufficient, but does not prevent the transaction from being accepted.
+	WarnCheck
+
+	// EnforceCheck prevents the transaction from being accepted if the payer's balance is insufficient to cover transaction fees.
+	EnforceCheck
+)
+
+func ParsePayerBalanceMode(s string) (PayerBalanceMode, error) {
+	switch s {
+	case Disabled.String():
+		return Disabled, nil
+	case WarnCheck.String():
+		return WarnCheck, nil
+	case EnforceCheck.String():
+		return EnforceCheck, nil
+	default:
+		return 0, errors.New("invalid payer balance mode")
+	}
+}
+
+func (m PayerBalanceMode) String() string {
+	switch m {
+	case Disabled:
+		return "disabled"
+	case WarnCheck:
+		return "warn"
+	case EnforceCheck:
+		return "enforce"
+	default:
+		return ""
+	}
+}
+
 type TransactionValidationOptions struct {
 	Expiry                       uint
 	ExpiryBuffer                 uint
@@ -87,7 +140,7 @@ type TransactionValidationOptions struct {
 	CheckScriptsParse            bool
 	MaxTransactionByteSize       uint64
 	MaxCollectionByteSize        uint64
-	CheckPayerBalance            bool
+	CheckPayerBalanceMode        PayerBalanceMode
 }
 
 type ValidationStep struct {
@@ -115,7 +168,7 @@ func NewTransactionValidator(
 	options TransactionValidationOptions,
 	executor execution.ScriptExecutor,
 ) (*TransactionValidator, error) {
-	if options.CheckPayerBalance && executor == nil {
+	if options.CheckPayerBalanceMode != Disabled && executor == nil {
 		return nil, errors.New("transaction validator cannot use checkPayerBalance with nil executor")
 	}
 
@@ -193,7 +246,11 @@ func (v *TransactionValidator) Validate(ctx context.Context, tx *flow.Transactio
 		// prevent the transaction from proceeding.
 		if IsInsufficientBalanceError(err) {
 			v.transactionValidationMetrics.TransactionValidationFailed(metrics.InsufficientBalance)
-			return err
+
+			if v.options.CheckPayerBalanceMode == EnforceCheck {
+				log.Warn().Err(err).Str("transactionID", tx.ID().String()).Str("payerAddress", tx.Payer.String()).Msg("enforce check error")
+				return err
+			}
 		}
 
 		// log and ignore all other errors
@@ -398,7 +455,7 @@ func (v *TransactionValidator) checkSignatureFormat(tx *flow.TransactionBody) er
 }
 
 func (v *TransactionValidator) checkSufficientBalanceToPayForTransaction(ctx context.Context, tx *flow.TransactionBody) error {
-	if !v.options.CheckPayerBalance {
+	if v.options.CheckPayerBalanceMode == Disabled {
 		return nil
 	}
 
diff --git a/access/validator_test.go b/access/validator_test.go
index 87b7ade62a9..5f7c30774e8 100644
--- a/access/validator_test.go
+++ b/access/validator_test.go
@@ -58,7 +58,7 @@ func (s *TransactionValidatorSuite) SetupTest() {
 
 	s.chain = flow.Testnet.Chain()
 	s.validatorOptions = access.TransactionValidationOptions{
-		CheckPayerBalance:      true,
+		CheckPayerBalanceMode:  access.EnforceCheck,
 		MaxTransactionByteSize: flow.DefaultMaxTransactionByteSize,
 		MaxCollectionByteSize:  flow.DefaultMaxCollectionByteSize,
 	}
@@ -144,25 +144,34 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_InsufficientBalance
 
 	scriptExecutor.
 		On("ExecuteAtBlockHeight", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
-		Return(actualResponse, nil).
-		Once()
+		Return(actualResponse, nil).Twice()
 
 	actualAccountResponse, err := unittest.AccountFixture()
 	assert.NoError(s.T(), err)
 	assert.NotNil(s.T(), actualAccountResponse)
 
-	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
-	assert.NoError(s.T(), err)
-	assert.NotNil(s.T(), validator)
-
-	txBody := unittest.TransactionBodyFixture()
+	validateTx := func() error {
+		txBody := unittest.TransactionBodyFixture()
+		validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
+		assert.NoError(s.T(), err)
+		assert.NotNil(s.T(), validator)
 
-	expectedError := access.InsufficientBalanceError{
-		Payer:           unittest.AddressFixture(),
-		RequiredBalance: requiredBalance,
+		return validator.Validate(context.Background(), &txBody)
 	}
 
-	actualErr := validator.Validate(context.Background(), &txBody)
-
-	assert.ErrorIs(s.T(), actualErr, expectedError)
+	s.Run("with enforce check", func() {
+		err := validateTx()
+
+		expectedError := access.InsufficientBalanceError{
+			Payer:           unittest.AddressFixture(),
+			RequiredBalance: requiredBalance,
+		}
+		assert.ErrorIs(s.T(), err, expectedError)
+	})
+
+	s.Run("with warn check", func() {
+		s.validatorOptions.CheckPayerBalanceMode = access.WarnCheck
+		err := validateTx()
+		assert.NoError(s.T(), err)
+	})
 }
diff --git a/cmd/access/node_builder/access_node_builder.go b/cmd/access/node_builder/access_node_builder.go
index 469c1f0462f..1dfb37aa9db 100644
--- a/cmd/access/node_builder/access_node_builder.go
+++ b/cmd/access/node_builder/access_node_builder.go
@@ -27,6 +27,7 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 
+	accessNode "github.com/onflow/flow-go/access"
 	"github.com/onflow/flow-go/admin/commands"
 	stateSyncCommands "github.com/onflow/flow-go/admin/commands/state_synchronization"
 	storageCommands "github.com/onflow/flow-go/admin/commands/storage"
@@ -171,7 +172,7 @@ type AccessNodeConfig struct {
 	registerCacheType                    string
 	registerCacheSize                    uint
 	programCacheSize                     uint
-	checkPayerBalance                    bool
+	checkPayerBalanceMode                string
 	versionControlEnabled                bool
 }
 
@@ -274,7 +275,7 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		registerCacheType:                    pstorage.CacheTypeTwoQueue.String(),
 		registerCacheSize:                    0,
 		programCacheSize:                     0,
-		checkPayerBalance:                    false,
+		checkPayerBalanceMode:                accessNode.Disabled.String(),
 		versionControlEnabled:                true,
 	}
 }
@@ -1402,10 +1403,12 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"program-cache-size",
 			defaultConfig.programCacheSize,
 			"[experimental] number of blocks to cache for cadence programs. use 0 to disable cache. default: 0. Note: this is an experimental feature and may cause nodes to become unstable under certain workloads. Use with caution.")
-		flags.BoolVar(&builder.checkPayerBalance,
-			"check-payer-balance",
-			defaultConfig.checkPayerBalance,
-			"checks that a transaction payer has sufficient balance to pay fees before submitting it to collection nodes")
+
+		// Payer Balance
+		flags.StringVar(&builder.checkPayerBalanceMode,
+			"check-payer-balance-mode",
+			defaultConfig.checkPayerBalanceMode,
+			"flag for payer balance validation that specifies whether or not to enforce the balance check. one of [disabled(default), warn, enforce]")
 	}).ValidateFlags(func() error {
 		if builder.supportsObserver && (builder.PublicNetworkConfig.BindAddress == cmd.NotSet || builder.PublicNetworkConfig.BindAddress == "") {
 			return errors.New("public-network-address must be set if supports-observer is true")
@@ -1469,7 +1472,7 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			return errors.New("transaction-error-messages-cache-size must be greater than 0")
 		}
 
-		if builder.checkPayerBalance && !builder.executionDataIndexingEnabled {
+		if builder.checkPayerBalanceMode != accessNode.Disabled.String() && !builder.executionDataIndexingEnabled {
 			return errors.New("execution-data-indexing-enabled must be set if check-payer-balance is enabled")
 		}
 
@@ -1889,6 +1892,11 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				return nil, fmt.Errorf("transaction result query mode 'compare' is not supported")
 			}
 
+			checkPayerBalanceMode, err := accessNode.ParsePayerBalanceMode(builder.checkPayerBalanceMode)
+			if err != nil {
+				return nil, fmt.Errorf("could not parse payer balance mode: %w", err)
+			}
+
 			nodeBackend, err := backend.New(backend.Params{
 				State:                     node.State,
 				CollectionRPC:             builder.CollectionRPC,
@@ -1913,7 +1921,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				TxErrorMessagesCacheSize:  builder.TxErrorMessagesCacheSize,
 				ScriptExecutor:            builder.ScriptExecutor,
 				ScriptExecutionMode:       scriptExecMode,
-				CheckPayerBalance:         builder.checkPayerBalance,
+				CheckPayerBalanceMode:     checkPayerBalanceMode,
 				EventQueryMode:            eventQueryMode,
 				BlockTracker:              blockTracker,
 				SubscriptionHandler: subscription.NewSubscriptionHandler(
diff --git a/engine/access/rpc/backend/backend.go b/engine/access/rpc/backend/backend.go
index d250b078d6b..9d45abab24b 100644
--- a/engine/access/rpc/backend/backend.go
+++ b/engine/access/rpc/backend/backend.go
@@ -110,7 +110,7 @@ type Params struct {
 	TxErrorMessagesCacheSize  uint
 	ScriptExecutor            execution.ScriptExecutor
 	ScriptExecutionMode       IndexQueryMode
-	CheckPayerBalance         bool
+	CheckPayerBalanceMode     access.PayerBalanceMode
 	EventQueryMode            IndexQueryMode
 	BlockTracker              subscription.BlockTracker
 	SubscriptionHandler       *subscription.SubscriptionHandler
@@ -246,7 +246,13 @@ func New(params Params) (*Backend, error) {
 		nodeInfo:          nodeInfo,
 	}
 
-	txValidator, err := configureTransactionValidator(params.State, params.ChainID, params.AccessMetrics, params.ScriptExecutor, params.CheckPayerBalance)
+	txValidator, err := configureTransactionValidator(
+		params.State,
+		params.ChainID,
+		params.AccessMetrics,
+		params.ScriptExecutor,
+		params.CheckPayerBalanceMode,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("could not create transaction validator: %w", err)
 	}
@@ -315,7 +321,7 @@ func configureTransactionValidator(
 	chainID flow.ChainID,
 	transactionMetrics module.TransactionValidationMetrics,
 	executor execution.ScriptExecutor,
-	checkPayerBalance bool,
+	checkPayerBalanceMode access.PayerBalanceMode,
 ) (*access.TransactionValidator, error) {
 	return access.NewTransactionValidator(
 		access.NewProtocolStateBlocks(state),
@@ -330,7 +336,7 @@ func configureTransactionValidator(
 			MaxGasLimit:                  flow.DefaultMaxTransactionGasLimit,
 			MaxTransactionByteSize:       flow.DefaultMaxTransactionByteSize,
 			MaxCollectionByteSize:        flow.DefaultMaxCollectionByteSize,
-			CheckPayerBalance:            checkPayerBalance,
+			CheckPayerBalanceMode:        checkPayerBalanceMode,
 		},
 		executor,
 	)
diff --git a/integration/go.mod b/integration/go.mod
index 4b956e824c3..af50df0d10e 100644
--- a/integration/go.mod
+++ b/integration/go.mod
@@ -24,11 +24,11 @@ require (
 	github.com/onflow/crypto v0.25.2
 	github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1
 	github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1
-	github.com/onflow/flow-emulator v1.0.0-preview.41.0.20240829134601-0be55d6970b5
-	github.com/onflow/flow-go v0.37.7-0.20240826193109-e211841b59f5
-	github.com/onflow/flow-go-sdk v1.0.0-preview.54
+	github.com/onflow/flow-emulator v1.0.1-0.20240930092334-2f46b2112195
+	github.com/onflow/flow-go v0.38.0-preview.0
+	github.com/onflow/flow-go-sdk v1.0.0
 	github.com/onflow/flow-go/insecure v0.0.0-00010101000000-000000000000
-	github.com/onflow/flow/protobuf/go/flow v0.4.6
+	github.com/onflow/flow/protobuf/go/flow v0.4.7
 	github.com/onflow/go-ethereum v1.14.7
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/client_model v0.5.0
diff --git a/integration/go.sum b/integration/go.sum
index df6b01ed2a5..d4159942746 100644
--- a/integration/go.sum
+++ b/integration/go.sum
@@ -2151,22 +2151,22 @@ github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1 h1:q9tXLIALwQ76bO4
 github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1/go.mod h1:u/mkP/B+PbV33tEG3qfkhhBlydSvAKxfLZSfB4lsJHg=
 github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1 h1:FfhMBAb78p6VAWkJ+iqdKLErGQVQgxk5w6DP5ZruWX8=
 github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1/go.mod h1:NgbMOYnMh0GN48VsNKZuiwK7uyk38Wyo8jN9+C9QE30=
-github.com/onflow/flow-emulator v1.0.0-preview.41.0.20240829134601-0be55d6970b5 h1:Z5PC3Sqvl2UemY27uwUwzkLb8EXUf+m/aEimxFzOXuc=
-github.com/onflow/flow-emulator v1.0.0-preview.41.0.20240829134601-0be55d6970b5/go.mod h1:gFafyGD4Qxs+XT2BRSIjQJ86ILSmgm1VYUoCr1nVxVI=
+github.com/onflow/flow-emulator v1.0.1-0.20240930092334-2f46b2112195 h1:buM9uEW5WhFiI9hMDA90lJhokItN1Cmac3ddb0GWSbY=
+github.com/onflow/flow-emulator v1.0.1-0.20240930092334-2f46b2112195/go.mod h1:b9gi9kvRfUVHmyz7cTXBsnT12oHOJisvrxpqwtFRMpM=
 github.com/onflow/flow-ft/lib/go/contracts v1.0.0 h1:mToacZ5NWqtlWwk/7RgIl/jeKB/Sy/tIXdw90yKHcV0=
 github.com/onflow/flow-ft/lib/go/contracts v1.0.0/go.mod h1:PwsL8fC81cjnUnTfmyL/HOIyHnyaw/JA474Wfj2tl6A=
 github.com/onflow/flow-ft/lib/go/templates v1.0.0 h1:6cMS/lUJJ17HjKBfMO/eh0GGvnpElPgBXx7h5aoWJhs=
 github.com/onflow/flow-ft/lib/go/templates v1.0.0/go.mod h1:uQ8XFqmMK2jxyBSVrmyuwdWjTEb+6zGjRYotfDJ5pAE=
 github.com/onflow/flow-go-sdk v1.0.0-M1/go.mod h1:TDW0MNuCs4SvqYRUzkbRnRmHQL1h4X8wURsCw9P9beo=
-github.com/onflow/flow-go-sdk v1.0.0-preview.54 h1:5GjCkyIyvE9KolOUUPTkGdEiV/8qOe1MGnLHOLBmthA=
-github.com/onflow/flow-go-sdk v1.0.0-preview.54/go.mod h1:u9oFiS25TpnU1EW62PQlq22jzkwBAj4VEiiCBM6nhHo=
+github.com/onflow/flow-go-sdk v1.0.0 h1:Ha4fQm1MMKsyaqMkQLCN3rA/yaQKG6DGwiIfx06j40c=
+github.com/onflow/flow-go-sdk v1.0.0/go.mod h1:iZkW2IWieVUZKK06mQCxpjJzPDgS0VtGpTaP/rKu6J4=
 github.com/onflow/flow-nft/lib/go/contracts v1.2.1 h1:woAAS5z651sDpi7ihAHll8NvRS9uFXIXkL6xR+bKFZY=
 github.com/onflow/flow-nft/lib/go/contracts v1.2.1/go.mod h1:2gpbza+uzs1k7x31hkpBPlggIRkI53Suo0n2AyA2HcE=
 github.com/onflow/flow-nft/lib/go/templates v1.2.0 h1:JSQyh9rg0RC+D1930BiRXN8lrtMs+ubVMK6aQPon6Yc=
 github.com/onflow/flow-nft/lib/go/templates v1.2.0/go.mod h1:p+2hRvtjLUR3MW1NsoJe5Gqgr2eeH49QB6+s6ze00w0=
 github.com/onflow/flow/protobuf/go/flow v0.3.2-0.20231121210617-52ee94b830c2/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
-github.com/onflow/flow/protobuf/go/flow v0.4.6 h1:KE/CsRVfyG5lGBtm1aNcjojMciQyS5GfPF3ixOWRfi0=
-github.com/onflow/flow/protobuf/go/flow v0.4.6/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
+github.com/onflow/flow/protobuf/go/flow v0.4.7 h1:iP6DFx4wZ3ETORsyeqzHu7neFT3d1CXF6wdK+AOOjmc=
+github.com/onflow/flow/protobuf/go/flow v0.4.7/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c h1:T0jDCm7k7uqDo26JiiujQ5oryl30itPnlmZQywTu9ng=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c/go.mod h1:XYnWtulwJvHVOr2B0WVA/UC3dvRgFevjp8Pn9a3E1xo=
 github.com/onflow/go-ethereum v1.14.7 h1:gg3awYqI02e3AypRdpJKEvNTJ6kz/OhAqRti0h54Wlc=

From b2e273e2c49b27236c66769a9f863c26eca8456b Mon Sep 17 00:00:00 2001
From: Peter Argue <89119817+peterargue@users.noreply.github.com>
Date: Wed, 2 Oct 2024 12:40:48 +0000
Subject: [PATCH 3/5] Merge pull request #6292 from
 AndriiDiachuk/use-indexed-height-checking-payer-balance

[Access] Use Indexed height when checking payer balance
---
 access/errors.go                              | 15 +++++++
 access/mock/blocks.go                         | 28 ++++++++++++
 access/validator.go                           | 45 ++++++++++++++++---
 access/validator_test.go                      | 33 ++++++++++++++
 .../node_builder/access_node_builder.go       | 15 ++++---
 cmd/bootstrap/utils/md5.go                    |  2 +-
 cmd/observer/node_builder/observer_builder.go | 14 +++---
 engine/access/rpc/backend/backend.go          | 15 +++----
 engine/access/rpc/engine.go                   |  3 +-
 engine/collection/ingest/engine.go            |  2 +-
 integration/go.mod                            |  4 +-
 integration/go.sum                            |  4 +-
 12 files changed, 147 insertions(+), 33 deletions(-)

diff --git a/access/errors.go b/access/errors.go
index a663179f018..310c86b43b5 100644
--- a/access/errors.go
+++ b/access/errors.go
@@ -12,6 +12,10 @@ import (
 // ErrUnknownReferenceBlock indicates that a transaction references an unknown block.
 var ErrUnknownReferenceBlock = errors.New("unknown reference block")
 
+// IndexReporterNotInitialized is returned when indexReporter is nil because
+// execution data syncing and indexing is disabled
+var IndexReporterNotInitialized = errors.New("index reported not initialized")
+
 // IncompleteTransactionError indicates that a transaction is missing one or more required fields.
 type IncompleteTransactionError struct {
 	MissingFields []string
@@ -115,3 +119,14 @@ func IsInsufficientBalanceError(err error) bool {
 	var balanceError InsufficientBalanceError
 	return errors.As(err, &balanceError)
 }
+
+// IndexedHeightFarBehindError indicates that a node is far behind on indexing.
+type IndexedHeightFarBehindError struct {
+	SealedHeight  uint64
+	IndexedHeight uint64
+}
+
+func (e IndexedHeightFarBehindError) Error() string {
+	return fmt.Sprintf("the difference between the latest sealed height (%d) and indexed height (%d) exceeds the maximum gap allowed",
+		e.SealedHeight, e.IndexedHeight)
+}
diff --git a/access/mock/blocks.go b/access/mock/blocks.go
index 153c2160321..088a50aa155 100644
--- a/access/mock/blocks.go
+++ b/access/mock/blocks.go
@@ -72,6 +72,34 @@ func (_m *Blocks) HeaderByID(id flow.Identifier) (*flow.Header, error) {
 	return r0, r1
 }
 
+// IndexedHeight provides a mock function with given fields:
+func (_m *Blocks) IndexedHeight() (uint64, error) {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for IndexedHeight")
+	}
+
+	var r0 uint64
+	var r1 error
+	if rf, ok := ret.Get(0).(func() (uint64, error)); ok {
+		return rf()
+	}
+	if rf, ok := ret.Get(0).(func() uint64); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(uint64)
+	}
+
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // SealedHeader provides a mock function with given fields:
 func (_m *Blocks) SealedHeader() (*flow.Header, error) {
 	ret := _m.Called()
diff --git a/access/validator.go b/access/validator.go
index 61a58069c88..d3169d3a41f 100644
--- a/access/validator.go
+++ b/access/validator.go
@@ -20,22 +20,32 @@ import (
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/execution"
 	"github.com/onflow/flow-go/module/metrics"
+	"github.com/onflow/flow-go/module/state_synchronization"
 	"github.com/onflow/flow-go/state"
 	"github.com/onflow/flow-go/state/protocol"
 )
 
+// DefaultSealedIndexedHeightThreshold is the default number of blocks between sealed and indexed height
+// this sets a limit on how far into the past the payer validator will allow for checking the payer's balance.
+const DefaultSealedIndexedHeightThreshold = 30
+
 type Blocks interface {
 	HeaderByID(id flow.Identifier) (*flow.Header, error)
 	FinalizedHeader() (*flow.Header, error)
 	SealedHeader() (*flow.Header, error)
+	IndexedHeight() (uint64, error)
 }
 
 type ProtocolStateBlocks struct {
-	state protocol.State
+	state         protocol.State
+	indexReporter state_synchronization.IndexReporter
 }
 
-func NewProtocolStateBlocks(state protocol.State) *ProtocolStateBlocks {
-	return &ProtocolStateBlocks{state: state}
+func NewProtocolStateBlocks(state protocol.State, indexReporter state_synchronization.IndexReporter) *ProtocolStateBlocks {
+	return &ProtocolStateBlocks{
+		state:         state,
+		indexReporter: indexReporter,
+	}
 }
 
 func (b *ProtocolStateBlocks) HeaderByID(id flow.Identifier) (*flow.Header, error) {
@@ -56,7 +66,19 @@ func (b *ProtocolStateBlocks) FinalizedHeader() (*flow.Header, error) {
 }
 
 func (b *ProtocolStateBlocks) SealedHeader() (*flow.Header, error) {
+
 	return b.state.Sealed().Head()
+
+}
+
+// IndexedHeight returns the highest indexed height by calling corresponding function of indexReporter.
+// Expected errors during normal operation:
+// - access.IndexReporterNotInitialized - indexed reporter was not initialized.
+func (b *ProtocolStateBlocks) IndexedHeight() (uint64, error) {
+	if b.indexReporter != nil {
+		return b.indexReporter.HighestIndexedHeight()
+	}
+	return 0, IndexReporterNotInitialized
 }
 
 // RateLimiter is an interface for checking if an address is rate limited.
@@ -464,6 +486,19 @@ func (v *TransactionValidator) checkSufficientBalanceToPayForTransaction(ctx con
 		return fmt.Errorf("could not fetch block header: %w", err)
 	}
 
+	indexedHeight, err := v.blocks.IndexedHeight()
+	if err != nil {
+		return fmt.Errorf("could not get indexed height: %w", err)
+	}
+
+	// we use latest indexed block to get the most up-to-date state data available for executing scripts.
+	// check here to make sure indexing is within an acceptable tolerance of sealing to avoid issues
+	// if indexing falls behind
+	sealedHeight := header.Height
+	if indexedHeight < sealedHeight-DefaultSealedIndexedHeightThreshold {
+		return IndexedHeightFarBehindError{SealedHeight: sealedHeight, IndexedHeight: indexedHeight}
+	}
+
 	payerAddress := cadence.NewAddress(tx.Payer)
 	inclusionEffort := cadence.UFix64(tx.InclusionEffort())
 	gasLimit := cadence.UFix64(tx.GasLimit)
@@ -473,7 +508,7 @@ func (v *TransactionValidator) checkSufficientBalanceToPayForTransaction(ctx con
 		return fmt.Errorf("failed to encode cadence args for script executor: %w", err)
 	}
 
-	result, err := v.scriptExecutor.ExecuteAtBlockHeight(ctx, v.verifyPayerBalanceScript, args, header.Height)
+	result, err := v.scriptExecutor.ExecuteAtBlockHeight(ctx, v.verifyPayerBalanceScript, args, indexedHeight)
 	if err != nil {
 		return fmt.Errorf("script finished with error: %w", err)
 	}
@@ -489,7 +524,7 @@ func (v *TransactionValidator) checkSufficientBalanceToPayForTransaction(ctx con
 	}
 
 	// return no error if payer has sufficient balance
-	if bool(canExecuteTransaction) {
+	if canExecuteTransaction {
 		return nil
 	}
 
diff --git a/access/validator_test.go b/access/validator_test.go
index 5f7c30774e8..58c8bd1e1ab 100644
--- a/access/validator_test.go
+++ b/access/validator_test.go
@@ -88,6 +88,10 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_ScriptExecutorInter
 	scriptExecutor := execmock.NewScriptExecutor(s.T())
 	assert.NotNil(s.T(), scriptExecutor)
 
+	s.blocks.
+		On("IndexedHeight").
+		Return(s.header.Height, nil)
+
 	scriptExecutor.
 		On("ExecuteAtBlockHeight", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(nil, errors.New("script executor internal error")).
@@ -115,6 +119,10 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_SufficientBalance()
 	actualResponse, err := jsoncdc.Encode(actualResponseValue)
 	assert.NoError(s.T(), err)
 
+	s.blocks.
+		On("IndexedHeight").
+		Return(s.header.Height, nil)
+
 	scriptExecutor.
 		On("ExecuteAtBlockHeight", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(actualResponse, nil).
@@ -142,6 +150,10 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_InsufficientBalance
 	actualResponse, err := jsoncdc.Encode(actualResponseValue)
 	assert.NoError(s.T(), err)
 
+	s.blocks.
+		On("IndexedHeight").
+		Return(s.header.Height, nil)
+
 	scriptExecutor.
 		On("ExecuteAtBlockHeight", mock.Anything, mock.Anything, mock.Anything, mock.Anything).
 		Return(actualResponse, nil).Twice()
@@ -175,3 +187,24 @@ func (s *TransactionValidatorSuite) TestTransactionValidator_InsufficientBalance
 		assert.NoError(s.T(), err)
 	})
 }
+
+func (s *TransactionValidatorSuite) TestTransactionValidator_SealedIndexedHeightThresholdLimit() {
+	scriptExecutor := execmock.NewScriptExecutor(s.T())
+
+	// setting indexed height to be behind of sealed by bigger number than allowed(DefaultSealedIndexedHeightThreshold)
+	indexedHeight := s.header.Height - 40
+
+	s.blocks.
+		On("IndexedHeight").
+		Return(indexedHeight, nil)
+
+	validator, err := access.NewTransactionValidator(s.blocks, s.chain, s.metrics, s.validatorOptions, scriptExecutor)
+	assert.NoError(s.T(), err)
+	assert.NotNil(s.T(), validator)
+
+	txBody := unittest.TransactionBodyFixture()
+
+	err = validator.Validate(context.Background(), &txBody)
+	assert.NoError(s.T(), err)
+
+}
diff --git a/cmd/access/node_builder/access_node_builder.go b/cmd/access/node_builder/access_node_builder.go
index 1dfb37aa9db..5d956721dcb 100644
--- a/cmd/access/node_builder/access_node_builder.go
+++ b/cmd/access/node_builder/access_node_builder.go
@@ -1892,9 +1892,16 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				return nil, fmt.Errorf("transaction result query mode 'compare' is not supported")
 			}
 
+			// If execution data syncing and indexing is disabled, pass nil indexReporter
+			var indexReporter state_synchronization.IndexReporter
+			if builder.executionDataSyncEnabled && builder.executionDataIndexingEnabled {
+				indexReporter = builder.Reporter
+			}
+
 			checkPayerBalanceMode, err := accessNode.ParsePayerBalanceMode(builder.checkPayerBalanceMode)
 			if err != nil {
 				return nil, fmt.Errorf("could not parse payer balance mode: %w", err)
+
 			}
 
 			nodeBackend, err := backend.New(backend.Params{
@@ -1935,17 +1942,13 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				TxResultQueryMode:   txResultQueryMode,
 				TxResultsIndex:      builder.TxResultsIndex,
 				LastFullBlockHeight: lastFullBlockHeight,
+				IndexReporter:       indexReporter,
+				VersionControl:      builder.VersionControl,
 			})
 			if err != nil {
 				return nil, fmt.Errorf("could not initialize backend: %w", err)
 			}
 
-			// If execution data syncing and indexing is disabled, pass nil indexReporter
-			var indexReporter state_synchronization.IndexReporter
-			if builder.executionDataSyncEnabled && builder.executionDataIndexingEnabled {
-				indexReporter = builder.Reporter
-			}
-
 			engineBuilder, err := rpc.NewBuilder(
 				node.Logger,
 				node.State,
diff --git a/cmd/bootstrap/utils/md5.go b/cmd/bootstrap/utils/md5.go
index 3abe9c42948..e885ed891e2 100644
--- a/cmd/bootstrap/utils/md5.go
+++ b/cmd/bootstrap/utils/md5.go
@@ -3,7 +3,7 @@ package utils
 // The google storage API only provides md5 and crc32 hence overriding the linter flag for md5
 // #nosec
 import (
-	"crypto/md5"
+	"crypto/md5" //nolint:gosec
 	"io"
 	"os"
 )
diff --git a/cmd/observer/node_builder/observer_builder.go b/cmd/observer/node_builder/observer_builder.go
index 80c702f2967..7daffed6268 100644
--- a/cmd/observer/node_builder/observer_builder.go
+++ b/cmd/observer/node_builder/observer_builder.go
@@ -1904,6 +1904,12 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			return nil, fmt.Errorf("failed to initialize block tracker: %w", err)
 		}
 
+		// If execution data syncing and indexing is disabled, pass nil indexReporter
+		var indexReporter state_synchronization.IndexReporter
+		if builder.executionDataSyncEnabled && builder.executionDataIndexingEnabled {
+			indexReporter = builder.Reporter
+		}
+
 		backendParams := backend.Params{
 			State:                     node.State,
 			Blocks:                    node.Storage.Blocks,
@@ -1930,6 +1936,8 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 				builder.stateStreamConf.ResponseLimit,
 				builder.stateStreamConf.ClientSendBufferSize,
 			),
+			IndexReporter:  indexReporter,
+			VersionControl: builder.versionControl,
 		}
 
 		if builder.localServiceAPIEnabled {
@@ -1957,12 +1965,6 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 			return nil, err
 		}
 
-		// If execution data syncing and indexing is disabled, pass nil indexReporter
-		var indexReporter state_synchronization.IndexReporter
-		if builder.executionDataSyncEnabled && builder.executionDataIndexingEnabled {
-			indexReporter = builder.Reporter
-		}
-
 		engineBuilder, err := rpc.NewBuilder(
 			node.Logger,
 			node.State,
diff --git a/engine/access/rpc/backend/backend.go b/engine/access/rpc/backend/backend.go
index 9d45abab24b..f26b200540f 100644
--- a/engine/access/rpc/backend/backend.go
+++ b/engine/access/rpc/backend/backend.go
@@ -16,12 +16,14 @@ import (
 	"github.com/onflow/flow-go/engine/access/rpc/connection"
 	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/engine/common/rpc"
+	"github.com/onflow/flow-go/engine/common/version"
 	"github.com/onflow/flow-go/fvm/blueprints"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
 	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/counters"
 	"github.com/onflow/flow-go/module/execution"
+	"github.com/onflow/flow-go/module/state_synchronization"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/storage"
 )
@@ -119,6 +121,8 @@ type Params struct {
 	TxResultQueryMode   IndexQueryMode
 	TxResultsIndex      *index.TransactionResultsIndex
 	LastFullBlockHeight *counters.PersistentStrictMonotonicCounter
+	IndexReporter       state_synchronization.IndexReporter
+	VersionControl      *version.VersionControl
 }
 
 var _ TransactionErrorMessage = (*Backend)(nil)
@@ -246,13 +250,7 @@ func New(params Params) (*Backend, error) {
 		nodeInfo:          nodeInfo,
 	}
 
-	txValidator, err := configureTransactionValidator(
-		params.State,
-		params.ChainID,
-		params.AccessMetrics,
-		params.ScriptExecutor,
-		params.CheckPayerBalanceMode,
-	)
+	txValidator, err := configureTransactionValidator(params.State, params.ChainID, params.IndexReporter, params.AccessMetrics, params.ScriptExecutor, params.CheckPayerBalanceMode)
 	if err != nil {
 		return nil, fmt.Errorf("could not create transaction validator: %w", err)
 	}
@@ -319,12 +317,13 @@ func identifierList(ids []string) (flow.IdentifierList, error) {
 func configureTransactionValidator(
 	state protocol.State,
 	chainID flow.ChainID,
+	indexReporter state_synchronization.IndexReporter,
 	transactionMetrics module.TransactionValidationMetrics,
 	executor execution.ScriptExecutor,
 	checkPayerBalanceMode access.PayerBalanceMode,
 ) (*access.TransactionValidator, error) {
 	return access.NewTransactionValidator(
-		access.NewProtocolStateBlocks(state),
+		access.NewProtocolStateBlocks(state, indexReporter),
 		chainID.Chain(),
 		transactionMetrics,
 		access.TransactionValidationOptions{
diff --git a/engine/access/rpc/engine.go b/engine/access/rpc/engine.go
index a42c8495345..145e3d62143 100644
--- a/engine/access/rpc/engine.go
+++ b/engine/access/rpc/engine.go
@@ -8,8 +8,6 @@ import (
 	"net/http"
 	"sync"
 
-	"github.com/onflow/flow-go/module/state_synchronization"
-
 	"github.com/rs/zerolog"
 	"google.golang.org/grpc/credentials"
 
@@ -25,6 +23,7 @@ import (
 	"github.com/onflow/flow-go/module/events"
 	"github.com/onflow/flow-go/module/grpcserver"
 	"github.com/onflow/flow-go/module/irrecoverable"
+	"github.com/onflow/flow-go/module/state_synchronization"
 	"github.com/onflow/flow-go/state/protocol"
 )
 
diff --git a/engine/collection/ingest/engine.go b/engine/collection/ingest/engine.go
index 438e6bea88b..ae21f71253f 100644
--- a/engine/collection/ingest/engine.go
+++ b/engine/collection/ingest/engine.go
@@ -61,7 +61,7 @@ func New(
 	logger := log.With().Str("engine", "ingest").Logger()
 
 	transactionValidator := access.NewTransactionValidatorWithLimiter(
-		access.NewProtocolStateBlocks(state),
+		access.NewProtocolStateBlocks(state, nil),
 		chain,
 		access.TransactionValidationOptions{
 			Expiry:                 flow.DefaultTransactionExpiry,
diff --git a/integration/go.mod b/integration/go.mod
index af50df0d10e..fed4bda3c3b 100644
--- a/integration/go.mod
+++ b/integration/go.mod
@@ -24,8 +24,8 @@ require (
 	github.com/onflow/crypto v0.25.2
 	github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1
 	github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1
-	github.com/onflow/flow-emulator v1.0.1-0.20240930092334-2f46b2112195
-	github.com/onflow/flow-go v0.38.0-preview.0
+	github.com/onflow/flow-emulator v1.0.1-0.20241002100151-fa253c380189
+	github.com/onflow/flow-go v0.38.0-preview.0.0.20241001140429-ec4ad1cf1c8a
 	github.com/onflow/flow-go-sdk v1.0.0
 	github.com/onflow/flow-go/insecure v0.0.0-00010101000000-000000000000
 	github.com/onflow/flow/protobuf/go/flow v0.4.7
diff --git a/integration/go.sum b/integration/go.sum
index d4159942746..39b7cb8c05b 100644
--- a/integration/go.sum
+++ b/integration/go.sum
@@ -2151,8 +2151,8 @@ github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1 h1:q9tXLIALwQ76bO4
 github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1/go.mod h1:u/mkP/B+PbV33tEG3qfkhhBlydSvAKxfLZSfB4lsJHg=
 github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1 h1:FfhMBAb78p6VAWkJ+iqdKLErGQVQgxk5w6DP5ZruWX8=
 github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1/go.mod h1:NgbMOYnMh0GN48VsNKZuiwK7uyk38Wyo8jN9+C9QE30=
-github.com/onflow/flow-emulator v1.0.1-0.20240930092334-2f46b2112195 h1:buM9uEW5WhFiI9hMDA90lJhokItN1Cmac3ddb0GWSbY=
-github.com/onflow/flow-emulator v1.0.1-0.20240930092334-2f46b2112195/go.mod h1:b9gi9kvRfUVHmyz7cTXBsnT12oHOJisvrxpqwtFRMpM=
+github.com/onflow/flow-emulator v1.0.1-0.20241002100151-fa253c380189 h1:UCVla50Y50Q2b+o6l22um8nHrD35XYRveLFHQg9EOv0=
+github.com/onflow/flow-emulator v1.0.1-0.20241002100151-fa253c380189/go.mod h1:DEfNNXJuEOWqG/NS3RJ8jI+5BOhbENZ2hzKOz14ZPJ0=
 github.com/onflow/flow-ft/lib/go/contracts v1.0.0 h1:mToacZ5NWqtlWwk/7RgIl/jeKB/Sy/tIXdw90yKHcV0=
 github.com/onflow/flow-ft/lib/go/contracts v1.0.0/go.mod h1:PwsL8fC81cjnUnTfmyL/HOIyHnyaw/JA474Wfj2tl6A=
 github.com/onflow/flow-ft/lib/go/templates v1.0.0 h1:6cMS/lUJJ17HjKBfMO/eh0GGvnpElPgBXx7h5aoWJhs=

From 049d753d1afb6ecd31835c95ac69dcef6df44c32 Mon Sep 17 00:00:00 2001
From: Peter Argue <89119817+peterargue@users.noreply.github.com>
Date: Thu, 12 Sep 2024 13:34:00 +0000
Subject: [PATCH 4/5] Merge pull request #6202 from
 The-K-R-O-K/AndriiSlisarchuk/5790-add-stop-control

[Access] Stop Control feature for AN
---
 .../node_builder/access_node_builder.go       |  32 ++++
 cmd/observer/node_builder/observer_builder.go |  43 ++++-
 engine/access/access_test.go                  |   9 +-
 engine/access/ingestion/engine_test.go        |   3 +-
 engine/common/stop/stop_control.go            | 157 ++++++++++++++++++
 engine/common/stop/stop_control_test.go       | 138 +++++++++++++++
 .../execution_data/mock/downloader.go         |   5 +
 .../processed_height_recorder.go              |  24 +++
 storage/pebble/open.go                        |   3 +-
 9 files changed, 399 insertions(+), 15 deletions(-)
 create mode 100644 engine/common/stop/stop_control.go
 create mode 100644 engine/common/stop/stop_control_test.go

diff --git a/cmd/access/node_builder/access_node_builder.go b/cmd/access/node_builder/access_node_builder.go
index 5d956721dcb..595f797d600 100644
--- a/cmd/access/node_builder/access_node_builder.go
+++ b/cmd/access/node_builder/access_node_builder.go
@@ -56,6 +56,7 @@ import (
 	"github.com/onflow/flow-go/engine/access/subscription"
 	followereng "github.com/onflow/flow-go/engine/common/follower"
 	"github.com/onflow/flow-go/engine/common/requester"
+	"github.com/onflow/flow-go/engine/common/stop"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/common/version"
 	"github.com/onflow/flow-go/engine/execution/computation/query"
@@ -174,6 +175,7 @@ type AccessNodeConfig struct {
 	programCacheSize                     uint
 	checkPayerBalanceMode                string
 	versionControlEnabled                bool
+	stopControlEnabled                   bool
 }
 
 type PublicNetworkConfig struct {
@@ -277,6 +279,7 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		programCacheSize:                     0,
 		checkPayerBalanceMode:                accessNode.Disabled.String(),
 		versionControlEnabled:                true,
+		stopControlEnabled:                   false,
 	}
 }
 
@@ -326,6 +329,7 @@ type FlowAccessNodeBuilder struct {
 	ExecutionDatastoreManager    edstorage.DatastoreManager
 	ExecutionDataTracker         tracker.Storage
 	VersionControl               *version.VersionControl
+	StopControl                  *stop.StopControl
 
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
@@ -995,6 +999,10 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					return nil, err
 				}
 
+				if builder.stopControlEnabled {
+					builder.StopControl.RegisterHeightRecorder(builder.ExecutionIndexer)
+				}
+
 				return builder.ExecutionIndexer, nil
 			}, builder.IndexerDependencies)
 	}
@@ -1261,6 +1269,10 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"version-control-enabled",
 			defaultConfig.versionControlEnabled,
 			"whether to enable the version control feature. Default value is true")
+		flags.BoolVar(&builder.stopControlEnabled,
+			"stop-control-enabled",
+			defaultConfig.stopControlEnabled,
+			"whether to enable the stop control feature. Default value is false")
 		// ExecutionDataRequester config
 		flags.BoolVar(&builder.executionDataSyncEnabled,
 			"execution-data-sync-enabled",
@@ -1593,6 +1605,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 	builder.IndexerDependencies.Add(ingestionDependable)
 	versionControlDependable := module.NewProxiedReadyDoneAware()
 	builder.IndexerDependencies.Add(versionControlDependable)
+	stopControlDependable := module.NewProxiedReadyDoneAware()
+	builder.IndexerDependencies.Add(stopControlDependable)
 	var lastFullBlockHeight *counters.PersistentStrictMonotonicCounter
 
 	builder.
@@ -1827,6 +1841,24 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			return versionControl, nil
 		}).
+		Component("stop control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+			if !builder.stopControlEnabled {
+				noop := &module.NoopReadyDoneAware{}
+				stopControlDependable.Init(noop)
+				return noop, nil
+			}
+
+			stopControl := stop.NewStopControl(
+				builder.Logger,
+			)
+
+			builder.VersionControl.AddVersionUpdatesConsumer(stopControl.OnVersionUpdate)
+
+			builder.StopControl = stopControl
+			stopControlDependable.Init(builder.StopControl)
+
+			return stopControl, nil
+		}).
 		Component("RPC engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			config := builder.rpcConf
 			backendConfig := config.BackendConfig
diff --git a/cmd/observer/node_builder/observer_builder.go b/cmd/observer/node_builder/observer_builder.go
index 7daffed6268..27cc55dc432 100644
--- a/cmd/observer/node_builder/observer_builder.go
+++ b/cmd/observer/node_builder/observer_builder.go
@@ -54,6 +54,7 @@ import (
 	statestreambackend "github.com/onflow/flow-go/engine/access/state_stream/backend"
 	"github.com/onflow/flow-go/engine/access/subscription"
 	"github.com/onflow/flow-go/engine/common/follower"
+	"github.com/onflow/flow-go/engine/common/stop"
 	synceng "github.com/onflow/flow-go/engine/common/synchronization"
 	"github.com/onflow/flow-go/engine/common/version"
 	"github.com/onflow/flow-go/engine/execution/computation/query"
@@ -164,6 +165,7 @@ type ObserverServiceConfig struct {
 	executionDataPruningInterval         time.Duration
 	localServiceAPIEnabled               bool
 	versionControlEnabled                bool
+	stopControlEnabled                   bool
 	executionDataDir                     string
 	executionDataStartHeight             uint64
 	executionDataConfig                  edrequester.ExecutionDataConfig
@@ -239,6 +241,7 @@ func DefaultObserverServiceConfig() *ObserverServiceConfig {
 		executionDataPruningInterval:         pruner.DefaultPruningInterval,
 		localServiceAPIEnabled:               false,
 		versionControlEnabled:                true,
+		stopControlEnabled:                   false,
 		executionDataDir:                     filepath.Join(homedir, ".flow", "execution_data"),
 		executionDataStartHeight:             0,
 		executionDataConfig: edrequester.ExecutionDataConfig{
@@ -279,7 +282,8 @@ type ObserverServiceBuilder struct {
 	ExecutionIndexerCore *indexer.IndexerCore
 	TxResultsIndex       *index.TransactionResultsIndex
 	IndexerDependencies  *cmd.DependencyList
-	versionControl       *version.VersionControl
+	VersionControl       *version.VersionControl
+	StopControl          *stop.StopControl
 
 	ExecutionDataDownloader   execution_data.Downloader
 	ExecutionDataRequester    state_synchronization.ExecutionDataRequester
@@ -681,6 +685,10 @@ func (builder *ObserverServiceBuilder) extraFlags() {
 			"version-control-enabled",
 			defaultConfig.versionControlEnabled,
 			"whether to enable the version control feature. Default value is true")
+		flags.BoolVar(&builder.stopControlEnabled,
+			"stop-control-enabled",
+			defaultConfig.stopControlEnabled,
+			"whether to enable the stop control feature. Default value is false")
 		flags.BoolVar(&builder.localServiceAPIEnabled, "local-service-api-enabled", defaultConfig.localServiceAPIEnabled, "whether to use local indexed data for api queries")
 		flags.StringVar(&builder.registersDBPath, "execution-state-dir", defaultConfig.registersDBPath, "directory to use for execution-state database")
 		flags.StringVar(&builder.checkpointFile, "execution-state-checkpoint", defaultConfig.checkpointFile, "execution-state checkpoint file")
@@ -1513,7 +1521,7 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.programCacheSize > 0,
 			)
 
-			err = builder.ScriptExecutor.Initialize(builder.ExecutionIndexer, scripts, builder.versionControl)
+			err = builder.ScriptExecutor.Initialize(builder.ExecutionIndexer, scripts, builder.VersionControl)
 			if err != nil {
 				return nil, err
 			}
@@ -1523,6 +1531,10 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				return nil, err
 			}
 
+			if builder.stopControlEnabled {
+				builder.StopControl.RegisterHeightRecorder(builder.ExecutionIndexer)
+			}
+
 			return builder.ExecutionIndexer, nil
 		}, builder.IndexerDependencies)
 	}
@@ -1826,6 +1838,8 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 
 	versionControlDependable := module.NewProxiedReadyDoneAware()
 	builder.IndexerDependencies.Add(versionControlDependable)
+	stopControlDependable := module.NewProxiedReadyDoneAware()
+	builder.IndexerDependencies.Add(stopControlDependable)
 
 	builder.Component("version control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 		if !builder.versionControlEnabled {
@@ -1854,11 +1868,30 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 		// VersionControl needs to consume BlockFinalized events.
 		node.ProtocolEvents.AddConsumer(versionControl)
 
-		builder.versionControl = versionControl
-		versionControlDependable.Init(builder.versionControl)
+		builder.VersionControl = versionControl
+		versionControlDependable.Init(builder.VersionControl)
 
 		return versionControl, nil
 	})
+	builder.Component("stop control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+		if !builder.stopControlEnabled {
+			noop := &module.NoopReadyDoneAware{}
+			stopControlDependable.Init(noop)
+			return noop, nil
+		}
+
+		stopControl := stop.NewStopControl(
+			builder.Logger,
+		)
+
+		builder.VersionControl.AddVersionUpdatesConsumer(stopControl.OnVersionUpdate)
+
+		builder.StopControl = stopControl
+		stopControlDependable.Init(builder.StopControl)
+
+		return stopControl, nil
+	})
+
 	builder.Component("RPC engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 		accessMetrics := builder.AccessMetrics
 		config := builder.rpcConf
@@ -1937,7 +1970,7 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 				builder.stateStreamConf.ClientSendBufferSize,
 			),
 			IndexReporter:  indexReporter,
-			VersionControl: builder.versionControl,
+			VersionControl: builder.VersionControl,
 		}
 
 		if builder.localServiceAPIEnabled {
diff --git a/engine/access/access_test.go b/engine/access/access_test.go
index 4aa8fc047ca..d7ea12e613c 100644
--- a/engine/access/access_test.go
+++ b/engine/access/access_test.go
@@ -686,8 +686,7 @@ func (suite *Suite) TestGetSealedTransaction() {
 		// create the ingest engine
 		processedHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineBlockHeight)
 
-		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections,
-			transactions, results, receipts, collectionExecutedMetric, processedHeight, lastFullBlockHeight)
+		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections, transactions, results, receipts, collectionExecutedMetric, processedHeight, lastFullBlockHeight)
 		require.NoError(suite.T(), err)
 
 		// 1. Assume that follower engine updated the block storage and the protocol state. The block is reported as sealed
@@ -848,8 +847,7 @@ func (suite *Suite) TestGetTransactionResult() {
 		require.NoError(suite.T(), err)
 
 		// create the ingest engine
-		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections,
-			transactions, results, receipts, collectionExecutedMetric, processedHeight, lastFullBlockHeight)
+		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections, transactions, results, receipts, collectionExecutedMetric, processedHeight, lastFullBlockHeight)
 		require.NoError(suite.T(), err)
 
 		background, cancel := context.WithCancel(context.Background())
@@ -1078,8 +1076,7 @@ func (suite *Suite) TestExecuteScript() {
 		require.NoError(suite.T(), err)
 
 		// create the ingest engine
-		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections,
-			transactions, results, receipts, collectionExecutedMetric, processedHeight, lastFullBlockHeight)
+		ingestEng, err := ingestion.New(suite.log, suite.net, suite.state, suite.me, suite.request, all.Blocks, all.Headers, collections, transactions, results, receipts, collectionExecutedMetric, processedHeight, lastFullBlockHeight)
 		require.NoError(suite.T(), err)
 
 		// create another block as a predecessor of the block created earlier
diff --git a/engine/access/ingestion/engine_test.go b/engine/access/ingestion/engine_test.go
index 5e00e720334..d93056bd80c 100644
--- a/engine/access/ingestion/engine_test.go
+++ b/engine/access/ingestion/engine_test.go
@@ -189,8 +189,7 @@ func (s *Suite) initIngestionEngine(ctx irrecoverable.SignalerContext) *Engine {
 	)
 	require.NoError(s.T(), err)
 
-	eng, err := New(s.log, s.net, s.proto.state, s.me, s.request, s.blocks, s.headers, s.collections,
-		s.transactions, s.results, s.receipts, s.collectionExecutedMetric, processedHeight, s.lastFullBlockHeight)
+	eng, err := New(s.log, s.net, s.proto.state, s.me, s.request, s.blocks, s.headers, s.collections, s.transactions, s.results, s.receipts, s.collectionExecutedMetric, processedHeight, s.lastFullBlockHeight)
 	require.NoError(s.T(), err)
 
 	eng.ComponentManager.Start(ctx)
diff --git a/engine/common/stop/stop_control.go b/engine/common/stop/stop_control.go
new file mode 100644
index 00000000000..81d5eea52d2
--- /dev/null
+++ b/engine/common/stop/stop_control.go
@@ -0,0 +1,157 @@
+package stop
+
+import (
+	"fmt"
+
+	"github.com/coreos/go-semver/semver"
+	"github.com/rs/zerolog"
+	"go.uber.org/atomic"
+
+	"github.com/onflow/flow-go/module/component"
+	"github.com/onflow/flow-go/module/counters"
+	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
+	"github.com/onflow/flow-go/module/irrecoverable"
+)
+
+type VersionMetadata struct {
+	// incompatibleBlockHeight is the height of the block that is incompatible with the current node version.
+	incompatibleBlockHeight uint64
+	// updatedVersion is the expected node version to continue working with new blocks.
+	updatedVersion string
+}
+
+// StopControl is responsible for managing the stopping behavior of the node
+// when an incompatible block height is encountered.
+type StopControl struct {
+	component.Component
+	cm *component.ComponentManager
+
+	log zerolog.Logger
+
+	versionData *atomic.Pointer[VersionMetadata]
+
+	// Notifier for new processed block height
+	processedHeightChannel chan uint64
+	// Signal channel to notify when processing is done
+	doneProcessingEvents chan struct{}
+
+	// Stores latest processed block height
+	lastProcessedHeight counters.StrictMonotonousCounter
+}
+
+// NewStopControl creates a new StopControl instance.
+//
+// Parameters:
+//   - log: The logger used for logging.
+//
+// Returns:
+//   - A pointer to the newly created StopControl instance.
+func NewStopControl(
+	log zerolog.Logger,
+) *StopControl {
+	sc := &StopControl{
+		log: log.With().
+			Str("component", "stop_control").
+			Logger(),
+		lastProcessedHeight:    counters.NewMonotonousCounter(0),
+		versionData:            atomic.NewPointer[VersionMetadata](nil),
+		processedHeightChannel: make(chan uint64),
+		doneProcessingEvents:   make(chan struct{}),
+	}
+
+	sc.cm = component.NewComponentManagerBuilder().
+		AddWorker(sc.processEvents).
+		Build()
+	sc.Component = sc.cm
+
+	return sc
+}
+
+// OnVersionUpdate is called when a version update occurs.
+//
+// It updates the incompatible block height and the expected node version
+// based on the provided height and semver.
+//
+// Parameters:
+//   - height: The block height that is incompatible with the current node version.
+//   - version: The new semantic version object that is expected for compatibility.
+func (sc *StopControl) OnVersionUpdate(height uint64, version *semver.Version) {
+	// If the version was updated, store new version information
+	if version != nil {
+		sc.log.Info().
+			Uint64("height", height).
+			Str("semver", version.String()).
+			Msg("Received version update")
+
+		sc.versionData.Store(&VersionMetadata{
+			incompatibleBlockHeight: height,
+			updatedVersion:          version.String(),
+		})
+		return
+	}
+
+	// If semver is 0, but notification was received, this means that the version update was deleted.
+	sc.versionData.Store(nil)
+}
+
+// onProcessedBlock is called when a new block is processed block.
+// when the last compatible block is processed, the StopControl will cause the node to crash
+//
+// Parameters:
+//   - ctx: The context used to signal an irrecoverable error.
+func (sc *StopControl) onProcessedBlock(ctx irrecoverable.SignalerContext) {
+	versionData := sc.versionData.Load()
+	if versionData == nil {
+		return
+	}
+
+	newHeight := sc.lastProcessedHeight.Value()
+	if newHeight >= versionData.incompatibleBlockHeight-1 {
+		ctx.Throw(fmt.Errorf("processed block at height %d is incompatible with the current node version, please upgrade to version %s starting from block height %d",
+			newHeight, versionData.updatedVersion, versionData.incompatibleBlockHeight))
+	}
+}
+
+// updateProcessedHeight updates the last processed height and triggers notifications.
+//
+// Parameters:
+//   - height: The height of the latest processed block.
+func (sc *StopControl) updateProcessedHeight(height uint64) {
+	select {
+	case sc.processedHeightChannel <- height: // Successfully sent the height to the channel
+	case <-sc.doneProcessingEvents: // Process events are done, do not block
+	}
+}
+
+// RegisterHeightRecorder registers an execution data height recorder with the StopControl.
+//
+// Parameters:
+//   - recorder: The execution data height recorder to register.
+func (sc *StopControl) RegisterHeightRecorder(recorder execution_data.ProcessedHeightRecorder) {
+	recorder.SetHeightUpdatesConsumer(sc.updateProcessedHeight)
+}
+
+// processEvents processes incoming events related to block heights and version updates.
+//
+// Parameters:
+//   - ctx: The context used to handle irrecoverable errors.
+//   - ready: A function to signal that the component is ready to start processing events.
+func (sc *StopControl) processEvents(ctx irrecoverable.SignalerContext, ready component.ReadyFunc) {
+	ready()
+
+	defer close(sc.doneProcessingEvents) // Ensure the signal channel is closed when done
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case height, ok := <-sc.processedHeightChannel:
+			if !ok {
+				return
+			}
+			if sc.lastProcessedHeight.Set(height) {
+				sc.onProcessedBlock(ctx)
+			}
+		}
+	}
+}
diff --git a/engine/common/stop/stop_control_test.go b/engine/common/stop/stop_control_test.go
new file mode 100644
index 00000000000..75d5c6e8026
--- /dev/null
+++ b/engine/common/stop/stop_control_test.go
@@ -0,0 +1,138 @@
+package stop
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/coreos/go-semver/semver"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	"github.com/onflow/flow-go/module/irrecoverable"
+	"github.com/onflow/flow-go/utils/unittest"
+)
+
+// RunWithStopControl is a helper function that creates a StopControl instance and runs the provided test function with it.
+//
+// Parameters:
+//   - t: The testing context.
+//   - f: A function that takes a MockSignalerContext and a StopControl, used to run the test logic.
+func RunWithStopControl(t *testing.T, f func(ctx *irrecoverable.MockSignalerContext, sc *StopControl)) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	signalerContext := irrecoverable.NewMockSignalerContext(t, ctx)
+
+	f(signalerContext, createStopControl(t, signalerContext))
+}
+
+// createStopControl creates and starts a new StopControl instance.
+//
+// Parameters:
+//   - t: The testing context.
+//   - signalerContext: The mock context used to simulate signaler behavior.
+//
+// Returns:
+//   - A pointer to the newly created and started StopControl instance.
+func createStopControl(t *testing.T, signalerContext *irrecoverable.MockSignalerContext) *StopControl {
+	sc := NewStopControl(zerolog.Nop())
+	assert.NotNil(t, sc)
+
+	// Start the StopControl component.
+	sc.Start(signalerContext)
+
+	return sc
+}
+
+// TestNewStopControl verifies that a new StopControl instance is created correctly and its components are ready.
+//
+// This test ensures that the StopControl can be initialized and started properly, and that all components are ready
+// within a specified time frame.
+func TestNewStopControl(t *testing.T) {
+	RunWithStopControl(t, func(_ *irrecoverable.MockSignalerContext, sc *StopControl) {
+		unittest.RequireComponentsReadyBefore(t, 2*time.Second, sc)
+	})
+}
+
+// TestStopControl_OnVersionUpdate tests the OnVersionUpdate method of the StopControl.
+//
+// This test covers two scenarios:
+// 1. When a valid version update is received, it checks that the version data is stored correctly.
+// 2. When a nil version is provided, it checks that the version data is cleared.
+func TestStopControl_OnVersionUpdate(t *testing.T) {
+	RunWithStopControl(t, func(_ *irrecoverable.MockSignalerContext, sc *StopControl) {
+
+		// Case 1: Version is updated
+		height := uint64(100)
+		version := semver.New("1.0.0")
+
+		sc.OnVersionUpdate(height, version)
+
+		// Verify that the version data is correctly stored.
+		versionData := sc.versionData.Load()
+		assert.NotNil(t, versionData)
+		assert.Equal(t, height, versionData.incompatibleBlockHeight)
+		assert.Equal(t, "1.0.0", versionData.updatedVersion)
+
+		// Case 2: Version update is deleted (nil version)
+		sc.OnVersionUpdate(0, nil)
+
+		// Verify that the version data is cleared.
+		versionData = sc.versionData.Load()
+		assert.Nil(t, versionData)
+	})
+}
+
+// TestStopControl_OnProcessedBlock tests the onProcessedBlock method of the StopControl.
+//
+// This test covers multiple scenarios related to processing block heights:
+// 1. Verifying that the processed height is updated correctly.
+// 2. Ensuring that a lower processed height cannot overwrite a higher one.
+// 3. Testing that the StopControl correctly triggers an irrecoverable error (via Throw) when the incompatible block height is reached.
+func TestStopControl_OnProcessedBlock(t *testing.T) {
+	RunWithStopControl(t, func(ctx *irrecoverable.MockSignalerContext, sc *StopControl) {
+		// Initial block height
+		height := uint64(10)
+
+		// Update processed height and verify it's stored correctly.
+		sc.updateProcessedHeight(height)
+		assert.Equal(t, height, sc.lastProcessedHeight.Value())
+
+		// Attempt to set a lower processed height, which should not be allowed.
+		sc.updateProcessedHeight(height - 1)
+		assert.Equal(t, height, sc.lastProcessedHeight.Value())
+
+		// Set version metadata with an incompatible height and verify the processed height behavior.
+		incompatibleHeight := uint64(13)
+		version := semver.New("1.0.0")
+
+		sc.OnVersionUpdate(incompatibleHeight, version)
+		height = incompatibleHeight - 2
+		sc.updateProcessedHeight(height)
+		assert.Equal(t, height, sc.lastProcessedHeight.Value())
+
+		// Prepare to trigger the Throw method when the incompatible block height is processed.
+		height = incompatibleHeight - 1
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+
+		// Expected error message when the incompatible block height is processed.
+		expectedError := fmt.Errorf("processed block at height %d is incompatible with the current node version, please upgrade to version %s starting from block height %d", height, version.String(), incompatibleHeight)
+
+		// Set expectation that the Throw method will be called with the expected error.
+		ctx.On("Throw", expectedError).Run(func(args mock.Arguments) { wg.Done() }).Return().Once()
+
+		// Update the processed height to the incompatible height and wait for Throw to be called.
+		sc.updateProcessedHeight(height)
+		unittest.RequireReturnsBefore(t, wg.Wait, 100*time.Millisecond, "expect for ctx.Throw before timeout")
+
+		// Verify that the processed height and the Throw method call are correct.
+		assert.Equal(t, height, sc.lastProcessedHeight.Value())
+		ctx.AssertCalled(t, "Throw", expectedError)
+	})
+}
diff --git a/module/executiondatasync/execution_data/mock/downloader.go b/module/executiondatasync/execution_data/mock/downloader.go
index 114e2e4bba5..6d384730d7a 100644
--- a/module/executiondatasync/execution_data/mock/downloader.go
+++ b/module/executiondatasync/execution_data/mock/downloader.go
@@ -109,6 +109,11 @@ func (_m *Downloader) Ready() <-chan struct{} {
 	return r0
 }
 
+// SetHeightUpdatesConsumer provides a mock function with given fields: _a0
+func (_m *Downloader) SetHeightUpdatesConsumer(_a0 execution_data.HeightUpdatesConsumer) {
+	_m.Called(_a0)
+}
+
 // NewDownloader creates a new instance of Downloader. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewDownloader(t interface {
diff --git a/module/executiondatasync/execution_data/processed_height_recorder.go b/module/executiondatasync/execution_data/processed_height_recorder.go
index fd5966cb2c2..4a59effdff4 100644
--- a/module/executiondatasync/execution_data/processed_height_recorder.go
+++ b/module/executiondatasync/execution_data/processed_height_recorder.go
@@ -1,9 +1,13 @@
 package execution_data
 
 import (
+	"sync"
+
 	"go.uber.org/atomic"
 )
 
+type HeightUpdatesConsumer func(height uint64)
+
 // ProcessedHeightRecorder is an interface for tracking the highest execution data processed
 // height when a block is processed and for providing this height.
 type ProcessedHeightRecorder interface {
@@ -11,6 +15,10 @@ type ProcessedHeightRecorder interface {
 	OnBlockProcessed(uint64)
 	// HighestCompleteHeight returns the highest complete processed block height.
 	HighestCompleteHeight() uint64
+
+	// SetHeightUpdatesConsumer subscribe consumer for processed height updates.
+	// Callback are called synchronously and must be non-blocking
+	SetHeightUpdatesConsumer(HeightUpdatesConsumer)
 }
 
 var _ ProcessedHeightRecorder = (*ProcessedHeightRecorderManager)(nil)
@@ -18,7 +26,9 @@ var _ ProcessedHeightRecorder = (*ProcessedHeightRecorderManager)(nil)
 // ProcessedHeightRecorderManager manages an execution data height recorder
 // and tracks the highest processed block height.
 type ProcessedHeightRecorderManager struct {
+	lock                  sync.RWMutex
 	highestCompleteHeight *atomic.Uint64
+	consumer              HeightUpdatesConsumer
 }
 
 // NewProcessedHeightRecorderManager creates a new ProcessedHeightRecorderManager with the given initial height.
@@ -32,6 +42,12 @@ func NewProcessedHeightRecorderManager(initHeight uint64) *ProcessedHeightRecord
 func (e *ProcessedHeightRecorderManager) OnBlockProcessed(height uint64) {
 	if height > e.highestCompleteHeight.Load() {
 		e.highestCompleteHeight.Store(height)
+
+		e.lock.RLock()
+		defer e.lock.RUnlock()
+		if e.consumer != nil {
+			e.consumer(height)
+		}
 	}
 }
 
@@ -39,3 +55,11 @@ func (e *ProcessedHeightRecorderManager) OnBlockProcessed(height uint64) {
 func (e *ProcessedHeightRecorderManager) HighestCompleteHeight() uint64 {
 	return e.highestCompleteHeight.Load()
 }
+
+// SetHeightUpdatesConsumer subscribe consumers for processed height updates.
+func (e *ProcessedHeightRecorderManager) SetHeightUpdatesConsumer(consumer HeightUpdatesConsumer) {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+
+	e.consumer = consumer
+}
diff --git a/storage/pebble/open.go b/storage/pebble/open.go
index 80f328ce87a..4e81f1155e1 100644
--- a/storage/pebble/open.go
+++ b/storage/pebble/open.go
@@ -1,9 +1,8 @@
 package pebble
 
 import (
-	"fmt"
-
 	"errors"
+	"fmt"
 
 	"github.com/cockroachdb/pebble"
 	"github.com/hashicorp/go-multierror"

From ca923834eac6bf78d96e755840e2e3e9fbee80c0 Mon Sep 17 00:00:00 2001
From: Andrii Slisarchuk <andriyslisarchuk@gmail.com>
Date: Thu, 7 Nov 2024 16:11:38 +0200
Subject: [PATCH 5/5] Fixed compiler errors

---
 engine/access/mock/execution_api_client.go | 37 ++++++++++++++++++++++
 engine/access/mock/execution_api_server.go | 30 ++++++++++++++++++
 engine/execution/rpc/engine.go             |  8 +++++
 go.mod                                     |  2 +-
 go.sum                                     |  4 +--
 insecure/go.mod                            |  2 +-
 insecure/go.sum                            |  4 +--
 7 files changed, 81 insertions(+), 6 deletions(-)

diff --git a/engine/access/mock/execution_api_client.go b/engine/access/mock/execution_api_client.go
index fd5fc20c718..7aa04d143c7 100644
--- a/engine/access/mock/execution_api_client.go
+++ b/engine/access/mock/execution_api_client.go
@@ -349,6 +349,43 @@ func (_m *ExecutionAPIClient) GetTransactionErrorMessagesByBlockID(ctx context.C
 	return r0, r1
 }
 
+// GetTransactionExecutionMetricsAfter provides a mock function with given fields: ctx, in, opts
+func (_m *ExecutionAPIClient) GetTransactionExecutionMetricsAfter(ctx context.Context, in *execution.GetTransactionExecutionMetricsAfterRequest, opts ...grpc.CallOption) (*execution.GetTransactionExecutionMetricsAfterResponse, error) {
+	_va := make([]interface{}, len(opts))
+	for _i := range opts {
+		_va[_i] = opts[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, in)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetTransactionExecutionMetricsAfter")
+	}
+
+	var r0 *execution.GetTransactionExecutionMetricsAfterResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *execution.GetTransactionExecutionMetricsAfterRequest, ...grpc.CallOption) (*execution.GetTransactionExecutionMetricsAfterResponse, error)); ok {
+		return rf(ctx, in, opts...)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *execution.GetTransactionExecutionMetricsAfterRequest, ...grpc.CallOption) *execution.GetTransactionExecutionMetricsAfterResponse); ok {
+		r0 = rf(ctx, in, opts...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*execution.GetTransactionExecutionMetricsAfterResponse)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *execution.GetTransactionExecutionMetricsAfterRequest, ...grpc.CallOption) error); ok {
+		r1 = rf(ctx, in, opts...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetTransactionResult provides a mock function with given fields: ctx, in, opts
 func (_m *ExecutionAPIClient) GetTransactionResult(ctx context.Context, in *execution.GetTransactionResultRequest, opts ...grpc.CallOption) (*execution.GetTransactionResultResponse, error) {
 	_va := make([]interface{}, len(opts))
diff --git a/engine/access/mock/execution_api_server.go b/engine/access/mock/execution_api_server.go
index e61517cb617..2c3f1f3b5a7 100644
--- a/engine/access/mock/execution_api_server.go
+++ b/engine/access/mock/execution_api_server.go
@@ -284,6 +284,36 @@ func (_m *ExecutionAPIServer) GetTransactionErrorMessagesByBlockID(_a0 context.C
 	return r0, r1
 }
 
+// GetTransactionExecutionMetricsAfter provides a mock function with given fields: _a0, _a1
+func (_m *ExecutionAPIServer) GetTransactionExecutionMetricsAfter(_a0 context.Context, _a1 *execution.GetTransactionExecutionMetricsAfterRequest) (*execution.GetTransactionExecutionMetricsAfterResponse, error) {
+	ret := _m.Called(_a0, _a1)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetTransactionExecutionMetricsAfter")
+	}
+
+	var r0 *execution.GetTransactionExecutionMetricsAfterResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *execution.GetTransactionExecutionMetricsAfterRequest) (*execution.GetTransactionExecutionMetricsAfterResponse, error)); ok {
+		return rf(_a0, _a1)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *execution.GetTransactionExecutionMetricsAfterRequest) *execution.GetTransactionExecutionMetricsAfterResponse); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*execution.GetTransactionExecutionMetricsAfterResponse)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *execution.GetTransactionExecutionMetricsAfterRequest) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetTransactionResult provides a mock function with given fields: _a0, _a1
 func (_m *ExecutionAPIServer) GetTransactionResult(_a0 context.Context, _a1 *execution.GetTransactionResultRequest) (*execution.GetTransactionResultResponse, error) {
 	ret := _m.Called(_a0, _a1)
diff --git a/engine/execution/rpc/engine.go b/engine/execution/rpc/engine.go
index 260495f1bf1..1579bc6522f 100644
--- a/engine/execution/rpc/engine.go
+++ b/engine/execution/rpc/engine.go
@@ -179,6 +179,14 @@ func (h *handler) Ping(
 	return &execution.PingResponse{}, nil
 }
 
+func (h *handler) GetTransactionExecutionMetricsAfter(
+	ctx context.Context,
+	request *execution.GetTransactionExecutionMetricsAfterRequest,
+) (*execution.GetTransactionExecutionMetricsAfterResponse, error) {
+	// Implementation should be backported
+	return nil, nil
+}
+
 func (h *handler) ExecuteScriptAtBlockID(
 	ctx context.Context,
 	req *execution.ExecuteScriptAtBlockIDRequest,
diff --git a/go.mod b/go.mod
index de885fb1d90..44e94ac193d 100644
--- a/go.mod
+++ b/go.mod
@@ -54,7 +54,7 @@ require (
 	github.com/onflow/flow-core-contracts/lib/go/contracts v1.3.1
 	github.com/onflow/flow-core-contracts/lib/go/templates v1.3.1
 	github.com/onflow/flow-go-sdk v1.0.0-preview.54
-	github.com/onflow/flow/protobuf/go/flow v0.4.5
+	github.com/onflow/flow/protobuf/go/flow v0.4.7
 	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58
 	github.com/pierrec/lz4 v2.6.1+incompatible
 	github.com/pkg/errors v0.9.1
diff --git a/go.sum b/go.sum
index bd046d60233..ab1e9de0246 100644
--- a/go.sum
+++ b/go.sum
@@ -2192,8 +2192,8 @@ github.com/onflow/flow-nft/lib/go/contracts v1.2.1/go.mod h1:2gpbza+uzs1k7x31hkp
 github.com/onflow/flow-nft/lib/go/templates v1.2.0 h1:JSQyh9rg0RC+D1930BiRXN8lrtMs+ubVMK6aQPon6Yc=
 github.com/onflow/flow-nft/lib/go/templates v1.2.0/go.mod h1:p+2hRvtjLUR3MW1NsoJe5Gqgr2eeH49QB6+s6ze00w0=
 github.com/onflow/flow/protobuf/go/flow v0.3.2-0.20231121210617-52ee94b830c2/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
-github.com/onflow/flow/protobuf/go/flow v0.4.5 h1:6o+pgYGqwXdEhqSJxu2BdnDXkOQVOkfGAb6IiXB+NPM=
-github.com/onflow/flow/protobuf/go/flow v0.4.5/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
+github.com/onflow/flow/protobuf/go/flow v0.4.7 h1:iP6DFx4wZ3ETORsyeqzHu7neFT3d1CXF6wdK+AOOjmc=
+github.com/onflow/flow/protobuf/go/flow v0.4.7/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c h1:T0jDCm7k7uqDo26JiiujQ5oryl30itPnlmZQywTu9ng=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c/go.mod h1:XYnWtulwJvHVOr2B0WVA/UC3dvRgFevjp8Pn9a3E1xo=
 github.com/onflow/go-ethereum v1.14.7 h1:gg3awYqI02e3AypRdpJKEvNTJ6kz/OhAqRti0h54Wlc=
diff --git a/insecure/go.mod b/insecure/go.mod
index e99fd5c109f..9dcd977429c 100644
--- a/insecure/go.mod
+++ b/insecure/go.mod
@@ -211,7 +211,7 @@ require (
 	github.com/onflow/flow-go-sdk v1.0.0-preview.54 // indirect
 	github.com/onflow/flow-nft/lib/go/contracts v1.2.1 // indirect
 	github.com/onflow/flow-nft/lib/go/templates v1.2.0 // indirect
-	github.com/onflow/flow/protobuf/go/flow v0.4.5 // indirect
+	github.com/onflow/flow/protobuf/go/flow v0.4.7 // indirect
 	github.com/onflow/go-ethereum v1.14.7 // indirect
 	github.com/onflow/sdks v0.6.0-preview.1 // indirect
 	github.com/onflow/wal v1.0.2 // indirect
diff --git a/insecure/go.sum b/insecure/go.sum
index b9c19f5364a..1279d79b70c 100644
--- a/insecure/go.sum
+++ b/insecure/go.sum
@@ -2178,8 +2178,8 @@ github.com/onflow/flow-nft/lib/go/contracts v1.2.1/go.mod h1:2gpbza+uzs1k7x31hkp
 github.com/onflow/flow-nft/lib/go/templates v1.2.0 h1:JSQyh9rg0RC+D1930BiRXN8lrtMs+ubVMK6aQPon6Yc=
 github.com/onflow/flow-nft/lib/go/templates v1.2.0/go.mod h1:p+2hRvtjLUR3MW1NsoJe5Gqgr2eeH49QB6+s6ze00w0=
 github.com/onflow/flow/protobuf/go/flow v0.3.2-0.20231121210617-52ee94b830c2/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
-github.com/onflow/flow/protobuf/go/flow v0.4.5 h1:6o+pgYGqwXdEhqSJxu2BdnDXkOQVOkfGAb6IiXB+NPM=
-github.com/onflow/flow/protobuf/go/flow v0.4.5/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
+github.com/onflow/flow/protobuf/go/flow v0.4.7 h1:iP6DFx4wZ3ETORsyeqzHu7neFT3d1CXF6wdK+AOOjmc=
+github.com/onflow/flow/protobuf/go/flow v0.4.7/go.mod h1:NA2pX2nw8zuaxfKphhKsk00kWLwfd+tv8mS23YXO4Sk=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c h1:T0jDCm7k7uqDo26JiiujQ5oryl30itPnlmZQywTu9ng=
 github.com/onflow/go-ds-pebble v0.0.0-20240731130313-f186539f382c/go.mod h1:XYnWtulwJvHVOr2B0WVA/UC3dvRgFevjp8Pn9a3E1xo=
 github.com/onflow/go-ethereum v1.14.7 h1:gg3awYqI02e3AypRdpJKEvNTJ6kz/OhAqRti0h54Wlc=