move flatKV hashing to a background thread

sei-cosmos/storev2/rootmulti/flatkv_helpers_test.go

@@ -369,7 +369,7 @@ func rollbackFlatKV(t *testing.T, dir string, cfg seidbconfig.StateCommitConfig,
 	flatkvCfg.DataDir = utils.GetFlatKVPath(dir)
 	stateWAL, err := flatkv.OpenStateWAL(&flatkvCfg)
 	require.NoError(t, err)
-	evmStore, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL)
+	evmStore, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL, nil)
 	require.NoError(t, err)
 	err = evmStore.LoadLatest()
 	require.NoError(t, err)
@@ -397,7 +397,7 @@ func openFlatKVReadOnly(t *testing.T, dir string, cfg seidbconfig.StateCommitCon
 	flatkvCfg.DataDir = utils.GetFlatKVPath(dir)
 	stateWAL, err := flatkv.OpenStateWAL(&flatkvCfg)
 	require.NoError(t, err)
-	store, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL)
+	store, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL, nil)
 	require.NoError(t, err)
 	ro, err := store.LoadVersionReadOnly(version)
 	require.NoError(t, err)
@@ -462,7 +462,7 @@ func collectFlatKVEVM(t *testing.T, dir string, cfg seidbconfig.StateCommitConfi
 
 	stateWAL, err := flatkv.OpenStateWAL(&flatkvCfg)
 	require.NoError(t, err)
-	s, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL)
+	s, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL, nil)
 	require.NoError(t, err)
 	defer func() { require.NoError(t, s.Close()) }()
 

sei-cosmos/storev2/rootmulti/flatkv_migration_test.go

@@ -38,7 +38,7 @@ func migrationVersionInFlatKV(t *testing.T, dir string, cfg seidbconfig.StateCom
 	flatkvCfg.DataDir = utils.GetFlatKVPath(dir)
 	stateWAL, err := flatkv.OpenStateWAL(&flatkvCfg)
 	require.NoError(t, err)
-	s, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL)
+	s, err := flatkv.NewCommitStore(context.Background(), &flatkvCfg, stateWAL, nil)
 	require.NoError(t, err)
 	err = s.LoadLatest()
 	require.NoError(t, err)

sei-cosmos/storev2/rootmulti/flatkv_workload_test.go

@@ -57,9 +57,9 @@ func TestFlatKVFullScanLtHashVerification(t *testing.T) {
 
 	require.NoError(t, flatkv.VerifyLtHash(ro), "full-scan LtHash verification failed")
 
-	roHash, _ := ro.RootHash()
-	require.Equal(t, expectedLatticeHash, roHash,
-		"flatkv RootHash should match evm_lattice in CommitInfo")
+	roHash := ro.PublishedHash().Global.Checksum()
+	require.Equal(t, expectedLatticeHash, roHash[:],
+		"flatkv's published root should match evm_lattice in CommitInfo")
 }
 
 // ---------------------------------------------------------------------------

sei-cosmos/storev2/rootmulti/hashlog.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"path/filepath"
 
+	"github.com/sei-protocol/sei-chain/sei-db/config"
 	"github.com/sei-protocol/sei-chain/sei-db/proto"
 	"github.com/sei-protocol/sei-chain/sei-db/state_db/sc/hashlog"
 )
@@ -55,11 +56,11 @@ func (rs *Store) SetNextResultHash(resultHash []byte) {
 // hashLogDir returns the directory hash log files are written to, defaulting to a "hash.log" directory
 // under the state-commit store's data directory (sibling of committer.db / receipt.db). The ".log"
 // suffix mirrors the data/ naming convention (.db, .wal); the files inside keep the .hlog format.
-func (rs *Store) hashLogDir() string {
-	if rs.hashLoggerConfig.Directory != "" {
-		return rs.hashLoggerConfig.Directory
+func hashLogDir(scDir string, cfg config.HashLoggerConfig) string {
+	if cfg.Directory != "" {
+		return cfg.Directory
 	}
-	return filepath.Join(rs.scDir, "data", "hash.log")
+	return filepath.Join(scDir, "data", "hash.log")
 }
 
 // desiredHashCategories computes the full caller-reported category set for the current backend state:
@@ -83,29 +84,30 @@ func (rs *Store) desiredHashCategories() map[string]struct{} {
 	return categories
 }
 
-// openHashLogger constructs the logger once. It starts with no caller columns (just the changeset
-// column); syncHashCategories then registers the live categories, which the logger handles as runtime
-// column changes (each new column rotates to a fresh file, but the empty initial files are dropped and
-// their indexes reused, so the first file with data starts at index 0).
-func (rs *Store) openHashLogger() error {
-	loggerVersion := rs.hashLoggerConfig.Version
+// openHashLogger constructs the logger once, with no caller columns beyond the changeset column.
+//
+// The columns arrive afterwards, from two directions: a backend registers the ones it reports when it is
+// handed this logger, and syncHashCategories registers whatever else the live backend set calls for and
+// removes what it drops. The logger treats each change as a file rotation, but an empty file is dropped
+// and its index reused, so the first file with data still starts at index 0.
+func openHashLogger(scDir string, hashLoggerConfig config.HashLoggerConfig) (hashlog.HashLogger, error) {
+	loggerVersion := hashLoggerConfig.Version
 	if loggerVersion == "" {
 		loggerVersion = "unknown"
 	}
-	cfg := hashlog.DefaultHashLoggerConfig(rs.hashLogDir(), loggerVersion)
+	cfg := hashlog.DefaultHashLoggerConfig(hashLogDir(scDir, hashLoggerConfig), loggerVersion)
 	// Propagate the operator-configured retention tunables verbatim. A configured 0 must reach the logger
 	// (where it disables that dimension); the old `if > 0` guards swallowed it. Defaults are applied at
 	// config construction (config.DefaultHashLoggerConfig), so these always carry a meaningful value.
-	cfg.BlocksToRetain = rs.hashLoggerConfig.BlocksToRetain
-	cfg.TargetFileSize = rs.hashLoggerConfig.TargetFileSize
-	cfg.MaxDiskSize = rs.hashLoggerConfig.MaxDiskSize
+	cfg.BlocksToRetain = hashLoggerConfig.BlocksToRetain
+	cfg.TargetFileSize = hashLoggerConfig.TargetFileSize
+	cfg.MaxDiskSize = hashLoggerConfig.MaxDiskSize
 
 	hl, err := hashlog.NewHashLogger(cfg)
 	if err != nil {
-		return fmt.Errorf("failed to create hash logger: %w", err)
+		return nil, fmt.Errorf("failed to create hash logger: %w", err)
 	}
-	rs.hashLogger = hl
-	return nil
+	return hl, nil
 }
 
 // syncHashCategories brings the logger's column set in line with the desired set for the current backend
@@ -145,21 +147,12 @@ func (rs *Store) disableHashLogger() {
 	}
 }
 
-// recordBlockHashes reports every hash for the just-committed block at the given version. It opens the
-// logger on first use and keeps its column set in sync with the live backends. On open failure it
-// disables hash logging rather than disrupting commit. Must be called with rs.mtx held (from Commit).
+// recordBlockHashes reports every hash for the just-committed block at the given version, keeping the
+// logger's column set in sync with the live backends. Must be called with rs.mtx held (from Commit).
 func (rs *Store) recordBlockHashes(version int64) {
 	if rs.hashLoggerDisabled {
 		return
 	}
-
-	if rs.hashLogger == nil {
-		if err := rs.openHashLogger(); err != nil {
-			logger.Error("failed to open hash logger; disabling hash logging", "err", err)
-			rs.disableHashLogger()
-			return
-		}
-	}
 	rs.syncHashCategories()
 
 	blockNumber := uint64(version) //nolint:gosec // commit versions are non-negative

sei-cosmos/storev2/rootmulti/store.go

@@ -143,8 +143,22 @@ func NewStore(
 	if scConfig.HistoricalProofRateLimit > 0 {
 		limiter = rate.NewLimiter(rate.Limit(scConfig.HistoricalProofRateLimit), burst)
 	}
+	// Opened before the store it is handed to: flatKV reports its hashes from its own finalization
+	// goroutine, so it needs the logger at construction rather than per block.
+	hashLoggingOn := scConfig.HashLogger.Enable
+	var hashLogger hashlog.HashLogger
+	if hashLoggingOn {
+		hl, err := openHashLogger(scDir, scConfig.HashLogger)
+		if err != nil {
+			logger.Error("failed to open hash logger; disabling hash logging", "err", err)
+			hashLoggingOn = false
+		} else {
+			hashLogger = hl
+		}
+	}
+
 	ctx := context.Background()
-	scStore, err := composite.NewCompositeCommitStore(ctx, scDir, scConfig)
+	scStore, err := composite.NewCompositeCommitStore(ctx, scDir, scConfig, hashLogger)
 	if err != nil {
 		panic(err)
 	}
@@ -169,7 +183,8 @@ func NewStore(
 			MaxBytes: scConfig.SubspaceMaxBytes,
 		},
 		hashLoggerConfig:   scConfig.HashLogger,
-		hashLoggerDisabled: !scConfig.HashLogger.Enable,
+		hashLogger:         hashLogger,
+		hashLoggerDisabled: !hashLoggingOn,
 		scDir:              scDir,
 		// No height has been flushed yet, and the first block is 1, so -1 cannot collide with it.
 		flushedVersion: -1,

sei-db/bootstrap/storage_manager.go

@@ -102,7 +102,7 @@ func (m *GigaStorageManager) openDBs(ctx context.Context, cfg config.GigaStorage
 	}
 
 	// StateDB writes the WAL; a store that held one would record every block twice.
-	sc, err := flatkv.NewCommitStore(ctx, cfg.FlatKVConfig, nil)
+	sc, err := flatkv.NewCommitStore(ctx, cfg.FlatKVConfig, nil, nil)
 	if err != nil {
 		return fmt.Errorf("open state commit store: %w", err)
 	}

sei-db/state_db/bench/wrappers/db_implementations.go

@@ -81,7 +81,7 @@ func newFlatKVCommitStore(ctx context.Context, dbDir string, config *flatkvConfi
 	if err != nil {
 		return nil, fmt.Errorf("failed to open FlatKV state WAL: %w", err)
 	}
-	cs, err := flatkv.NewCommitStore(ctx, config, stateWAL)
+	cs, err := flatkv.NewCommitStore(ctx, config, stateWAL, nil)
 	if err != nil {
 		_ = stateWAL.Close()
 		return nil, fmt.Errorf("failed to create FlatKV commit store: %w", err)
@@ -101,7 +101,7 @@ func newCompositeCommitStore(ctx context.Context, dbDir string, writeMode sctype
 	cfg.MemIAVLConfig.AsyncCommitBuffer = 10
 	cfg.MemIAVLConfig.SnapshotInterval = 100
 
-	cs, err := composite.NewCompositeCommitStore(ctx, dbDir, cfg)
+	cs, err := composite.NewCompositeCommitStore(ctx, dbDir, cfg, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create composite commit store: %w", err)
 	}

sei-db/state_db/giga/live_state_store.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/sei-protocol/sei-chain/sei-db/common/metrics"
 	"github.com/sei-protocol/sei-chain/sei-db/proto"
-	"github.com/sei-protocol/sei-chain/sei-db/state_db/sc/hashlog"
+	"github.com/sei-protocol/sei-chain/sei-db/state_db/sc/flatkv/lthash"
 	"github.com/sei-protocol/sei-chain/sei-db/state_db/sc/types"
 )
 
@@ -122,18 +122,38 @@ type LiveStateStore interface {
 		ascending bool,
 	) (dbm.Iterator, error)
 
-	// RootHash returns the 32-byte checksum of the committed LtHash and the height that checksum
-	// describes. Note: the checksum is the Blake3-256 digest of the underlying 2048-byte raw LtHash
-	// vector.
-	RootHash() ([]byte, int64)
+	// PublishedHash returns the most recent block hash the store has published: its height, its
+	// lattice hash root, and each database's root. Hashing is asynchronous, so on a committing store
+	// this lags the committed version; use FlushHashes to make it describe the version just committed.
+	// On a freshly loaded or read-only store it is the height that was loaded.
+	PublishedHash() *lthash.BlockHash
+
+	// HashChan returns a channel producing the hash of each block: exactly one per block committed, in
+	// block order, with no gaps or duplicates, closed once the store stops hashing.
+	//
+	// The channel has finite depth, so failure to dequeue hashes for long enough blocks commit. Every
+	// store that returns one therefore needs a consumer.
+	//
+	// A store that will never carry a stream reports why instead of handing back one that stays empty:
+	// one that is not open, and one that hashes only in order to replay and so consumes its own.
+	HashChan() (<-chan *lthash.BlockHash, error)
+
+	// FlushHashes blocks until the store has published a hash for every block committed so far, and
+	// recorded each one's metadata alongside the block it describes.
+	FlushHashes() error
+
+	// CommitPendingBlock commits the block currently being applied, if any, so that it has a hash. A
+	// no-op on a store with no pending writes.
+	//
+	// A block that has not been committed has no hash, so a caller wanting one mid-block is asking for
+	// the block to be committed. This is that request, made explicitly. Post-Cosmos nothing asks for a
+	// hash mid-block and this goes away.
+	CommitPendingBlock() error
 
 	// HashCategories returns the hash logger category names this store reports (the global root plus one
 	// per data DB). The set is fixed. The caller registers these on the logger.
 	HashCategories() []string
 
-	// RecordHashes reports this store's hashes (root + per-DB) for blockNumber. Call right after Commit.
-	RecordHashes(hl hashlog.HashLogger, blockNumber uint64) error
-
 	// Version returns the latest committed version.
 	Version() int64
 

sei-db/state_db/giga/state_db_impl_test.go

@@ -62,7 +62,7 @@ func (w *fakeStateWAL) SignalEndOfBlock() error {
 func newTestStateDB(t *testing.T) (giga.StateDB, *fakeStateWAL, *flatkv.CommitStore) {
 	t.Helper()
 
-	liveStateDB, err := flatkv.NewCommitStore(t.Context(), config.DefaultTestConfig(t), nil)
+	liveStateDB, err := flatkv.NewCommitStore(t.Context(), config.DefaultTestConfig(t), nil, nil)
 	require.NoError(t, err)
 	require.NoError(t, liveStateDB.LoadLatest())
 	t.Cleanup(func() { require.NoError(t, liveStateDB.Close()) })

sei-db/state_db/sc/composite/commit_info_stored_test.go

@@ -25,7 +25,7 @@ func storedInfoConfig() config.StateCommitConfig {
 
 func openStoredInfoStore(t *testing.T, dir string) *CompositeCommitStore {
 	t.Helper()
-	cs, err := NewCompositeCommitStore(t.Context(), dir, storedInfoConfig())
+	cs, err := NewCompositeCommitStore(t.Context(), dir, storedInfoConfig(), nil)
 	require.NoError(t, err)
 	require.NoError(t, cs.Initialize([]string{keys.BankStoreKey, keys.EVMStoreKey}))
 	require.NoError(t, cs.LoadLatest())
@@ -83,7 +83,7 @@ func TestLastCommitInfoUnmovedByWorkingHash(t *testing.T) {
 	require.NoError(t, cs.ApplyChangeSets(storedInfoChangeset(2)))
 	require.NotNil(t, cs.WorkingCommitInfo(cs.Version()+1))
 
-	_, flatKVVersion := cs.flatKV.RootHash()
+	flatKVVersion := cs.flatKV.Version()
 	require.Equal(t, committed+1, flatKVVersion, "flatkv should be a block ahead for this test to mean anything")
 
 	after := cs.LastCommitInfo()