Merge branch 'master' into leo/refactor-stored-chunk-data-pack

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,
@@ -2060,7 +2061,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			// handles block-related operations.
 			blockTracker, err := subscriptiontracker.NewBlockTracker(
 				node.State,
-				builder.FinalizedRootBlock.Height,
+				builder.SealedRootBlock.Height,
 				node.Storage.Headers,
 				broadcaster,
 			)

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()
 
@@ -1921,7 +1908,7 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 		// handles block-related operations.
 		blockTracker, err := subscriptiontracker.NewBlockTracker(
 			node.State,
-			builder.FinalizedRootBlock.Height,
+			builder.SealedRootBlock.Height,
 			node.Storage.Headers,
 			broadcaster,
 		)

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/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)
 		storedChunkDataPacks := store.NewStoredChunkDataPacks(metrics, pebbleimpl.ToDB(pdb), store.DefaultCacheSize)
 		chunkDataPacks := store.NewChunkDataPacks(metrics, pebbleimpl.ToDB(pdb), storedChunkDataPacks, store.NewCollections(db, store.NewTransactions(metrics, db)), store.DefaultCacheSize)
@@ -203,7 +204,8 @@ func TestReExecuteBlockWithDifferentResult(t *testing.T) {
 		collections := store.NewCollections(db, transactions)
 		storedChunkDataPacks := store.NewStoredChunkDataPacks(metrics, pebbleimpl.ToDB(pdb), bstorage.DefaultCacheSize)
 		chunkDataPacks := store.NewChunkDataPacks(metrics, pebbleimpl.ToDB(pdb), storedChunkDataPacks, 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

engine/access/handle_irrecoverable_state_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"os"
 	"testing"
 	"time"
 
@@ -45,10 +44,12 @@ import (
 // IrrecoverableStateTestSuite tests that Access node indicate an inconsistent or corrupted node state
 type IrrecoverableStateTestSuite struct {
 	suite.Suite
+	log    zerolog.Logger
+	cancel context.CancelFunc
+
 	state      *protocol.State
 	snapshot   *protocol.Snapshot
 	epochQuery *protocol.EpochQuery
-	log        zerolog.Logger
 	net        *mocknetwork.EngineRegistry
 	request    *module.Requester
 	collClient *accessmock.AccessAPIClient
@@ -72,7 +73,7 @@ type IrrecoverableStateTestSuite struct {
 }
 
 func (suite *IrrecoverableStateTestSuite) SetupTest() {
-	suite.log = zerolog.New(os.Stdout)
+	suite.log = unittest.Logger()
 	suite.net = mocknetwork.NewEngineRegistry(suite.T())
 	suite.state = protocol.NewState(suite.T())
 	suite.snapshot = protocol.NewSnapshot(suite.T())
@@ -186,13 +187,17 @@ func (suite *IrrecoverableStateTestSuite) SetupTest() {
 	assert.NoError(suite.T(), err)
 
 	err = fmt.Errorf("inconsistent node's state")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	suite.cancel = cancel
+
 	signCtxErr := irrecoverable.NewExceptionf("failed to lookup sealed header: %w", err)
-	ctx := irrecoverable.NewMockSignalerContextExpectError(suite.T(), context.Background(), signCtxErr)
+	signalCtx := irrecoverable.NewMockSignalerContextExpectError(suite.T(), ctx, signCtxErr)
 
-	suite.rpcEng.Start(ctx)
+	suite.rpcEng.Start(signalCtx)
 
-	suite.secureGrpcServer.Start(ctx)
-	suite.unsecureGrpcServer.Start(ctx)
+	suite.secureGrpcServer.Start(signalCtx)
+	suite.unsecureGrpcServer.Start(signalCtx)
 
 	// wait for the servers to startup
 	unittest.AssertClosesBefore(suite.T(), suite.secureGrpcServer.Ready(), 2*time.Second)
@@ -202,6 +207,13 @@ func (suite *IrrecoverableStateTestSuite) SetupTest() {
 	unittest.AssertClosesBefore(suite.T(), suite.rpcEng.Ready(), 2*time.Second)
 }
 
+func (suite *IrrecoverableStateTestSuite) TearDownTest() {
+	suite.cancel()
+	unittest.AssertClosesBefore(suite.T(), suite.secureGrpcServer.Done(), 2*time.Second)
+	unittest.AssertClosesBefore(suite.T(), suite.unsecureGrpcServer.Done(), 2*time.Second)
+	unittest.AssertClosesBefore(suite.T(), suite.rpcEng.Done(), 2*time.Second)
+}
+
 func TestIrrecoverableState(t *testing.T) {
 	suite.Run(t, new(IrrecoverableStateTestSuite))
 }