Merge branch 'master' into leo/refactor-index-protocol-kv-store

Author: zhangchiqing

cmd/access/node_builder/access_node_builder.go

@@ -161,7 +161,6 @@ type AccessNodeConfig struct {
 	rpcMetricsEnabled                    bool
 	executionDataSyncEnabled             bool
 	publicNetworkExecutionDataEnabled    bool
-	executionDataDBMode                  string
 	executionDataPrunerHeightRangeTarget uint64
 	executionDataPrunerThreshold         uint64
 	executionDataPruningInterval         time.Duration
@@ -277,7 +276,6 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 			MaxRetryDelay:      edrequester.DefaultMaxRetryDelay,
 		},
 		executionDataIndexingEnabled:         false,
-		executionDataDBMode:                  execution_data.ExecutionDataDBModePebble.String(),
 		executionDataPrunerHeightRangeTarget: 0,
 		executionDataPrunerThreshold:         pruner.DefaultThreshold,
 		executionDataPruningInterval:         pruner.DefaultPruningInterval,
@@ -591,7 +589,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 		Module("execution data datastore and blobstore", func(node *cmd.NodeConfig) error {
 			var err error
 			builder.ExecutionDatastoreManager, err = edstorage.CreateDatastoreManager(
-				node.Logger, builder.executionDataDir, builder.executionDataDBMode)
+				node.Logger, builder.executionDataDir)
 			if err != nil {
 				return fmt.Errorf("could not create execution data datastore manager: %w", err)
 			}
@@ -1362,10 +1360,13 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"execution-data-max-retry-delay",
 			defaultConfig.executionDataConfig.MaxRetryDelay,
 			"maximum delay for exponential backoff when fetching execution data fails e.g. 5m")
-		flags.StringVar(&builder.executionDataDBMode,
+
+		var builderexecutionDataDBMode string
+		flags.StringVar(&builderexecutionDataDBMode,
 			"execution-data-db",
-			defaultConfig.executionDataDBMode,
-			"[experimental] the DB type for execution datastore. One of [badger, pebble]")
+			"pebble",
+			"[deprecated] the DB type for execution datastore")
+
 		flags.Uint64Var(&builder.executionDataPrunerHeightRangeTarget,
 			"execution-data-height-range-target",
 			defaultConfig.executionDataPrunerHeightRangeTarget,

cmd/bootstrap/cmd/key.go

@@ -100,27 +100,30 @@ func keyCmdRun(_ *cobra.Command, _ []string) {
 	}
 	log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathNodeID)
 
-	err = common.WriteJSON(fmt.Sprintf(model.PathNodeInfoPriv, nodeInfo.NodeID), flagOutdir, private)
+	privKeyPath := fmt.Sprintf(model.PathNodeInfoPriv, nodeInfo.NodeID)
+	err = common.WriteJSON(privKeyPath, flagOutdir, private)
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to write json")
 	}
-	log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathNodeInfoPriv)
+	log.Info().Msgf("wrote file %s/%s", flagOutdir, privKeyPath)
 
-	err = common.WriteText(fmt.Sprintf(model.PathSecretsEncryptionKey, nodeInfo.NodeID), flagOutdir, secretsDBKey)
+	secretsKeyPath := fmt.Sprintf(model.PathSecretsEncryptionKey, nodeInfo.NodeID)
+	err = common.WriteText(secretsKeyPath, flagOutdir, secretsDBKey)
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to write file")
 	}
-	log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathSecretsEncryptionKey)
+	log.Info().Msgf("wrote file %s/%s", flagOutdir, secretsKeyPath)
 
 	public, err := nodeInfo.Public()
 	if err != nil {
 		log.Fatal().Err(err).Msg("could not access public keys")
 	}
-	err = common.WriteJSON(fmt.Sprintf(model.PathNodeInfoPub, nodeInfo.NodeID), flagOutdir, public)
+	pubNodeInfoPath := fmt.Sprintf(model.PathNodeInfoPub, nodeInfo.NodeID)
+	err = common.WriteJSON(pubNodeInfoPath, flagOutdir, public)
 	if err != nil {
 		log.Fatal().Err(err).Msg("failed to write json")
 	}
-	log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathNodeInfoPub)
+	log.Info().Msgf("wrote file %s/%s", flagOutdir, pubNodeInfoPath)
 
 	// write machine account info
 	if role == flow.RoleCollection || role == flow.RoleConsensus {
@@ -134,11 +137,12 @@ func keyCmdRun(_ *cobra.Command, _ []string) {
 		log.Debug().Str("address", flagAddress).Msg("assembling machine account information")
 		// write the public key to terminal for entry in Flow Port
 		machineAccountPriv := assembleNodeMachineAccountKey(machineKey)
-		err = common.WriteJSON(fmt.Sprintf(model.PathNodeMachineAccountPrivateKey, nodeInfo.NodeID), flagOutdir, machineAccountPriv)
+		privateKeyPath := fmt.Sprintf(model.PathNodeMachineAccountPrivateKey, nodeInfo.NodeID)
+		err = common.WriteJSON(privateKeyPath, flagOutdir, machineAccountPriv)
 		if err != nil {
 			log.Fatal().Err(err).Msg("failed to write json")
 		}
-		log.Info().Msgf("wrote file %s/%s", flagOutdir, model.PathNodeMachineAccountPrivateKey)
+		log.Info().Msgf("wrote file %s/%s", flagOutdir, privateKeyPath)
 	}
 }
 

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
@@ -716,7 +720,7 @@ func (exeNode *ExecutionNode) LoadExecutionDataDatastore(
 	node *NodeConfig,
 ) (err error) {
 	exeNode.executionDataDatastore, err = edstorage.CreateDatastoreManager(
-		node.Logger, exeNode.exeConf.executionDataDir, exeNode.exeConf.executionDataDBMode)
+		node.Logger, exeNode.exeConf.executionDataDir)
 	if err != nil {
 		return fmt.Errorf("could not create execution data datastore manager: %w", err)
 	}

cmd/execution_config.go

@@ -59,7 +59,6 @@ type ExecutionConfig struct {
 	importCheckpointWorkerCount           int
 	transactionExecutionMetricsEnabled    bool
 	transactionExecutionMetricsBufferSize uint
-	executionDataDBMode                   string
 	scheduleCallbacksEnabled              bool
 
 	computationConfig        computation.ComputationConfig
@@ -136,10 +135,12 @@ func (exeConf *ExecutionConfig) SetupFlags(flags *pflag.FlagSet) {
 	flags.IntVar(&exeConf.importCheckpointWorkerCount, "import-checkpoint-worker-count", 10, "number of workers to import checkpoint file during bootstrap")
 	flags.BoolVar(&exeConf.transactionExecutionMetricsEnabled, "tx-execution-metrics", true, "enable collection of transaction execution metrics")
 	flags.UintVar(&exeConf.transactionExecutionMetricsBufferSize, "tx-execution-metrics-buffer-size", 200, "buffer size for transaction execution metrics. The buffer size is the number of blocks that are kept in memory by the metrics provider engine")
-	flags.StringVar(&exeConf.executionDataDBMode,
+
+	var exeConfExecutionDataDBMode string
+	flags.StringVar(&exeConfExecutionDataDBMode,
 		"execution-data-db",
 		execution_data.ExecutionDataDBModePebble.String(),
-		"[experimental] the DB type for execution datastore. One of [badger, pebble]")
+		"[deprecated] the DB type for execution datastore. it's been deprecated")
 
 	flags.BoolVar(&exeConf.onflowOnlyLNs, "temp-onflow-only-lns", false, "do not use unless required. forces node to only request collections from onflow collection nodes")
 	flags.BoolVar(&exeConf.enableStorehouse, "enable-storehouse", false, "enable storehouse to store registers on disk, default is false")

cmd/observer/node_builder/observer_builder.go

@@ -154,7 +154,6 @@ type ObserverServiceConfig struct {
 	logTxTimeToSealed                    bool
 	executionDataSyncEnabled             bool
 	executionDataIndexingEnabled         bool
-	executionDataDBMode                  string
 	executionDataPrunerHeightRangeTarget uint64
 	executionDataPrunerThreshold         uint64
 	executionDataPruningInterval         time.Duration
@@ -234,7 +233,6 @@ func DefaultObserverServiceConfig() *ObserverServiceConfig {
 		logTxTimeToSealed:                    false,
 		executionDataSyncEnabled:             false,
 		executionDataIndexingEnabled:         false,
-		executionDataDBMode:                  execution_data.ExecutionDataDBModePebble.String(),
 		executionDataPrunerHeightRangeTarget: 0,
 		executionDataPrunerThreshold:         pruner.DefaultThreshold,
 		executionDataPruningInterval:         pruner.DefaultPruningInterval,
@@ -712,10 +710,9 @@ func (builder *ObserverServiceBuilder) extraFlags() {
 		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")
-		flags.StringVar(&builder.executionDataDBMode,
-			"execution-data-db",
-			defaultConfig.executionDataDBMode,
-			"[experimental] the DB type for execution datastore. One of [badger, pebble]")
+
+		var builderExecutionDataDBMode string
+		flags.StringVar(&builderExecutionDataDBMode, "execution-data-db", "pebble", "[deprecated] the DB type for execution datastore.")
 
 		// Execution data pruner
 		flags.Uint64Var(&builder.executionDataPrunerHeightRangeTarget,
@@ -1110,7 +1107,6 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 	var execDataDistributor *edrequester.ExecutionDataDistributor
 	var execDataCacheBackend *herocache.BlockExecutionData
 	var executionDataStoreCache *execdatacache.ExecutionDataCache
-	var executionDataDBMode execution_data.ExecutionDataDBMode
 
 	// setup dependency chain to ensure indexer starts after the requester
 	requesterDependable := module.NewProxiedReadyDoneAware()
@@ -1129,20 +1125,11 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				return err
 			}
 
-			executionDataDBMode, err = execution_data.ParseExecutionDataDBMode(builder.executionDataDBMode)
+			builder.ExecutionDatastoreManager, err = edstorage.NewPebbleDatastoreManager(
+				node.Logger.With().Str("pebbledb", "endata").Logger(),
+				datastoreDir, nil)
 			if err != nil {
-				return fmt.Errorf("could not parse execution data DB mode: %w", err)
-			}
-
-			if executionDataDBMode == execution_data.ExecutionDataDBModePebble {
-				builder.ExecutionDatastoreManager, err = edstorage.NewPebbleDatastoreManager(
-					node.Logger.With().Str("pebbledb", "endata").Logger(),
-					datastoreDir, nil)
-				if err != nil {
-					return fmt.Errorf("could not create PebbleDatastoreManager for execution data: %w", err)
-				}
-			} else {
-				return fmt.Errorf("datastore with badger has been deprecated, please use pebble instead")
+				return fmt.Errorf("could not create PebbleDatastoreManager for execution data: %w", err)
 			}
 			ds = builder.ExecutionDatastoreManager.Datastore()
 

cmd/util/cmd/read-badger/cmd/transaction_results.go

@@ -25,7 +25,10 @@ var transactionResultsCmd = &cobra.Command{
 	Short: "get transaction-result by block ID",
 	RunE: func(cmd *cobra.Command, args []string) error {
 		return common.WithStorage(flagDatadir, func(db storage.DB) error {
-			transactionResults := store.NewTransactionResults(metrics.NewNoopCollector(), db, 1)
+			transactionResults, err := store.NewTransactionResults(metrics.NewNoopCollector(), db, 1)
+			if err != nil {
+				return err
+			}
 			storages := common.InitStorages(db)
 			log.Info().Msgf("got flag block id: %s", flagBlockID)
 			blockID, err := flow.HexStringToIdentifier(flagBlockID)

cmd/util/cmd/read-light-block/read_light_block_test.go

@@ -25,14 +25,14 @@ func TestReadClusterRange(t *testing.T) {
 		err := unittest.WithLock(t, lockManager, storage.LockInsertOrFinalizeClusterBlock, func(lctx lockctx.Context) error {
 			// add parent as boundary
 			err := db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-				return operation.IndexClusterBlockHeight(lctx, rw.Writer(), parent.ChainID, parent.Height, parent.ID())
+				return operation.IndexClusterBlockHeight(lctx, rw, parent.ChainID, parent.Height, parent.ID())
 			})
 			if err != nil {
 				return err
 			}
 
 			return db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {
-				return operation.UpsertClusterFinalizedHeight(lctx, rw.Writer(), parent.ChainID, parent.Height)
+				return operation.BootstrapClusterFinalizedHeight(lctx, rw, parent.ChainID, parent.Height)
 			})
 		})
 		require.NoError(t, err)

cmd/util/cmd/rollback-executed-height/cmd/rollback_executed_height.go

@@ -67,7 +67,10 @@ func runE(*cobra.Command, []string) error {
 
 		metrics := &metrics.NoopCollector{}
 
-		transactionResults := store.NewTransactionResults(metrics, db, badger.DefaultCacheSize)
+		transactionResults, err := store.NewTransactionResults(metrics, db, badger.DefaultCacheSize)
+		if err != nil {
+			return err
+		}
 		commits := store.NewCommits(metrics, db)
 		results := store.NewExecutionResults(metrics, db)
 		receipts := store.NewExecutionReceipts(metrics, db, results, badger.DefaultCacheSize)

cmd/util/cmd/rollback-executed-height/cmd/rollback_executed_height_test.go

@@ -41,7 +41,8 @@ func TestReExecuteBlock(t *testing.T) {
 		all := store.InitAll(metrics, db)
 		headers := all.Headers
 		blocks := all.Blocks
-		txResults := store.NewTransactionResults(metrics, db, store.DefaultCacheSize)
+		txResults, err := store.NewTransactionResults(metrics, db, store.DefaultCacheSize)
+		require.NoError(t, err)
 		commits := store.NewCommits(metrics, db)
 		chunkDataPacks := store.NewChunkDataPacks(metrics, pebbleimpl.ToDB(pdb), store.NewCollections(db, store.NewTransactions(metrics, db)), store.DefaultCacheSize)
 		results := all.Results
@@ -201,7 +202,8 @@ func TestReExecuteBlockWithDifferentResult(t *testing.T) {
 		transactions := store.NewTransactions(metrics, db)
 		collections := store.NewCollections(db, transactions)
 		chunkDataPacks := store.NewChunkDataPacks(metrics, pebbleimpl.ToDB(pdb), collections, bstorage.DefaultCacheSize)
-		txResults := store.NewTransactionResults(metrics, db, bstorage.DefaultCacheSize)
+		txResults, err := store.NewTransactionResults(metrics, db, bstorage.DefaultCacheSize)
+		require.NoError(t, err)
 
 		err = unittest.WithLock(t, lockManager, storage.LockInsertBlock, func(lctx lockctx.Context) error {
 			return db.WithReaderBatchWriter(func(rw storage.ReaderBatchWriter) error {

engine/access/access_test.go

@@ -3,7 +3,6 @@ package access_test
 import (
 	"context"
 	"encoding/json"
-	"os"
 	"testing"
 
 	"github.com/cockroachdb/pebble/v2"
@@ -93,7 +92,7 @@ func TestAccess(t *testing.T) {
 
 func (suite *Suite) SetupTest() {
 	suite.lockManager = storage.NewTestingLockManager()
-	suite.log = zerolog.New(os.Stderr)
+	suite.log = unittest.Logger()
 	suite.net = new(mocknetwork.EngineRegistry)
 	suite.state = new(protocol.State)
 	suite.finalSnapshot = new(protocol.Snapshot)
@@ -774,6 +773,10 @@ func (suite *Suite) TestGetSealedTransaction() {
 		ctx := irrecoverable.NewMockSignalerContext(suite.T(), background)
 		ingestEng.Start(ctx)
 		<-ingestEng.Ready()
+		defer func() {
+			cancel()
+			<-ingestEng.Done()
+		}()
 
 		// 2. Ingest engine was notified by the follower engine about a new block.
 		// Follower engine --> Ingest engine