[Storage Refactor] Refactor saving execution results

cmd/access/node_builder/access_node_builder.go

@@ -346,6 +346,11 @@ type FlowAccessNodeBuilder struct {
 	VersionControl               *version.VersionControl
 	StopControl                  *stop.StopControl
 
+	// storage
+	events                         storage.Events
+	lightTransactionResults        storage.LightTransactionResults
+	transactionResultErrorMessages storage.TransactionResultErrorMessages
+
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
 	// Hence, a factory function that needs to be called just before creating the sync engine
@@ -872,7 +877,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				return nil
 			}).
 			Module("transaction results storage", func(node *cmd.NodeConfig) error {
-				builder.Storage.LightTransactionResults = bstorage.NewLightTransactionResults(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)
+				builder.lightTransactionResults = store.NewLightTransactionResults(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
 				return nil
 			}).
 			DependableComponent("execution data indexer", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
@@ -961,13 +966,13 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				indexerCore, err := indexer.New(
 					builder.Logger,
 					metrics.NewExecutionStateIndexerCollector(),
-					builder.DB,
+					builder.ProtocolDB,
 					builder.Storage.RegisterIndex,
 					builder.Storage.Headers,
-					builder.Storage.Events,
+					builder.events,
 					builder.Storage.Collections,
 					builder.Storage.Transactions,
-					builder.Storage.LightTransactionResults,
+					builder.lightTransactionResults,
 					builder.RootChainID.Chain(),
 					indexerDerivedChainData,
 					builder.collectionExecutedMetric,
@@ -1834,19 +1839,19 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			return nil
 		}).
 		Module("events storage", func(node *cmd.NodeConfig) error {
-			builder.Storage.Events = bstorage.NewEvents(node.Metrics.Cache, node.DB)
+			builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
 			return nil
 		}).
 		Module("reporter", func(node *cmd.NodeConfig) error {
 			builder.Reporter = index.NewReporter()
 			return nil
 		}).
 		Module("events index", func(node *cmd.NodeConfig) error {
-			builder.EventsIndex = index.NewEventsIndex(builder.Reporter, builder.Storage.Events)
+			builder.EventsIndex = index.NewEventsIndex(builder.Reporter, builder.events)
 			return nil
 		}).
 		Module("transaction result index", func(node *cmd.NodeConfig) error {
-			builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.Storage.LightTransactionResults)
+			builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.lightTransactionResults)
 			return nil
 		}).
 		Module("processed finalized block height consumer progress", func(node *cmd.NodeConfig) error {
@@ -1870,7 +1875,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 		}).
 		Module("transaction result error messages storage", func(node *cmd.NodeConfig) error {
 			if builder.storeTxResultErrorMessages {
-				builder.Storage.TransactionResultErrorMessages = bstorage.NewTransactionResultErrorMessages(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)
+				builder.transactionResultErrorMessages = store.NewTransactionResultErrorMessages(node.Metrics.Cache, node.ProtocolDB, bstorage.DefaultCacheSize)
 			}
 
 			return nil
@@ -2030,7 +2035,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				Transactions:          node.Storage.Transactions,
 				ExecutionReceipts:     node.Storage.Receipts,
 				ExecutionResults:      node.Storage.Results,
-				TxResultErrorMessages: node.Storage.TransactionResultErrorMessages,
+				TxResultErrorMessages: builder.transactionResultErrorMessages,
 				ChainID:               node.RootChainID,
 				AccessMetrics:         builder.AccessMetrics,
 				ConnFactory:           connFactory,
@@ -2116,7 +2121,7 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 				builder.TxResultErrorMessagesCore = tx_error_messages.NewTxErrorMessagesCore(
 					node.Logger,
 					builder.nodeBackend,
-					node.Storage.TransactionResultErrorMessages,
+					builder.transactionResultErrorMessages,
 					builder.ExecNodeIdentitiesProvider,
 				)
 			}

cmd/execution_builder.go

@@ -91,7 +91,7 @@ import (
 	"github.com/onflow/flow-go/state/protocol/blocktimer"
 	storageerr "github.com/onflow/flow-go/storage"
 	storage "github.com/onflow/flow-go/storage/badger"
-	"github.com/onflow/flow-go/storage/badger/procedure"
+	"github.com/onflow/flow-go/storage/operation"
 	"github.com/onflow/flow-go/storage/operation/pebbleimpl"
 	storagepebble "github.com/onflow/flow-go/storage/pebble"
 	"github.com/onflow/flow-go/storage/store"
@@ -128,17 +128,22 @@ type ExecutionNode struct {
 
 	ingestionUnit *engine.Unit
 
-	collector              *metrics.ExecutionCollector
-	executionState         state.ExecutionState
-	followerState          protocol.FollowerState
-	committee              hotstuff.DynamicCommittee
-	ledgerStorage          *ledger.Ledger
-	registerStore          *storehouse.RegisterStore
-	events                 *storage.Events
-	serviceEvents          *storage.ServiceEvents
-	txResults              *storage.TransactionResults
-	results                *storage.ExecutionResults
-	myReceipts             *storage.MyExecutionReceipts
+	collector      *metrics.ExecutionCollector
+	executionState state.ExecutionState
+	followerState  protocol.FollowerState
+	committee      hotstuff.DynamicCommittee
+	ledgerStorage  *ledger.Ledger
+	registerStore  *storehouse.RegisterStore
+
+	// storage
+	events        storageerr.Events
+	serviceEvents storageerr.ServiceEvents
+	txResults     storageerr.TransactionResults
+	results       storageerr.ExecutionResults
+	receipts      storageerr.ExecutionReceipts
+	myReceipts    storageerr.MyExecutionReceipts
+	commits       storageerr.Commits
+
 	providerEngine         exeprovider.ProviderEngine
 	checkerEng             *checker.Engine
 	syncCore               *chainsync.Core
@@ -202,7 +207,7 @@ func (builder *ExecutionNodeBuilder) LoadComponentsAndModules() {
 		Module("system specs", exeNode.LoadSystemSpecs).
 		Module("execution metrics", exeNode.LoadExecutionMetrics).
 		Module("sync core", exeNode.LoadSyncCore).
-		Module("execution receipts storage", exeNode.LoadExecutionReceiptsStorage).
+		Module("execution storage", exeNode.LoadExecutionStorage).
 		Module("follower distributor", exeNode.LoadFollowerDistributor).
 		Module("authorization checking function", exeNode.LoadAuthorizationCheckingFunction).
 		Module("execution data datastore", exeNode.LoadExecutionDataDatastore).
@@ -279,9 +284,9 @@ func (exeNode *ExecutionNode) LoadExecutionMetrics(node *NodeConfig) error {
 	// report the highest executed block height as soon as possible
 	// this is guaranteed to exist because LoadBootstrapper has inserted
 	// the root block as executed block
-	var height uint64
 	var blockID flow.Identifier
-	err := node.DB.View(procedure.GetLastExecutedBlock(&height, &blockID))
+	reader := node.ProtocolDB.Reader()
+	err := operation.RetrieveExecutedBlock(reader, &blockID)
 	if err != nil {
 		// database has not been bootstrapped yet
 		if errors.Is(err, storageerr.ErrNotFound) {
@@ -290,7 +295,12 @@ func (exeNode *ExecutionNode) LoadExecutionMetrics(node *NodeConfig) error {
 		return fmt.Errorf("could not get highest executed block: %w", err)
 	}
 
-	exeNode.collector.ExecutionLastExecutedBlockHeight(height)
+	executed, err := node.Storage.Headers.ByBlockID(blockID)
+	if err != nil {
+		return fmt.Errorf("could not get header by id: %v: %w", blockID, err)
+	}
+
+	exeNode.collector.ExecutionLastExecutedBlockHeight(executed.Height)
 	return nil
 }
 
@@ -300,11 +310,19 @@ func (exeNode *ExecutionNode) LoadSyncCore(node *NodeConfig) error {
 	return err
 }
 
-func (exeNode *ExecutionNode) LoadExecutionReceiptsStorage(
+func (exeNode *ExecutionNode) LoadExecutionStorage(
 	node *NodeConfig,
 ) error {
-	exeNode.results = storage.NewExecutionResults(node.Metrics.Cache, node.DB)
-	exeNode.myReceipts = storage.NewMyExecutionReceipts(node.Metrics.Cache, node.DB, node.Storage.Receipts.(*storage.ExecutionReceipts))
+	db := node.ProtocolDB
+	exeNode.commits = store.NewCommits(node.Metrics.Cache, db)
+	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)
+
+	// Needed for gRPC server, make sure to assign to main scoped vars
+	exeNode.events = store.NewEvents(node.Metrics.Cache, db)
+	exeNode.serviceEvents = store.NewServiceEvents(node.Metrics.Cache, db)
+	exeNode.txResults = store.NewTransactionResults(node.Metrics.Cache, db, exeNode.exeConf.transactionResultsCacheSize)
 	return nil
 }
 
@@ -438,7 +456,7 @@ func (exeNode *ExecutionNode) LoadGCPBlockDataUploader(
 	retryableUploader := uploader.NewBadgerRetryableUploaderWrapper(
 		asyncUploader,
 		node.Storage.Blocks,
-		node.Storage.Commits,
+		exeNode.commits,
 		node.Storage.Collections,
 		exeNode.events,
 		exeNode.results,
@@ -743,14 +761,18 @@ func (exeNode *ExecutionNode) LoadExecutionState(
 	chunkDataPacks := store.NewChunkDataPacks(node.Metrics.Cache,
 		pebbleimpl.ToDB(chunkDataPackDB), node.Storage.Collections, exeNode.exeConf.chunkDataPackCacheSize)
 
-	// Needed for gRPC server, make sure to assign to main scoped vars
-	exeNode.events = storage.NewEvents(node.Metrics.Cache, node.DB)
-	exeNode.serviceEvents = storage.NewServiceEvents(node.Metrics.Cache, node.DB)
-	exeNode.txResults = storage.NewTransactionResults(node.Metrics.Cache, node.DB, exeNode.exeConf.transactionResultsCacheSize)
+	getLatestFinalized := func() (uint64, error) {
+		final, err := node.State.Final().Head()
+		if err != nil {
+			return 0, err
+		}
+
+		return final.Height, nil
+	}
 
 	exeNode.executionState = state.NewExecutionState(
 		exeNode.ledgerStorage,
-		node.Storage.Commits,
+		exeNode.commits,
 		node.Storage.Blocks,
 		node.Storage.Headers,
 		node.Storage.Collections,
@@ -760,7 +782,8 @@ func (exeNode *ExecutionNode) LoadExecutionState(
 		exeNode.events,
 		exeNode.serviceEvents,
 		exeNode.txResults,
-		node.DB,
+		node.ProtocolDB,
+		getLatestFinalized,
 		node.Tracer,
 		exeNode.registerStore,
 		exeNode.exeConf.enableStorehouse,
@@ -1316,7 +1339,7 @@ func (exeNode *ExecutionNode) LoadGrpcServer(
 		exeNode.events,
 		exeNode.results,
 		exeNode.txResults,
-		node.Storage.Commits,
+		exeNode.commits,
 		exeNode.metricsProvider,
 		node.RootChainID,
 		signature.NewBlockSignerDecoder(exeNode.committee),
@@ -1330,7 +1353,7 @@ func (exeNode *ExecutionNode) LoadBootstrapper(node *NodeConfig) error {
 	// check if the execution database already exists
 	bootstrapper := bootstrap.NewBootstrapper(node.Logger)
 
-	commit, bootstrapped, err := bootstrapper.IsBootstrapped(node.DB)
+	commit, bootstrapped, err := bootstrapper.IsBootstrapped(node.ProtocolDB)
 	if err != nil {
 		return fmt.Errorf("could not query database to know whether database has been bootstrapped: %w", err)
 	}
@@ -1355,7 +1378,7 @@ func (exeNode *ExecutionNode) LoadBootstrapper(node *NodeConfig) error {
 			return fmt.Errorf("could not load bootstrap state from checkpoint file: %w", err)
 		}
 
-		err = bootstrapper.BootstrapExecutionDatabase(node.DB, node.RootSeal)
+		err = bootstrapper.BootstrapExecutionDatabase(node.ProtocolDB, node.RootSeal)
 		if err != nil {
 			return fmt.Errorf("could not bootstrap execution database: %w", err)
 		}

cmd/node_builder.go

@@ -27,6 +27,7 @@ import (
 	"github.com/onflow/flow-go/network/p2p"
 	"github.com/onflow/flow-go/state/protocol"
 	"github.com/onflow/flow-go/state/protocol/events"
+	"github.com/onflow/flow-go/storage"
 	bstorage "github.com/onflow/flow-go/storage/badger"
 	"github.com/onflow/flow-go/utils/grpcutils"
 )
@@ -202,6 +203,7 @@ type NodeConfig struct {
 	Metrics           Metrics
 	DB                *badger.DB
 	PebbleDB          *pebble.DB
+	ProtocolDB        storage.DB
 	SecretsDB         *badger.DB
 	Storage           Storage
 	ProtocolEvents    *events.Distributor

cmd/observer/node_builder/observer_builder.go

@@ -299,6 +299,10 @@ type ObserverServiceBuilder struct {
 	EventsIndex         *index.EventsIndex
 	ScriptExecutor      *backend.ScriptExecutor
 
+	// storage
+	events                  storage.Events
+	lightTransactionResults storage.LightTransactionResults
+
 	// available until after the network has started. Hence, a factory function that needs to be called just before
 	// creating the sync engine
 	SyncEngineParticipantsProviderFactory func() module.IdentifierProvider
@@ -1326,7 +1330,7 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 			indexedBlockHeight = store.NewConsumerProgress(badgerimpl.ToDB(builder.DB), module.ConsumeProgressExecutionDataIndexerBlockHeight)
 			return nil
 		}).Module("transaction results storage", func(node *cmd.NodeConfig) error {
-			builder.Storage.LightTransactionResults = bstorage.NewLightTransactionResults(node.Metrics.Cache, node.DB, bstorage.DefaultCacheSize)
+			builder.lightTransactionResults = store.NewLightTransactionResults(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
@@ -1415,13 +1419,13 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 			indexerCore, err := indexer.New(
 				builder.Logger,
 				metrics.NewExecutionStateIndexerCollector(),
-				builder.DB,
+				builder.ProtocolDB,
 				builder.Storage.RegisterIndex,
 				builder.Storage.Headers,
-				builder.Storage.Events,
+				builder.events,
 				builder.Storage.Collections,
 				builder.Storage.Transactions,
-				builder.Storage.LightTransactionResults,
+				builder.lightTransactionResults,
 				builder.RootChainID.Chain(),
 				indexerDerivedChainData,
 				collectionExecutedMetric,
@@ -1769,19 +1773,19 @@ func (builder *ObserverServiceBuilder) enqueueRPCServer() {
 		return nil
 	})
 	builder.Module("events storage", func(node *cmd.NodeConfig) error {
-		builder.Storage.Events = bstorage.NewEvents(node.Metrics.Cache, node.DB)
+		builder.events = store.NewEvents(node.Metrics.Cache, node.ProtocolDB)
 		return nil
 	})
 	builder.Module("reporter", func(node *cmd.NodeConfig) error {
 		builder.Reporter = index.NewReporter()
 		return nil
 	})
 	builder.Module("events index", func(node *cmd.NodeConfig) error {
-		builder.EventsIndex = index.NewEventsIndex(builder.Reporter, builder.Storage.Events)
+		builder.EventsIndex = index.NewEventsIndex(builder.Reporter, builder.events)
 		return nil
 	})
 	builder.Module("transaction result index", func(node *cmd.NodeConfig) error {
-		builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.Storage.LightTransactionResults)
+		builder.TxResultsIndex = index.NewTransactionResultsIndex(builder.Reporter, builder.lightTransactionResults)
 		return nil
 	})
 	builder.Module("script executor", func(node *cmd.NodeConfig) error {

cmd/scaffold.go

@@ -83,6 +83,7 @@ import (
 	"github.com/onflow/flow-go/storage"
 	bstorage "github.com/onflow/flow-go/storage/badger"
 	"github.com/onflow/flow-go/storage/badger/operation"
+	"github.com/onflow/flow-go/storage/operation/badgerimpl"
 	sutil "github.com/onflow/flow-go/storage/util"
 	"github.com/onflow/flow-go/utils/logging"
 )
@@ -1108,9 +1109,12 @@ func (fnb *FlowNodeBuilder) initBadgerDB() error {
 		return fmt.Errorf("could not open public db: %w", err)
 	}
 	fnb.DB = publicDB
+	// set badger db as protocol db
+	// TODO: making it dynamic to switch between badger and pebble
+	fnb.ProtocolDB = badgerimpl.ToDB(publicDB)
 
 	fnb.ShutdownFunc(func() error {
-		if err := fnb.DB.Close(); err != nil {
+		if err := publicDB.Close(); err != nil {
 			return fmt.Errorf("error closing protocol database: %w", err)
 		}
 		return nil
@@ -1222,7 +1226,6 @@ func (fnb *FlowNodeBuilder) initStorage() error {
 	collections := bstorage.NewCollections(fnb.DB, transactions)
 	setups := bstorage.NewEpochSetups(fnb.Metrics.Cache, fnb.DB)
 	epochCommits := bstorage.NewEpochCommits(fnb.Metrics.Cache, fnb.DB)
-	commits := bstorage.NewCommits(fnb.Metrics.Cache, fnb.DB)
 	protocolState := bstorage.NewEpochProtocolStateEntries(fnb.Metrics.Cache, setups, epochCommits, fnb.DB,
 		bstorage.DefaultEpochProtocolStateCacheSize, bstorage.DefaultProtocolStateIndexCacheSize)
 	protocolKVStores := bstorage.NewProtocolKVStore(fnb.Metrics.Cache, fnb.DB,
@@ -1232,8 +1235,6 @@ func (fnb *FlowNodeBuilder) initStorage() error {
 	fnb.Storage = Storage{
 		Headers:                   headers,
 		Guarantees:                guarantees,
-		Receipts:                  receipts,
-		Results:                   results,
 		Seals:                     seals,
 		Index:                     index,
 		Payloads:                  payloads,
@@ -1246,7 +1247,9 @@ func (fnb *FlowNodeBuilder) initStorage() error {
 		VersionBeacons:            versionBeacons,
 		EpochProtocolStateEntries: protocolState,
 		ProtocolKVStore:           protocolKVStores,
-		Commits:                   commits,
+
+		Results:  results,
+		Receipts: receipts,
 	}
 
 	return nil