[Access] Add registerDB pruning module

cmd/access/node_builder/access_node_builder.go

@@ -71,7 +71,7 @@ import (
 	"github.com/onflow/flow-go/module/execution"
 	"github.com/onflow/flow-go/module/executiondatasync/execution_data"
 	execdatacache "github.com/onflow/flow-go/module/executiondatasync/execution_data/cache"
-	"github.com/onflow/flow-go/module/executiondatasync/pruner"
+	edpruner "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"
@@ -173,6 +173,9 @@ type AccessNodeConfig struct {
 	programCacheSize                     uint
 	checkPayerBalance                    bool
 	versionControlEnabled                bool
+	registerDBPruningEnabled             bool
+	registerDBPruneTickerInterval        time.Duration
+	registerDBPruneThrottleDelay         time.Duration
 }
 
 type PublicNetworkConfig struct {
@@ -264,8 +267,8 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		executionDataIndexingEnabled:         false,
 		executionDataDBMode:                  execution_data.ExecutionDataDBModeBadger.String(),
 		executionDataPrunerHeightRangeTarget: 0,
-		executionDataPrunerThreshold:         pruner.DefaultThreshold,
-		executionDataPruningInterval:         pruner.DefaultPruningInterval,
+		executionDataPrunerThreshold:         edpruner.DefaultThreshold,
+		executionDataPruningInterval:         edpruner.DefaultPruningInterval,
 		registersDBPath:                      filepath.Join(homedir, ".flow", "execution_state"),
 		checkpointFile:                       cmd.NotSet,
 		scriptExecutorConfig:                 query.NewDefaultConfig(),
@@ -276,6 +279,9 @@ func DefaultAccessNodeConfig() *AccessNodeConfig {
 		programCacheSize:                     0,
 		checkPayerBalance:                    false,
 		versionControlEnabled:                true,
+		registerDBPruningEnabled:             false,
+		registerDBPruneTickerInterval:        pstorage.DefaultPruneTickerInterval,
+		registerDBPruneThrottleDelay:         pstorage.DefaultPruneThrottleDelay,
 	}
 }
 
@@ -287,43 +293,46 @@ type FlowAccessNodeBuilder struct {
 	*AccessNodeConfig
 
 	// components
-	FollowerState              protocol.FollowerState
-	SyncCore                   *chainsync.Core
-	RpcEng                     *rpc.Engine
-	FollowerDistributor        *consensuspubsub.FollowerDistributor
-	CollectionRPC              access.AccessAPIClient
-	TransactionTimings         *stdmap.TransactionTimings
-	CollectionsToMarkFinalized *stdmap.Times
-	CollectionsToMarkExecuted  *stdmap.Times
-	BlocksToMarkExecuted       *stdmap.Times
-	TransactionMetrics         *metrics.TransactionCollector
-	RestMetrics                *metrics.RestCollector
-	AccessMetrics              module.AccessMetrics
-	PingMetrics                module.PingMetrics
-	Committee                  hotstuff.DynamicCommittee
-	Finalized                  *flow.Header // latest finalized block that the node knows of at startup time
-	Pending                    []*flow.Header
-	FollowerCore               module.HotStuffFollower
-	Validator                  hotstuff.Validator
-	ExecutionDataDownloader    execution_data.Downloader
-	PublicBlobService          network.BlobService
-	ExecutionDataRequester     state_synchronization.ExecutionDataRequester
-	ExecutionDataStore         execution_data.ExecutionDataStore
-	ExecutionDataBlobstore     blobs.Blobstore
-	ExecutionDataCache         *execdatacache.ExecutionDataCache
-	ExecutionIndexer           *indexer.Indexer
-	ExecutionIndexerCore       *indexer.IndexerCore
-	ScriptExecutor             *backend.ScriptExecutor
-	RegistersAsyncStore        *execution.RegistersAsyncStore
-	Reporter                   *index.Reporter
-	EventsIndex                *index.EventsIndex
-	TxResultsIndex             *index.TransactionResultsIndex
-	IndexerDependencies        *cmd.DependencyList
-	collectionExecutedMetric   module.CollectionExecutedMetric
-	ExecutionDataPruner        *pruner.Pruner
-	ExecutionDatastoreManager  edstorage.DatastoreManager
-	ExecutionDataTracker       tracker.Storage
-	versionControl             *version.VersionControl
+	FollowerState                protocol.FollowerState
+	SyncCore                     *chainsync.Core
+	RpcEng                       *rpc.Engine
+	FollowerDistributor          *consensuspubsub.FollowerDistributor
+	CollectionRPC                access.AccessAPIClient
+	TransactionTimings           *stdmap.TransactionTimings
+	CollectionsToMarkFinalized   *stdmap.Times
+	CollectionsToMarkExecuted    *stdmap.Times
+	BlocksToMarkExecuted         *stdmap.Times
+	TransactionMetrics           *metrics.TransactionCollector
+	RestMetrics                  *metrics.RestCollector
+	RegisterDBPrunerMetrics      *metrics.RegisterDBPrunerCollector
+	AccessMetrics                module.AccessMetrics
+	PingMetrics                  module.PingMetrics
+	Committee                    hotstuff.DynamicCommittee
+	Finalized                    *flow.Header // latest finalized block that the node knows of at startup time
+	Pending                      []*flow.Header
+	FollowerCore                 module.HotStuffFollower
+	Validator                    hotstuff.Validator
+	ExecutionDataDownloader      execution_data.Downloader
+	PublicBlobService            network.BlobService
+	ExecutionDataRequester       state_synchronization.ExecutionDataRequester
+	ExecutionDataStore           execution_data.ExecutionDataStore
+	ExecutionDataBlobstore       blobs.Blobstore
+	ExecutionDataCache           *execdatacache.ExecutionDataCache
+	ExecutionIndexer             *indexer.Indexer
+	ExecutionIndexerCore         *indexer.IndexerCore
+	ScriptExecutor               *backend.ScriptExecutor
+	RegistersAsyncStore          *execution.RegistersAsyncStore
+	Reporter                     *index.Reporter
+	EventsIndex                  *index.EventsIndex
+	TxResultsIndex               *index.TransactionResultsIndex
+	IndexerDependencies          *cmd.DependencyList
+	collectionExecutedMetric     module.CollectionExecutedMetric
+	ExecutionDataPruner          *edpruner.Pruner
+	ExecutionDatastoreManager    edstorage.DatastoreManager
+	ExecutionDataTracker         tracker.Storage
+	versionControl               *version.VersionControl
+	RegisterDB                   *pebble.DB
+	RegisterDBPrunerDependencies *cmd.DependencyList
 
 	// The sync engine participants provider is the libp2p peer store for the access node
 	// which is not available until after the network has started.
@@ -542,6 +551,10 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 	requesterDependable := module.NewProxiedReadyDoneAware()
 	builder.IndexerDependencies.Add(requesterDependable)
 
+	// setup dependency chain to ensure register db pruner starts after the indexer
+	indexerDependable := module.NewProxiedReadyDoneAware()
+	builder.RegisterDBPrunerDependencies.Add(indexerDependable)
+
 	executionDataPrunerEnabled := builder.executionDataPrunerHeightRangeTarget != 0
 
 	builder.
@@ -771,16 +784,16 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 			}
 
 			var err error
-			builder.ExecutionDataPruner, err = pruner.NewPruner(
+			builder.ExecutionDataPruner, err = edpruner.NewPruner(
 				node.Logger,
 				prunerMetrics,
 				builder.ExecutionDataTracker,
-				pruner.WithPruneCallback(func(ctx context.Context) error {
+				edpruner.WithPruneCallback(func(ctx context.Context) error {
 					return builder.ExecutionDatastoreManager.CollectGarbage(ctx)
 				}),
-				pruner.WithHeightRangeTarget(builder.executionDataPrunerHeightRangeTarget),
-				pruner.WithThreshold(builder.executionDataPrunerThreshold),
-				pruner.WithPruningInterval(builder.executionDataPruningInterval),
+				edpruner.WithHeightRangeTarget(builder.executionDataPrunerHeightRangeTarget),
+				edpruner.WithThreshold(builder.executionDataPrunerThreshold),
+				edpruner.WithPruningInterval(builder.executionDataPruningInterval),
 			)
 			if err != nil {
 				return nil, fmt.Errorf("failed to create execution data pruner: %w", err)
@@ -845,16 +858,16 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 				// 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
 				// take hours to complete.
-
-				pdb, err := pstorage.OpenRegisterPebbleDB(builder.registersDBPath)
+				var err error
+				builder.RegisterDB, err = pstorage.OpenRegisterPebbleDB(builder.registersDBPath)
 				if err != nil {
 					return nil, fmt.Errorf("could not open registers db: %w", err)
 				}
 				builder.ShutdownFunc(func() error {
-					return pdb.Close()
+					return builder.RegisterDB.Close()
 				})
 
-				bootstrapped, err := pstorage.IsBootstrapped(pdb)
+				bootstrapped, err := pstorage.IsBootstrapped(builder.RegisterDB)
 				if err != nil {
 					return nil, fmt.Errorf("could not check if registers db is bootstrapped: %w", err)
 				}
@@ -886,7 +899,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					}
 
 					rootHash := ledger.RootHash(builder.RootSeal.FinalState)
-					bootstrap, err := pstorage.NewRegisterBootstrap(pdb, checkpointFile, checkpointHeight, rootHash, builder.Logger)
+					bootstrap, err := pstorage.NewRegisterBootstrap(builder.RegisterDB, checkpointFile, checkpointHeight, rootHash, builder.Logger)
 					if err != nil {
 						return nil, fmt.Errorf("could not create registers bootstrap: %w", err)
 					}
@@ -899,7 +912,7 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					}
 				}
 
-				registers, err := pstorage.NewRegisters(pdb)
+				registers, err := pstorage.NewRegisters(builder.RegisterDB)
 				if err != nil {
 					return nil, fmt.Errorf("could not create registers storage: %w", err)
 				}
@@ -993,8 +1006,31 @@ func (builder *FlowAccessNodeBuilder) BuildExecutionSyncComponents() *FlowAccess
 					return nil, err
 				}
 
+				// add indexer into ReadyDoneAware dependency passed to pruner. This allows the register db pruner
+				// to wait for the indexer to be ready before starting.
+				indexerDependable.Init(builder.ExecutionIndexer)
+
 				return builder.ExecutionIndexer, nil
-			}, builder.IndexerDependencies)
+			}, builder.IndexerDependencies).
+			DependableComponent("register db pruner", func(node *cmd.NodeConfig) (module.ReadyDoneAware, error) {
+				if !builder.registerDBPruningEnabled {
+					return &module.NoopReadyDoneAware{}, nil
+				}
+
+				registerDBPruner, err := pstorage.NewRegisterPruner(
+					node.Logger,
+					builder.RegisterDB,
+					pstorage.WithPrunerMetrics(builder.RegisterDBPrunerMetrics),
+					//pstorage.WithPruneThreshold(builder.registerDBPruneThreshold),
+					pstorage.WithPruneThrottleDelay(builder.registerDBPruneThrottleDelay),
+					pstorage.WithPruneTickerInterval(builder.registerDBPruneTickerInterval),
+				)
+				if err != nil {
+					return nil, fmt.Errorf("failed to create register db pruner: %w", err)
+				}
+
+				return registerDBPruner, nil
+			}, builder.RegisterDBPrunerDependencies)
 	}
 
 	if builder.stateStreamConf.ListenAddr != "" {
@@ -1121,10 +1157,11 @@ func FlowAccessNode(nodeBuilder *cmd.FlowNodeBuilder) *FlowAccessNodeBuilder {
 	dist := consensuspubsub.NewFollowerDistributor()
 	dist.AddProposalViolationConsumer(notifications.NewSlashingViolationsConsumer(nodeBuilder.Logger))
 	return &FlowAccessNodeBuilder{
-		AccessNodeConfig:    DefaultAccessNodeConfig(),
-		FlowNodeBuilder:     nodeBuilder,
-		FollowerDistributor: dist,
-		IndexerDependencies: cmd.NewDependencyList(),
+		AccessNodeConfig:             DefaultAccessNodeConfig(),
+		FlowNodeBuilder:              nodeBuilder,
+		FollowerDistributor:          dist,
+		IndexerDependencies:          cmd.NewDependencyList(),
+		RegisterDBPrunerDependencies: cmd.NewDependencyList(),
 	}
 }
 
@@ -1297,6 +1334,8 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			"execution-data-db",
 			defaultConfig.executionDataDBMode,
 			"[experimental] the DB type for execution datastore. One of [badger, pebble]")
+
+		// Execution data pruner
 		flags.Uint64Var(&builder.executionDataPrunerHeightRangeTarget,
 			"execution-data-height-range-target",
 			defaultConfig.executionDataPrunerHeightRangeTarget,
@@ -1310,6 +1349,20 @@ func (builder *FlowAccessNodeBuilder) extraFlags() {
 			defaultConfig.executionDataPruningInterval,
 			"duration after which the pruner tries to prune execution data. The default value is 10 minutes")
 
+		// RegisterDB pruning
+		flags.BoolVar(&builder.registerDBPruningEnabled,
+			"registerdb-pruning-enabled",
+			defaultConfig.registerDBPruningEnabled,
+			"whether to enable the pruning for register db")
+		flags.DurationVar(&builder.registerDBPruneThrottleDelay,
+			"registerdb-prune-throttle-delay",
+			defaultConfig.registerDBPruneThrottleDelay,
+			"delay for controlling a pause between batches of registers inspected and pruned")
+		flags.DurationVar(&builder.registerDBPruneTickerInterval,
+			"registerdb-prune-ticker-interval",
+			defaultConfig.registerDBPruneTickerInterval,
+			"duration after which the pruner tries to prune data. The default value is 10 minutes")
+
 		// Execution State Streaming API
 		flags.Uint32Var(&builder.stateStreamConf.ExecutionDataCacheSize, "execution-data-cache-size", defaultConfig.stateStreamConf.ExecutionDataCacheSize, "block execution data cache size")
 		flags.Uint32Var(&builder.stateStreamConf.MaxGlobalStreams, "state-stream-global-max-streams", defaultConfig.stateStreamConf.MaxGlobalStreams, "global maximum number of concurrent streams")
@@ -1673,11 +1726,16 @@ func (builder *FlowAccessNodeBuilder) Build() (cmd.Node, error) {
 			builder.RestMetrics = m
 			return nil
 		}).
+		Module("register db metrics", func(node *cmd.NodeConfig) error {
+			builder.RegisterDBPrunerMetrics = metrics.NewRegisterDBPrunerCollector()
+			return nil
+		}).
 		Module("access metrics", func(node *cmd.NodeConfig) error {
 			builder.AccessMetrics = metrics.NewAccessCollector(
 				metrics.WithTransactionMetrics(builder.TransactionMetrics),
 				metrics.WithBackendScriptsMetrics(builder.TransactionMetrics),
 				metrics.WithRestMetrics(builder.RestMetrics),
+				metrics.WithRegisterDBPrunerMetrics(builder.RegisterDBPrunerMetrics),
 			)
 			return nil
 		}).