feat: ADR-040: Implement KV Store with decoupled storage and SMT (#9892)

## Description

Resolves: #10117

Implements a `CommitKVStore` which separates the concerns of state storage and state commitment according to [ADR-040](https://github.com/cosmos/cosmos-sdk/blob/eb7d939f86c6cd7b4218492364cdda3f649f06b5/docs/architecture/adr-040-storage-and-smt-state-commitments.md).

---

### Author Checklist

*All items are required. Please add a note to the item if the item is not applicable and
please add links to any relevant follow up issues.*

I have...

- [x] included the correct [type prefix](https://github.com/commitizen/conventional-commit-types/blob/v3.0.0/index.json) in the PR title
- [ ] added `!` to the type prefix if API or client breaking change
- [x] targeted the correct branch (see [PR Targeting](https://github.com/cosmos/cosmos-sdk/blob/master/CONTRIBUTING.md#pr-targeting))
- [x] provided a link to the relevant issue or specification
- [ ] followed the guidelines for [building modules](https://github.com/cosmos/cosmos-sdk/blob/master/docs/building-modules) - n/a
- [x] included the necessary unit and integration [tests](https://github.com/cosmos/cosmos-sdk/blob/master/CONTRIBUTING.md#testing)
- [x] added a changelog entry to `CHANGELOG.md`
- [x] included comments for [documenting Go code](https://blog.golang.org/godoc)
- [x] updated the relevant documentation or specification
- [x] reviewed "Files changed" and left comments if necessary
- [ ] confirmed all CI checks have passed

### Reviewers Checklist

*All items are required. Please add a note if the item is not applicable and please add
your handle next to the items reviewed if you only reviewed selected items.*

I have...

- [ ] confirmed the correct [type prefix](https://github.com/commitizen/conventional-commit-types/blob/v3.0.0/index.json) in the PR title
- [ ] confirmed `!` in the type prefix if API or client breaking change
- [ ] confirmed all author checklist items have been addressed 
- [ ] reviewed state machine logic
- [ ] reviewed API design and naming
- [ ] reviewed documentation is accurate
- [ ] reviewed tests and test coverage
- [ ] manually tested (if applicable)

CHANGELOG.md

@@ -200,6 +200,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 * [\#9848](https://github.com/cosmos/cosmos-sdk/pull/9848) ADR-040: Implement BadgerDB backend
 * [\#9851](https://github.com/cosmos/cosmos-sdk/pull/9851) ADR-040: Implement RocksDB backend
 * [\#10308](https://github.com/cosmos/cosmos-sdk/pull/10308) ADR-040: Implement DBConnection.Revert
+* [\#9892](https://github.com/cosmos/cosmos-sdk/pull/9892) ADR-040: KV Store with decoupled storage and state commitment
 
 
 ### Client Breaking Changes

db/adapter.go

@@ -0,0 +1,23 @@
+package db
+
+type readerRWAdapter struct{ DBReader }
+
+// ReaderAsReadWriter returns a ReadWriter that forwards to a reader and errors if writes are
+// attempted. Can be used to pass a Reader when a ReadWriter is expected
+// but no writes will actually occur.
+func ReaderAsReadWriter(r DBReader) DBReadWriter {
+	return readerRWAdapter{r}
+}
+
+func (readerRWAdapter) Set([]byte, []byte) error {
+	return ErrReadOnly
+}
+
+func (readerRWAdapter) Delete([]byte) error {
+	return ErrReadOnly
+}
+
+func (rw readerRWAdapter) Commit() error {
+	rw.Discard()
+	return nil
+}

db/dbtest/util.go

@@ -0,0 +1,52 @@
+package dbtest
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	dbm "github.com/cosmos/cosmos-sdk/db"
+)
+
+func AssertNext(t *testing.T, itr dbm.Iterator, expected bool) {
+	t.Helper()
+	require.Equal(t, expected, itr.Next())
+}
+
+func AssertDomain(t *testing.T, itr dbm.Iterator, start, end []byte) {
+	t.Helper()
+	ds, de := itr.Domain()
+	assert.Equal(t, start, ds, "checkDomain domain start incorrect")
+	assert.Equal(t, end, de, "checkDomain domain end incorrect")
+}
+
+func AssertItem(t *testing.T, itr dbm.Iterator, key, value []byte) {
+	t.Helper()
+	assert.Exactly(t, itr.Key(), k)
+	assert.Exactly(t, itr.Value(), v)
+}
+
+func AssertInvalid(t *testing.T, itr dbm.Iterator) {
+	t.Helper()
+	AssertNext(t, itr, false)
+	AssertKeyPanics(t, itr)
+	AssertValuePanics(t, itr)
+}
+
+func AssertKeyPanics(t *testing.T, itr dbm.Iterator) {
+	t.Helper()
+	assert.Panics(t, func() { itr.Key() }, "checkKeyPanics expected panic but didn't")
+}
+
+func AssertValue(t *testing.T, db dbm.DBReader, key, valueWanted []byte) {
+	t.Helper()
+	valueGot, err := db.Get(key)
+	assert.NoError(t, err)
+	assert.Equal(t, valueWanted, valueGot)
+}
+
+func AssertValuePanics(t *testing.T, itr dbm.Iterator) {
+	t.Helper()
+	assert.Panics(t, func() { itr.Value() })
+}

db/internal/util.go

@@ -18,7 +18,7 @@ func ValidateKv(key, value []byte) error {
 func CombineErrors(ret error, also error, desc string) error {
 	if also != nil {
 		if ret != nil {
-			ret = fmt.Errorf("%w; %v: %v", ret, desc, also)
+			ret = fmt.Errorf("%w; %s: %v", ret, desc, also)
 		} else {
 			ret = also
 		}

db/prefix/prefix.go

@@ -0,0 +1,159 @@
+package prefix
+
+import (
+	dbm "github.com/cosmos/cosmos-sdk/db"
+)
+
+// Prefix Reader/Writer lets you namespace multiple DBs within a single DB.
+type prefixR struct {
+	db     dbm.DBReader
+	prefix []byte
+}
+
+type prefixRW struct {
+	db     dbm.DBReadWriter
+	prefix []byte
+}
+
+var _ dbm.DBReader = (*prefixR)(nil)
+var _ dbm.DBReadWriter = (*prefixRW)(nil)
+
+func NewPrefixReader(db dbm.DBReader, prefix []byte) prefixR {
+	return prefixR{
+		prefix: prefix,
+		db:     db,
+	}
+}
+
+func NewPrefixReadWriter(db dbm.DBReadWriter, prefix []byte) prefixRW {
+	return prefixRW{
+		prefix: prefix,
+		db:     db,
+	}
+}
+
+func prefixed(prefix, key []byte) []byte {
+	return append(prefix, key...)
+}
+
+// Get implements DBReader.
+func (pdb prefixR) Get(key []byte) ([]byte, error) {
+	if len(key) == 0 {
+		return nil, dbm.ErrKeyEmpty
+	}
+	return pdb.db.Get(prefixed(pdb.prefix, key))
+}
+
+// Has implements DBReader.
+func (pdb prefixR) Has(key []byte) (bool, error) {
+	if len(key) == 0 {
+		return false, dbm.ErrKeyEmpty
+	}
+	return pdb.db.Has(prefixed(pdb.prefix, key))
+}
+
+// Iterator implements DBReader.
+func (pdb prefixR) Iterator(start, end []byte) (dbm.Iterator, error) {
+	if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) {
+		return nil, dbm.ErrKeyEmpty
+	}
+
+	var pend []byte
+	if end == nil {
+		pend = cpIncr(pdb.prefix)
+	} else {
+		pend = prefixed(pdb.prefix, end)
+	}
+	itr, err := pdb.db.Iterator(prefixed(pdb.prefix, start), pend)
+	if err != nil {
+		return nil, err
+	}
+	return newPrefixIterator(pdb.prefix, start, end, itr), nil
+}
+
+// ReverseIterator implements DBReader.
+func (pdb prefixR) ReverseIterator(start, end []byte) (dbm.Iterator, error) {
+	if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) {
+		return nil, dbm.ErrKeyEmpty
+	}
+
+	var pend []byte
+	if end == nil {
+		pend = cpIncr(pdb.prefix)
+	} else {
+		pend = prefixed(pdb.prefix, end)
+	}
+	ritr, err := pdb.db.ReverseIterator(prefixed(pdb.prefix, start), pend)
+	if err != nil {
+		return nil, err
+	}
+	return newPrefixIterator(pdb.prefix, start, end, ritr), nil
+}
+
+// Discard implements DBReader.
+func (pdb prefixR) Discard() error { return pdb.db.Discard() }
+
+// Set implements DBReadWriter.
+func (pdb prefixRW) Set(key []byte, value []byte) error {
+	if len(key) == 0 {
+		return dbm.ErrKeyEmpty
+	}
+	return pdb.db.Set(prefixed(pdb.prefix, key), value)
+}
+
+// Delete implements DBReadWriter.
+func (pdb prefixRW) Delete(key []byte) error {
+	if len(key) == 0 {
+		return dbm.ErrKeyEmpty
+	}
+	return pdb.db.Delete(prefixed(pdb.prefix, key))
+}
+
+// Get implements DBReadWriter.
+func (pdb prefixRW) Get(key []byte) ([]byte, error) {
+	return NewPrefixReader(pdb.db, pdb.prefix).Get(key)
+}
+
+// Has implements DBReadWriter.
+func (pdb prefixRW) Has(key []byte) (bool, error) {
+	return NewPrefixReader(pdb.db, pdb.prefix).Has(key)
+}
+
+// Iterator implements DBReadWriter.
+func (pdb prefixRW) Iterator(start, end []byte) (dbm.Iterator, error) {
+	return NewPrefixReader(pdb.db, pdb.prefix).Iterator(start, end)
+}
+
+// ReverseIterator implements DBReadWriter.
+func (pdb prefixRW) ReverseIterator(start, end []byte) (dbm.Iterator, error) {
+	return NewPrefixReader(pdb.db, pdb.prefix).ReverseIterator(start, end)
+}
+
+// Close implements DBReadWriter.
+func (pdb prefixRW) Commit() error { return pdb.db.Commit() }
+
+// Discard implements DBReadWriter.
+func (pdb prefixRW) Discard() error { return pdb.db.Discard() }
+
+// Returns a slice of the same length (big endian), but incremented by one.
+// Returns nil on overflow (e.g. if bz bytes are all 0xFF)
+// CONTRACT: len(bz) > 0
+func cpIncr(bz []byte) (ret []byte) {
+	if len(bz) == 0 {
+		panic("cpIncr expects non-zero bz length")
+	}
+	ret = make([]byte, len(bz))
+	copy(ret, bz)
+	for i := len(bz) - 1; i >= 0; i-- {
+		if ret[i] < byte(0xFF) {
+			ret[i]++
+			return
+		}
+		ret[i] = byte(0x00)
+		if i == 0 {
+			// Overflow
+			return nil
+		}
+	}
+	return nil
+}

db/prefix/prefix_iterator.go

@@ -0,0 +1,112 @@
+package prefix
+
+import (
+	"bytes"
+	"fmt"
+
+	dbm "github.com/cosmos/cosmos-sdk/db"
+)
+
+// IteratePrefix is a convenience function for iterating over a key domain
+// restricted by prefix.
+func IteratePrefix(db dbm.DBReader, prefix []byte) (dbm.Iterator, error) {
+	var start, end []byte
+	if len(prefix) != 0 {
+		start = prefix
+		end = cpIncr(prefix)
+	}
+	itr, err := db.Iterator(start, end)
+	if err != nil {
+		return nil, err
+	}
+	return itr, nil
+}
+
+// Strips prefix while iterating from Iterator.
+type prefixDBIterator struct {
+	prefix []byte
+	start  []byte
+	end    []byte
+	source dbm.Iterator
+	err    error
+}
+
+var _ dbm.Iterator = (*prefixDBIterator)(nil)
+
+func newPrefixIterator(prefix, start, end []byte, source dbm.Iterator) *prefixDBIterator {
+	return &prefixDBIterator{
+		prefix: prefix,
+		start:  start,
+		end:    end,
+		source: source,
+	}
+}
+
+// Domain implements Iterator.
+func (itr *prefixDBIterator) Domain() (start, end []byte) {
+	return itr.start, itr.end
+}
+
+func (itr *prefixDBIterator) valid() bool {
+	if itr.err != nil {
+		return false
+	}
+
+	key := itr.source.Key()
+	if len(key) < len(itr.prefix) || !bytes.Equal(key[:len(itr.prefix)], itr.prefix) {
+		itr.err = fmt.Errorf("received invalid key from backend: %x (expected prefix %x)",
+			key, itr.prefix)
+		return false
+	}
+
+	return true
+}
+
+// Next implements Iterator.
+func (itr *prefixDBIterator) Next() bool {
+	if !itr.source.Next() {
+		return false
+	}
+	key := itr.source.Key()
+	if !bytes.HasPrefix(key, itr.prefix) {
+		return false
+	}
+	// Empty keys are not allowed, so if a key exists in the database that exactly matches the
+	// prefix we need to skip it.
+	if bytes.Equal(key, itr.prefix) {
+		return itr.Next()
+	}
+	return true
+}
+
+// Next implements Iterator.
+func (itr *prefixDBIterator) Key() []byte {
+	itr.assertIsValid()
+	key := itr.source.Key()
+	return key[len(itr.prefix):] // we have checked the key in Valid()
+}
+
+// Value implements Iterator.
+func (itr *prefixDBIterator) Value() []byte {
+	itr.assertIsValid()
+	return itr.source.Value()
+}
+
+// Error implements Iterator.
+func (itr *prefixDBIterator) Error() error {
+	if err := itr.source.Error(); err != nil {
+		return err
+	}
+	return itr.err
+}
+
+// Close implements Iterator.
+func (itr *prefixDBIterator) Close() error {
+	return itr.source.Close()
+}
+
+func (itr *prefixDBIterator) assertIsValid() {
+	if !itr.valid() {
+		panic("iterator is invalid")
+	}
+}