Fix DeleteAll for CacheKV/MultiVersionKV so that all keys are properl…

…y deleted (#464)

## Describe your changes and provide context
If we have a cacheKV that has a dirty cache of 1->a and a parent with an
entry 2->b, the previous implementation of `DeleteAll` would only call
cacheKV.Delete on key 1 but not affecting 2 at all (i.e. if someone
queries key 2 on the cacheKV they would still get `b`, but the
expectation is for 2 to be deleted as appearing on the cacheKV level).
This PR fixes that by first getting all the keys to be (marked as)
deleted recursively and then delete all those keys one-by-one; note that
this recursion should be fairly efficient since it's not subject to the
exponential latency that merge iterators have.

This PR also fixed DeleteAll for `mvkv` so that it doesn't actually
touch parent store's value but only change the writeset (as a single
`Delete` would have done).

## Testing performed to validate your change
unit test

server/mock/store.go

@@ -234,6 +234,10 @@ func (kv kvStore) DeleteAll(start, end []byte) error {
 	panic("not implemented")
 }
 
+func (kv kvStore) GetAllKeyStrsInRange(start, end []byte) []string {
+	panic("not implemented")
+}
+
 func NewCommitMultiStore() sdk.CommitMultiStore {
 	return multiStore{kv: make(map[sdk.StoreKey]kvStore)}
 }

store/cachekv/store.go

@@ -369,16 +369,28 @@ func (store *Store) GetParent() types.KVStore {
 }
 
 func (store *Store) DeleteAll(start, end []byte) error {
-	store.dirtyItems(start, end)
-	// memdb iterator
-	cachedIter, err := store.sortedCache.Iterator(start, end)
-	if err != nil {
-		return err
-	}
-	defer cachedIter.Close()
-	for ; cachedIter.Valid(); cachedIter.Next() {
-		// `Delete` would not touch sortedCache so it's okay to perform inside iterator
-		store.Delete(cachedIter.Key())
+	for _, k := range store.GetAllKeyStrsInRange(start, end) {
+		store.Delete([]byte(k))
 	}
 	return nil
 }
+
+func (store *Store) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	keyStrs := map[string]struct{}{}
+	for _, pk := range store.parent.GetAllKeyStrsInRange(start, end) {
+		keyStrs[pk] = struct{}{}
+	}
+	store.cache.Range(func(key, value any) bool {
+		cv := value.(*types.CValue)
+		if cv.Value() == nil {
+			delete(keyStrs, key.(string))
+		} else {
+			keyStrs[key.(string)] = struct{}{}
+		}
+		return true
+	})
+	for k := range keyStrs {
+		res = append(res, k)
+	}
+	return res
+}

store/cachekv/store_test.go

@@ -67,14 +67,17 @@ func TestCacheKVStore(t *testing.T) {
 	require.NotNil(t, st.GetParent())
 
 	// DeleteAll deletes all entries in cache but not affect mem
-	st = cachekv.NewStore(mem, types.NewKVStoreKey("CacheKvTest"), types.DefaultCacheSizeLimit)
 	mem.Set(keyFmt(1), valFmt(1))
+	mem.Set(keyFmt(3), valFmt(4))
+	st = cachekv.NewStore(mem, types.NewKVStoreKey("CacheKvTest"), types.DefaultCacheSizeLimit)
 	st.Set(keyFmt(1), valFmt(2))
 	st.Set(keyFmt(2), valFmt(3))
 	require.Nil(t, st.DeleteAll(nil, nil))
 	require.Nil(t, st.Get(keyFmt(1)))
 	require.Nil(t, st.Get(keyFmt(2)))
+	require.Nil(t, st.Get(keyFmt(3)))
 	require.Equal(t, valFmt(1), mem.Get(keyFmt(1)))
+	require.Equal(t, valFmt(4), mem.Get(keyFmt(3)))
 }
 
 func TestCacheKVStoreNoNilSet(t *testing.T) {

store/dbadapter/store.go

@@ -112,5 +112,14 @@ func (dsa Store) DeleteAll(start, end []byte) error {
 	return nil
 }
 
+func (dsa Store) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	iter := dsa.Iterator(start, end)
+	defer iter.Close()
+	for ; iter.Valid(); iter.Next() {
+		res = append(res, string(iter.Key()))
+	}
+	return
+}
+
 // dbm.DB implements KVStore so we can CacheKVStore it.
 var _ types.KVStore = Store{}

store/gaskv/store.go

@@ -134,6 +134,10 @@ func (gs *Store) DeleteAll(start, end []byte) error {
 	return gs.parent.DeleteAll(start, end)
 }
 
+func (gs *Store) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	return gs.parent.GetAllKeyStrsInRange(start, end)
+}
+
 type gasIterator struct {
 	gasMeter  types.GasMeter
 	gasConfig types.GasConfig

store/iavl/store.go

@@ -436,6 +436,15 @@ func (st *Store) DeleteAll(start, end []byte) error {
 	return nil
 }
 
+func (st *Store) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	iter := st.Iterator(start, end)
+	defer iter.Close()
+	for ; iter.Valid(); iter.Next() {
+		res = append(res, string(iter.Key()))
+	}
+	return
+}
+
 // Takes a MutableTree, a key, and a flag for creating existence or absence proof and returns the
 // appropriate merkle.Proof. Since this must be called after querying for the value, this function should never error
 // Thus, it will panic on error rather than returning it

store/listenkv/store.go

@@ -165,3 +165,7 @@ func (s *Store) onWrite(delete bool, key, value []byte) {
 func (s *Store) DeleteAll(start, end []byte) error {
 	return s.parent.DeleteAll(start, end)
 }
+
+func (s *Store) GetAllKeyStrsInRange(start, end []byte) []string {
+	return s.parent.GetAllKeyStrsInRange(start, end)
+}

store/multiversion/mvkv.go

@@ -316,7 +316,19 @@ func (v *VersionIndexedStore) VersionExists(version int64) bool {
 }
 
 func (v *VersionIndexedStore) DeleteAll(start, end []byte) error {
-	return v.parent.DeleteAll(start, end)
+	for _, k := range v.GetAllKeyStrsInRange(start, end) {
+		v.Delete([]byte(k))
+	}
+	return nil
+}
+
+func (v *VersionIndexedStore) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	iter := v.Iterator(start, end)
+	defer iter.Close()
+	for ; iter.Valid(); iter.Next() {
+		res = append(res, string(iter.Key()))
+	}
+	return
 }
 
 // GetStoreType implements types.KVStore.

store/multiversion/mvkv_test.go

@@ -474,3 +474,50 @@ func TestRemoveLastEntry(t *testing.T) {
 	require.False(t, valid)
 	require.Empty(t, conflicts)
 }
+
+func TestVersionIndexedStoreGetAllKeyStrsInRange(t *testing.T) {
+	mem := dbadapter.Store{DB: dbm.NewMemDB()}
+	parentKVStore := cachekv.NewStore(mem, types.NewKVStoreKey("mock"), 1000)
+	mvs := multiversion.NewMultiVersionStore(parentKVStore)
+	// initialize a new VersionIndexedStore
+	vis := multiversion.NewVersionIndexedStore(parentKVStore, mvs, 1, 2, make(chan scheduler.Abort, 1))
+
+	parentKVStore.Set([]byte("key1"), []byte("value1")) // unchanged
+	parentKVStore.Set([]byte("key2"), []byte("value2")) // updated by index 0 incar 1
+	parentKVStore.Set([]byte("key3"), []byte("value3")) // deleted by index 0 incar 1
+	parentKVStore.Set([]byte("key4"), []byte("value4")) // deleted by index 2
+	parentKVStore.Set([]byte("key5"), []byte("value5")) // updated by vis
+	parentKVStore.Set([]byte("key6"), []byte("value6")) // deleted by vis
+	mvs.SetWriteset(0, 1, map[string][]byte{
+		"key2": []byte("value8"),
+		"key3": nil,
+	})
+	mvs.SetWriteset(2, 1, map[string][]byte{
+		"key4": nil,
+	})
+	vis.Set([]byte("key5"), []byte("value7"))
+	vis.Delete([]byte("key6"))
+	require.Equal(t, []string{"key1", "key2", "key4", "key5"}, vis.GetAllKeyStrsInRange(nil, nil))
+}
+
+func TestVersionIndexedStoreGetAllKeyStrsInRangeError(t *testing.T) {
+	mem := dbadapter.Store{DB: dbm.NewMemDB()}
+	parentKVStore := cachekv.NewStore(mem, types.NewKVStoreKey("mock"), 1000)
+	mvs := multiversion.NewMultiVersionStore(parentKVStore)
+	// initialize a new VersionIndexedStore
+	vis2 := multiversion.NewVersionIndexedStore(parentKVStore, mvs, 2, 1, make(chan scheduler.Abort, 1))
+	vis3 := multiversion.NewVersionIndexedStore(parentKVStore, mvs, 3, 1, make(chan scheduler.Abort, 1))
+
+	parentKVStore.Set([]byte("key1"), []byte("value1"))
+	parentKVStore.Set([]byte("key2"), []byte("value2"))
+
+	mvs.SetWriteset(1, 1, map[string][]byte{
+		"key1": []byte("value3"),
+	})
+	vis2.Set([]byte("key3"), []byte("value4"))
+	vis3.DeleteAll(nil, nil)
+	vis2.WriteToMultiVersionStore()
+	vis3.WriteToMultiVersionStore()
+	valid, _ := mvs.ValidateTransactionState(3)
+	require.False(t, valid)
+}

store/prefix/store.go

@@ -102,6 +102,18 @@ func (s Store) DeleteAll(start, end []byte) error {
 	return s.parent.DeleteAll(newstart, newend)
 }
 
+func (s Store) GetAllKeyStrsInRange(start, end []byte) []string {
+	newstart := cloneAppend(s.prefix, start)
+
+	var newend []byte
+	if end == nil {
+		newend = cpIncr(s.prefix)
+	} else {
+		newend = cloneAppend(s.prefix, end)
+	}
+	return s.parent.GetAllKeyStrsInRange(newstart, newend)
+}
+
 // Implements KVStore
 // Check https://github.com/tendermint/tendermint/blob/master/libs/db/prefix_db.go#L106
 func (s Store) Iterator(start, end []byte) types.Iterator {

store/tracekv/store.go

@@ -190,6 +190,10 @@ func (tkv *Store) DeleteAll(start, end []byte) error {
 	return tkv.parent.DeleteAll(start, end)
 }
 
+func (tkv *Store) GetAllKeyStrsInRange(start, end []byte) []string {
+	return tkv.parent.GetAllKeyStrsInRange(start, end)
+}
+
 // writeOperation writes a KVStore operation to the underlying io.Writer as
 // JSON-encoded data where the key/value pair is base64 encoded.
 func writeOperation(w io.Writer, op operation, tc types.TraceContext, key, value []byte) {

store/types/store.go

@@ -257,6 +257,8 @@ type KVStore interface {
 	VersionExists(version int64) bool
 
 	DeleteAll(start, end []byte) error
+
+	GetAllKeyStrsInRange(start, end []byte) []string
 }
 
 // Iterator is an alias db's Iterator for convenience.

storev2/commitment/store.go

@@ -195,3 +195,12 @@ func (st *Store) DeleteAll(start, end []byte) error {
 	}
 	return nil
 }
+
+func (st *Store) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	iter := st.Iterator(start, end)
+	defer iter.Close()
+	for ; iter.Valid(); iter.Next() {
+		res = append(res, string(iter.Key()))
+	}
+	return
+}

storev2/state/store.go

@@ -147,3 +147,12 @@ func (st *Store) DeleteAll(start, end []byte) error {
 	}
 	return nil
 }
+
+func (st *Store) GetAllKeyStrsInRange(start, end []byte) (res []string) {
+	iter := st.Iterator(start, end)
+	defer iter.Close()
+	for ; iter.Valid(); iter.Next() {
+		res = append(res, string(iter.Key()))
+	}
+	return
+}