feat(dot/state): Implement online pruning of state tries.

cmd/gossamer/config.go

@@ -443,6 +443,9 @@ func setDotGlobalConfigFromToml(tomlCfg *ctoml.Config, cfg *dot.GlobalConfig) {
 		}
 
 		cfg.MetricsPort = tomlCfg.Global.MetricsPort
+
+		cfg.RetainBlocks = tomlCfg.Global.RetainBlocks
+		cfg.GCMode = tomlCfg.Global.GCMode
 	}
 }
 
@@ -472,6 +475,8 @@ func setDotGlobalConfigFromFlags(ctx *cli.Context, cfg *dot.GlobalConfig) {
 		cfg.MetricsPort = uint32(metricsPort)
 	}
 
+	cfg.RetainBlocks = ctx.GlobalInt64(RetainBlockNumberFlag.Name)
+	cfg.GCMode = ctx.GlobalString(GCModeFlag.Name)
 	cfg.NoTelemetry = ctx.Bool("no-telemetry")
 }
 

cmd/gossamer/export.go

@@ -79,11 +79,13 @@ func dotConfigToToml(dcfg *dot.Config) *ctoml.Config {
 	cfg := &ctoml.Config{}
 
 	cfg.Global = ctoml.GlobalConfig{
-		Name:        dcfg.Global.Name,
-		ID:          dcfg.Global.ID,
-		BasePath:    dcfg.Global.BasePath,
-		LogLvl:      dcfg.Global.LogLvl.String(),
-		MetricsPort: dcfg.Global.MetricsPort,
+		Name:         dcfg.Global.Name,
+		ID:           dcfg.Global.ID,
+		BasePath:     dcfg.Global.BasePath,
+		LogLvl:       dcfg.Global.LogLvl.String(),
+		MetricsPort:  dcfg.Global.MetricsPort,
+		RetainBlocks: dcfg.Global.RetainBlocks,
+		GCMode:       dcfg.Global.GCMode,
 	}
 
 	cfg.Log = ctoml.LogConfig{

cmd/gossamer/flags.go

@@ -285,6 +285,13 @@ var (
 		Usage: "Retain number of block from latest block while pruning",
 		Value: 256,
 	}
+
+	// GCModeFlag sets the blockchain GC mode. It's either full or archive.
+	GCModeFlag = cli.StringFlag{
+		Name:  "gcmode",
+		Usage: `Blockchain garbage collection mode ("full", "archive")`,
+		Value: "full",
+	}
 )
 
 // flag sets that are shared by multiple commands
@@ -302,6 +309,7 @@ var (
 		DBPathFlag,
 		BloomFilterSizeFlag,
 		RetainBlockNumberFlag,
+		GCModeFlag,
 	}
 
 	// StartupFlags are flags that are valid for use with the root command and the export subcommand

cmd/gossamer/main.go

@@ -234,6 +234,14 @@ func gossamerAction(ctx *cli.Context) error {
 		return err
 	}
 
+	if cfg.Global.RetainBlocks < 256 {
+		return fmt.Errorf("--%s cannot be less than 256", RetainBlockNumberFlag.Name)
+	}
+
+	if cfg.Global.GCMode != "full" && cfg.Global.GCMode != "archive" {
+		return fmt.Errorf("--%s must be either 'full' or 'archive'", GCModeFlag.Name)
+	}
+
 	cfg.Global.LogLvl = lvl
 
 	// expand data directory and update node configuration (performed separately
@@ -448,7 +456,7 @@ func pruneState(ctx *cli.Context) error {
 		return err
 	}
 
-	logger.Info("Pruner initialised")
+	logger.Info("Offline pruner initialised")
 
 	err = pruner.SetBloomFilter()
 	if err != nil {

dot/config.go

@@ -52,6 +52,8 @@ type GlobalConfig struct {
 	PublishMetrics bool
 	MetricsPort    uint32
 	NoTelemetry    bool
+	RetainBlocks   int64
+	GCMode         string
 }
 
 // LogConfig represents the log levels for individual packages

dot/config/toml/config.go

@@ -29,11 +29,13 @@ type Config struct {
 
 // GlobalConfig is to marshal/unmarshal toml global config vars
 type GlobalConfig struct {
-	Name        string `toml:"name,omitempty"`
-	ID          string `toml:"id,omitempty"`
-	BasePath    string `toml:"basepath,omitempty"`
-	LogLvl      string `toml:"log,omitempty"`
-	MetricsPort uint32 `toml:"metrics-port,omitempty"`
+	Name         string `toml:"name,omitempty"`
+	ID           string `toml:"id,omitempty"`
+	BasePath     string `toml:"basepath,omitempty"`
+	LogLvl       string `toml:"log,omitempty"`
+	MetricsPort  uint32 `toml:"metrics-port,omitempty"`
+	RetainBlocks int64  `toml:"retain-blocks,omitempty"`
+	GCMode       string `toml:"gc-mode,omitempty"`
 }
 
 // LogConfig represents the log levels for individual packages

dot/node.go

@@ -237,6 +237,16 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 	// Network Service
 	var networkSrvc *network.Service
 
+	var pruner state.Pruner
+	if cfg.Global.GCMode == "full" {
+		pruner, err = state.CreatePruner(stateSrvc.DB(), cfg.Global.RetainBlocks)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		pruner = &state.ArchivalNodePruner{}
+	}
+
 	// check if network service is enabled
 	if enabled := networkServiceEnabled(cfg); enabled {
 		// create network service and append network service to node services
@@ -262,7 +272,7 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 	}
 
 	// create BABE service
-	bp, err := createBABEService(cfg, rt, stateSrvc, ks.Babe)
+	bp, err := createBABEService(cfg, rt, stateSrvc, ks.Babe, pruner)
 	if err != nil {
 		return nil, err
 	}
@@ -283,7 +293,7 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 	nodeSrvcs = append(nodeSrvcs, fg)
 
 	// Syncer
-	syncer, err := createSyncService(cfg, stateSrvc, bp, fg, dh, ver, rt)
+	syncer, err := createSyncService(cfg, stateSrvc, bp, fg, dh, ver, rt, pruner)
 	if err != nil {
 		return nil, err
 	}

dot/services.go

@@ -165,7 +165,7 @@ func createRuntime(cfg *Config, st *state.Service, ks *keystore.GlobalKeystore,
 	return rt, nil
 }
 
-func createBABEService(cfg *Config, rt runtime.Instance, st *state.Service, ks keystore.Keystore) (*babe.Service, error) {
+func createBABEService(cfg *Config, rt runtime.Instance, st *state.Service, ks keystore.Keystore, pruner state.Pruner) (*babe.Service, error) {
 	logger.Info(
 		"creating BABE service...",
 		"authority", cfg.Core.BabeAuthority,
@@ -192,6 +192,7 @@ func createBABEService(cfg *Config, rt runtime.Instance, st *state.Service, ks k
 		SlotDuration:     cfg.Core.SlotDuration, // TODO: remove this, should only be modified via runtime constant
 		Authority:        cfg.Core.BabeAuthority,
 		IsDev:            cfg.Global.ID == "dev",
+		Pruner:           pruner,
 	}
 
 	if cfg.Core.BabeAuthority {
@@ -376,7 +377,7 @@ func createBlockVerifier(st *state.Service) (*babe.VerificationManager, error) {
 	return ver, nil
 }
 
-func createSyncService(cfg *Config, st *state.Service, bp sync.BlockProducer, fg sync.FinalityGadget, dh *core.DigestHandler, verifier *babe.VerificationManager, rt runtime.Instance) (*sync.Service, error) {
+func createSyncService(cfg *Config, st *state.Service, bp sync.BlockProducer, fg sync.FinalityGadget, dh *core.DigestHandler, verifier *babe.VerificationManager, rt runtime.Instance, pruner state.Pruner) (*sync.Service, error) {
 	syncCfg := &sync.Config{
 		LogLvl:           cfg.Log.SyncLvl,
 		BlockState:       st.Block,
@@ -387,6 +388,7 @@ func createSyncService(cfg *Config, st *state.Service, bp sync.BlockProducer, fg
 		Verifier:         verifier,
 		Runtime:          rt,
 		DigestHandler:    dh,
+		Pruner:           pruner,
 	}
 
 	return sync.NewService(syncCfg)

dot/services_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/ChainSafe/gossamer/dot/network"
+	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/keystore"
 	"github.com/ChainSafe/gossamer/lib/utils"
@@ -136,7 +137,7 @@ func TestCreateSyncService(t *testing.T) {
 	ver, err := createBlockVerifier(stateSrvc)
 	require.NoError(t, err)
 
-	_, err = createSyncService(cfg, stateSrvc, nil, nil, nil, ver, rt)
+	_, err = createSyncService(cfg, stateSrvc, nil, nil, nil, ver, rt, &state.ArchivalNodePruner{})
 	require.NoError(t, err)
 }
 
@@ -233,7 +234,7 @@ func TestCreateBABEService(t *testing.T) {
 	rt, err := createRuntime(cfg, stateSrvc, ks, &network.Service{})
 	require.NoError(t, err)
 
-	bs, err := createBABEService(cfg, rt, stateSrvc, ks.Babe)
+	bs, err := createBABEService(cfg, rt, stateSrvc, ks.Babe, &state.ArchivalNodePruner{})
 	require.NoError(t, err)
 	require.NotNil(t, bs)
 }

dot/state/prune.go → dot/state/offline_pruner.go

@@ -14,11 +14,11 @@ import (
 	"github.com/dgraph-io/badger/v2/pb"
 )
 
-// Pruner is an offline tool to prune the stale state with the help of
-// bloom filter, The workflow of pruner is very simple:
+// OfflinePruner is a tool to prune the stale state with the help of
+// bloom filter, The workflow of Pruner is very simple:
 // - iterate the storage state, reconstruct the relevant state tries
 // - iterate the database, stream all the targeted keys to new DB
-type Pruner struct {
+type OfflinePruner struct {
 	inputDB        *chaindb.BadgerDB
 	storageState   *StorageState
 	blockState     *BlockState
@@ -30,8 +30,8 @@ type Pruner struct {
 	prunedDBPath string
 }
 
-// NewPruner creates an instance of Pruner.
-func NewPruner(inputDBPath, prunedDBPath string, bloomSize uint64, retainBlockNum int64) (*Pruner, error) {
+// NewPruner creates an instance of OfflinePruner.
+func NewPruner(inputDBPath, prunedDBPath string, bloomSize uint64, retainBlockNum int64) (*OfflinePruner, error) {
 	db, err := utils.LoadChainDB(inputDBPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to load DB %w", err)
@@ -67,7 +67,7 @@ func NewPruner(inputDBPath, prunedDBPath string, bloomSize uint64, retainBlockNu
 		return nil, fmt.Errorf("failed to create new storage state %w", err)
 	}
 
-	return &Pruner{
+	return &OfflinePruner{
 		inputDB:        db,
 		storageState:   storageState,
 		blockState:     blockState,
@@ -80,7 +80,7 @@ func NewPruner(inputDBPath, prunedDBPath string, bloomSize uint64, retainBlockNu
 }
 
 // SetBloomFilter loads keys with storage prefix of last `retainBlockNum` blocks into the bloom filter
-func (p *Pruner) SetBloomFilter() error {
+func (p *OfflinePruner) SetBloomFilter() error {
 	defer p.inputDB.Close() // nolint: errcheck
 	finalisedHash, err := p.blockState.GetFinalizedHash(0, 0)
 	if err != nil {
@@ -134,7 +134,7 @@ func (p *Pruner) SetBloomFilter() error {
 }
 
 // Prune starts streaming the data from input db to the pruned db.
-func (p *Pruner) Prune() error {
+func (p *OfflinePruner) Prune() error {
 	inputDB, err := utils.LoadBadgerDB(p.inputDBPath)
 	if err != nil {
 		return fmt.Errorf("failed to load DB %w", err)