[Access] Add support for pebbleDB to execution data tracker/pruner

Makefile

@@ -211,7 +211,6 @@ generate-mocks: install-mock-generators
 	mockery --name 'ExecForkActor' --structname 'ExecForkActorMock' --dir=module/mempool/consensus/mock/ --case=underscore --output="./module/mempool/consensus/mock/" --outpkg="mock"
 	mockery --name '.*' --dir=engine/verification/fetcher/ --case=underscore --output="./engine/verification/fetcher/mock" --outpkg="mockfetcher"
 	mockery --name '.*' --dir=./cmd/util/ledger/reporters --case=underscore --output="./cmd/util/ledger/reporters/mock" --outpkg="mock"
-	mockery --name 'Storage' --dir=module/executiondatasync/tracker --case=underscore --output="module/executiondatasync/tracker/mock" --outpkg="mocktracker"
 	mockery --name 'ScriptExecutor' --dir=module/execution --case=underscore --output="module/execution/mock" --outpkg="mock"
 	mockery --name 'StorageSnapshot' --dir=fvm/storage/snapshot --case=underscore --output="fvm/storage/snapshot/mock" --outpkg="mock"
 

cmd/access/node_builder/access_node_builder.go

@@ -73,7 +73,6 @@ import (
 	execdatacache "github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
 	"github.com/onflow/flow-go/module/executiondatasync/pruner"
 	edstorage "github.com/onflow/flow-go/module/executiondatasync/storage"
-	"github.com/onflow/flow-go/module/executiondatasync/tracker"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
 	"github.com/onflow/flow-go/module/grpcserver"
 	"github.com/onflow/flow-go/module/id"
@@ -322,7 +321,7 @@ type FlowAccessNodeBuilder struct {
 	collectionExecutedMetric   module.CollectionExecutedMetric
 	ExecutionDataPruner        *pruner.Pruner
 	ExecutionDatastoreManager  edstorage.DatastoreManager
-	ExecutionDataTracker       tracker.Storage
+	ExecutionDataTracker       storage.ExecutionDataTracker
 	versionControl             *version.VersionControl
 
 	// The sync engine participants provider is the libp2p peer store for the access node
@@ -670,15 +669,27 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				}
 
 				trackerDir := filepath.Join(builder.executionDataDir, "tracker")
-				builder.ExecutionDataTracker, err = tracker.OpenStorage(
-					trackerDir,
-					sealed.Height,
-					node.Logger,
-					tracker.WithPruneCallback(func(c cid.Cid) error {
-						// TODO: use a proper context here
-						return builder.ExecutionDataBlobstore.DeleteBlob(context.TODO(), c)
-					}),
-				)
+				if executionDataDBMode == execution_data.ExecutionDataDBModeBadger {
+					builder.ExecutionDataTracker, err = bstorage.NewExecutionDataTracker(
+						node.Logger,
+						trackerDir,
+						sealed.Height,
+						bstorage.WithPruneCallback(func(c cid.Cid) error {
+							// TODO: use a proper context here
+							return builder.ExecutionDataBlobstore.DeleteBlob(context.TODO(), c)
+						}),
+					)
+				} else {
+					builder.ExecutionDataTracker, err = pstorage.NewExecutionDataTracker(
+						node.Logger,
+						trackerDir,
+						sealed.Height,
+						pstorage.WithPruneCallback(func(c cid.Cid) error {
+							// TODO: use a proper context here
+							return builder.ExecutionDataBlobstore.DeleteBlob(context.TODO(), c)
+						}),
+					)
+				}
 				if err != nil {
 					return nil, fmt.Errorf("failed to create execution data tracker: %w", err)
 				}

cmd/execution_builder.go

@@ -78,7 +78,6 @@ import (
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	exedataprovider "github.com/onflow/flow-go/module/executiondatasync/provider"
 	"github.com/onflow/flow-go/module/executiondatasync/pruner"
-	"github.com/onflow/flow-go/module/executiondatasync/tracker"
 	"github.com/onflow/flow-go/module/finalizedreader"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
 	"github.com/onflow/flow-go/module/mempool/queue"
@@ -90,8 +89,8 @@ import (
 	"github.com/onflow/flow-go/state/protocol"
 	badgerState "github.com/onflow/flow-go/state/protocol/badger"
 	"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"
+	bstorage "github.com/onflow/flow-go/storage/badger"
 	"github.com/onflow/flow-go/storage/badger/procedure"
 	storagepebble "github.com/onflow/flow-go/storage/pebble"
 	sutil "github.com/onflow/flow-go/storage/util"
@@ -133,11 +132,11 @@ type ExecutionNode struct {
 	committee              hotstuff.DynamicCommittee
 	ledgerStorage          *ledger.Ledger
 	registerStore          *storehouse.RegisterStore
-	events                 *storage.Events
-	serviceEvents          *storage.ServiceEvents
-	txResults              *storage.TransactionResults
-	results                *storage.ExecutionResults
-	myReceipts             *storage.MyExecutionReceipts
+	events                 *bstorage.Events
+	serviceEvents          *bstorage.ServiceEvents
+	txResults              *bstorage.TransactionResults
+	results                *bstorage.ExecutionResults
+	myReceipts             *bstorage.MyExecutionReceipts
 	providerEngine         exeprovider.ProviderEngine
 	checkerEng             *checker.Engine
 	syncCore               *chainsync.Core
@@ -157,7 +156,7 @@ type ExecutionNode struct {
 	executionDataDatastore *badgerds.Datastore
 	executionDataPruner    *pruner.Pruner
 	executionDataBlobstore blobs.Blobstore
-	executionDataTracker   tracker.Storage
+	executionDataTracker   storage.ExecutionDataTracker
 	blobService            network.BlobService
 	blobserviceDependable  *module.ProxiedReadyDoneAware
 }
@@ -281,7 +280,7 @@ func (exeNode *ExecutionNode) LoadExecutionMetrics(node *NodeConfig) error {
 	err := node.DB.View(procedure.GetHighestExecutedBlock(&height, &blockID))
 	if err != nil {
 		// database has not been bootstrapped yet
-		if errors.Is(err, storageerr.ErrNotFound) {
+		if errors.Is(err, storage.ErrNotFound) {
 			return nil
 		}
 		return fmt.Errorf("could not get highest executed block: %w", err)
@@ -300,8 +299,8 @@ func (exeNode *ExecutionNode) LoadSyncCore(node *NodeConfig) error {
 func (exeNode *ExecutionNode) LoadExecutionReceiptsStorage(
 	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))
+	exeNode.results = bstorage.NewExecutionResults(node.Metrics.Cache, node.DB)
+	exeNode.myReceipts = bstorage.NewMyExecutionReceipts(node.Metrics.Cache, node.DB, node.Storage.Receipts.(*bstorage.ExecutionReceipts))
 	return nil
 }
 
@@ -440,7 +439,7 @@ func (exeNode *ExecutionNode) LoadGCPBlockDataUploader(
 		exeNode.events,
 		exeNode.results,
 		exeNode.txResults,
-		storage.NewComputationResultUploadStatus(node.DB),
+		bstorage.NewComputationResultUploadStatus(node.DB),
 		execution_data.NewDownloader(exeNode.blobService),
 		exeNode.collector)
 	if retryableUploader == nil {
@@ -739,9 +738,9 @@ func (exeNode *ExecutionNode) LoadExecutionState(
 		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)
+	exeNode.events = bstorage.NewEvents(node.Metrics.Cache, node.DB)
+	exeNode.serviceEvents = bstorage.NewServiceEvents(node.Metrics.Cache, node.DB)
+	exeNode.txResults = bstorage.NewTransactionResults(node.Metrics.Cache, node.DB, exeNode.exeConf.transactionResultsCacheSize)
 
 	exeNode.executionState = state.NewExecutionState(
 		exeNode.ledgerStorage,
@@ -946,11 +945,11 @@ func (exeNode *ExecutionNode) LoadExecutionDataPruner(
 	}
 
 	trackerDir := filepath.Join(exeNode.exeConf.executionDataDir, "tracker")
-	exeNode.executionDataTracker, err = tracker.OpenStorage(
+	exeNode.executionDataTracker, err = bstorage.NewExecutionDataTracker(
+		node.Logger,
 		trackerDir,
 		sealed.Height,
-		node.Logger,
-		tracker.WithPruneCallback(func(c cid.Cid) error {
+		bstorage.WithPruneCallback(func(c cid.Cid) error {
 			// TODO: use a proper context here
 			return exeNode.executionDataBlobstore.DeleteBlob(context.TODO(), c)
 		}),

cmd/observer/node_builder/observer_builder.go

@@ -72,7 +72,6 @@ import (
 	execdatacache "github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
 	"github.com/onflow/flow-go/module/executiondatasync/pruner"
 	edstorage "github.com/onflow/flow-go/module/executiondatasync/storage"
-	"github.com/onflow/flow-go/module/executiondatasync/tracker"
 	finalizer "github.com/onflow/flow-go/module/finalizer/consensus"
 	"github.com/onflow/flow-go/module/grpcserver"
 	"github.com/onflow/flow-go/module/id"
@@ -287,7 +286,7 @@ type ObserverServiceBuilder struct {
 	ExecutionDataBlobstore    blobs.Blobstore
 	ExecutionDataPruner       *pruner.Pruner
 	ExecutionDatastoreManager edstorage.DatastoreManager
-	ExecutionDataTracker      tracker.Storage
+	ExecutionDataTracker      storage.ExecutionDataTracker
 
 	RegistersAsyncStore *execution.RegistersAsyncStore
 	Reporter            *index.Reporter
@@ -1239,15 +1238,27 @@ func (builder *ObserverServiceBuilder) BuildExecutionSyncComponents() *ObserverS
 				}
 
 				trackerDir := filepath.Join(builder.executionDataDir, "tracker")
-				builder.ExecutionDataTracker, err = tracker.OpenStorage(
-					trackerDir,
-					sealed.Height,
-					node.Logger,
-					tracker.WithPruneCallback(func(c cid.Cid) error {
-						// TODO: use a proper context here
-						return builder.ExecutionDataBlobstore.DeleteBlob(context.TODO(), c)
-					}),
-				)
+				if executionDataDBMode == execution_data.ExecutionDataDBModeBadger {
+					builder.ExecutionDataTracker, err = bstorage.NewExecutionDataTracker(
+						node.Logger,
+						trackerDir,
+						sealed.Height,
+						bstorage.WithPruneCallback(func(c cid.Cid) error {
+							// TODO: use a proper context here
+							return builder.ExecutionDataBlobstore.DeleteBlob(context.TODO(), c)
+						}),
+					)
+				} else {
+					builder.ExecutionDataTracker, err = pstorage.NewExecutionDataTracker(
+						node.Logger,
+						trackerDir,
+						sealed.Height,
+						pstorage.WithPruneCallback(func(c cid.Cid) error {
+							// TODO: use a proper context here
+							return builder.ExecutionDataBlobstore.DeleteBlob(context.TODO(), c)
+						}),
+					)
+				}
 				if err != nil {
 					return nil, fmt.Errorf("failed to create execution data tracker: %w", err)
 				}

engine/execution/computation/computer/computer_test.go

@@ -49,12 +49,12 @@ import (
 	"github.com/onflow/flow-go/module/epochs"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	"github.com/onflow/flow-go/module/executiondatasync/provider"
-	mocktracker "github.com/onflow/flow-go/module/executiondatasync/tracker/mock"
 	"github.com/onflow/flow-go/module/mempool/entity"
 	"github.com/onflow/flow-go/module/metrics"
 	modulemock "github.com/onflow/flow-go/module/mock"
 	requesterunit "github.com/onflow/flow-go/module/state_synchronization/requester/unittest"
 	"github.com/onflow/flow-go/module/trace"
+	mocktracker "github.com/onflow/flow-go/storage/mocks"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -179,7 +179,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			Times(1) // 1 block
 
 		bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -317,7 +317,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		committer := new(computermock.ViewCommitter)
 
 		bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -421,7 +421,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		comm := new(computermock.ViewCommitter)
 
 		bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -477,7 +477,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		committer := new(computermock.ViewCommitter)
 
 		bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -710,7 +710,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 			vm := fvm.NewVirtualMachine()
 
 			bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-			trackerStorage := mocktracker.NewMockStorage()
+			trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 			prov := provider.NewProvider(
 				zerolog.Nop(),
@@ -821,7 +821,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		vm := fvm.NewVirtualMachine()
 
 		bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -934,7 +934,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		vm := fvm.NewVirtualMachine()
 
 		bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -980,7 +980,7 @@ func TestBlockExecutor_ExecuteBlock(t *testing.T) {
 		bservice := requesterunit.MockBlobService(
 			blockstore.NewBlockstore(
 				dssync.MutexWrap(datastore.NewMapDatastore())))
-		trackerStorage := mocktracker.NewMockStorage()
+		trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 		prov := provider.NewProvider(
 			zerolog.Nop(),
@@ -1297,7 +1297,7 @@ func Test_ExecutingSystemCollection(t *testing.T) {
 		Return(nil)
 
 	bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-	trackerStorage := mocktracker.NewMockStorage()
+	trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 	prov := provider.NewProvider(
 		zerolog.Nop(),

engine/execution/computation/execution_verification_test.go

@@ -39,10 +39,10 @@ import (
 	"github.com/onflow/flow-go/module/chunks"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	exedataprovider "github.com/onflow/flow-go/module/executiondatasync/provider"
-	mocktracker "github.com/onflow/flow-go/module/executiondatasync/tracker/mock"
 	"github.com/onflow/flow-go/module/metrics"
 	requesterunit "github.com/onflow/flow-go/module/state_synchronization/requester/unittest"
 	"github.com/onflow/flow-go/module/trace"
+	mocktracker "github.com/onflow/flow-go/storage/mocks"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -753,7 +753,7 @@ func executeBlockAndVerifyWithParameters(t *testing.T,
 	ledgerCommiter := committer.NewLedgerViewCommitter(ledger, tracer)
 
 	bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-	trackerStorage := mocktracker.NewMockStorage()
+	trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 	prov := exedataprovider.NewProvider(
 		zerolog.Nop(),

engine/execution/computation/manager_benchmark_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 	"time"
 
-	blockstore "github.com/ipfs/boxo/blockstore"
+	"github.com/ipfs/boxo/blockstore"
 	"github.com/ipfs/go-datastore"
 	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/onflow/cadence/runtime"
@@ -25,12 +25,12 @@ import (
 	"github.com/onflow/flow-go/model/flow"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	exedataprovider "github.com/onflow/flow-go/module/executiondatasync/provider"
-	mocktracker "github.com/onflow/flow-go/module/executiondatasync/tracker/mock"
 	"github.com/onflow/flow-go/module/mempool/entity"
 	"github.com/onflow/flow-go/module/metrics"
 	module "github.com/onflow/flow-go/module/mock"
 	requesterunit "github.com/onflow/flow-go/module/state_synchronization/requester/unittest"
 	"github.com/onflow/flow-go/module/trace"
+	mocktracker "github.com/onflow/flow-go/storage/mocks"
 	"github.com/onflow/flow-go/utils/unittest"
 )
 
@@ -180,7 +180,7 @@ func benchmarkComputeBlock(
 		Return(nil, nil)
 
 	bservice := requesterunit.MockBlobService(blockstore.NewBlockstore(dssync.MutexWrap(datastore.NewMapDatastore())))
-	trackerStorage := mocktracker.NewMockStorage()
+	trackerStorage := mocktracker.NewMockExecutionDataTracker()
 
 	prov := exedataprovider.NewProvider(
 		zerolog.Nop(),