State WAL replacement

sei-db/state_db/statewal/metrics.go

@@ -0,0 +1,48 @@
+package statewal
+
+import (
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric"
+)
+
+// The name of the OpenTelemetry meter for state WAL metrics.
+const walMeterName = "seidb_statewal"
+
+var (
+	walMeter = otel.Meter(walMeterName)
+
+	// The number of blocks (end-of-block markers) written to the WAL.
+	walBlocksWritten = must(walMeter.Int64Counter(
+		"state_wal_blocks_written",
+		metric.WithDescription("Number of blocks written to the state WAL"),
+		metric.WithUnit("{count}"),
+	))
+
+	// The number of record bytes appended to the WAL (including framing).
+	walBytesWritten = must(walMeter.Int64Counter(
+		"state_wal_bytes_written",
+		metric.WithDescription("Number of bytes written to the state WAL"),
+		metric.WithUnit("By"),
+	))
+
+	// The number of WAL files sealed (rotated) after reaching the target size.
+	walFilesSealed = must(walMeter.Int64Counter(
+		"state_wal_files_sealed",
+		metric.WithDescription("Number of state WAL files sealed on rotation"),
+		metric.WithUnit("{count}"),
+	))
+
+	// The number of sealed WAL files deleted by pruning.
+	walFilesPruned = must(walMeter.Int64Counter(
+		"state_wal_files_pruned",
+		metric.WithDescription("Number of state WAL files removed by pruning"),
+		metric.WithUnit("{count}"),
+	))
+)
+
+func must[V any](v V, err error) V {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}

sei-db/state_db/statewal/state_wal.go

@@ -0,0 +1,91 @@
+package statewal
+
+import "github.com/sei-protocol/sei-chain/sei-db/proto"
+
+// A WAL for state.
+type StateWAL interface {
+
+	// Write a set of changes to the WAL.
+	//
+	// This method only schedules the write, it does not block until the write is complete.
+	//
+	// cs, and every byte slice reachable through it (changeset keys and values), must not be modified after
+	// this call. Callers that need to modify those buffers must copy them first.
+	//
+	// The StateWAL rejects writes for blocks if provided out of order. To avoid errors, observe
+	// the following rules:
+	//
+	// - The block numbers passed to Write() may never decrease.
+	// - If data has been written for block N, you cannot write data for block N+1 until you have called
+	//   SignalEndOfBlock().
+	Write(
+		// The block number associated with the changeset.
+		blockNumber uint64,
+		// The changeset to write.
+		cs []*proto.NamedChangeSet,
+	) error
+
+	// Signal that there will be no more writes for the current block number. Attempting to write additional
+	// changes for the same block number after calling this method may result in an error.
+	//
+	// Similar to Write(), this method is asynchronous. Calling this method does not, by itself, make data immediately
+	// crash durable.
+	SignalEndOfBlock() error
+
+	// Flush the WAL to disk. All data previously passed to Write() before this call will be crash durable
+	// after this call returns.
+	Flush() error
+
+	// Get the range of block numbers stored in the WAL.
+	GetStoredRange() (
+		// If true, there is data in the WAL. If false, the WAL is empty and startBlockNumber and
+		// endBlockNumber are undefined.
+		ok bool,
+		// The lowest block number stored in the WAL, inclusive. Only valid if ok is true.
+		startBlockNumber uint64,
+		// The highest block number stored in the WAL, inclusive. Only valid if ok is true.
+		endBlockNumber uint64,
+		// Any error encountered while retrieving the range.
+		err error,
+	)
+
+	// Prune the WAL, removing all entries with block numbers less than lowestBlockNumberToKeep.
+	//
+	// This method merely schedules the prune operation, it does not block until the prune is complete. Pruning
+	// is async and lazy, and implementations are free to delay pruning arbitrarily long. If crashed or closed
+	// before the prune is complete, the WAL may not attempt to prune again on the next open unless Prune() is
+	// called again or for a higher block number.
+	Prune(lowestBlockNumberToKeep uint64) error
+
+	// Create an iterator over the WAL, starting at the given block number.
+	//
+	// 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. Data written before that instant is included; data written after it
+	// is not. For data written concurrently with this call, whether it is included is unspecified.
+	Iterator(startingBlockNumber uint64) (StateWALIterator, error)
+
+	// Close the WAL, flushing any pending writes and releasing resources.
+	Close() error
+}
+
+// Iterates over data in a state WAL, in ascending block order, yielding one entry per block. All changesets
+// written for a block (across one or more Write calls) are coalesced, in write order, into that block's single
+// entry; the entry's EndOfBlock field is always false. Incomplete trailing blocks (those with no end-of-block
+// marker) are not yielded.
+type StateWALIterator interface {
+	// Next advances the iterator to the next block. It returns false when iteration is complete (no more
+	// blocks), 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 coalesced entry for the block at the iterator's current position. It is only valid to
+	// call Entry after Next has returned (true, nil).
+	//
+	// The returned entry, and every byte slice reachable through it (changeset keys and values), 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() *Entry
+
+	// Close releases the resources held by the iterator.
+	Close() error
+}

sei-db/state_db/statewal/state_wal_config.go

@@ -0,0 +1,60 @@
+package statewal
+
+import (
+	"fmt"
+
+	"github.com/sei-protocol/sei-chain/sei-db/common/unit"
+)
+
+// Configuration for a state WAL.
+type Config struct {
+	// The directory where the WAL writes its files.
+	Path string
+
+	// The size of the channel used to send work from the caller to the serialization goroutine.
+	RequestBufferSize uint
+
+	// The size of the channel used to send serialized records from the serialization goroutine to the
+	// writer goroutine.
+	WriteBufferSize uint
+
+	// The size a WAL file may reach before it is sealed and a fresh one is opened. Rotation only happens on
+	// block boundaries, so a file may exceed this by the size of a single block. 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 blocks an iterator's reader thread may prefetch ahead of the consumer. A larger value
+	// keeps the reader busy while the consumer processes blocks, which matters for startup replay speed.
+	// Must be greater than 0.
+	IteratorPrefetchSize uint
+}
+
+// Constructor for a default state WAL configuration.
+func DefaultConfig(path string) *Config {
+	return &Config{
+		Path:                 path,
+		RequestBufferSize:    16,
+		WriteBufferSize:      16,
+		TargetFileSize:       64 * unit.MB,
+		FsyncOnFlush:         true,
+		IteratorPrefetchSize: 32,
+	}
+}
+
+// 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 c.TargetFileSize == 0 {
+		// A zero target would seal and rotate a fresh file after every single block.
+		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/state_db/statewal/state_wal_config_test.go

@@ -0,0 +1,30 @@
+package statewal
+
+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").Validate())
+	})
+
+	t.Run("empty path is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("")
+		require.Error(t, cfg.Validate())
+	})
+
+	t.Run("zero target file size is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("/tmp/wal")
+		cfg.TargetFileSize = 0
+		require.Error(t, cfg.Validate())
+	})
+
+	t.Run("zero iterator prefetch size is rejected", func(t *testing.T) {
+		cfg := DefaultConfig("/tmp/wal")
+		cfg.IteratorPrefetchSize = 0
+		require.Error(t, cfg.Validate())
+	})
+}

sei-db/state_db/statewal/state_wal_entry.go

@@ -0,0 +1,158 @@
+package statewal
+
+import (
+	"encoding/binary"
+	"fmt"
+
+	"github.com/sei-protocol/sei-chain/sei-db/proto"
+)
+
+// The kind of a WAL record. Every serialized entry begins with one of these bytes.
+type entryKind byte
+
+const (
+	// A changeset record: a block number plus the set of changes written for that block.
+	kindChangeset entryKind = 1
+	// An end-of-block record: marks that no more changes will be written for a block number. On reload, a
+	// block whose changeset records are not followed by an end-of-block marker is discarded.
+	kindEndOfBlock entryKind = 2
+)
+
+// A WAL entry for state.
+//
+// An entry is either a changeset record (EndOfBlock is false, Changeset holds the block's changes) or an
+// end-of-block marker (EndOfBlock is true, Changeset is nil). A single block may be described by several
+// changeset records followed by exactly one end-of-block marker.
+type Entry struct {
+
+	// The block number associated with this entry.
+	BlockNumber uint64
+
+	// The changeset associated with this entry. Nil for end-of-block markers.
+	Changeset []*proto.NamedChangeSet
+
+	// True if this entry marks the end of a block. End-of-block entries carry no changeset.
+	EndOfBlock bool
+}
+
+// Constructor for a changeset entry.
+func NewEntry(blockNumber uint64, changeset []*proto.NamedChangeSet) *Entry {
+	return &Entry{
+		BlockNumber: blockNumber,
+		Changeset:   changeset,
+	}
+}
+
+// Constructor for an end-of-block marker entry.
+func NewEndOfBlockEntry(blockNumber uint64) *Entry {
+	return &Entry{
+		BlockNumber: blockNumber,
+		EndOfBlock:  true,
+	}
+}
+
+// Serialize the WAL entry to bytes. The returned bytes are the record payload; the file layer is responsible
+// for framing (length prefix and checksum). The layout is:
+//
+//	[1-byte kind][uvarint block number]
+//
+// followed, for changeset records only, by:
+//
+//	[uvarint changeset count]([uvarint marshaled length][marshaled NamedChangeSet])*
+func (e *Entry) Serialize() ([]byte, error) {
+	var buf []byte
+	var scratch [binary.MaxVarintLen64]byte
+
+	if e.EndOfBlock {
+		buf = append(buf, byte(kindEndOfBlock))
+		n := binary.PutUvarint(scratch[:], e.BlockNumber)
+		buf = append(buf, scratch[:n]...)
+		return buf, nil
+	}
+
+	buf = append(buf, byte(kindChangeset))
+	n := binary.PutUvarint(scratch[:], e.BlockNumber)
+	buf = append(buf, scratch[:n]...)
+
+	n = binary.PutUvarint(scratch[:], uint64(len(e.Changeset)))
+	buf = append(buf, scratch[:n]...)
+
+	for i, ncs := range e.Changeset {
+		if ncs == nil {
+			return nil, fmt.Errorf("changeset %d is nil", i)
+		}
+		marshaled, err := ncs.Marshal()
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal changeset %d: %w", i, err)
+		}
+		n = binary.PutUvarint(scratch[:], uint64(len(marshaled)))
+		buf = append(buf, scratch[:n]...)
+		buf = append(buf, marshaled...)
+	}
+
+	return buf, nil
+}
+
+// DeserializeEntry parses a record payload previously produced by Serialize.
+func DeserializeEntry(data []byte) (
+	// The resulting WAL entry.
+	entry *Entry,
+	// If true, the WAL entry was successfully deserialized.
+	// If false, the data was truncated or otherwise incomplete and entry is nil.
+	ok bool,
+	// Returns an error if the data could not be deserialized due to an unexpected error (e.g. a corrupt
+	// protobuf payload). Does not return an error if the data is simply truncated.
+	err error,
+) {
+	if len(data) == 0 {
+		return nil, false, nil
+	}
+
+	kind := entryKind(data[0])
+	rest := data[1:]
+
+	blockNumber, n := binary.Uvarint(rest)
+	if n <= 0 {
+		return nil, false, nil
+	}
+	rest = rest[n:]
+
+	switch kind {
+	case kindEndOfBlock:
+		return NewEndOfBlockEntry(blockNumber), true, nil
+	case kindChangeset:
+		count, n := binary.Uvarint(rest)
+		if n <= 0 {
+			return nil, false, nil
+		}
+		rest = rest[n:]
+
+		// Each changeset entry occupies at least one byte in rest (its length prefix), so a count larger
+		// than the remaining bytes cannot be valid. Reject it before allocating, to avoid a panic/OOM on a
+		// corrupt payload that survived the CRC32 check. Mirrors the length bound in the loop below.
+		if count > uint64(len(rest)) {
+			return nil, false, nil
+		}
+
+		changeset := make([]*proto.NamedChangeSet, 0, count)
+		for i := uint64(0); i < count; i++ {
+			length, n := binary.Uvarint(rest)
+			if n <= 0 {
+				return nil, false, nil
+			}
+			rest = rest[n:]
+			if uint64(len(rest)) < length {
+				return nil, false, nil
+			}
+			ncs := &proto.NamedChangeSet{}
+			if err := ncs.Unmarshal(rest[:length]); err != nil {
+				return nil, false, fmt.Errorf("failed to unmarshal changeset %d: %w", i, err)
+			}
+			rest = rest[length:]
+			changeset = append(changeset, ncs)
+		}
+		return NewEntry(blockNumber, changeset), true, nil
+	default:
+		return nil, false, fmt.Errorf("unknown WAL entry kind %d", kind)
+	}
+}