[Access] Add schedule transaction indexing

Author: peterargue

cmd/access/node_builder/access_node_builder.go

@@ -351,6 +351,7 @@ type FlowAccessNodeBuilder struct {
 	transactionResultErrorMessages storage.TransactionResultErrorMessages
 	transactions                   storage.Transactions
 	collections                    storage.Collections
+	scheduledTransactions          storage.ScheduledTransactions
 
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
@@ -852,6 +853,10 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				builder.lightTransactionResults = store.NewLightTransactionResults(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
 				return nil
 			}).
+			Module("scheduled transactions storage", func(node *cmd.NodeConfig) error {
+				builder.scheduledTransactions = store.NewScheduledTransactions(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
+				return nil
+			}).
 			DependableComponent("execution data indexer", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 				// Note: using a DependableComponent here to ensure that the indexer does not block
 				// other components from starting while bootstrapping the register db since it may
@@ -937,7 +942,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					return nil, fmt.Errorf("could not create derived chain data: %w", err)
 				}
 
-				indexerCore, err := indexer.New(
+				builder.ExecutionIndexerCore = indexer.New(
 					builder.Logger,
 					metrics.NewExecutionStateIndexerCollector(),
 					notNil(builder.ProtocolDB),
@@ -947,22 +952,19 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					notNil(builder.collections),
 					notNil(builder.transactions),
 					notNil(builder.lightTransactionResults),
-					builder.RootChainID.Chain(),
+					notNil(builder.scheduledTransactions),
+					builder.RootChainID,
 					indexerDerivedChainData,
 					notNil(builder.collectionExecutedMetric),
 					node.StorageLockMgr,
 				)
-				if err != nil {
-					return nil, err
-				}
-				builder.ExecutionIndexerCore = indexerCore
 
 				// execution state worker uses a jobqueue to process new execution data and indexes it by using the indexer.
 				builder.ExecutionIndexer, err = indexer.NewIndexer(
 					builder.Logger,
 					registers.FirstHeight(),
 					registers,
-					indexerCore,
+					builder.ExecutionIndexerCore,
 					executionDataStoreCache,
 					builder.ExecutionDataRequester.HighestConsecutiveHeight,
 					indexedBlockHeight,

cmd/observer/node_builder/observer_builder.go

@@ -299,6 +299,7 @@ type ObserverServiceBuilder struct {
 	// storage
 	events                  storage.Events
 	lightTransactionResults storage.LightTransactionResults
+	scheduledTransactions   storage.ScheduledTransactions
 
 	// available until after the network has started. Hence, a factory function that needs to be called just before
 	// creating the sync engine
@@ -1349,6 +1350,9 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 		}).Module("transaction results storage", func(node *cmd.NodeConfig) error {
 			builder.lightTransactionResults = store.NewLightTransactionResults(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
 			return nil
+		}).Module("scheduled transactions storage", func(node *cmd.NodeConfig) error {
+			builder.scheduledTransactions = store.NewScheduledTransactions(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
+			return nil
 		}).DependableComponent("execution data indexer", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			// Note: using a DependableComponent here to ensure that the indexer does not block
 			// other components from starting while bootstrapping the register db since it may
@@ -1435,7 +1439,7 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 			}
 
 			var collectionExecutedMetric module.CollectionExecutedMetric = metrics.NewNoopCollector()
-			indexerCore, err := indexer.New(
+			builder.ExecutionIndexerCore = indexer.New(
 				builder.Logger,
 				metrics.NewExecutionStateIndexerCollector(),
 				builder.ProtocolDB,
@@ -1445,22 +1449,19 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				builder.Storage.Collections,
 				builder.Storage.Transactions,
 				builder.lightTransactionResults,
-				builder.RootChainID.Chain(),
+				builder.scheduledTransactions,
+				builder.RootChainID,
 				indexerDerivedChainData,
 				collectionExecutedMetric,
 				node.StorageLockMgr,
 			)
-			if err != nil {
-				return nil, err
-			}
-			builder.ExecutionIndexerCore = indexerCore
 
 			// execution state worker uses a jobqueue to process new execution data and indexes it by using the indexer.
 			builder.ExecutionIndexer, err = indexer.NewIndexer(
 				builder.Logger,
 				registers.FirstHeight(),
 				registers,
-				indexerCore,
+				builder.ExecutionIndexerCore,
 				executionDataStoreCache,
 				builder.ExecutionDataRequester.HighestConsecutiveHeight,
 				indexedBlockHeight,

engine/access/rpc/backend/script_executor_test.go

@@ -129,7 +129,7 @@ func (s *ScriptExecutorSuite) SetupTest() {
 	derivedChainData, err := derived.NewDerivedChainData(derived.DefaultDerivedDataCacheSize)
 	s.Require().NoError(err)
 
-	indexerCore, err := indexer.New(
+	indexerCore := indexer.New(
 		s.log,
 		module.ExecutionStateIndexerMetrics(metrics.NewNoopCollector()),
 		nil,
@@ -139,12 +139,12 @@ func (s *ScriptExecutorSuite) SetupTest() {
 		nil,
 		nil,
 		nil,
-		s.chain,
+		nil,
+		s.chain.ChainID(),
 		derivedChainData,
 		module.CollectionExecutedMetric(metrics.NewNoopCollector()),
 		lockManager,
 	)
-	s.Require().NoError(err)
 
 	s.scripts = execution.NewScripts(
 		s.log,

fvm/blueprints/scheduled_callback.go

@@ -73,7 +73,7 @@ func ExecuteCallbacksTransactions(chainID flow.Chain, processEvents flow.EventsL
 		// event.EventIndex
 
 		// skip any fee events or other events that are not pending execution events
-		if !isPendingExecutionEvent(env, event) {
+		if !IsPendingExecutionEvent(env, event) {
 			continue
 		}
 
@@ -114,19 +114,41 @@ func executeCallbackTransaction(
 // callback scheduler contract and has the following signature:
 // event PendingExecution(id: UInt64, priority: UInt8, executionEffort: UInt64, fees: UFix64, callbackOwner: Address)
 func callbackArgsFromEvent(event flow.Event) ([]byte, uint64, error) {
+	cadenceId, cadenceEffort, err := CallbackDetailsFromEvent(event)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	effort := uint64(cadenceEffort)
+
+	if effort > flow.DefaultMaxTransactionGasLimit {
+		log.Warn().Uint64("effort", effort).Msg("effort is greater than max transaction gas limit, setting to max")
+		effort = flow.DefaultMaxTransactionGasLimit
+	}
+
+	encID, err := jsoncdc.Encode(cadenceId)
+	if err != nil {
+		return nil, 0, fmt.Errorf("failed to encode id: %w", err)
+	}
+
+	return encID, uint64(effort), nil
+}
+
+// CallbackDetailsFromEvent decodes the event payload and returns the callback ID and effort.
+func CallbackDetailsFromEvent(event flow.Event) (cadence.UInt64, cadence.UInt64, error) {
 	const (
 		processedCallbackIDFieldName     = "id"
 		processedCallbackEffortFieldName = "executionEffort"
 	)
 
 	eventData, err := ccf.Decode(nil, event.Payload)
 	if err != nil {
-		return nil, 0, fmt.Errorf("failed to decode event: %w", err)
+		return 0, 0, fmt.Errorf("failed to decode event: %w", err)
 	}
 
 	cadenceEvent, ok := eventData.(cadence.Event)
 	if !ok {
-		return nil, 0, fmt.Errorf("event data is not a cadence event")
+		return 0, 0, fmt.Errorf("event data is not a cadence event")
 	}
 
 	idValue := cadence.SearchFieldByName(
@@ -139,32 +161,21 @@ func callbackArgsFromEvent(event flow.Event) ([]byte, uint64, error) {
 		processedCallbackEffortFieldName,
 	)
 
-	id, ok := idValue.(cadence.UInt64)
+	cadenceId, ok := idValue.(cadence.UInt64)
 	if !ok {
-		return nil, 0, fmt.Errorf("id is not uint64")
+		return 0, 0, fmt.Errorf("id is not uint64")
 	}
 
 	cadenceEffort, ok := effortValue.(cadence.UInt64)
 	if !ok {
-		return nil, 0, fmt.Errorf("effort is not uint64")
-	}
-
-	effort := uint64(cadenceEffort)
-
-	if effort > flow.DefaultMaxTransactionGasLimit {
-		log.Warn().Uint64("effort", effort).Msg("effort is greater than max transaction gas limit, setting to max")
-		effort = flow.DefaultMaxTransactionGasLimit
+		return 0, 0, fmt.Errorf("effort is not uint64")
 	}
 
-	encID, err := jsoncdc.Encode(id)
-	if err != nil {
-		return nil, 0, fmt.Errorf("failed to encode id: %w", err)
-	}
-
-	return encID, uint64(effort), nil
+	return cadenceId, cadenceEffort, nil
 }
 
-func isPendingExecutionEvent(env templates.Environment, event flow.Event) bool {
+// IsPendingExecutionEvent returns true if the event is a pending execution event.
+func IsPendingExecutionEvent(env templates.Environment, event flow.Event) bool {
 	processedEventType := PendingExecutionEventType(env)
 	return event.Type == processedEventType
 }

fvm/blueprints/scheduled_callback_test.go

@@ -183,6 +183,27 @@ func TestExecuteCallbackTransaction(t *testing.T) {
 	assert.Equal(t, tx.GasLimit, uint64(effort))
 }
 
+func TestIsPendingExecutionEvent(t *testing.T) {
+	t.Parallel()
+
+	chain := flow.Mainnet.Chain()
+	env := systemcontracts.SystemContractsForChain(chain.ChainID()).AsTemplateEnv()
+	assert.True(t, blueprints.IsPendingExecutionEvent(env, createValidCallbackEvent(t, 1, 100)))
+}
+
+func TestCallbackDetailsFromEvent(t *testing.T) {
+	t.Parallel()
+
+	expectedID := uint64(123)
+	expectedEffort := uint64(456)
+	event := createValidCallbackEvent(t, expectedID, expectedEffort)
+
+	actualID, actualEffort, err := blueprints.CallbackDetailsFromEvent(event)
+	require.NoError(t, err)
+	assert.Equal(t, expectedID, uint64(actualID))
+	assert.Equal(t, expectedEffort, uint64(actualEffort))
+}
+
 func createValidCallbackEvent(t *testing.T, id uint64, effort uint64) flow.Event {
 	const processedEventTypeTemplate = "A.%v.FlowTransactionScheduler.PendingExecution"
 	env := systemcontracts.SystemContractsForChain(flow.Mainnet.Chain().ChainID()).AsTemplateEnv()

module/execution/scripts_test.go

@@ -178,7 +178,7 @@ func (s *scriptTestSuite) SetupTest() {
 	derivedChainData, err := derived.NewDerivedChainData(derived.DefaultDerivedDataCacheSize)
 	s.Require().NoError(err)
 
-	index, err := indexer.New(
+	index := indexer.New(
 		logger,
 		metrics.NewNoopCollector(),
 		nil,
@@ -188,12 +188,12 @@ func (s *scriptTestSuite) SetupTest() {
 		nil,
 		nil,
 		nil,
-		flow.Testnet.Chain(),
+		nil,
+		flow.Testnet,
 		derivedChainData,
 		nil,
 		lockManager,
 	)
-	s.Require().NoError(err)
 
 	s.scripts = NewScripts(
 		logger,