WAL Benchmark

.gitignore

@@ -67,4 +67,5 @@ sei-db/state_db/bench/cryptosim/data/**
 sei-db/state_db/bench/cryptosim/bin/
 sei-db/state_db/bench/cryptosim/logs/
 sei-db/ledger_db/block/blocksim/bin/
+sei-db/seiwal/walsim/bin/
 sei-db/db_engine/litt/bin/

sei-db/seiwal/walsim/Makefile

@@ -0,0 +1,14 @@
+BINARY := bin/walsim
+LOGGER_BINARY := bin/configure-logger
+MODULE_ROOT ?= $(shell git -C "$(CURDIR)" rev-parse --show-toplevel)
+CMD_DIR := ./sei-db/seiwal/walsim/cmd
+
+.PHONY: build
+build:
+	@mkdir -p "$(CURDIR)/bin"
+	cd "$(MODULE_ROOT)" && go build -o "$(CURDIR)/$(BINARY)" "$(CMD_DIR)/walsim"
+	cd "$(MODULE_ROOT)" && go build -o "$(CURDIR)/$(LOGGER_BINARY)" "$(CMD_DIR)/configure-logger"
+
+.PHONY: clean
+clean:
+	rm -f "$(CURDIR)/$(BINARY)" "$(CURDIR)/$(LOGGER_BINARY)"

sei-db/seiwal/walsim/cmd/configure-logger/main.go

@@ -0,0 +1,52 @@
+// configure-logger reads a walsim config file and prints shell export
+// statements that configure seilog's environment variables. Intended to be
+// called via eval in a shell script before launching the benchmark binary.
+//
+// Usage:
+//
+//	eval "$(configure-logger config.json)"
+package main
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/sei-protocol/sei-chain/sei-db/common/utils"
+	"github.com/sei-protocol/sei-chain/sei-db/seiwal/walsim"
+)
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintf(os.Stderr, "configure-logger: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	if len(os.Args) != 2 {
+		return fmt.Errorf("usage: configure-logger <config-file>")
+	}
+
+	cfg := walsim.DefaultWalsimConfig()
+	if err := utils.LoadConfigFromFile(os.Args[1], cfg); err != nil {
+		return fmt.Errorf("load config: %w", err)
+	}
+
+	logDir, err := utils.ResolveAndCreateDir(cfg.LogDir)
+	if err != nil {
+		return fmt.Errorf("resolve log dir: %w", err)
+	}
+
+	logFile := filepath.Join(logDir, "walsim.log")
+
+	fmt.Printf("export SEI_LOG_OUTPUT=%s\n", shellQuote(logFile))
+	fmt.Printf("export SEI_LOG_LEVEL=%s\n", shellQuote(strings.ToLower(cfg.LogLevel)))
+
+	return nil
+}
+
+func shellQuote(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
+}

sei-db/seiwal/walsim/cmd/walsim/main.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+
+	"github.com/sei-protocol/sei-chain/sei-db/common/metrics"
+	"github.com/sei-protocol/sei-chain/sei-db/common/utils"
+	"github.com/sei-protocol/sei-chain/sei-db/seiwal/walsim"
+)
+
+func main() {
+	err := run()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	if len(os.Args) != 2 {
+		fmt.Fprintf(os.Stderr, "Usage: %s <config-file>\n", os.Args[0])
+		os.Exit(1)
+	}
+	config := walsim.DefaultWalsimConfig()
+	if err := utils.LoadConfigFromFile(os.Args[1], config); err != nil {
+		return err
+	}
+
+	configString, err := utils.StringifyConfig(config)
+	if err != nil {
+		return fmt.Errorf("failed to stringify config: %w", err)
+	}
+	fmt.Printf("%s\n", configString)
+
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer stop()
+
+	reg, shutdown, err := metrics.SetupOtelPrometheus()
+	if err != nil {
+		return fmt.Errorf("setup metrics: %w", err)
+	}
+	defer func() {
+		_ = shutdown(context.Background())
+	}()
+
+	wsMetrics := walsim.NewWalsimMetrics(ctx, config)
+
+	ws, err := walsim.NewWalSim(ctx, config, wsMetrics)
+	if err != nil {
+		return fmt.Errorf("failed to create walsim: %w", err)
+	}
+	defer func() {
+		if err := ws.Close(); err != nil {
+			fmt.Fprintf(os.Stderr, "Error closing walsim: %v\n", err)
+		}
+	}()
+
+	metrics.StartMetricsServer(ctx, reg, config.MetricsAddr)
+	metrics.StartSystemMetrics(ctx, "walsim", config.BackgroundMetricsScrapeInterval,
+		[]metrics.MonitoredDir{
+			{Name: "data_dir", Path: config.DataDir, TrackAvailableSpace: true},
+			{Name: "log_dir", Path: config.LogDir},
+		})
+
+	if config.EnableSuspension {
+		go func() {
+			scanner := bufio.NewScanner(os.Stdin)
+			suspended := false
+			for scanner.Scan() {
+				if suspended {
+					ws.Resume()
+					suspended = false
+				} else {
+					ws.Suspend()
+					suspended = true
+				}
+			}
+		}()
+	}
+
+	ws.BlockUntilHalted()
+
+	return nil
+}

sei-db/seiwal/walsim/config/debug.json

@@ -0,0 +1,9 @@
+{
+  "Comment": "Configuration for local smoke testing and debugging.",
+  "DataDir": "~/walsim/data",
+  "LogDir": "~/walsim/logs",
+  "CleanDataOnStart": true,
+  "CleanLogsOnStart": true,
+  "CleanDataOnExit": true,
+  "CleanLogsOnExit": true
+}

sei-db/seiwal/walsim/config/standard.json

@@ -0,0 +1,5 @@
+{
+  "Comment": "Standard walsim configuration.",
+  "DataDir": "~/walsim/data",
+  "LogDir": "~/walsim/logs"
+}

sei-db/seiwal/walsim/record_generator.go

@@ -0,0 +1,50 @@
+package walsim
+
+import (
+	"context"
+
+	crand "github.com/sei-protocol/sei-chain/sei-db/common/rand"
+)
+
+// RecordGenerator asynchronously produces fixed-size opaque records and feeds them into a channel.
+// Each record is a zero-copy sub-slice of a pre-generated, immutable CannedRandom buffer, so the
+// generator never runs math/rand or allocates payload bytes on the hot path. The generator stops
+// when the context is cancelled.
+//
+// The CannedRandom buffer is never mutated, so the sub-slices remain valid indefinitely; this makes
+// it safe for the WAL to retain a record slice and serialize it asynchronously.
+type RecordGenerator struct {
+	ctx        context.Context
+	rand       *crand.CannedRandom
+	recordSize int
+	recordChan chan []byte
+}
+
+// NewRecordGenerator creates a RecordGenerator and immediately starts its background goroutine. rand
+// must not be shared with any other goroutine (this generator owns it).
+func NewRecordGenerator(
+	ctx context.Context,
+	rng *crand.CannedRandom,
+	recordSize int,
+	queueSize int,
+) *RecordGenerator {
+	g := &RecordGenerator{
+		ctx:        ctx,
+		rand:       rng,
+		recordSize: recordSize,
+		recordChan: make(chan []byte, queueSize),
+	}
+	go g.mainLoop()
+	return g
+}
+
+func (g *RecordGenerator) mainLoop() {
+	for {
+		record := g.rand.Bytes(g.recordSize)
+		select {
+		case <-g.ctx.Done():
+			return
+		case g.recordChan <- record:
+		}
+	}
+}

sei-db/seiwal/walsim/wal_store.go

@@ -0,0 +1,147 @@
+package walsim
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/sei-protocol/sei-chain/sei-db/seiwal"
+	"github.com/sei-protocol/sei-chain/sei-db/wal"
+)
+
+// walStore is the minimal WAL surface the benchmark drives. The new seiwal WAL (seiwal.WAL[[]byte])
+// satisfies it directly with no wrapper; the legacy sei-db/wal WAL is adapted by legacyWALShim.
+type walStore interface {
+	// Append schedules a record with the given index and payload.
+	Append(index uint64, data []byte) error
+
+	// Flush blocks until previously scheduled appends are durable.
+	Flush() error
+
+	// Bounds reports whether any record is stored and, if so, the lowest and highest stored indices.
+	Bounds() (ok bool, first uint64, last uint64, err error)
+
+	// PruneBefore requests removal of all records with an index below lowestIndexToKeep.
+	PruneBefore(lowestIndexToKeep uint64) error
+
+	// Close flushes pending appends and releases resources.
+	Close() error
+}
+
+// The seiwal WAL implements walStore directly.
+var _ walStore = (seiwal.WAL[[]byte])(nil)
+
+// The legacy shim implements walStore.
+var _ walStore = (*legacyWALShim)(nil)
+
+// openWALStore opens the WAL backend selected by the configuration.
+func openWALStore(ctx context.Context, config *WalsimConfig) (walStore, error) {
+	switch config.Backend {
+	case "seiwal":
+		cfg := config.Seiwal
+		// walsim owns the storage path and metric name.
+		cfg.Path = config.DataDir
+		cfg.Name = "walsim"
+		w, err := seiwal.NewWAL(&cfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to open seiwal WAL: %w", err)
+		}
+		return w, nil
+	case "legacy":
+		return newLegacyWALShim(ctx, config)
+	default:
+		return nil, fmt.Errorf("unknown WAL backend: %q", config.Backend)
+	}
+}
+
+// legacyWALShim adapts the legacy sei-db/wal WAL to the walStore interface so the benchmark can
+// drive it exactly like the new seiwal WAL. It is throwaway code: the legacy WAL is scheduled for
+// deletion, and this shim goes with it.
+//
+// The legacy WAL's TruncateBefore rewrites segment files and is O(segments), so forwarding it on
+// every request (as the size-target prune loop wants) would be catastrophic. The shim therefore
+// coalesces prune requests, forwarding only every pruneRelaxationFactor-th PruneBefore to the
+// underlying TruncateBefore.
+type legacyWALShim struct {
+	inner *wal.WAL[[]byte]
+
+	// Forward a TruncateBefore only once per this many PruneBefore calls.
+	pruneRelaxationFactor uint64
+
+	// The number of PruneBefore calls received so far.
+	pruneRequestCount uint64
+}
+
+func newLegacyWALShim(ctx context.Context, config *WalsimConfig) (*legacyWALShim, error) {
+	identity := func(b []byte) ([]byte, error) { return b, nil }
+
+	cfg := config.Legacy
+	// Force the legacy WAL's own background pruning off so pruning is driven solely through
+	// PruneBefore (and coalesced below).
+	cfg.KeepRecent = 0
+	cfg.PruneInterval = 0
+
+	inner, err := wal.NewWAL[[]byte](ctx, identity, identity, config.DataDir, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open legacy WAL: %w", err)
+	}
+
+	return &legacyWALShim{
+		inner:                 inner,
+		pruneRelaxationFactor: config.PruneRelaxationFactor,
+	}, nil
+}
+
+// Append writes data to the legacy WAL. The legacy WAL assigns its own 1-based, contiguous index,
+// so the caller-supplied index is ignored; walsim seeds its counter from Bounds, keeping the two
+// aligned.
+func (s *legacyWALShim) Append(_ uint64, data []byte) error {
+	if err := s.inner.Write(data); err != nil {
+		return fmt.Errorf("failed to write to legacy WAL: %w", err)
+	}
+	return nil
+}
+
+// Flush is a no-op. The legacy WAL exposes no explicit sync, so durability is governed entirely by
+// its FsyncEnabled and batching configuration.
+func (s *legacyWALShim) Flush() error {
+	return nil
+}
+
+// Bounds reports the stored index range. The legacy WAL reports 0/0 for an empty log; real indices
+// are 1-based, so a last offset of 0 means empty.
+func (s *legacyWALShim) Bounds() (bool, uint64, uint64, error) {
+	first, err := s.inner.FirstOffset()
+	if err != nil {
+		return false, 0, 0, fmt.Errorf("failed to read first offset: %w", err)
+	}
+	last, err := s.inner.LastOffset()
+	if err != nil {
+		return false, 0, 0, fmt.Errorf("failed to read last offset: %w", err)
+	}
+	if last == 0 {
+		return false, 0, 0, nil
+	}
+	return true, first, last, nil
+}
+
+// PruneBefore coalesces prune requests: only every pruneRelaxationFactor-th call reaches the
+// underlying TruncateBefore, which is expensive in the legacy WAL. The most recent
+// lowestIndexToKeep is used when the forward fires.
+func (s *legacyWALShim) PruneBefore(lowestIndexToKeep uint64) error {
+	s.pruneRequestCount++
+	if s.pruneRequestCount%s.pruneRelaxationFactor != 0 {
+		return nil
+	}
+	if err := s.inner.TruncateBefore(lowestIndexToKeep); err != nil {
+		return fmt.Errorf("failed to truncate legacy WAL: %w", err)
+	}
+	return nil
+}
+
+// Close shuts down the legacy WAL.
+func (s *legacyWALShim) Close() error {
+	if err := s.inner.Close(); err != nil {
+		return fmt.Errorf("failed to close legacy WAL: %w", err)
+	}
+	return nil
+}