LaconicNetwork/cosmos-sdk
10
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

feat(store/v2): add the commitment store (#18526)

Browse Source
This commit is contained in:
cool-developer 2023-11-22 09:17:48 -05:00 committed by GitHub
parent 6ca0b2d1aa
commit fc4ce6c778
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 349 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
store/changeset.go
View File

@ -4,17 +4,19 @@ package store
// track writes. Deletion can be denoted by a nil value or explicitly by the
// Delete field.
type KVPair struct {
Key []byte
Value []byte
StoreKey string // optional
Key []byte
Value []byte
}
// Changeset defines a set of KVPair entries.
type KVPairs []KVPair
// Changeset defines a set of KVPair entries by maintaining a map
// from store key to a slice of KVPair objects.
type Changeset struct {
Pairs []KVPair
Pairs map[string]KVPairs
}
func NewChangeset(pairs ...KVPair) *Changeset {
func NewChangeset(pairs map[string]KVPairs) *Changeset {
return &Changeset{
Pairs: pairs,
}
@ -22,18 +24,23 @@ func NewChangeset(pairs ...KVPair) *Changeset {
// Size returns the number of key-value pairs in the batch.
func (cs *Changeset) Size() int {
return len(cs.Pairs)
cnt := 0
for _, pairs := range cs.Pairs {
cnt += len(pairs)
}
return cnt
}
// Add adds a key-value pair to the ChangeSet.
func (cs *Changeset) Add(key, value []byte) {
cs.Pairs = append(cs.Pairs, KVPair{
func (cs *Changeset) Add(storeKey string, key, value []byte) {
cs.Pairs[storeKey] = append(cs.Pairs[storeKey], KVPair{
Key: key,
Value: value,
})
}
// AddKVPair adds a KVPair to the ChangeSet.
func (cs *Changeset) AddKVPair(pair KVPair) {
cs.Pairs = append(cs.Pairs, pair)
func (cs *Changeset) AddKVPair(storeKey string, pair KVPair) {
cs.Pairs[storeKey] = append(cs.Pairs[storeKey], pair)
}

34
store/commitment/iavl/tree.go
View File

@ -8,10 +8,10 @@ import (
ics23 "github.com/cosmos/ics23/go"
log "cosmossdk.io/log"
"cosmossdk.io/store/v2"
"cosmossdk.io/store/v2/commitment"
)
var _ store.Committer = (*IavlTree)(nil)
var _ commitment.Tree = (*IavlTree)(nil)
// IavlTree is a wrapper around iavl.MutableTree.
type IavlTree struct {
@ -26,25 +26,19 @@ func NewIavlTree(db dbm.DB, logger log.Logger, cfg *Config) *IavlTree {
}
}
// WriteBatch writes a batch of key-value pairs to the database.
func (t *IavlTree) WriteBatch(cs *store.Changeset) error {
for _, kv := range cs.Pairs {
if kv.Value == nil {
_, res, err := t.tree.Remove(kv.Key)
if err != nil {
return err
}
if !res {
return fmt.Errorf("failed to delete key %X", kv.Key)
}
} else {
_, err := t.tree.Set(kv.Key, kv.Value)
if err != nil {
return err
}
}
// Remove removes the given key from the tree.
func (t *IavlTree) Remove(key []byte) error {
_, res, err := t.tree.Remove(key)
if !res {
return fmt.Errorf("key %x not found", key)
}
return nil
return err
}
// Set sets the given key-value pair in the tree.
func (t *IavlTree) Set(key, value []byte) error {
_, err := t.tree.Set(key, value)
return err
}
// WorkingHash returns the working hash of the database.

28
store/commitment/iavl/tree_test.go
View File

@ -7,7 +7,6 @@ import (
"github.com/stretchr/testify/require"
"cosmossdk.io/log"
"cosmossdk.io/store/v2"
)
func generateTree(treeType string) *IavlTree {
@ -25,13 +24,9 @@ func TestIavlTree(t *testing.T) {
require.Equal(t, uint64(0), initVersion)
// write a batch of version 1
cs1 := store.NewChangeset()
cs1.Add([]byte("key1"), []byte("value1"))
cs1.Add([]byte("key2"), []byte("value2"))
cs1.Add([]byte("key3"), []byte("value3"))
err := tree.WriteBatch(cs1)
require.NoError(t, err)
require.NoError(t, tree.Set([]byte("key1"), []byte("value1")))
require.NoError(t, tree.Set([]byte("key2"), []byte("value2")))
require.NoError(t, tree.Set([]byte("key3"), []byte("value3")))
workingHash := tree.WorkingHash()
require.NotNil(t, workingHash)
@ -44,13 +39,10 @@ func TestIavlTree(t *testing.T) {
require.Equal(t, uint64(1), tree.GetLatestVersion())
// write a batch of version 2
cs2 := store.NewChangeset()
cs2.Add([]byte("key4"), []byte("value4"))
cs2.Add([]byte("key5"), []byte("value5"))
cs2.Add([]byte("key6"), []byte("value6"))
cs2.Add([]byte("key1"), nil) // delete key1
err = tree.WriteBatch(cs2)
require.NoError(t, err)
require.NoError(t, tree.Set([]byte("key4"), []byte("value4")))
require.NoError(t, tree.Set([]byte("key5"), []byte("value5")))
require.NoError(t, tree.Set([]byte("key6"), []byte("value6")))
require.NoError(t, tree.Remove([]byte("key1"))) // delete key1
version2Hash := tree.WorkingHash()
require.NotNil(t, version2Hash)
commitHash, err = tree.Commit()
@ -67,10 +59,8 @@ func TestIavlTree(t *testing.T) {
require.NotNil(t, proof.GetNonexist())
// write a batch of version 3
cs3 := store.NewChangeset()
cs3.Add([]byte("key7"), []byte("value7"))
cs3.Add([]byte("key8"), []byte("value8"))
err = tree.WriteBatch(cs3)
require.NoError(t, tree.Set([]byte("key7"), []byte("value7")))
require.NoError(t, tree.Set([]byte("key8"), []byte("value8")))
require.NoError(t, err)
_, err = tree.Commit()
require.NoError(t, err)

138
store/commitment/store.go Normal file
View File

@ -0,0 +1,138 @@
package commitment
import (
"errors"
"fmt"
ics23 "github.com/cosmos/ics23/go"
"cosmossdk.io/log"
"cosmossdk.io/store/v2"
)
var _ store.Committer = (*CommitStore)(nil)
// CommitStore is a wrapper around multiple Tree objects mapped by a unique store
// key. Each store key reflects dedicated and unique usage within a module. A caller
// can construct a CommitStore with one or more store keys. It is expected that a
// RootStore use a CommitStore as an abstraction to handle multiple store keys
// and trees.
type CommitStore struct {
logger log.Logger
multiTrees map[string]Tree
}
// NewCommitStore creates a new CommitStore instance.
func NewCommitStore(multiTrees map[string]Tree, logger log.Logger) (*CommitStore, error) {
return &CommitStore{
logger: logger,
multiTrees: multiTrees,
}, nil
}
func (c *CommitStore) WriteBatch(cs *store.Changeset) error {
for storeKey, pairs := range cs.Pairs {
tree, ok := c.multiTrees[storeKey]
if !ok {
return fmt.Errorf("store key %s not found in multiTrees", storeKey)
}
for _, kv := range pairs {
if kv.Value == nil {
if err := tree.Remove(kv.Key); err != nil {
return err
}
} else if err := tree.Set(kv.Key, kv.Value); err != nil {
return err
}
}
}
return nil
}
func (c *CommitStore) WorkingStoreInfos(version uint64) []store.StoreInfo {
storeInfos := make([]store.StoreInfo, 0, len(c.multiTrees))
for storeKey, tree := range c.multiTrees {
storeInfos = append(storeInfos, store.StoreInfo{
Name: storeKey,
CommitID: store.CommitID{
Version: version,
Hash: tree.WorkingHash(),
},
})
}
return storeInfos
}
func (c *CommitStore) GetLatestVersion() (uint64, error) {
latestVersion := uint64(0)
for storeKey, tree := range c.multiTrees {
version := tree.GetLatestVersion()
if latestVersion != 0 && version != latestVersion {
return 0, fmt.Errorf("store %s has version %d, not equal to latest version %d", storeKey, version, latestVersion)
}
latestVersion = version
}
return latestVersion, nil
}
func (c *CommitStore) LoadVersion(targetVersion uint64) error {
for _, tree := range c.multiTrees {
if err := tree.LoadVersion(targetVersion); err != nil {
return err
}
}
return nil
}
func (c *CommitStore) Commit() ([]store.StoreInfo, error) {
storeInfos := make([]store.StoreInfo, 0, len(c.multiTrees))
for storeKey, tree := range c.multiTrees {
hash, err := tree.Commit()
if err != nil {
return nil, err
}
storeInfos = append(storeInfos, store.StoreInfo{
Name: storeKey,
CommitID: store.CommitID{
Version: tree.GetLatestVersion(),
Hash: hash,
},
})
}
return storeInfos, nil
}
func (c *CommitStore) GetProof(storeKey string, version uint64, key []byte) (*ics23.CommitmentProof, error) {
tree, ok := c.multiTrees[storeKey]
if !ok {
return nil, fmt.Errorf("store %s not found", storeKey)
}
return tree.GetProof(version, key)
}
func (c *CommitStore) Prune(version uint64) (ferr error) {
for _, tree := range c.multiTrees {
if err := tree.Prune(version); err != nil {
ferr = errors.Join(ferr, err)
}
}
return ferr
}
func (c *CommitStore) Close() (ferr error) {
for _, tree := range c.multiTrees {
if err := tree.Close(); err != nil {
ferr = errors.Join(ferr, err)
}
}
return ferr
}

21
store/commitment/tree.go Normal file
View File

@ -0,0 +1,21 @@
package commitment
import (
"io"
ics23 "github.com/cosmos/ics23/go"
)
// Tree is the interface that wraps the basic Tree methods.
type Tree interface {
Set(key, value []byte) error
Remove(key []byte) error
GetLatestVersion() uint64
WorkingHash() []byte
LoadVersion(version uint64) error
Commit() ([]byte, error)
GetProof(version uint64, key []byte) (*ics23.CommitmentProof, error)
Prune(version uint64) error
io.Closer
}

8
store/database.go
View File

@ -68,11 +68,11 @@ type VersionedDatabase interface {
// Committer defines an API for committing state.
type Committer interface {
WriteBatch(cs *Changeset) error
WorkingHash() []byte
GetLatestVersion() uint64
WorkingStoreInfos(version uint64) []StoreInfo
GetLatestVersion() (uint64, error)
LoadVersion(targetVersion uint64) error
Commit() ([]byte, error)
GetProof(version uint64, key []byte) (*ics23.CommitmentProof, error)
Commit() ([]StoreInfo, error)
GetProof(storeKey string, version uint64, key []byte) (*ics23.CommitmentProof, error)
// Prune attempts to prune all versions up to and including the provided
// version argument. The operation should be idempotent. An error should be

2
store/kv/branch/iterator.go
View File

@ -22,7 +22,7 @@ type iterator struct {
key []byte
value []byte
keys []string
values []store.KVPair
values store.KVPairs
reverse bool
exhausted bool // exhausted reflects if the parent iterator is exhausted or not
}

15
store/kv/branch/store.go
View File

@ -72,17 +72,16 @@ func (s *Store) GetChangeset() *store.Changeset {
keys := maps.Keys(s.changeset)
slices.Sort(keys)
pairs := make([]store.KVPair, len(keys))
pairs := make(store.KVPairs, len(keys))
for i, key := range keys {
kvPair := s.changeset[key]
pairs[i] = store.KVPair{
Key: []byte(key),
Value: slices.Clone(kvPair.Value),
StoreKey: kvPair.StoreKey,
Key: []byte(key),
Value: slices.Clone(kvPair.Value),
}
}
return store.NewChangeset(pairs...)
return store.NewChangeset(map[string]store.KVPairs{s.storeKey: pairs})
}
func (s *Store) Reset(toVersion uint64) error {
@ -170,7 +169,7 @@ func (s *Store) Set(key, value []byte) {
defer s.mu.Unlock()
// omit the key as that can be inferred from the map key
s.changeset[string(key)] = store.KVPair{Value: slices.Clone(value), StoreKey: s.storeKey}
s.changeset[string(key)] = store.KVPair{Value: slices.Clone(value)}
}
func (s *Store) Delete(key []byte) {
@ -180,7 +179,7 @@ func (s *Store) Delete(key []byte) {
defer s.mu.Unlock()
// omit the key as that can be inferred from the map key
s.changeset[string(key)] = store.KVPair{Value: nil, StoreKey: s.storeKey}
s.changeset[string(key)] = store.KVPair{Value: nil}
}
func (s *Store) Write() {
@ -290,7 +289,7 @@ func (s *Store) newIterator(parentItr store.Iterator, start, end []byte, reverse
slices.Reverse(keys)
}
values := make([]store.KVPair, len(keys))
values := make(store.KVPairs, len(keys))
for i, key := range keys {
values[i] = s.changeset[key]
}

6
store/kv/branch/store_test.go
View File

@ -28,12 +28,12 @@ func (s *StoreTestSuite) SetupTest() {
storage, err := sqlite.New(s.T().TempDir())
s.Require().NoError(err)
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey: {}})
for i := 0; i < 100; i++ {
key := fmt.Sprintf("key%03d", i) // key000, key001, ..., key099
val := fmt.Sprintf("val%03d", i) // val000, val001, ..., val099
cs.AddKVPair(store.KVPair{StoreKey: storeKey, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
s.Require().NoError(storage.ApplyChangeset(1, cs))
@ -128,7 +128,7 @@ func (s *StoreTestSuite) TestBranch() {
s.Require().Equal([]byte("updated_val001"), s.kvStore.Get([]byte("key001")))
s.Require().Equal(1, b.GetChangeset().Size())
s.Require().Equal([]byte("key001"), b.GetChangeset().Pairs[0].Key)
s.Require().Equal([]byte("key001"), b.GetChangeset().Pairs[storeKey][0].Key)
// write the branched store and ensure all writes are flushed to the parent
b.Write()

15
store/kv/mem/store.go
View File

@ -45,7 +45,7 @@ func (s *Store) GetStoreType() store.StoreType {
func (s *Store) Get(key []byte) []byte {
store.AssertValidKey(key)
kvPair, ok := s.tree.Get(store.KVPair{Key: key, StoreKey: s.storeKey})
kvPair, ok := s.tree.Get(store.KVPair{Key: key})
if !ok || kvPair.Value == nil {
return nil
}
@ -63,29 +63,28 @@ func (s *Store) Set(key, value []byte) {
store.AssertValidKey(key)
store.AssertValidValue(value)
s.tree.Set(store.KVPair{Key: key, Value: value, StoreKey: s.storeKey})
s.tree.Set(store.KVPair{Key: key, Value: value})
}
func (s *Store) Delete(key []byte) {
store.AssertValidKey(key)
s.tree.Set(store.KVPair{Key: key, StoreKey: s.storeKey, Value: nil})
s.tree.Set(store.KVPair{Key: key, Value: nil})
}
func (s *Store) GetChangeset() *store.Changeset {
itr := s.Iterator(nil, nil)
defer itr.Close()
var kvPairs []store.KVPair
var kvPairs store.KVPairs
for ; itr.Valid(); itr.Next() {
kvPairs = append(kvPairs, store.KVPair{
StoreKey: s.storeKey,
Key: itr.Key(),
Value: itr.Value(),
Key: itr.Key(),
Value: itr.Value(),
})
}
return store.NewChangeset(kvPairs...)
return store.NewChangeset(map[string]store.KVPairs{s.storeKey: kvPairs})
}
func (s *Store) Reset(_ uint64) error {

2
store/kv/trace/store_test.go
View File

@ -15,7 +15,7 @@ import (
const storeKey = "storeKey"
var kvPairs = []store.KVPair{
var kvPairs = store.KVPairs{
{Key: []byte(fmt.Sprintf("key%0.8d", 1)), Value: []byte(fmt.Sprintf("value%0.8d", 1))},
{Key: []byte(fmt.Sprintf("key%0.8d", 2)), Value: []byte(fmt.Sprintf("value%0.8d", 2))},
{Key: []byte(fmt.Sprintf("key%0.8d", 3)), Value: []byte(fmt.Sprintf("value%0.8d", 3))},

23
store/pruning/manager_test.go
View File

@ -9,10 +9,13 @@ import (
"cosmossdk.io/log"
"cosmossdk.io/store/v2"
"cosmossdk.io/store/v2/commitment"
"cosmossdk.io/store/v2/commitment/iavl"
"cosmossdk.io/store/v2/storage/sqlite"
)
const defaultStoreKey = "default"
type PruningTestSuite struct {
suite.Suite
@ -34,7 +37,9 @@ func (s *PruningTestSuite) SetupTest() {
ss, err := sqlite.New(s.T().TempDir())
s.Require().NoError(err)
sc := iavl.NewIavlTree(dbm.NewMemDB(), log.NewNopLogger(), iavl.DefaultConfig())
tree := iavl.NewIavlTree(dbm.NewMemDB(), log.NewNopLogger(), iavl.DefaultConfig())
sc, err := commitment.NewCommitStore(map[string]commitment.Tree{"default": tree}, logger)
s.Require().NoError(err)
s.manager = NewManager(logger, ss, sc)
s.ss = ss
@ -57,9 +62,11 @@ func (s *PruningTestSuite) TestPruning() {
for i := uint64(0); i < latestVersion; i++ {
version := i + 1
cs := store.NewChangeset()
cs.Add([]byte("key"), []byte(fmt.Sprintf("value%d", version)))
cs := store.NewChangeset(map[string]store.KVPairs{defaultStoreKey: {}})
cs.AddKVPair(defaultStoreKey, store.KVPair{
Key: []byte("key"),
Value: []byte(fmt.Sprintf("value%d", version)),
})
err := s.sc.WriteBatch(cs)
s.Require().NoError(err)
@ -75,22 +82,22 @@ func (s *PruningTestSuite) TestPruning() {
s.manager.Stop()
// check the store for the version 96
val, err := s.ss.Get("", latestVersion-4, []byte("key"))
val, err := s.ss.Get(defaultStoreKey, latestVersion-4, []byte("key"))
s.Require().NoError(err)
s.Require().Equal([]byte("value96"), val)
// check the store for the version 50
val, err = s.ss.Get("", 50, []byte("key"))
val, err = s.ss.Get(defaultStoreKey, 50, []byte("key"))
s.Require().Error(err)
s.Require().Nil(val)
// check the commitment for the version 96
proof, err := s.sc.GetProof(latestVersion-4, []byte("key"))
proof, err := s.sc.GetProof(defaultStoreKey, latestVersion-4, []byte("key"))
s.Require().NoError(err)
s.Require().NotNil(proof.GetExist())
// check the commitment for the version 95
proof, err = s.sc.GetProof(latestVersion-5, []byte("key"))
proof, err = s.sc.GetProof(defaultStoreKey, latestVersion-5, []byte("key"))
s.Require().Error(err)
s.Require().Nil(proof)
}

31
store/root/store.go
View File

@ -137,7 +137,11 @@ func (s *Store) LastCommitID() (store.CommitID, error) {
}
// sanity check: ensure integrity of latest version against SC
scVersion := s.stateCommitment.GetLatestVersion()
scVersion, err := s.stateCommitment.GetLatestVersion()
if err != nil {
return store.CommitID{}, err
}
if scVersion != latestVersion {
return store.CommitID{}, fmt.Errorf("SC and SS version mismatch; got: %d, expected: %d", scVersion, latestVersion)
}
@ -170,7 +174,7 @@ func (s *Store) Query(storeKey string, version uint64, key []byte, prove bool) (
}
if prove {
proof, err := s.stateCommitment.GetProof(version, key)
proof, err := s.stateCommitment.GetProof(storeKey, version, key)
if err != nil {
return store.QueryResult{}, err
}
@ -368,16 +372,8 @@ func (s *Store) writeSC() error {
}
s.lastCommitInfo = &store.CommitInfo{
Version: version,
StoreInfos: []store.StoreInfo{
{
Name: defaultStoreKey,
CommitID: store.CommitID{
Version: version,
Hash: s.stateCommitment.WorkingHash(),
},
},
},
Version: version,
StoreInfos: s.stateCommitment.WorkingStoreInfos(version),
}
return nil
@ -388,18 +384,23 @@ func (s *Store) writeSC() error {
// solely commits that batch. An error is returned if commit fails or if the
// resulting commit hash is not equivalent to the working hash.
func (s *Store) commitSC() error {
commitBz, err := s.stateCommitment.Commit()
commitStoreInfos, err := s.stateCommitment.Commit()
if err != nil {
return fmt.Errorf("failed to commit SC store: %w", err)
}
commitHash := store.CommitInfo{
Version: s.lastCommitInfo.Version,
StoreInfos: commitStoreInfos,
}.Hash()
workingHash, err := s.WorkingHash()
if err != nil {
return fmt.Errorf("failed to get working hash: %w", err)
}
if bytes.Equal(commitBz, workingHash) {
return fmt.Errorf("unexpected commit hash; got: %X, expected: %X", commitBz, workingHash)
if !bytes.Equal(commitHash, workingHash) {
return fmt.Errorf("unexpected commit hash; got: %X, expected: %X", commitHash, workingHash)
}
return nil

11
store/root/store_test.go
View File

@ -10,6 +10,7 @@ import (
"cosmossdk.io/log"
"cosmossdk.io/store/v2"
"cosmossdk.io/store/v2/commitment"
"cosmossdk.io/store/v2/commitment/iavl"
"cosmossdk.io/store/v2/storage/sqlite"
)
@ -30,7 +31,9 @@ func (s *RootStoreTestSuite) SetupTest() {
ss, err := sqlite.New(s.T().TempDir())
s.Require().NoError(err)
sc := iavl.NewIavlTree(dbm.NewMemDB(), noopLog, iavl.DefaultConfig())
tree := iavl.NewIavlTree(dbm.NewMemDB(), noopLog, iavl.DefaultConfig())
sc, err := commitment.NewCommitStore(map[string]commitment.Tree{"default": tree}, noopLog)
s.Require().NoError(err)
rs, err := New(noopLog, 1, ss, sc)
s.Require().NoError(err)
@ -64,7 +67,7 @@ func (s *RootStoreTestSuite) TestGetKVStore() {
func (s *RootStoreTestSuite) TestGetBranchedKVStore() {
bs := s.rootStore.GetBranchedKVStore("")
s.Require().NotNil(bs)
s.Require().Empty(bs.GetChangeset().Pairs)
s.Require().Empty(bs.GetChangeset().Size())
}
func (s *RootStoreTestSuite) TestQuery() {
@ -85,7 +88,7 @@ func (s *RootStoreTestSuite) TestQuery() {
s.Require().Equal(workingHash, commitHash)
// ensure the proof is non-nil for the corresponding version
result, err := s.rootStore.Query("", 1, []byte("foo"), true)
result, err := s.rootStore.Query(defaultStoreKey, 1, []byte("foo"), true)
s.Require().NoError(err)
s.Require().NotNil(result.Proof)
s.Require().Equal([]byte("foo"), result.Proof.GetExist().Key)
@ -271,7 +274,7 @@ func (s *RootStoreTestSuite) TestCommit() {
s.Require().Equal(uint64(1), lv)
// ensure the root KVStore is cleared
s.Require().Empty(s.rootStore.(*Store).rootKVStore.GetChangeset().Pairs)
s.Require().Empty(s.rootStore.(*Store).rootKVStore.GetChangeset().Size())
// perform reads on the updated root store
bs := s.rootStore.GetKVStore("")

18
store/storage/pebbledb/db.go
View File

@ -181,14 +181,16 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
return err
}
for _, kvPair := range cs.Pairs {
if kvPair.Value == nil {
if err := b.Delete(kvPair.StoreKey, kvPair.Key); err != nil {
return err
}
} else {
if err := b.Set(kvPair.StoreKey, kvPair.Key, kvPair.Value); err != nil {
return err
for storeKey, pairs := range cs.Pairs {
for _, kvPair := range pairs {
if kvPair.Value == nil {
if err := b.Delete(storeKey, kvPair.Key); err != nil {
return err
}
} else {
if err := b.Set(storeKey, kvPair.Key, kvPair.Value); err != nil {
return err
}
}
}
}

18
store/storage/rocksdb/db.go
View File

@ -144,14 +144,16 @@ func (db *Database) Get(storeKey string, version uint64, key []byte) ([]byte, er
func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
b := NewBatch(db, version)
for _, kvPair := range cs.Pairs {
if kvPair.Value == nil {
if err := b.Delete(kvPair.StoreKey, kvPair.Key); err != nil {
return err
}
} else {
if err := b.Set(kvPair.StoreKey, kvPair.Key, kvPair.Value); err != nil {
return err
for storeKey, pairs := range cs.Pairs {
for _, kvPair := range pairs {
if kvPair.Value == nil {
if err := b.Delete(storeKey, kvPair.Key); err != nil {
return err
}
} else {
if err := b.Set(storeKey, kvPair.Key, kvPair.Value); err != nil {
return err
}
}
}
}

4
store/storage/rocksdb/db_test.go
View File

@ -33,12 +33,12 @@ func TestDatabase_ReverseIterator(t *testing.T) {
require.NoError(t, err)
defer db.Close()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 100; i++ {
key := fmt.Sprintf("key%03d", i) // key000, key001, ..., key099
val := fmt.Sprintf("val%03d", i) // val000, val001, ..., val099
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
require.NoError(t, db.ApplyChangeset(1, cs))

18
store/storage/sqlite/db.go
View File

@ -174,14 +174,16 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
return err
}
for _, kvPair := range cs.Pairs {
if kvPair.Value == nil {
if err := b.Delete(kvPair.StoreKey, kvPair.Key); err != nil {
return err
}
} else {
if err := b.Set(kvPair.StoreKey, kvPair.Key, kvPair.Value); err != nil {
return err
for storeKey, pairs := range cs.Pairs {
for _, kvPair := range pairs {
if kvPair.Value == nil {
if err := b.Delete(storeKey, kvPair.Key); err != nil {
return err
}
} else {
if err := b.Set(storeKey, kvPair.Key, kvPair.Value); err != nil {
return err
}
}
}
}

12
store/storage/sqlite/db_test.go
View File

@ -31,12 +31,12 @@ func TestDatabase_ReverseIterator(t *testing.T) {
require.NoError(t, err)
defer db.Close()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 100; i++ {
key := fmt.Sprintf("key%03d", i) // key000, key001, ..., key099
val := fmt.Sprintf("val%03d", i) // val000, val001, ..., val099
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
require.NoError(t, db.ApplyChangeset(1, cs))
@ -106,12 +106,12 @@ func TestParallelWrites(t *testing.T) {
go func(i int) {
<-triggerStartCh
defer wg.Done()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for j := 0; j < kvCount; j++ {
key := fmt.Sprintf("key-%d-%03d", i, j)
val := fmt.Sprintf("val-%d-%03d", i, j)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
require.NoError(t, db.ApplyChangeset(uint64(i+1), cs))
@ -155,12 +155,12 @@ func TestParallelWriteAndPruning(t *testing.T) {
<-triggerStartCh
defer wg.Done()
for i := 0; i < latestVersion; i++ {
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for j := 0; j < kvCount; j++ {
key := fmt.Sprintf("key-%d-%03d", i, j)
val := fmt.Sprintf("val-%d-%03d", i, j)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
require.NoError(t, db.ApplyChangeset(uint64(i+1), cs))

12
store/storage/storage_bench_test.go
View File

@ -57,9 +57,9 @@ func BenchmarkGet(b *testing.B) {
_ = db.Close()
}()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < numKeyVals; i++ {
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: keys[i], Value: vals[i]})
cs.AddKVPair(storeKey1, store.KVPair{Key: keys[i], Value: vals[i]})
}
require.NoError(b, db.ApplyChangeset(1, cs))
@ -93,7 +93,7 @@ func BenchmarkApplyChangeset(b *testing.B) {
for i := 0; i < b.N; i++ {
b.StopTimer()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for j := 0; j < 1000; j++ {
key := make([]byte, 128)
val := make([]byte, 128)
@ -103,7 +103,7 @@ func BenchmarkApplyChangeset(b *testing.B) {
_, err = rng.Read(val)
require.NoError(b, err)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: key, Value: val})
cs.AddKVPair(storeKey1, store.KVPair{Key: key, Value: val})
}
b.StartTimer()
@ -140,9 +140,9 @@ func BenchmarkIterate(b *testing.B) {
b.StopTimer()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < numKeyVals; i++ {
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: keys[i], Value: vals[i]})
cs.AddKVPair(storeKey1, store.KVPair{Key: keys[i], Value: vals[i]})
}
require.NoError(b, db.ApplyChangeset(1, cs))

68
store/storage/storage_test_suite.go
View File

@ -57,7 +57,9 @@ func (s *StorageTestSuite) TestDatabase_VersionedKeys() {
for i := uint64(1); i <= 100; i++ {
s.Require().NoError(db.ApplyChangeset(i, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key"), Value: []byte(fmt.Sprintf("value%03d", i))},
map[string]store.KVPairs{
storeKey1: {{Key: []byte("key"), Value: []byte(fmt.Sprintf("value%03d", i))}},
},
)))
}
@ -75,7 +77,9 @@ func (s *StorageTestSuite) TestDatabase_GetVersionedKey() {
// store a key at version 1
s.Require().NoError(db.ApplyChangeset(1, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key"), Value: []byte("value001")},
map[string]store.KVPairs{
storeKey1: {{Key: []byte("key"), Value: []byte("value001")}},
},
)))
// assume chain progresses to version 10 w/o any changes to key
@ -89,7 +93,9 @@ func (s *StorageTestSuite) TestDatabase_GetVersionedKey() {
// chain progresses to version 11 with an update to key
s.Require().NoError(db.ApplyChangeset(11, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key"), Value: []byte("value011")},
map[string]store.KVPairs{
storeKey1: {{Key: []byte("key"), Value: []byte("value011")}},
},
)))
bz, err = db.Get(storeKey1, 10, []byte("key"))
@ -112,7 +118,7 @@ func (s *StorageTestSuite) TestDatabase_GetVersionedKey() {
// chain progresses to version 15 with a delete to key
s.Require().NoError(db.ApplyChangeset(15, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key")},
map[string]store.KVPairs{storeKey1: {{Key: []byte("key")}}},
)))
// all queries up to version 14 should return the latest value
@ -143,14 +149,14 @@ func (s *StorageTestSuite) TestDatabase_ApplyChangeset() {
s.Require().NoError(err)
defer db.Close()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 100; i++ {
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(fmt.Sprintf("key%03d", i)), Value: []byte("value")})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(fmt.Sprintf("key%03d", i)), Value: []byte("value")})
}
for i := 0; i < 100; i++ {
if i%10 == 0 {
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(fmt.Sprintf("key%03d", i))})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(fmt.Sprintf("key%03d", i))})
}
}
@ -230,12 +236,12 @@ func (s *StorageTestSuite) TestDatabase_Iterator() {
s.Require().NoError(err)
defer db.Close()
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 100; i++ {
key := fmt.Sprintf("key%03d", i) // key000, key001, ..., key099
val := fmt.Sprintf("val%03d", i) // val000, val001, ..., val099
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
s.Require().NoError(db.ApplyChangeset(1, cs))
@ -298,16 +304,24 @@ func (s *StorageTestSuite) TestDatabase_Iterator_RangedDeletes() {
defer db.Close()
s.Require().NoError(db.ApplyChangeset(1, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key001"), Value: []byte("value001")},
store.KVPair{StoreKey: storeKey1, Key: []byte("key002"), Value: []byte("value001")},
map[string]store.KVPairs{
storeKey1: {
{Key: []byte("key001"), Value: []byte("value001")},
{Key: []byte("key002"), Value: []byte("value001")},
},
},
)))
s.Require().NoError(db.ApplyChangeset(5, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key002"), Value: []byte("value002")},
map[string]store.KVPairs{
storeKey1: {{Key: []byte("key002"), Value: []byte("value002")}},
},
)))
s.Require().NoError(db.ApplyChangeset(10, store.NewChangeset(
store.KVPair{StoreKey: storeKey1, Key: []byte("key002")},
map[string]store.KVPairs{
storeKey1: {{Key: []byte("key002")}},
},
)))
itr, err := db.Iterator(storeKey1, 11, []byte("key001"), nil)
@ -332,12 +346,12 @@ func (s *StorageTestSuite) TestDatabase_IteratorMultiVersion() {
// for versions 1-49, set all 10 keys
for v := uint64(1); v < 50; v++ {
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 10; i++ {
key := fmt.Sprintf("key%03d", i)
val := fmt.Sprintf("val%03d-%03d", i, v)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
s.Require().NoError(db.ApplyChangeset(v, cs))
@ -345,13 +359,13 @@ func (s *StorageTestSuite) TestDatabase_IteratorMultiVersion() {
// for versions 50-100, only update even keys
for v := uint64(50); v <= 100; v++ {
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 10; i++ {
if i%2 == 0 {
key := fmt.Sprintf("key%03d", i)
val := fmt.Sprintf("val%03d-%03d", i, v)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
}
@ -390,12 +404,12 @@ func (s *StorageTestSuite) TestDatabase_IteratorNoDomain() {
// for versions 1-50, set all 10 keys
for v := uint64(1); v <= 50; v++ {
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 10; i++ {
key := fmt.Sprintf("key%03d", i)
val := fmt.Sprintf("val%03d-%03d", i, v)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
s.Require().NoError(db.ApplyChangeset(v, cs))
@ -430,12 +444,12 @@ func (s *StorageTestSuite) TestDatabase_Prune() {
// for versions 1-50, set 10 keys
for v := uint64(1); v <= 50; v++ {
cs := new(store.Changeset)
cs := store.NewChangeset(map[string]store.KVPairs{storeKey1: {}})
for i := 0; i < 10; i++ {
key := fmt.Sprintf("key%03d", i)
val := fmt.Sprintf("val%03d-%03d", i, v)
cs.AddKVPair(store.KVPair{StoreKey: storeKey1, Key: []byte(key), Value: []byte(val)})
cs.AddKVPair(storeKey1, store.KVPair{Key: []byte(key), Value: []byte(val)})
}
s.Require().NoError(db.ApplyChangeset(v, cs))
@ -496,9 +510,15 @@ func (s *StorageTestSuite) TestDatabase_Prune_KeepRecent() {
key := []byte("key")
// write a key at three different versions
s.Require().NoError(db.ApplyChangeset(1, store.NewChangeset(store.KVPair{StoreKey: storeKey1, Key: key, Value: []byte("val001")})))
s.Require().NoError(db.ApplyChangeset(100, store.NewChangeset(store.KVPair{StoreKey: storeKey1, Key: key, Value: []byte("val100")})))
s.Require().NoError(db.ApplyChangeset(200, store.NewChangeset(store.KVPair{StoreKey: storeKey1, Key: key, Value: []byte("val200")})))
s.Require().NoError(db.ApplyChangeset(1, store.NewChangeset(
map[string]store.KVPairs{storeKey1: {{Key: key, Value: []byte("val001")}}},
)))
s.Require().NoError(db.ApplyChangeset(100, store.NewChangeset(
map[string]store.KVPairs{storeKey1: {{Key: key, Value: []byte("val100")}}},
)))
s.Require().NoError(db.ApplyChangeset(200, store.NewChangeset(
map[string]store.KVPairs{storeKey1: {{Key: key, Value: []byte("val200")}}},
)))
// prune version 50
s.Require().NoError(db.Prune(50))