Merge branch 'master' into jord/7397-storage-readme

Author: jordanschalm

admin/commands/storage/backfill_tx_error_messages.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/rs/zerolog"
+
 	"github.com/onflow/flow-go/admin"
 	"github.com/onflow/flow-go/admin/commands"
 	"github.com/onflow/flow-go/engine/access/ingestion/tx_error_messages"
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/model/flow/filter"
+	"github.com/onflow/flow-go/module/util"
 	"github.com/onflow/flow-go/state/protocol"
 )
 
@@ -25,16 +28,19 @@ type backfillTxErrorMessagesRequest struct {
 // BackfillTxErrorMessagesCommand executes a command to backfill
 // transaction error messages by fetching them from execution nodes.
 type BackfillTxErrorMessagesCommand struct {
+	log                 zerolog.Logger
 	state               protocol.State
 	txErrorMessagesCore *tx_error_messages.TxErrorMessagesCore
 }
 
 // NewBackfillTxErrorMessagesCommand creates a new instance of BackfillTxErrorMessagesCommand
 func NewBackfillTxErrorMessagesCommand(
+	log zerolog.Logger,
 	state protocol.State,
 	txErrorMessagesCore *tx_error_messages.TxErrorMessagesCore,
 ) commands.AdminCommand {
 	return &BackfillTxErrorMessagesCommand{
+		log:                 log.With().Str("command", "backfill-tx-error-messages").Logger(),
 		state:               state,
 		txErrorMessagesCore: txErrorMessagesCore,
 	}
@@ -147,6 +153,17 @@ func (b *BackfillTxErrorMessagesCommand) Handler(ctx context.Context, request *a
 
 	data := request.ValidatorData.(*backfillTxErrorMessagesRequest)
 
+	total := data.endHeight - data.startHeight + 1
+	progress := util.LogProgress(b.log,
+		util.DefaultLogProgressConfig("backfilling", int(total)),
+	)
+
+	b.log.Info().
+		Uint64("start_height", data.startHeight).
+		Uint64("end_height", data.endHeight).
+		Uint64("blocks", total).
+		Msgf("starting backfill")
+
 	for height := data.startHeight; height <= data.endHeight; height++ {
 		header, err := b.state.AtHeight(height).Head()
 		if err != nil {
@@ -158,6 +175,8 @@ func (b *BackfillTxErrorMessagesCommand) Handler(ctx context.Context, request *a
 		if err != nil {
 			return nil, fmt.Errorf("error encountered while processing transaction result error message for block: %d, %w", height, err)
 		}
+
+		progress(1)
 	}
 
 	return nil, nil
@@ -170,13 +189,18 @@ func (b *BackfillTxErrorMessagesCommand) Handler(ctx context.Context, request *a
 // - admin.InvalidAdminReqParameterError - if execution-node-ids is empty or has an invalid format.
 func (b *BackfillTxErrorMessagesCommand) parseExecutionNodeIds(executionNodeIdsIn interface{}, allIdentities flow.IdentityList) (flow.IdentitySkeletonList, error) {
 	var ids flow.IdentityList
-
 	switch executionNodeIds := executionNodeIdsIn.(type) {
-	case []string:
+	case []any:
 		if len(executionNodeIds) == 0 {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", "must be a non empty list of strings", executionNodeIdsIn)
 		}
-		requestedENIdentifiers, err := flow.IdentifierListFromHex(executionNodeIds)
+
+		idStrings := make([]string, len(executionNodeIds))
+		for i, id := range executionNodeIds {
+			idStrings[i] = id.(string)
+		}
+
+		requestedENIdentifiers, err := flow.IdentifierListFromHex(idStrings)
 		if err != nil {
 			return nil, admin.NewInvalidAdminReqParameterError("execution-node-ids", err.Error(), executionNodeIdsIn)
 		}

admin/commands/storage/backfill_tx_error_messages_test.go

@@ -77,6 +77,7 @@ func TestBackfillTxErrorMessages(t *testing.T) {
 func (suite *BackfillTxErrorMessagesSuite) SetupTest() {
 	suite.log = zerolog.New(os.Stderr)
 
+	lockManager := storage.NewTestingLockManager()
 	suite.state = new(protocolmock.State)
 	suite.headers = new(storagemock.Headers)
 	suite.receipts = new(storagemock.ExecutionReceipts)
@@ -160,9 +161,11 @@ func (suite *BackfillTxErrorMessagesSuite) SetupTest() {
 		errorMessageProvider,
 		suite.txErrorMessages,
 		executionNodeIdentitiesProvider,
+		lockManager,
 	)
 
 	suite.command = NewBackfillTxErrorMessagesCommand(
+		suite.log,
 		suite.state,
 		suite.txResultErrorMessagesCore,
 	)
@@ -237,7 +240,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestValidateInvalidFormat() {
 			Data: map[string]interface{}{
 				"start-height":       float64(1),         // raw json parses to float64
 				"end-height":         float64(endHeight), // raw json parses to float64
-				"execution-node-ids": []string{suite.allENIDs[0].NodeID.String()},
+				"execution-node-ids": []any{suite.allENIDs[0].NodeID.String()},
 			},
 		})
 		suite.Error(err)
@@ -290,7 +293,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestValidateInvalidFormat() {
 			Data: map[string]interface{}{
 				"start-height":       float64(1), // raw json parses to float64
 				"end-height":         float64(4), // raw json parses to float64
-				"execution-node-ids": []string{invalidENID.String()},
+				"execution-node-ids": []any{invalidENID.String()},
 			},
 		})
 		suite.Error(err)
@@ -321,7 +324,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestValidateValidFormat() {
 			Data: map[string]interface{}{
 				"start-height":       float64(1), // raw json parses to float64
 				"end-height":         float64(3), // raw json parses to float64
-				"execution-node-ids": []string{suite.allENIDs[0].NodeID.String()},
+				"execution-node-ids": []any{suite.allENIDs[0].NodeID.String()},
 			},
 		})
 		suite.NoError(err)
@@ -389,7 +392,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestHandleBackfillTxErrorMessages() {
 			Data: map[string]interface{}{
 				"start-height":       float64(startHeight), // raw json parses to float64
 				"end-height":         float64(endHeight),   // raw json parses to float64
-				"execution-node-ids": []string{executorID.String()},
+				"execution-node-ids": []any{executorID.String()},
 			},
 		}
 		suite.Require().NoError(suite.command.Validator(req))
@@ -432,6 +435,7 @@ func (suite *BackfillTxErrorMessagesSuite) TestHandleBackfillTxErrorMessagesErro
 	suite.Run("error when txErrorMessagesCore is nil", func() {
 		req := &admin.CommandRequest{Data: map[string]interface{}{}}
 		command := NewBackfillTxErrorMessagesCommand(
+			suite.log,
 			suite.state,
 			nil,
 		)
@@ -530,7 +534,7 @@ func (suite *BackfillTxErrorMessagesSuite) mockStoreTxErrorMessages(
 		}
 	}
 
-	suite.txErrorMessages.On("Store", blockID, txErrorMessages).Return(nil).Once()
+	suite.txErrorMessages.On("Store", mock.Anything, blockID, txErrorMessages).Return(nil).Once()
 }
 
 // assertAllExpectations asserts that all the expectations set on various mocks are met,

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,
@@ -1601,6 +1603,11 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			return errors.New("rpc-max-execution-response-message-size must be greater than 0")
 		}
 
+		// indexing tx error messages is only supported when tx results are also indexed
+		if builder.storeTxResultErrorMessages && !builder.executionDataIndexingEnabled {
+			return errors.New("execution-data-indexing-enabled must be set if store-tx-result-error-messages is enabled")
+		}
+
 		return nil
 	})
 }
@@ -1956,13 +1963,6 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 
 			return nil
 		}).
-		Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
-			if builder.storeTxResultErrorMessages {
-				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
-			}
-
-			return nil
-		}).
 		Component("version control", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
 			if !builder.versionControlEnabled {
 				noop := &module.NoopReadyDoneAware{}
@@ -2217,6 +2217,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 					notNil(builder.txResultErrorMessageProvider),
 					builder.transactionResultErrorMessages,
 					notNil(builder.ExecNodeIdentitiesProvider),
+					node.StorageLockMgr,
 				)
 			}
 
@@ -2262,34 +2263,45 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		}).
 		AdminCommand("backfill-tx-error-messages", func(config *cmd.NodeConfig) commands.AdminCommand {
 			return storageCommands.NewBackfillTxErrorMessagesCommand(
+				builder.Logger,
 				builder.State,
 				builder.TxResultErrorMessagesCore,
 			)
 		})
 
 	if builder.storeTxResultErrorMessages {
-		builder.Module("processed error messages block height consumer progress", func(node *cmd.NodeConfig) error {
-			processedTxErrorMessagesBlockHeight = store.NewConsumerProgress(
-				builder.ProtocolDB,
-				module.ConsumeProgressEngineTxErrorMessagesBlockHeight,
-			)
-			return nil
-		})
-		builder.Component("transaction result error messages engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
-			engine, err := tx_error_messages.New(
-				node.Logger,
-				node.State,
-				node.Storage.Headers,
-				processedTxErrorMessagesBlockHeight,
-				builder.TxResultErrorMessagesCore,
-			)
-			if err != nil {
-				return nil, err
-			}
-			builder.FollowerDistributor.AddOnBlockFinalizedConsumer(engine.OnFinalizedBlock)
+		builder.
+			Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
+				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(
+					node.Metrics.Cache,
+					node.ProtocolDB,
+					bstorage.DefaultCacheSize,
+				)
+				return nil
+			}).
+			Module("processed error messages block height consumer progress", func(node *cmd.NodeConfig) error {
+				processedTxErrorMessagesBlockHeight = store.NewConsumerProgress(
+					builder.ProtocolDB,
+					module.ConsumeProgressEngineTxErrorMessagesBlockHeight,
+				)
+				return nil
+			}).
+			Component("transaction result error messages engine", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+				engine, err := tx_error_messages.New(
+					node.Logger,
+					metrics.NewTransactionErrorMessagesCollector(),
+					node.State,
+					node.Storage.Headers,
+					processedTxErrorMessagesBlockHeight,
+					builder.TxResultErrorMessagesCore,
+				)
+				if err != nil {
+					return nil, err
+				}
+				builder.FollowerDistributor.AddOnBlockFinalizedConsumer(engine.OnFinalizedBlock)
 
-			return engine, nil
-		})
+				return engine, nil
+			})
 	}
 
 	if builder.supportsObserver {

cmd/collection/main.go

@@ -636,6 +636,7 @@ func main() {
 				node.Logger,
 				node.Me,
 				node.ProtocolDB,
+				node.StorageLockMgr,
 				node.State,
 				node.Metrics.Engine,
 				node.Metrics.Mempool,

cmd/consensus/main.go

@@ -396,6 +396,7 @@ func main() {
 				multipleReceiptsFilterMempool,
 				consensusMempools.LogForkAndCrash(node.Logger),
 				node.ProtocolDB,
+				node.StorageLockMgr,
 				node.Logger,
 			)
 			if err != nil {
@@ -621,7 +622,7 @@ func main() {
 			notifier.AddFollowerConsumer(followerDistributor)
 
 			// initialize the persister
-			persist, err := persister.New(node.ProtocolDB, node.RootChainID)
+			persist, err := persister.New(node.ProtocolDB, node.RootChainID, node.StorageLockMgr)
 			if err != nil {
 				return nil, err
 			}

cmd/execution_builder.go

@@ -334,6 +334,7 @@ func (exeNode *ExecutionNode) LoadSyncCore(node *NodeConfig) error {
 func (exeNode *ExecutionNode) LoadExecutionStorage(
 	node *NodeConfig,
 ) error {
+	var err error
 	db := node.ProtocolDB
 
 	exeNode.events = store.NewEvents(node.Metrics.Cache, db)
@@ -342,7 +343,10 @@ func (exeNode *ExecutionNode) LoadExecutionStorage(
 	exeNode.results = store.NewExecutionResults(node.Metrics.Cache, db)
 	exeNode.receipts = store.NewExecutionReceipts(node.Metrics.Cache, db, exeNode.results, storage.DefaultCacheSize)
 	exeNode.myReceipts = store.NewMyExecutionReceipts(node.Metrics.Cache, db, exeNode.receipts)
-	exeNode.txResults = store.NewTransactionResults(node.Metrics.Cache, db, exeNode.exeConf.transactionResultsCacheSize)
+	exeNode.txResults, err = store.NewTransactionResults(node.Metrics.Cache, db, exeNode.exeConf.transactionResultsCacheSize)
+	if err != nil {
+		return err
+	}
 	exeNode.eventsReader = exeNode.events
 	exeNode.commitsReader = exeNode.commits
 	exeNode.resultsReader = exeNode.results
@@ -758,8 +762,12 @@ func (exeNode *ExecutionNode) LoadExecutionState(
 		}
 		return nil
 	})
+
+	chunkDB := pebbleimpl.ToDB(chunkDataPackDB)
+	storedChunkDataPacks := store.NewStoredChunkDataPacks(
+		node.Metrics.Cache, chunkDB, exeNode.exeConf.chunkDataPackCacheSize)
 	chunkDataPacks := store.NewChunkDataPacks(node.Metrics.Cache,
-		pebbleimpl.ToDB(chunkDataPackDB), exeNode.collections, exeNode.exeConf.chunkDataPackCacheSize)
+		chunkDB, storedChunkDataPacks, exeNode.collections, exeNode.exeConf.chunkDataPackCacheSize)
 
 	getLatestFinalized := func() (uint64, error) {
 		final, err := node.State.Final().Head()