Replace benchmark DB wrapper

.github/workflows/sei-db-tests.yml

@@ -66,7 +66,7 @@ jobs:
             -run '^$' \
             -bench . \
             -benchtime=1x \
-            ./sei-db/state_db/bench/...
+            ./sei-db/bench/...
 
   coverage:
     name: Coverage

sei-db/bench/cryptosim/config/basic-config.json

@@ -1,17 +1,24 @@
 {
   "Comment": "Basic configuration for the cryptosim benchmark. Intended for basic correctness/sanity testing.",
-  "Backend": "FlatKV",
   "StateStoreConfig": {
-    "Enable": true,
+    "Enable": false,
     "DBDirectory": "",
     "Backend": "pebbledb",
     "AsyncWriteBuffer": 100,
     "KeepRecent": 100000,
     "PruneIntervalSeconds": 600,
     "ImportNumWorkers": 1,
     "KeepLastVersion": true,
-    "UseDefaultComparer": false,
-    "EVMDBDirectory": ""
+    "UseDefaultComparer": false
+  },
+  "CheckpointConfig": {
+    "TimeInterval": 600000000000,
+    "BlockInterval": 0
+  },
+  "PruningConfig": {
+    "RollbackWindow": 1000,
+    "LookbackWindow": 0,
+    "PruneInterval": 300000000000
   },
   "CannedRandomSize": 1073741824,
   "ConstantThreadCount": 0,

sei-db/bench/cryptosim/cryptosim.go

@@ -3,14 +3,17 @@ package cryptosim
 import (
 	"context"
 	"fmt"
+	"path/filepath"
 	"runtime"
 	"time"
 
-	"github.com/sei-protocol/sei-chain/sei-db/bench/wrappers"
+	"golang.org/x/time/rate"
+
 	"github.com/sei-protocol/sei-chain/sei-db/common/keys"
 	crand "github.com/sei-protocol/sei-chain/sei-db/common/rand"
 	"github.com/sei-protocol/sei-chain/sei-db/common/utils"
-	"golang.org/x/time/rate"
+	"github.com/sei-protocol/sei-chain/sei-db/controller"
+	"github.com/sei-protocol/sei-chain/sei-db/state_db/giga"
 )
 
 const (
@@ -133,23 +136,50 @@ func NewCryptoSim(
 	fmt.Printf("Running cryptosim benchmark from data directory: %s\n", config.DataDir)
 	fmt.Printf("Logs are being routed to: %s\n", config.LogDir)
 
-	var dbConfig any
-	switch config.Backend {
-	case wrappers.FlatKV:
-		dbConfig = config.FlatKVConfig
-	case wrappers.SSComposite, wrappers.CompositeDual_SSComposite:
-		dbConfig = config.StateStoreConfig
-	case wrappers.SSHistoricalOffload:
-		dbConfig = config.HistoricalOffload
+	config.FlatKVConfig.DataDir = config.DataDir
+	config.StateStoreConfig.EVMDBDirectory = filepath.Join(
+		config.DataDir, "state_store", "evm", config.StateStoreConfig.Backend)
+
+	// Every store the state DB opens is pruned by the collector started below, so each one stands its
+	// own pruner down. This is the same handover bootstrap.GigaStorageManager performs for a node.
+	config.FlatKVConfig.ExternalPruning = true
+	config.StateStoreConfig.ExternalPruning = true
+
+	// giga.NewStateDB is the node's own entry point, and the only one that leaves the state WAL
+	// outside the live state DB: it opens the WAL itself and writes each block to it ahead of the
+	// commit. A live state DB opened directly would own its WAL and write it inline instead.
+	db, err := giga.NewStateDB(ctx, config.FlatKVConfig, config.StateStoreConfig, config.CheckpointConfig)
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("failed to open the state DB: %w", err)
 	}
 
-	db, err := wrappers.NewDBImpl(ctx, config.Backend, config.DataDir, dbConfig)
+	// Nothing the state DB opens prunes itself on this path: the state WAL, and the historical state
+	// DB when it is enabled, shrink only when a collector tells them to.
+	garbageCollector, err := controller.NewStorageGarbageCollector(
+		ctx, config.PruningConfig, db.PrunableStores())
 	if err != nil {
 		cancel()
-		return nil, fmt.Errorf("failed to create database: %w", err)
+		if closeErr := db.Close(); closeErr != nil {
+			fmt.Printf("failed to close the state DB during error recovery: %v\n", closeErr)
+		}
+		return nil, fmt.Errorf("failed to start the storage garbage collector: %w", err)
 	}
 
-	metrics := NewCryptosimMetrics(ctx, db.GetPhaseTimer(), config)
+	// Every construction failure past this point releases through here. The state DB holds the state
+	// WAL directory's exclusive lock, so a handle left open makes an in-process retry fail to open the
+	// WAL rather than only leaking descriptors.
+	releaseStorage := func() {
+		cancel()
+		if closeErr := garbageCollector.Close(); closeErr != nil {
+			fmt.Printf("failed to close the garbage collector during error recovery: %v\n", closeErr)
+		}
+		if closeErr := db.Close(); closeErr != nil {
+			fmt.Printf("failed to close the state DB during error recovery: %v\n", closeErr)
+		}
+	}
+
+	metrics := NewCryptosimMetrics(ctx, db.SC().GetPhaseTimer(), config)
 	// Server start deferred until after DataGenerator loads DB state and sets gauges,
 	// avoiding rate() spikes when restarting with a preserved DB.
 
@@ -160,24 +190,15 @@ func NewCryptoSim(
 
 	start := time.Now()
 
-	database, err := NewDatabase(config, db, metrics, 0)
+	database, err := NewDatabase(config, db, garbageCollector, metrics)
 	if err != nil {
-		cancel()
-		if closeErr := db.Close(); closeErr != nil {
-			fmt.Printf("failed to close database during error recovery: %v\n", closeErr)
-		}
+		releaseStorage()
 		return nil, fmt.Errorf("failed to create database: %w", err)
 	}
 
-	dataGenerator, err := NewDataGenerator(config, database, rand, metrics)
-	if err != nil {
-		cancel()
-		if closeErr := db.Close(); closeErr != nil {
-			fmt.Printf("failed to close database during error recovery: %v\n", closeErr)
-		}
-		return nil, fmt.Errorf("failed to create data generator: %w", err)
-	}
-	database.nextBlockNumber = dataGenerator.InitialNextBlockNumber()
+	fmt.Printf("Next block number: %s.\n", int64Commas(database.nextBlockNumber))
+
+	dataGenerator := NewDataGenerator(config, database, rand, metrics)
 	threadCount := int(config.ThreadsPerCore)*runtime.NumCPU() + config.ConstantThreadCount
 	if threadCount < 1 {
 		threadCount = 1
@@ -195,7 +216,7 @@ func NewCryptoSim(
 		recieptsChan = make(chan *block, config.RecieptChannelCapacity)
 		_, err := NewRecieptStoreSimulator(ctx, config, recieptsChan, metrics, rand.Clone(false))
 		if err != nil {
-			cancel()
+			releaseStorage()
 			return nil, fmt.Errorf("failed to create receipt store simulator: %w", err)
 		}
 		metrics.startReceiptChannelDepthSampling(recieptsChan, config.BackgroundMetricsScrapeInterval)
@@ -230,6 +251,7 @@ func NewCryptoSim(
 
 	err = c.setup()
 	if err != nil {
+		releaseStorage()
 		return nil, fmt.Errorf("failed to setup benchmark: %w", err)
 	}
 

sei-db/bench/cryptosim/cryptosim_config.go

@@ -7,7 +7,6 @@ import (
 	"path/filepath"
 	"strings"
 
-	"github.com/sei-protocol/sei-chain/sei-db/bench/wrappers"
 	"github.com/sei-protocol/sei-chain/sei-db/config"
 	flatkvConfig "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/config"
 )
@@ -86,7 +85,7 @@ type CryptoSimConfig struct {
 	// How many blocks the benchmark may run ahead of block hashing. Databases hash committed blocks
 	// asynchronously, and the benchmark takes one block's hash per block committed once it is this far
 	// ahead — so a block's hash must arrive no later than this many blocks after it was committed, and
-	// the benchmark waits when it does not. A database that publishes no block hashes waits on nothing.
+	// the benchmark waits when it does not.
 	HashLagBlocks int
 
 	// The directory to store the benchmark data.
@@ -100,16 +99,14 @@ type CryptoSimConfig struct {
 	// in undefined behavior, don't change the size unless you are starting a new run from scratch.
 	CannedRandomSize int
 
-	// The backend to use for the benchmark database.
-	Backend wrappers.DBType
+	// Configures the historical state DB.
+	StateStoreConfig config.StateStoreConfig
 
-	// StateStoreConfig controls SS-backed benchmark backends such as SSComposite.
-	// The default preserves the benchmark SS defaults: pebbledb, async buffer 100.
-	StateStoreConfig *config.StateStoreConfig
+	// Configures the cadence the state DB checkpoints both halves of state on.
+	CheckpointConfig config.CheckpointConfig
 
-	// HistoricalOffload configures the transport used by the
-	// SSHistoricalOffload backend.
-	HistoricalOffload *wrappers.HistoricalOffloadConfig
+	// Configures the prune cycle that enforces retention across the state DB's stores.
+	PruningConfig *config.StorageGarbageCollectorConfig
 
 	// This field is ignored, but allows for a comment to be added to the config file.
 	// Something, something, why in the name of all things holy doesn't json support comments?
@@ -163,7 +160,7 @@ type CryptoSimConfig struct {
 	// If true, the log directory will be deleted on a clean shutdown.
 	DeleteLogDirOnShutdown bool
 
-	// Configures the FlatKV database. Ignored if Backend is not "FlatKV".
+	// Configures the live state DB.
 	FlatKVConfig *flatkvConfig.Config
 
 	// The capacity of the channel that holds blocks awaiting execution.
@@ -243,6 +240,11 @@ func DefaultCryptoSimConfig() *CryptoSimConfig {
 	// Note: if you add new fields or modify default values, be sure to keep config/basic-config.json in sync.
 	// That file should contain every available config set to its default value, as a reference.
 
+	ssConfig := config.DefaultStateStoreConfig()
+	// Nothing in the benchmark reads the historical state DB, so a run pays to write it only when
+	// the config asks for it.
+	ssConfig.Enable = false
+
 	cfg := &CryptoSimConfig{
 		NumberOfHotAccounts:               100,
 		MinimumNumberOfColdAccounts:       1_000_000,
@@ -262,8 +264,9 @@ func DefaultCryptoSimConfig() *CryptoSimConfig {
 		HashLagBlocks:                     32,
 		Seed:                              1337,
 		CannedRandomSize:                  1024 * 1024 * 1024, // 1GB
-		Backend:                           wrappers.FlatKV,
-		StateStoreConfig:                  wrappers.DefaultBenchStateStoreConfig(),
+		StateStoreConfig:                  ssConfig,
+		CheckpointConfig:                  config.DefaultCheckpointConfig(),
+		PruningConfig:                     config.DefaultStorageGarbageCollectorConfig(),
 		ConsoleUpdateIntervalSeconds:      1,
 		ConsoleUpdateIntervalTransactions: 1_000_000,
 		SetupUpdateIntervalCount:          100_000,
@@ -412,19 +415,14 @@ func (c *CryptoSimConfig) Validate() error {
 		return fmt.Errorf("ReceiptLogFilterMaxBlockRange must be >= ReceiptLogFilterMinBlockRange (got %d < %d)",
 			c.ReceiptLogFilterMaxBlockRange, c.ReceiptLogFilterMinBlockRange)
 	}
-	if c.StateStoreConfig == nil {
-		return fmt.Errorf("StateStoreConfig is required")
-	}
 	switch c.StateStoreConfig.Backend {
 	case config.PebbleDBBackend, config.RocksDBBackend:
 	default:
 		return fmt.Errorf("StateStoreConfig.Backend must be one of %q or %q (got %q)",
 			config.PebbleDBBackend, config.RocksDBBackend, c.StateStoreConfig.Backend)
 	}
-	if c.Backend == wrappers.SSHistoricalOffload {
-		if err := c.HistoricalOffload.Validate(); err != nil {
-			return err
-		}
+	if err := c.PruningConfig.Validate(); err != nil {
+		return fmt.Errorf("PruningConfig is invalid: %w", err)
 	}
 	switch strings.ToLower(c.LogLevel) {
 	case "debug", "info", "warn", "error":