[Access] Store Transaction Result error messages in db

cmd/access/node_builder/access_node_builder.go

@@ -163,7 +163,6 @@ type AccessNodeConfig struct {
 	executionDataConfig                  edrequester.ExecutionDataConfig
 	PublicNetworkConfig                  PublicNetworkConfig
 	TxResultCacheSize                    uint
-	TxErrorMessagesCacheSize             uint
 	executionDataIndexingEnabled         bool
 	registersDBPath                      string
 	checkpointFile                       string
@@ -175,6 +174,7 @@ type AccessNodeConfig struct {
 	programCacheSize                     uint
 	checkPayerBalanceMode                string
 	versionControlEnabled                bool
+	storeTxResultErrorMessages           bool
 	stopControlEnabled                   bool
 	registerDBPruneThreshold             uint64
 }
@@ -248,7 +248,6 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		apiRatelimits:                nil,
 		apiBurstlimits:               nil,
 		TxResultCacheSize:            0,
-		TxErrorMessagesCacheSize:     1000,
 		PublicNetworkConfig: PublicNetworkConfig{
 			BindAddress: cmd.NotSet,
 			Metrics:     metrics.NewNoopCollector(),
@@ -280,6 +279,7 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		programCacheSize:                     0,
 		checkPayerBalanceMode:                accessNode.Disabled.String(),
 		versionControlEnabled:                true,
+		storeTxResultErrorMessages:           false,
 		stopControlEnabled:                   false,
 		registerDBPruneThreshold:             pruner.DefaultThreshold,
 	}
@@ -351,6 +351,7 @@ type FlowAccessNodeBuilder struct {
 	stateStreamGrpcServer *grpcserver.GrpcServer
 
 	stateStreamBackend *statestreambackend.StateStreamBackend
+	nodeBackend        *backend.Backend
 }
 
 func (builder *FlowAccessNodeBuilder) buildFollowerState() *FlowAccessNodeBuilder {
@@ -1241,7 +1242,6 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 		flags.BoolVar(&builder.retryEnabled, "retry-enabled", defaultConfig.retryEnabled, "whether to enable the retry mechanism at the access node level")
 		flags.BoolVar(&builder.rpcMetricsEnabled, "rpc-metrics-enabled", defaultConfig.rpcMetricsEnabled, "whether to enable the rpc metrics")
 		flags.UintVar(&builder.TxResultCacheSize, "transaction-result-cache-size", defaultConfig.TxResultCacheSize, "transaction result cache size.(Disabled by default i.e 0)")
-		flags.UintVar(&builder.TxErrorMessagesCacheSize, "transaction-error-messages-cache-size", defaultConfig.TxErrorMessagesCacheSize, "transaction error messages cache size.(By default 1000)")
 		flags.StringVarP(&builder.nodeInfoFile,
 			"node-info-file",
 			"",
@@ -1375,7 +1375,10 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"tx-result-query-mode",
 			defaultConfig.rpcConf.BackendConfig.TxResultQueryMode,
 			"mode to use when querying transaction results. one of [local-only, execution-nodes-only(default), failover]")
-
+		flags.BoolVar(&builder.storeTxResultErrorMessages,
+			"store-tx-result-error-messages",
+			defaultConfig.storeTxResultErrorMessages,
+			"whether to enable storing transaction error messages into the db")
 		// Script Execution
 		flags.StringVar(&builder.rpcConf.BackendConfig.ScriptExecutionMode,
 			"script-execution-mode",
@@ -1488,9 +1491,6 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 				return errors.New("circuit-breaker-restore-timeout must be greater than 0")
 			}
 		}
-		if builder.TxErrorMessagesCacheSize == 0 {
-			return errors.New("transaction-error-messages-cache-size must be greater than 0")
-		}
 
 		if builder.checkPayerBalanceMode != accessNode.Disabled.String() && !builder.executionDataIndexingEnabled {
 			return errors.New("execution-data-indexing-enabled must be set if check-payer-balance is enabled")
@@ -1603,7 +1603,8 @@ func (builder *FlowAccessNodeBuilder) enqueueRelayNetwork() {
 }
 
 func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
-	var processedBlockHeight storage.ConsumerProgress
+	var processedFinalizedBlockHeight storage.ConsumerProgress
+	var processedTxErrorMessagesBlockHeight storage.ConsumerProgress
 
 	if builder.executionDataSyncEnabled {
 		builder.BuildExecutionSyncComponents()
@@ -1799,8 +1800,15 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.Storage.LightTransactionResults)
 			return nil
 		}).
-		Module("processed block height consumer progress", func(node *cmd.NodeConfig) error {
-			processedBlockHeight = bstorage.NewConsumerProgress(builder.DB, module.ConsumeProgressIngestionEngineBlockHeight)
+		Module("processed finalized block height consumer progress", func(node *cmd.NodeConfig) error {
+			processedFinalizedBlockHeight = bstorage.NewConsumerProgress(builder.DB, module.ConsumeProgressIngestionEngineBlockHeight)
+			return nil
+		}).
+		Module("processed error messages block height consumer progress", func(node *cmd.NodeConfig) error {
+			processedTxErrorMessagesBlockHeight = bstorage.NewConsumerProgress(
+				builder.DB,
+				module.ConsumeProgressIngestionEngineTxErrorMessagesBlockHeight,
+			)
 			return nil
 		}).
 		Module("processed last full block height monotonic consumer progress", func(node *cmd.NodeConfig) error {
@@ -1817,6 +1825,13 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			return nil
 		}).
+		Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
+			if builder.storeTxResultErrorMessages {
+				builder.Storage.TransactionResultErrorMessages = bstorage.NewTransactionResultErrorMessages(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)
+			}
+
+			return nil
+		}).
 		Component("version control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			if !builder.versionControlEnabled {
 				noop := &module.NoopReadyDoneAware{}
@@ -1944,7 +1959,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			}
 
-			nodeBackend, err := backend.New(backend.Params{
+			builder.nodeBackend, err = backend.New(backend.Params{
 				State:                     node.State,
 				CollectionRPC:             builder.CollectionRPC,
 				HistoricalAccessNodes:     builder.HistoricalAccessRPCs,
@@ -1954,6 +1969,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				Transactions:              node.Storage.Transactions,
 				ExecutionReceipts:         node.Storage.Receipts,
 				ExecutionResults:          node.Storage.Results,
+				TxResultErrorMessages:     node.Storage.TransactionResultErrorMessages,
 				ChainID:                   node.RootChainID,
 				AccessMetrics:             builder.AccessMetrics,
 				ConnFactory:               connFactory,
@@ -1965,7 +1981,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				SnapshotHistoryLimit:      backend.DefaultSnapshotHistoryLimit,
 				Communicator:              backend.NewNodeCommunicator(backendConfig.CircuitBreakerConfig.Enabled),
 				TxResultCacheSize:         builder.TxResultCacheSize,
-				TxErrorMessagesCacheSize:  builder.TxErrorMessagesCacheSize,
 				ScriptExecutor:            builder.ScriptExecutor,
 				ScriptExecutionMode:       scriptExecMode,
 				CheckPayerBalanceMode:     checkPayerBalanceMode,
@@ -1997,8 +2012,8 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.AccessMetrics,
 				builder.rpcMetricsEnabled,
 				builder.Me,
-				nodeBackend,
-				nodeBackend,
+				builder.nodeBackend,
+				builder.nodeBackend,
 				builder.secureGrpcServer,
 				builder.unsecureGrpcServer,
 				builder.stateStreamBackend,
@@ -2049,9 +2064,14 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				node.Storage.Transactions,
 				node.Storage.Results,
 				node.Storage.Receipts,
+				node.Storage.TransactionResultErrorMessages,
 				builder.collectionExecutedMetric,
-				processedBlockHeight,
+				processedFinalizedBlockHeight,
+				processedTxErrorMessagesBlockHeight,
 				lastFullBlockHeight,
+				builder.nodeBackend,
+				builder.rpcConf.BackendConfig.PreferredExecutionNodeIDs,
+				builder.rpcConf.BackendConfig.FixedExecutionNodeIDs,
 			)
 			if err != nil {
 				return nil, err

engine/access/access_test.go

@@ -329,16 +329,15 @@ func (suite *Suite) TestSendTransactionToRandomCollectionNode() {
 		connFactory.On("GetAccessAPIClient", collNode2.Address, nil).Return(col2ApiClient, &mockCloser{}, nil)
 
 		bnd, err := backend.New(backend.Params{State: suite.state,
-			Collections:              collections,
-			Transactions:             transactions,
-			ChainID:                  suite.chainID,
-			AccessMetrics:            metrics,
-			ConnFactory:              connFactory,
-			MaxHeightRange:           backend.DefaultMaxHeightRange,
-			Log:                      suite.log,
-			SnapshotHistoryLimit:     backend.DefaultSnapshotHistoryLimit,
-			Communicator:             backend.NewNodeCommunicator(false),
-			TxErrorMessagesCacheSize: 1000,
+			Collections:          collections,
+			Transactions:         transactions,
+			ChainID:              suite.chainID,
+			AccessMetrics:        metrics,
+			ConnFactory:          connFactory,
+			MaxHeightRange:       backend.DefaultMaxHeightRange,
+			Log:                  suite.log,
+			SnapshotHistoryLimit: backend.DefaultSnapshotHistoryLimit,
+			Communicator:         backend.NewNodeCommunicator(false),
 		})
 		require.NoError(suite.T(), err)
 
@@ -659,7 +658,6 @@ func (suite *Suite) TestGetSealedTransaction() {
 			Log:                       suite.log,
 			SnapshotHistoryLimit:      backend.DefaultSnapshotHistoryLimit,
 			Communicator:              backend.NewNodeCommunicator(false),
-			TxErrorMessagesCacheSize:  1000,
 			TxResultQueryMode:         backend.IndexQueryModeExecutionNodesOnly,
 		})
 		require.NoError(suite.T(), err)
@@ -685,8 +683,29 @@ func (suite *Suite) TestGetSealedTransaction() {
 
 		// create the ingest engine
 		processedHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineBlockHeight)
+		processedTxErrorMessagesBlockHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineTxErrorMessagesBlockHeight)
 
-		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,
+			nil,
+			collectionExecutedMetric,
+			processedHeight,
+			processedTxErrorMessagesBlockHeight,
+			lastFullBlockHeight,
+			bnd,
+			enNodeIDs.Strings(),
+			nil,
+		)
 		require.NoError(suite.T(), err)
 
 		// 1. Assume that follower engine updated the block storage and the protocol state. The block is reported as sealed
@@ -743,7 +762,6 @@ func (suite *Suite) TestGetTransactionResult() {
 		all := util.StorageLayer(suite.T(), db)
 		results := bstorage.NewExecutionResults(suite.metrics, db)
 		receipts := bstorage.NewExecutionReceipts(suite.metrics, db, results, bstorage.DefaultCacheSize)
-
 		originID := unittest.IdentifierFixture()
 
 		*suite.state = protocol.State{}
@@ -821,7 +839,6 @@ func (suite *Suite) TestGetTransactionResult() {
 			Log:                       suite.log,
 			SnapshotHistoryLimit:      backend.DefaultSnapshotHistoryLimit,
 			Communicator:              backend.NewNodeCommunicator(false),
-			TxErrorMessagesCacheSize:  1000,
 			TxResultQueryMode:         backend.IndexQueryModeExecutionNodesOnly,
 		})
 		require.NoError(suite.T(), err)
@@ -840,14 +857,36 @@ func (suite *Suite) TestGetTransactionResult() {
 		require.NoError(suite.T(), err)
 
 		processedHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineBlockHeight)
+		processedTxErrorMessagesBlockHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineTxErrorMessagesBlockHeight)
+
 		lastFullBlockHeight, err := counters.NewPersistentStrictMonotonicCounter(
 			bstorage.NewConsumerProgress(db, module.ConsumeProgressLastFullBlockHeight),
 			suite.rootBlock.Height,
 		)
 		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,
+			nil,
+			collectionExecutedMetric,
+			processedHeight,
+			processedTxErrorMessagesBlockHeight,
+			lastFullBlockHeight,
+			bnd,
+			enNodeIDs.Strings(),
+			nil,
+		)
 		require.NoError(suite.T(), err)
 
 		background, cancel := context.WithCancel(context.Background())
@@ -1009,7 +1048,6 @@ func (suite *Suite) TestExecuteScript() {
 		collections := bstorage.NewCollections(db, transactions)
 		results := bstorage.NewExecutionResults(suite.metrics, db)
 		receipts := bstorage.NewExecutionReceipts(suite.metrics, db, results, bstorage.DefaultCacheSize)
-
 		identities := unittest.IdentityListFixture(2, unittest.WithRole(flow.RoleExecution))
 		suite.sealedSnapshot.On("Identities", mock.Anything).Return(identities, nil)
 		suite.finalSnapshot.On("Identities", mock.Anything).Return(identities, nil)
@@ -1020,25 +1058,24 @@ func (suite *Suite) TestExecuteScript() {
 
 		var err error
 		suite.backend, err = backend.New(backend.Params{
-			State:                    suite.state,
-			CollectionRPC:            suite.collClient,
-			Blocks:                   all.Blocks,
-			Headers:                  all.Headers,
-			Collections:              collections,
-			Transactions:             transactions,
-			ExecutionReceipts:        receipts,
-			ExecutionResults:         results,
-			ChainID:                  suite.chainID,
-			AccessMetrics:            suite.metrics,
-			ConnFactory:              connFactory,
-			MaxHeightRange:           backend.DefaultMaxHeightRange,
-			FixedExecutionNodeIDs:    (identities.NodeIDs()).Strings(),
-			Log:                      suite.log,
-			SnapshotHistoryLimit:     backend.DefaultSnapshotHistoryLimit,
-			Communicator:             backend.NewNodeCommunicator(false),
-			ScriptExecutionMode:      backend.IndexQueryModeExecutionNodesOnly,
-			TxErrorMessagesCacheSize: 1000,
-			TxResultQueryMode:        backend.IndexQueryModeExecutionNodesOnly,
+			State:                 suite.state,
+			CollectionRPC:         suite.collClient,
+			Blocks:                all.Blocks,
+			Headers:               all.Headers,
+			Collections:           collections,
+			Transactions:          transactions,
+			ExecutionReceipts:     receipts,
+			ExecutionResults:      results,
+			ChainID:               suite.chainID,
+			AccessMetrics:         suite.metrics,
+			ConnFactory:           connFactory,
+			MaxHeightRange:        backend.DefaultMaxHeightRange,
+			FixedExecutionNodeIDs: (identities.NodeIDs()).Strings(),
+			Log:                   suite.log,
+			SnapshotHistoryLimit:  backend.DefaultSnapshotHistoryLimit,
+			Communicator:          backend.NewNodeCommunicator(false),
+			ScriptExecutionMode:   backend.IndexQueryModeExecutionNodesOnly,
+			TxResultQueryMode:     backend.IndexQueryModeExecutionNodesOnly,
 		})
 		require.NoError(suite.T(), err)
 
@@ -1069,14 +1106,36 @@ func (suite *Suite) TestExecuteScript() {
 			Once()
 
 		processedHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineBlockHeight)
+		processedTxErrorMessagesBlockHeight := bstorage.NewConsumerProgress(db, module.ConsumeProgressIngestionEngineTxErrorMessagesBlockHeight)
+
 		lastFullBlockHeight, err := counters.NewPersistentStrictMonotonicCounter(
 			bstorage.NewConsumerProgress(db, module.ConsumeProgressLastFullBlockHeight),
 			suite.rootBlock.Height,
 		)
 		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,
+			nil,
+			collectionExecutedMetric,
+			processedHeight,
+			processedTxErrorMessagesBlockHeight,
+			lastFullBlockHeight,
+			suite.backend,
+			nil,
+			identities.NodeIDs().Strings(),
+		)
 		require.NoError(suite.T(), err)
 
 		// create another block as a predecessor of the block created earlier