feat(store/v2): add the catch up process in the migration (#19454)

store/batch.go

@@ -13,7 +13,7 @@ type Batch interface {
 	Write() error
 
 	// Reset resets the batch.
-	Reset()
+	Reset() error
 }
 
 // RawBatch represents a group of writes. They may or may not be written atomically depending on the

store/internal/encoding/changeset.go

@@ -0,0 +1,125 @@
+package encoding
+
+import (
+	"bytes"
+	"fmt"
+
+	corestore "cosmossdk.io/core/store"
+)
+
+// encodedSize returns the size of the encoded Changeset.
+func encodedSize(cs *corestore.Changeset) int {
+	size := EncodeUvarintSize(uint64(len(cs.Changes)))
+	for _, changes := range cs.Changes {
+		size += EncodeBytesSize(changes.Actor)
+		size += EncodeUvarintSize(uint64(len(changes.StateChanges)))
+		for _, pair := range changes.StateChanges {
+			size += EncodeBytesSize(pair.Key)
+			size += EncodeUvarintSize(1) // pair.Remove
+			if !pair.Remove {
+				size += EncodeBytesSize(pair.Value)
+			}
+		}
+	}
+	return size
+}
+
+// MarshalChangeset returns the encoded byte representation of Changeset.
+// NOTE: The Changeset is encoded as follows:
+// - number of store keys (uvarint)
+// - for each store key:
+// -- store key (bytes)
+// -- number of pairs (uvarint)
+// -- for each pair:
+// --- key (bytes)
+// --- remove (1 byte)
+// --- value (bytes)
+func MarshalChangeset(cs *corestore.Changeset) ([]byte, error) {
+	var buf bytes.Buffer
+	buf.Grow(encodedSize(cs))
+
+	if err := EncodeUvarint(&buf, uint64(len(cs.Changes))); err != nil {
+		return nil, err
+	}
+	for _, changes := range cs.Changes {
+		if err := EncodeBytes(&buf, changes.Actor); err != nil {
+			return nil, err
+		}
+		if err := EncodeUvarint(&buf, uint64(len(changes.StateChanges))); err != nil {
+			return nil, err
+		}
+		for _, pair := range changes.StateChanges {
+			if err := EncodeBytes(&buf, pair.Key); err != nil {
+				return nil, err
+			}
+			if pair.Remove {
+				if err := EncodeUvarint(&buf, 1); err != nil {
+					return nil, err
+				}
+			} else {
+				if err := EncodeUvarint(&buf, 0); err != nil {
+					return nil, err
+				}
+				if err := EncodeBytes(&buf, pair.Value); err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+
+	return buf.Bytes(), nil
+}
+
+// UnmarshalChangeset decodes the Changeset from the given byte slice.
+func UnmarshalChangeset(cs *corestore.Changeset, buf []byte) error {
+	storeCount, n, err := DecodeUvarint(buf)
+	if err != nil {
+		return err
+	}
+	buf = buf[n:]
+	changes := make([]corestore.StateChanges, storeCount)
+	for i := uint64(0); i < storeCount; i++ {
+		storeKey, n, err := DecodeBytes(buf)
+		if err != nil {
+			return err
+		}
+		buf = buf[n:]
+
+		pairCount, n, err := DecodeUvarint(buf)
+		if err != nil {
+			return err
+		}
+		buf = buf[n:]
+
+		pairs := make([]corestore.KVPair, pairCount)
+		for j := uint64(0); j < pairCount; j++ {
+			pairs[j].Key, n, err = DecodeBytes(buf)
+			if err != nil {
+				return err
+			}
+			buf = buf[n:]
+
+			remove, n, err := DecodeUvarint(buf)
+			if err != nil {
+				return err
+			}
+			buf = buf[n:]
+			if remove == 0 {
+				pairs[j].Remove = false
+				pairs[j].Value, n, err = DecodeBytes(buf)
+				if err != nil {
+					return err
+				}
+				buf = buf[n:]
+			} else if remove == 1 {
+				pairs[j].Remove = true
+			} else {
+				return fmt.Errorf("invalid remove flag: %d", remove)
+			}
+		}
+		changes[i] = corestore.StateChanges{Actor: storeKey, StateChanges: pairs}
+	}
+	cs.Changes = changes
+
+	return nil
+}

store/internal/encoding/changeset_test.go

@@ -0,0 +1,94 @@
+package encoding
+
+import (
+	"testing"
+
+	corestore "cosmossdk.io/core/store"
+	"github.com/stretchr/testify/require"
+)
+
+func TestChangesetMarshal(t *testing.T) {
+	testcases := []struct {
+		name         string
+		changeset    *corestore.Changeset
+		encodedSize  int
+		encodedBytes []byte
+	}{
+		{
+			name:         "empty",
+			changeset:    corestore.NewChangeset(),
+			encodedSize:  1,
+			encodedBytes: []byte{0x0},
+		},
+		{
+			name: "one store",
+			changeset: &corestore.Changeset{Changes: []corestore.StateChanges{
+				{
+					Actor: []byte("storekey"),
+					StateChanges: corestore.KVPairs{
+						{Key: []byte("key"), Value: []byte("value"), Remove: false},
+					},
+				},
+			}},
+			encodedSize:  1 + 1 + 8 + 1 + 1 + 3 + 1 + 1 + 5,
+			encodedBytes: []byte{0x1, 0x8, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x6b, 0x65, 0x79, 0x1, 0x3, 0x6b, 0x65, 0x79, 0x0, 0x5, 0x76, 0x61, 0x6c, 0x75, 0x65},
+		},
+		{
+			name: "one remove store",
+			changeset: &corestore.Changeset{Changes: []corestore.StateChanges{
+				{
+					Actor: []byte("storekey"),
+					StateChanges: corestore.KVPairs{
+						{Key: []byte("key"), Remove: true},
+					},
+				},
+			}},
+			encodedSize:  1 + 1 + 8 + 1 + 1 + 3 + 1,
+			encodedBytes: []byte{0x1, 0x8, 0x73, 0x74, 0x6f, 0x72, 0x65, 0x6b, 0x65, 0x79, 0x1, 0x3, 0x6b, 0x65, 0x79, 0x1},
+		},
+		{
+			name: "two stores",
+			changeset: &corestore.Changeset{Changes: []corestore.StateChanges{
+				{
+					Actor: []byte("storekey1"),
+					StateChanges: corestore.KVPairs{
+						{Key: []byte("key1"), Value: []byte("value1"), Remove: false},
+					},
+				},
+				{
+					Actor: []byte("storekey2"),
+					StateChanges: corestore.KVPairs{
+						{Key: []byte("key2"), Value: []byte("value2"), Remove: false},
+						{Key: []byte("key1"), Remove: true},
+					},
+				},
+			}},
+			encodedSize: 2 + 1 + 9 + 1 + 1 + 4 + 1 + 6 + 1 + 9 + 1 + 1 + 4 + 1 + 1 + 6 + 1 + 4 + 1,
+			// encodedBytes: it is not deterministic,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			// check the encoded size
+			require.Equal(t, encodedSize(tc.changeset), tc.encodedSize, "encoded size mismatch")
+			// check the encoded bytes
+			encodedBytes, err := MarshalChangeset(tc.changeset)
+			require.NoError(t, err, "marshal error")
+			if len(tc.encodedBytes) != 0 {
+				require.Equal(t, encodedBytes, tc.encodedBytes, "encoded bytes mismatch")
+			}
+			// check the unmarshaled changeset
+			cs := corestore.NewChangeset()
+			require.NoError(t, UnmarshalChangeset(cs, encodedBytes), "unmarshal error")
+			require.Equal(t, len(tc.changeset.Changes), len(cs.Changes), "unmarshaled changeset store size mismatch")
+			for i, changes := range tc.changeset.Changes {
+				require.Equal(t, changes.Actor, cs.Changes[i].Actor, "unmarshaled changeset store key mismatch")
+				require.Equal(t, len(changes.StateChanges), len(cs.Changes[i].StateChanges), "unmarshaled changeset StateChanges size mismatch")
+				for j, pair := range changes.StateChanges {
+					require.Equal(t, pair, cs.Changes[i].StateChanges[j], "unmarshaled changeset pair mismatch")
+				}
+			}
+		})
+	}
+}