diff --git a/sei-db/state_db/bench/cryptosim/database.go b/sei-db/state_db/bench/cryptosim/database.go index 189bc0adb0..56b2684f29 100644 --- a/sei-db/state_db/bench/cryptosim/database.go +++ b/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) } diff --git a/sei-db/state_db/bench/helper.go b/sei-db/state_db/bench/helper.go index 2a3f7b6d59..eb33f6f16d 100644 --- a/sei-db/state_db/bench/helper.go +++ b/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) diff --git a/sei-db/state_db/bench/wrappers/combined_wrapper.go b/sei-db/state_db/bench/wrappers/combined_wrapper.go index a38a1aacec..ab086ffc03 100644 --- a/sei-db/state_db/bench/wrappers/combined_wrapper.go +++ b/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) { diff --git a/sei-db/state_db/bench/wrappers/composite_wrapper.go b/sei-db/state_db/bench/wrappers/composite_wrapper.go index bc76a1fdf1..00efc66692 100644 --- a/sei-db/state_db/bench/wrappers/composite_wrapper.go +++ b/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) { diff --git a/sei-db/state_db/bench/wrappers/db_wrapper.go b/sei-db/state_db/bench/wrappers/db_wrapper.go index 58c966f31d..bcdd90029d 100644 --- a/sei-db/state_db/bench/wrappers/db_wrapper.go +++ b/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) diff --git a/sei-db/state_db/bench/wrappers/flatkv_wrapper.go b/sei-db/state_db/bench/wrappers/flatkv_wrapper.go index 567936f593..a1e959c58f 100644 --- a/sei-db/state_db/bench/wrappers/flatkv_wrapper.go +++ b/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 } diff --git a/sei-db/state_db/bench/wrappers/memiavl_wrapper.go b/sei-db/state_db/bench/wrappers/memiavl_wrapper.go index 443ae70806..7c957a92c3 100644 --- a/sei-db/state_db/bench/wrappers/memiavl_wrapper.go +++ b/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) { diff --git a/sei-db/state_db/bench/wrappers/state_store_wrapper.go b/sei-db/state_db/bench/wrappers/state_store_wrapper.go index b74d2ccf36..59d551a9e6 100644 --- a/sei-db/state_db/bench/wrappers/state_store_wrapper.go +++ b/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) { diff --git a/sei-db/state_db/bench/wrappers/wrappers_test.go b/sei-db/state_db/bench/wrappers/wrappers_test.go new file mode 100644 index 0000000000..a1d4df27f0 --- /dev/null +++ b/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) +}