Pass versioned changeset entries to wrappers + ApplyChangesetasync

sei-db/state_db/bench/cryptosim/database.go

@@ -180,7 +180,11 @@ func (d *Database) FinalizeBlock(
 	})
 	d.nextBlockNumber++
 
-	err := d.db.ApplyChangeSets(changeSets)
+	entry := &proto.ChangelogEntry{
+		Version:    d.db.Version() + 1,
+		Changesets: changeSets,
+	}
+	err := d.db.ApplyChangeSets(entry)
 	if err != nil {
 		return fmt.Errorf("failed to apply change sets: %w", err)
 	}

sei-db/state_db/bench/helper.go

@@ -427,7 +427,11 @@ func runBenchmark(b *testing.B, scenario TestScenario, withProgress bool) {
 				if !ok {
 					break
 				}
-				err := cs.ApplyChangeSets([]*proto.NamedChangeSet{changeset})
+				entry := &proto.ChangelogEntry{
+					Version:    baseVersion + block,
+					Changesets: []*proto.NamedChangeSet{changeset},
+				}
+				err := cs.ApplyChangeSets(entry)
 				require.NoError(b, err)
 				version, err := cs.Commit()
 				require.NoError(b, err)

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

@@ -26,12 +26,12 @@ func NewCombinedWrapper(sc DBWrapper, ss dbTypes.StateStore) DBWrapper {
 	return w
 }
 
-func (c *combinedWrapper) ApplyChangeSets(cs []*proto.NamedChangeSet) error {
-	if err := c.sc.ApplyChangeSets(cs); err != nil {
+func (c *combinedWrapper) ApplyChangeSets(entry *proto.ChangelogEntry) error {
+	if err := c.sc.ApplyChangeSets(entry); err != nil {
 		return err
 	}
-	nextVersion := c.ssVersion.Add(1)
-	return c.ss.ApplyChangesetSync(nextVersion, cs)
+	c.ssVersion.Store(entry.Version)
+	return c.ss.ApplyChangesetAsync(entry.Version, entry.Changesets)
 }
 
 func (c *combinedWrapper) Read(key []byte) (data []byte, found bool, err error) {

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

@@ -21,8 +21,8 @@ func NewCompositeWrapper(store *composite.CompositeCommitStore) DBWrapper {
 	}
 }
 
-func (c *compositeWrapper) ApplyChangeSets(cs []*proto.NamedChangeSet) error {
-	return c.base.ApplyChangeSets(cs)
+func (c *compositeWrapper) ApplyChangeSets(entry *proto.ChangelogEntry) error {
+	return c.base.ApplyChangeSets(entry.Changesets)
 }
 
 func (c *compositeWrapper) Commit() (int64, error) {

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

@@ -8,9 +8,10 @@ import (
 
 // This benchmarking utility is capable of benchmarking a DB that implements this interface.
 type DBWrapper interface {
-	// ApplyChangeSets buffers EVM changesets (x/evm memiavl keys) and updates LtHash.
-	// Non-EVM modules are ignored. Call Commit to persist.
-	ApplyChangeSets(cs []*proto.NamedChangeSet) error
+	// ApplyChangeSets applies a versioned changelog entry. SC-backed wrappers buffer
+	// entry changesets until Commit, while SS-backed wrappers can use entry.Version
+	// to persist at the benchmark-assigned version immediately.
+	ApplyChangeSets(entry *proto.ChangelogEntry) error
 
 	// Read reads the value for the given key.
 	Read(key []byte) (data []byte, found bool, err error)

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

@@ -24,8 +24,8 @@ func NewFlatKVWrapper(store flatkv.Store) DBWrapper {
 	}
 }
 
-func (f *flatKVWrapper) ApplyChangeSets(cs []*proto.NamedChangeSet) error {
-	err := f.base.ApplyChangeSets(cs)
+func (f *flatKVWrapper) ApplyChangeSets(entry *proto.ChangelogEntry) error {
+	err := f.base.ApplyChangeSets(entry.Changesets)
 	if err == nil {
 		f.hasPending = true
 	}

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

@@ -34,8 +34,8 @@ func (m *memIAVLWrapper) Version() int64 {
 	return m.base.Version()
 }
 
-func (m *memIAVLWrapper) ApplyChangeSets(cs []*proto.NamedChangeSet) error {
-	return m.base.ApplyChangeSets(cs)
+func (m *memIAVLWrapper) ApplyChangeSets(entry *proto.ChangelogEntry) error {
+	return m.base.ApplyChangeSets(entry.Changesets)
 }
 
 func (m *memIAVLWrapper) Importer(version int64) (types.Importer, error) {

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

@@ -14,8 +14,8 @@ var _ DBWrapper = (*stateStoreWrapper)(nil)
 
 // stateStoreWrapper adapts a versioned StateStore (SS layer) to the DBWrapper
 // interface used by the cryptosim benchmark. Each ApplyChangeSets call maps to
-// a single ApplyChangesetSync at an incrementing version. The SS layer persists
-// on every apply, so Commit is a no-op.
+// a single ApplyChangesetSync at the benchmark-provided version. The SS layer
+// persists on every apply, so Commit is a no-op.
 type stateStoreWrapper struct {
 	base    dbTypes.StateStore
 	version atomic.Int64
@@ -29,9 +29,9 @@ func NewStateStoreWrapper(store dbTypes.StateStore) DBWrapper {
 	return w
 }
 
-func (s *stateStoreWrapper) ApplyChangeSets(cs []*proto.NamedChangeSet) error {
-	nextVersion := s.version.Add(1)
-	return s.base.ApplyChangesetSync(nextVersion, cs)
+func (s *stateStoreWrapper) ApplyChangeSets(entry *proto.ChangelogEntry) error {
+	s.version.Store(entry.Version)
+	return s.base.ApplyChangesetSync(entry.Version, entry.Changesets)
 }
 
 func (s *stateStoreWrapper) Read(key []byte) (data []byte, found bool, err error) {

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

@@ -0,0 +1,171 @@
+package wrappers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/sei-protocol/sei-chain/sei-db/common/metrics"
+	dbTypes "github.com/sei-protocol/sei-chain/sei-db/db_engine/types"
+	"github.com/sei-protocol/sei-chain/sei-db/proto"
+	scTypes "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/types"
+)
+
+type mockDBWrapper struct {
+	appliedEntries []*proto.ChangelogEntry
+	commitVersion  int64
+	applyErr       error
+	commitErr      error
+}
+
+func (m *mockDBWrapper) ApplyChangeSets(entry *proto.ChangelogEntry) error {
+	m.appliedEntries = append(m.appliedEntries, entry)
+	return m.applyErr
+}
+
+func (m *mockDBWrapper) Read(_ []byte) ([]byte, bool, error) {
+	return nil, false, nil
+}
+
+func (m *mockDBWrapper) Commit() (int64, error) {
+	return m.commitVersion, m.commitErr
+}
+
+func (m *mockDBWrapper) Close() error {
+	return nil
+}
+
+func (m *mockDBWrapper) Version() int64 {
+	return m.commitVersion
+}
+
+func (m *mockDBWrapper) LoadVersion(_ int64) error {
+	return nil
+}
+
+func (m *mockDBWrapper) Importer(_ int64) (scTypes.Importer, error) {
+	return nil, nil
+}
+
+func (m *mockDBWrapper) GetPhaseTimer() *metrics.PhaseTimer {
+	return nil
+}
+
+type mockStateStore struct {
+	latestVersion int64
+	asyncVersion  int64
+	asyncChanges  []*proto.NamedChangeSet
+	asyncCalls    int
+	syncVersion   int64
+	syncChanges   []*proto.NamedChangeSet
+	syncCalls     int
+}
+
+func (m *mockStateStore) Get(_ string, _ int64, _ []byte) ([]byte, error) {
+	return nil, nil
+}
+
+func (m *mockStateStore) Has(_ string, _ int64, _ []byte) (bool, error) {
+	return false, nil
+}
+
+func (m *mockStateStore) Iterator(_ string, _ int64, _, _ []byte) (dbTypes.DBIterator, error) {
+	return nil, nil
+}
+
+func (m *mockStateStore) ReverseIterator(_ string, _ int64, _, _ []byte) (dbTypes.DBIterator, error) {
+	return nil, nil
+}
+
+func (m *mockStateStore) RawIterate(_ string, _ func([]byte, []byte, int64) bool) (bool, error) {
+	return false, nil
+}
+
+func (m *mockStateStore) GetLatestVersion() int64 {
+	return m.latestVersion
+}
+
+func (m *mockStateStore) SetLatestVersion(version int64) error {
+	m.latestVersion = version
+	return nil
+}
+
+func (m *mockStateStore) GetEarliestVersion() int64 {
+	return 0
+}
+
+func (m *mockStateStore) SetEarliestVersion(_ int64, _ bool) error {
+	return nil
+}
+
+func (m *mockStateStore) ApplyChangesetSync(version int64, changesets []*proto.NamedChangeSet) error {
+	m.syncCalls++
+	m.syncVersion = version
+	m.syncChanges = changesets
+	m.latestVersion = version
+	return nil
+}
+
+func (m *mockStateStore) ApplyChangesetAsync(version int64, changesets []*proto.NamedChangeSet) error {
+	m.asyncCalls++
+	m.asyncVersion = version
+	m.asyncChanges = changesets
+	m.latestVersion = version
+	return nil
+}
+
+func (m *mockStateStore) Prune(_ int64) error {
+	return nil
+}
+
+func (m *mockStateStore) Import(_ int64, _ <-chan dbTypes.SnapshotNode) error {
+	return nil
+}
+
+func (m *mockStateStore) Close() error {
+	return nil
+}
+
+func TestCombinedWrapperApplyChangeSetsUsesAsyncSS(t *testing.T) {
+	sc := &mockDBWrapper{commitVersion: 7}
+	ss := &mockStateStore{latestVersion: 7}
+	wrapper := NewCombinedWrapper(sc, ss)
+
+	changesets := []*proto.NamedChangeSet{{Name: EVMStoreName}}
+	entry := &proto.ChangelogEntry{
+		Version:    8,
+		Changesets: changesets,
+	}
+
+	err := wrapper.ApplyChangeSets(entry)
+	require.NoError(t, err)
+	require.Len(t, sc.appliedEntries, 1)
+	require.Same(t, entry, sc.appliedEntries[0])
+	require.Equal(t, 1, ss.asyncCalls)
+	require.Equal(t, entry.Version, ss.asyncVersion)
+	require.Equal(t, changesets, ss.asyncChanges)
+	require.Zero(t, ss.syncCalls)
+	require.Equal(t, entry.Version, wrapper.Version())
+}
+
+func TestStateStoreWrapperApplyChangeSetsUsesEntryVersion(t *testing.T) {
+	store := &mockStateStore{latestVersion: 11}
+	wrapper := NewStateStoreWrapper(store)
+
+	changesets := []*proto.NamedChangeSet{{Name: EVMStoreName}}
+	entry := &proto.ChangelogEntry{
+		Version:    15,
+		Changesets: changesets,
+	}
+
+	err := wrapper.ApplyChangeSets(entry)
+	require.NoError(t, err)
+	require.Equal(t, 1, store.syncCalls)
+	require.Equal(t, entry.Version, store.syncVersion)
+	require.Equal(t, changesets, store.syncChanges)
+	require.Zero(t, store.asyncCalls)
+	require.Equal(t, entry.Version, wrapper.Version())
+	version, err := wrapper.Commit()
+	require.NoError(t, err)
+	require.Equal(t, entry.Version, version)
+}