[Access] Index Execution Data

model/convert/service_event.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 
 	"github.com/coreos/go-semver/semver"
-
 	"github.com/onflow/cadence"
 	"github.com/onflow/cadence/encoding/ccf"
 

module/irrecoverable/unittest.go

@@ -3,19 +3,26 @@ package irrecoverable
 import (
 	"context"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 // MockSignalerContext is a SignalerContext which will immediately fail a test if an error is thrown.
 type MockSignalerContext struct {
 	context.Context
-	t *testing.T
+	t           *testing.T
+	expectError error
 }
 
 var _ SignalerContext = &MockSignalerContext{}
 
 func (m MockSignalerContext) sealed() {}
 
 func (m MockSignalerContext) Throw(err error) {
+	if m.expectError != nil {
+		assert.EqualError(m.t, err, m.expectError.Error())
+		return
+	}
 	m.t.Fatalf("mock signaler context received error: %v", err)
 }
 
@@ -30,3 +37,11 @@ func NewMockSignalerContextWithCancel(t *testing.T, parent context.Context) (*Mo
 	ctx, cancel := context.WithCancel(parent)
 	return NewMockSignalerContext(t, ctx), cancel
 }
+
+func NewMockSignalerContextExpectError(t *testing.T, ctx context.Context, err error) *MockSignalerContext {
+	return &MockSignalerContext{
+		Context:     ctx,
+		t:           t,
+		expectError: err,
+	}
+}

module/jobqueue.go

@@ -10,6 +10,8 @@ const (
 
 	ConsumeProgressExecutionDataRequesterBlockHeight  = "ConsumeProgressExecutionDataRequesterBlockHeight"
 	ConsumeProgressExecutionDataRequesterNotification = "ConsumeProgressExecutionDataRequesterNotification"
+
+	ConsumeProgressExecutionDataIndexerBlockHeight = "ConsumeProgressExecutionDataIndexerBlockHeight"
 )
 
 // JobID is a unique ID of the job.

module/state_synchronization/indexer/indexer.go

@@ -0,0 +1,99 @@
+package indexer
+
+import (
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/onflow/flow-go/engine"
+	"github.com/onflow/flow-go/module"
+	"github.com/onflow/flow-go/module/component"
+	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
+	"github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
+	"github.com/onflow/flow-go/module/irrecoverable"
+	"github.com/onflow/flow-go/module/jobqueue"
+	"github.com/onflow/flow-go/module/state_synchronization/requester/jobs"
+	"github.com/onflow/flow-go/storage"
+)
+
+const (
+	workersCount = 1 // how many workers will concurrently process the tasks in the jobqueue
+	searchAhead  = 1 // how many block heights ahead of the current will be requested and tasked for jobqueue
+)
+
+// Indexer handles ingestion of new execution data available and uses the execution data indexer module
+// to index the data.
+// The processing of new available data is done by creating a jobqueue that uses the execution data reader to
+// obtain new jobs. The worker also implements the `highestConsecutiveHeight` method which is used by the execution
+// data reader, so it doesn't surpass the highest sealed block height when fetching the data.
+// The execution state worker has a callback that is used by the upstream queues which download new execution data to
+// notify new data is available and kick off indexing.
+type Indexer struct {
+	component.Component
+	log             zerolog.Logger
+	exeDataReader   *jobs.ExecutionDataReader
+	exeDataNotifier engine.Notifier
+	indexer         *IndexerCore
+}
+
+// NewIndexer creates a new execution worker.
+func NewIndexer(
+	log zerolog.Logger,
+	initHeight uint64,
+	fetchTimeout time.Duration,
+	indexer *IndexerCore,
+	executionCache *cache.ExecutionDataCache,
+	executionDataLatestHeight func() (uint64, error),
+	processedHeight storage.ConsumerProgress,
+) *Indexer {
+	r := &Indexer{
+		exeDataNotifier: engine.NewNotifier(),
+		indexer:         indexer,
+	}
+
+	r.exeDataReader = jobs.NewExecutionDataReader(executionCache, fetchTimeout, executionDataLatestHeight)
+
+	// create a jobqueue that will process new available block execution data. The `exeDataNotifier` is used to
+	// signal new work, which is being triggered on the `OnExecutionData` handler.
+	r.Component = jobqueue.NewComponentConsumer(
+		log.With().Str("module", "execution_indexer").Logger(),
+		r.exeDataNotifier.Channel(),
+		processedHeight,
+		r.exeDataReader,
+		initHeight,
+		r.processExecutionData,
+		workersCount,
+		searchAhead,
+	)
+
+	return r
+}
+
+// Start the worker jobqueue to consume the available data.
+func (i *Indexer) Start(ctx irrecoverable.SignalerContext) {
+	i.exeDataReader.AddContext(ctx)
+	i.Component.Start(ctx)
+}
+
+// OnExecutionData is used to notify when new execution data is downloaded by the execution data requester jobqueue.
+func (i *Indexer) OnExecutionData(_ *execution_data.BlockExecutionDataEntity) {
+	i.exeDataNotifier.Notify()
+}
+
+// processExecutionData is a worker method that is being called by the jobqueue when processing a new job.
+// The job data contains execution data which we provide to the execution indexer to index it.
+func (i *Indexer) processExecutionData(ctx irrecoverable.SignalerContext, job module.Job, done func()) {
+	entry, err := jobs.JobToBlockEntry(job)
+	if err != nil {
+		i.log.Error().Err(err).Str("job_id", string(job.ID())).Msg("error converting execution data job")
+		ctx.Throw(err)
+	}
+
+	err = i.indexer.IndexBlockData(entry.ExecutionData)
+	if err != nil {
+		i.log.Error().Err(err).Str("job_id", string(job.ID())).Msg("error during execution data index processing job")
+		ctx.Throw(err)
+	}
+
+	done()
+}

module/state_synchronization/indexer/indexer_core.go

@@ -0,0 +1,174 @@
+package indexer
+
+import (
+	"fmt"
+
+	"github.com/rs/zerolog"
+	"golang.org/x/exp/maps"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/onflow/flow-go/ledger"
+	"github.com/onflow/flow-go/ledger/common/convert"
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
+	"github.com/onflow/flow-go/storage"
+	"github.com/onflow/flow-go/utils/logging"
+)
+
+// IndexerCore indexes the execution state.
+type IndexerCore struct {
+	registers storage.RegisterIndex
+	headers   storage.Headers
+	events    storage.Events
+	log       zerolog.Logger
+}
+
+// New execution state indexer used to ingest block execution data and index it by height.
+// The passed RegisterIndex storage must be populated to include the first and last height otherwise the indexer
+// won't be initialized to ensure we have bootstrapped the storage first.
+func New(
+	log zerolog.Logger,
+	registers storage.RegisterIndex,
+	headers storage.Headers,
+	events storage.Events,
+) (*IndexerCore, error) {
+	return &IndexerCore{
+		registers: registers,
+		headers:   headers,
+		events:    events,
+		log:       log.With().Str("component", "execution_indexer").Logger(),
+	}, nil
+}
+
+// RegisterValues retrieves register values by the register IDs at the provided block height.
+// Even if the register wasn't indexed at the provided height, returns the highest height the register was indexed at.
+// Expected errors:
+// - storage.ErrNotFound if the register by the ID was never indexed
+func (c *IndexerCore) RegisterValues(IDs flow.RegisterIDs, height uint64) ([]flow.RegisterValue, error) {
+	values := make([]flow.RegisterValue, len(IDs))
+
+	for j, id := range IDs {
+		value, err := c.registers.Get(id, height)
+		if err != nil {
+			return nil, err
+		}
+
+		values[j] = value
+	}
+
+	return values, nil
+}
+
+// IndexBlockData indexes all execution block data by height.
+// This method shouldn't be used concurrently.
+// Expected errors:
+// - storage.ErrNotFound if the block for execution data was not found
+func (c *IndexerCore) IndexBlockData(data *execution_data.BlockExecutionDataEntity) error {
+	block, err := c.headers.ByBlockID(data.BlockID)
+	if err != nil {
+		return fmt.Errorf("could not get the block by ID %s: %w", data.BlockID, err)
+	}
+
+	lg := c.log.With().
+		Hex("block_id", logging.ID(data.BlockID)).
+		Uint64("height", block.Height).
+		Logger()
+
+	lg.Debug().Msgf("indexing new block")
+
+	// the height we are indexing must be exactly one bigger or same as the latest height indexed from the storage
+	latest := c.registers.LatestHeight()
+	if block.Height != latest+1 && block.Height != latest {
+		return fmt.Errorf("must store registers with the next height %d, but got %d", latest+1, block.Height)
+	}
+	// allow rerunning the indexer for same height since we are fetching height from register storage, but there are other storages
+	// for indexing resources which might fail to update the values, so this enables rerunning and reindexing those resources
+	if block.Height == latest {
+		lg.Warn().Msg("reindexing block data")
+	}
+
+	// concurrently process indexing of block data
+	g := errgroup.Group{}
+
+	g.Go(func() error {
+		events := make([]flow.Event, 0)
+		for _, chunk := range data.ChunkExecutionDatas {
+			events = append(events, chunk.Events...)
+		}
+
+		err := c.indexEvents(data.BlockID, events)
+		if err != nil {
+			return fmt.Errorf("could not index events at height %d: %w", block.Height, err)
+		}
+		return nil
+	})
+
+	g.Go(func() error {
+		updates := make([]*ledger.TrieUpdate, 0)
+
+		for _, chunk := range data.ChunkExecutionDatas {
+			if chunk.TrieUpdate != nil {
+				updates = append(updates, chunk.TrieUpdate)
+			}
+		}
+
+		// we are iterating all the registers and overwrite any existing register at the same path
+		// this will make sure if we have multiple register changes only the last change will get persisted
+		// if block has two chucks:
+		// first chunk updates: { X: 1, Y: 2 }
+		// second chunk updates: { X: 2 }
+		// then we should persist only {X: 2: Y: 2}
+		payloads := make(map[ledger.Path]*ledger.Payload)
+		for _, update := range updates {
+			for i, path := range update.Paths {
+				payloads[path] = update.Payloads[i]
+			}
+		}
+
+		err = c.indexRegisters(maps.Values(payloads), block.Height)
+		if err != nil {
+			return fmt.Errorf("could not index register payloads at height %d: %w", block.Height, err)
+		}
+
+		lg.Debug().
+			Int("register_count", len(payloads)).
+			Msg("indexed registers")
+
+		return nil
+	})
+
+	err = g.Wait()
+	if err != nil {
+		return fmt.Errorf("failed to index block data at height %d: %w", block.Height, err)
+	}
+	return nil
+}
+
+func (c *IndexerCore) indexEvents(blockID flow.Identifier, events flow.EventsList) error {
+	// Note: the last chunk in an execution data is the system chunk. All events in that ChunkExecutionData are service events.
+	err := c.events.Store(blockID, []flow.EventsList{events})
+	return err
+}
+
+func (c *IndexerCore) indexRegisters(payloads []*ledger.Payload, height uint64) error {
+	regEntries := make(flow.RegisterEntries, len(payloads))
+
+	for j, payload := range payloads {
+		k, err := payload.Key()
+		if err != nil {
+			return err
+		}
+
+		id, err := convert.LedgerKeyToRegisterID(k)
+		if err != nil {
+			return err
+		}
+
+		regEntries[j] = flow.RegisterEntry{
+			Key:   id,
+			Value: payload.Value(),
+		}
+	}
+
+	return c.registers.Store(regEntries, height)
+}