State WAL replacement

sei-db/seiwal/metrics.go

@@ -0,0 +1,95 @@
+package seiwal
+
+import (
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+
+	commonmetrics "github.com/sei-protocol/sei-chain/sei-db/common/metrics"
+)
+
+// The name of the OpenTelemetry meter for WAL metrics.
+const walMeterName = "seidb_seiwal"
+
+// Instruments are shared process-wide (created once); individual WAL instances are distinguished by the
+// "wal" attribute attached at each recording (see walNameAttr), mirroring LittDB's per-table labeling. This
+// keeps metrics from multiple instances in one process from clobbering each other.
+var (
+	walMeter = otel.Meter(walMeterName)
+
+	// The number of records appended to the WAL.
+	walRecordsWritten = must(walMeter.Int64Counter(
+		"seiwal_records_written",
+		metric.WithDescription("Number of records appended to the WAL"),
+		metric.WithUnit("{count}"),
+	))
+
+	// The number of record bytes appended to the WAL (including framing).
+	walBytesWritten = must(walMeter.Int64Counter(
+		"seiwal_bytes_written",
+		metric.WithDescription("Number of bytes written to the WAL"),
+		metric.WithUnit("By"),
+	))
+
+	// The number of WAL files sealed (rotated) after reaching the target size.
+	walFilesSealed = must(walMeter.Int64Counter(
+		"seiwal_files_sealed",
+		metric.WithDescription("Number of WAL files sealed on rotation"),
+		metric.WithUnit("{count}"),
+	))
+
+	// The number of sealed WAL files deleted by pruning.
+	walFilesPruned = must(walMeter.Int64Counter(
+		"seiwal_files_pruned",
+		metric.WithDescription("Number of WAL files removed by pruning"),
+		metric.WithUnit("{count}"),
+	))
+
+	// The time spent serializing a payload in the generic serializing WAL.
+	walSerializeDuration = must(walMeter.Float64Histogram(
+		"seiwal_serialize_duration_seconds",
+		metric.WithDescription("Time spent serializing a payload in the generic WAL"),
+		metric.WithUnit("s"),
+		metric.WithExplicitBucketBoundaries(commonmetrics.LatencyBuckets...),
+	))
+
+	// The number of payload bytes produced by serialization in the generic serializing WAL.
+	walSerializedBytes = must(walMeter.Int64Counter(
+		"seiwal_serialized_bytes",
+		metric.WithDescription("Number of payload bytes produced by serialization in the generic WAL"),
+		metric.WithUnit("By"),
+	))
+
+	// The number of serialization failures in the generic serializing WAL.
+	walSerializeErrors = must(walMeter.Int64Counter(
+		"seiwal_serialize_errors",
+		metric.WithDescription("Number of serialization failures in the generic WAL"),
+		metric.WithUnit("{count}"),
+	))
+
+	// The buffered depth of a WAL's internal channel, sampled periodically.
+	walQueueDepth = must(walMeter.Int64Gauge(
+		"seiwal_queue_depth",
+		metric.WithDescription("Buffered depth of a WAL internal channel, sampled periodically"),
+		metric.WithUnit("{count}"),
+	))
+)
+
+// walNameAttr returns the measurement option that tags an observation with a WAL instance's name, so metrics
+// from distinct instances in the same process remain distinguishable.
+func walNameAttr(name string) metric.MeasurementOption {
+	return metric.WithAttributeSet(attribute.NewSet(attribute.String("wal", name)))
+}
+
+// queueDepthAttrs tags a queue-depth observation with the WAL instance name and which internal channel
+// ("writer" or "serializer") is being measured.
+func queueDepthAttrs(name string, queue string) metric.MeasurementOption {
+	return metric.WithAttributeSet(attribute.NewSet(attribute.String("wal", name), attribute.String("queue", queue)))
+}
+
+func must[V any](v V, err error) V {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}

sei-db/seiwal/seiwal.go

@@ -0,0 +1,92 @@
+package seiwal
+
+// WAL is a generic, index-keyed, append-only write-ahead log over payloads of type T.
+//
+// Each record is tagged with a caller-provided monotonic index. The index is what makes garbage
+// collection ("drop everything below N"), iteration ("start at N"), and rollback ("drop everything
+// above N") expressible without the WAL ever interpreting a payload.
+//
+// A WAL instance is not safe for concurrent use: its methods must not be called from multiple
+// goroutines simultaneously. Callers that share a WAL across goroutines must serialize access
+// themselves.
+//
+// Slices are not copied at the call boundary. Any slice passed into a WAL method — the payload and every
+// slice reachable through it — must not be modified after the call: the WAL may retain it and read it
+// asynchronously, so mutating it races the WAL and can corrupt what is persisted. Likewise every slice
+// returned from a WAL or its iterator is owned by the WAL and must be treated as read-only. Callers that
+// need to mutate such data must copy it first.
+type WAL[T any] interface {
+
+	// Append a record with the given index and payload.
+	//
+	// The index must be strictly greater than the index of the most recently appended record (indices
+	// need not be contiguous, but they must strictly increase).
+	//
+	// This method only schedules the append; it does not block until the record is durable. Durability is
+	// achieved by a subsequent Flush.
+	//
+	// data, and every slice reachable through it, must not be modified after this call: the payload may be
+	// retained and serialized asynchronously, so mutating it races the WAL and can corrupt what is
+	// persisted. Callers that need to reuse or mutate the buffer must copy it first.
+	Append(index uint64, data T) error
+
+	// Flush blocks until all previously scheduled appends are durable.
+	Flush() error
+
+	// Bounds reports the range of record indices currently stored in the WAL.
+	Bounds() (
+		// If true, there is at least one record in the WAL and first/last are valid. If false, the WAL is
+		// empty and first/last are undefined.
+		ok bool,
+		// The lowest stored record index, inclusive. Only valid if ok is true.
+		first uint64,
+		// The highest stored record index, inclusive. Only valid if ok is true.
+		last uint64,
+		// Any error encountered while retrieving the range.
+		err error,
+	)
+
+	// Prune removes all records with an index less than lowestIndexToKeep.
+	//
+	// This method merely schedules the prune; it does not block until the prune is complete. Pruning is
+	// async and lazy, and implementations are free to delay it arbitrarily long. Pruning removes whole
+	// sealed files only, so records may survive above the requested threshold until their containing file
+	// is fully below it.
+	Prune(lowestIndexToKeep uint64) error
+
+	// Iterator returns an iterator over the WAL starting at the given index.
+	//
+	// The iterator reads a consistent, point-in-time snapshot of the WAL taken at some instant between the
+	// start and the return of this call. Records appended before that instant are included; records
+	// appended after it are not. For records appended concurrently with this call, whether they are
+	// included is unspecified.
+	Iterator(startIndex uint64) (Iterator[T], error)
+
+	// Close flushes pending appends, seals the current file, and releases resources.
+	Close() error
+}
+
+// Iterator iterates over the records of a WAL in ascending index order.
+//
+// An Iterator is single-consumer and not safe for concurrent use: all of its methods, including Close, must
+// be called from a single goroutine (or with external serialization). In particular, Close must not be
+// called concurrently with Next from another goroutine.
+//
+// Every payload returned by an Iterator — and every slice reachable through it — is owned by the WAL and
+// must be treated as read-only. Callers that need to retain or mutate the data must copy it first.
+type Iterator[T any] interface {
+	// Next advances the iterator to the next record. It returns false when iteration is complete (no more
+	// records), and returns an error if advancing failed. After Next returns (false, nil), iteration is
+	// complete; after it returns an error, the iterator must not be used further (other than Close).
+	Next() (bool, error)
+
+	// Entry returns the index and payload of the record at the iterator's current position. It is only
+	// valid to call Entry after Next has returned (true, nil).
+	//
+	// The returned payload must be treated as read-only and must not be modified. Callers that need to
+	// retain or mutate the data must copy it first.
+	Entry() (index uint64, data T)
+
+	// Close releases the resources held by the iterator.
+	Close() error
+}

sei-db/seiwal/seiwal_config.go

@@ -0,0 +1,81 @@
+package seiwal
+
+import (
+	"fmt"
+	"regexp"
+	"time"
+
+	"github.com/sei-protocol/sei-chain/sei-db/common/unit"
+)
+
+// The permitted shape of a WAL instance name: it becomes a metric attribute value, so it is restricted to
+// characters safe for label values.
+var nameRegex = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
+
+// Config configures a WAL.
+type Config struct {
+	// The directory where the WAL writes its files.
+	Path string
+
+	// A short identifier for this WAL instance, used to distinguish its metrics from those of other
+	// instances in the same process. Required; must match [a-zA-Z0-9_-]+.
+	Name string
+
+	// The size of the channel used to send framed records and control messages to the writer goroutine.
+	WriteBufferSize uint
+
+	// The depth of the serialization request queue. Used only by the generic serializing WAL
+	// (NewGenericWAL); the byte-oriented engine ignores it.
+	SerializerBufferSize uint
+
+	// The size a WAL file may reach before it is sealed and a fresh one is opened. Rotation happens after a
+	// record is appended, so a file may exceed this by the size of a single record — and because a record
+	// is written atomically to a single file, a record larger than this threshold produces a file that
+	// overshoots it by that record's size. Must be greater than 0.
+	TargetFileSize uint
+
+	// When true, Flush calls fsync on the underlying file so that flushed data survives a power loss, not
+	// merely a process crash. When false, Flush only flushes the in-process buffer to the OS.
+	FsyncOnFlush bool
+
+	// The number of records an iterator's reader thread may prefetch ahead of the consumer. A larger value
+	// keeps the reader busy while the consumer processes records, which matters for startup replay speed.
+	// Must be greater than 0.
+	IteratorPrefetchSize uint
+
+	// The interval at which the WAL samples the buffered depth of its internal channel into the
+	// seiwal_queue_depth gauge. Zero or negative disables sampling.
+	MetricsSampleInterval time.Duration
+}
+
+// DefaultConfig returns a default WAL configuration for the WAL at path, identified by name.
+func DefaultConfig(path string, name string) *Config {
+	return &Config{
+		Path:                  path,
+		Name:                  name,
+		WriteBufferSize:       16,
+		SerializerBufferSize:  16,
+		TargetFileSize:        64 * unit.MB,
+		FsyncOnFlush:          true,
+		IteratorPrefetchSize:  32,
+		MetricsSampleInterval: 15 * time.Second,
+	}
+}
+
+// Validate the configuration, returning nil if valid, or an error describing the problem if invalid.
+func (c *Config) Validate() error {
+	if c.Path == "" {
+		return fmt.Errorf("path is required")
+	}
+	if !nameRegex.MatchString(c.Name) {
+		return fmt.Errorf("name %q is required and must match %s", c.Name, nameRegex.String())
+	}
+	if c.TargetFileSize == 0 {
+		// A zero target would seal and rotate a fresh file after every single record.
+		return fmt.Errorf("target file size must be greater than 0")
+	}
+	if c.IteratorPrefetchSize == 0 {
+		return fmt.Errorf("iterator prefetch size must be greater than 0")
+	}
+	return nil
+}

sei-db/seiwal/seiwal_config_test.go

@@ -0,0 +1,40 @@
+package seiwal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigValidate(t *testing.T) {
+	t.Run("default config is valid", func(t *testing.T) {
+		require.NoError(t, DefaultConfig("/tmp/wal", "test").Validate())
+	})
+
+	t.Run("empty path is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("", "test")
+		require.Error(t, cfg.Validate())
+	})
+
+	t.Run("empty name is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("/tmp/wal", "")
+		require.Error(t, cfg.Validate())
+	})
+
+	t.Run("malformed name is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("/tmp/wal", "bad name!")
+		require.Error(t, cfg.Validate())
+	})
+
+	t.Run("zero target file size is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("/tmp/wal", "test")
+		cfg.TargetFileSize = 0
+		require.Error(t, cfg.Validate())
+	})
+
+	t.Run("zero iterator prefetch size is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("/tmp/wal", "test")
+		cfg.IteratorPrefetchSize = 0
+		require.Error(t, cfg.Validate())
+	})
+}