Fix edge case for iterating over tombstoned value

ss/pebbledb/iterator.go

@@ -63,9 +63,8 @@ func newPebbleDBIterator(src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte
 	if valid {
 		currKey, currKeyVersion, ok := SplitMVCCKey(itr.source.Key())
 		if !ok {
-			// XXX: This should not happen as that would indicate we have a malformed
-			// MVCC value.
-			panic(fmt.Sprintf("invalid PebbleDB MVCC value: %s", itr.source.Key()))
+			// XXX: This should not happen as that would indicate we have a malformed MVCC key.
+			panic(fmt.Sprintf("invalid PebbleDB MVCC key: %s", itr.source.Key()))
 		}
 
 		curKeyVersionDecoded, err := decodeUint64Ascending(currKeyVersion)
@@ -86,6 +85,15 @@ func newPebbleDBIterator(src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte
 		}
 	}
 
+	// Make sure we skip to the next key if the current one is tombstone
+	if valTombstoned(itr.source.Value()) {
+		if reverse {
+			itr.nextReverse()
+		} else {
+			itr.nextForward()
+		}
+	}
+
 	return itr
 }
 

ss/test/storage_test_suite.go

@@ -1,15 +1,13 @@
 package sstest
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/cosmos/iavl"
+	"github.com/sei-protocol/sei-db/ss/types"
 	"github.com/stretchr/testify/suite"
 	"golang.org/x/exp/slices"
-
-	"fmt"
-
-	"github.com/sei-protocol/sei-db/ss/types"
 )
 
 const (
@@ -305,7 +303,7 @@ func (s *StorageTestSuite) TestDatabaseIterator() {
 		s.Require().False(itr.Valid())
 	}
 
-	// iterator with with a start and end domain over multiple versions
+	// iterator with a start and end domain over multiple versions
 	for v := int64(1); v < 5; v++ {
 		itr2, err := db.Iterator(storeKey1, v, []byte("key010"), []byte("key019"))
 		s.Require().NoError(err)
@@ -333,7 +331,6 @@ func (s *StorageTestSuite) TestDatabaseIterator() {
 	s.Require().Error(err)
 	s.Require().Nil(iter3)
 }
-
 func (s *StorageTestSuite) TestDatabaseIteratorRangedDeletes() {
 	db, err := s.NewDB(s.T().TempDir())
 	s.Require().NoError(err)
@@ -359,6 +356,42 @@ func (s *StorageTestSuite) TestDatabaseIteratorRangedDeletes() {
 	s.Require().NoError(itr.Error())
 }
 
+func (s *StorageTestSuite) TestDatabaseIteratorDeletes() {
+	db, err := s.NewDB(s.T().TempDir())
+	s.Require().NoError(err)
+	defer db.Close()
+
+	s.Require().NoError(DBApplyChangeset(db, 1, storeKey1, [][]byte{[]byte("key001")}, [][]byte{[]byte("value001")}))
+	s.Require().NoError(DBApplyChangeset(db, 1, storeKey1, [][]byte{[]byte("key002")}, [][]byte{[]byte("value002")}))
+	s.Require().NoError(DBApplyDeleteChangeset(db, 5, storeKey1, [][]byte{[]byte("key001")}))
+
+	itr, err := db.Iterator(storeKey1, 11, []byte("key001"), nil)
+	s.Require().NoError(err)
+
+	// there should be only one valid key in the iterator
+	var count = 0
+	for ; itr.Valid(); itr.Next() {
+		s.Require().Equal([]byte("key002"), itr.Key())
+		count++
+	}
+	s.Require().Equal(1, count)
+	s.Require().NoError(itr.Error())
+	itr.Close()
+
+	s.Require().NoError(DBApplyChangeset(db, 10, storeKey1, [][]byte{[]byte("key001")}, [][]byte{[]byte("value001")}))
+	itr, err = db.Iterator(storeKey1, 11, []byte("key001"), nil)
+	s.Require().NoError(err)
+
+	// there should be two valid keys in the iterator
+	count = 0
+	for ; itr.Valid(); itr.Next() {
+		count++
+	}
+	s.Require().Equal(2, count)
+	s.Require().NoError(itr.Error())
+	itr.Close()
+}
+
 func (s *StorageTestSuite) TestDatabaseIteratorMultiVersion() {
 	db, err := s.NewDB(s.T().TempDir())
 	s.Require().NoError(err)

ss/test/utils.go

@@ -59,3 +59,19 @@ func DBApplyChangeset(db types.StateStore, version int64, storeKey string, key,
 
 	return db.ApplyChangeset(version, ncs)
 }
+
+// Helper for creating the changeset and applying it to db
+func DBApplyDeleteChangeset(db types.StateStore, version int64, storeKey string, key [][]byte) error {
+	cs := &iavl.ChangeSet{}
+	cs.Pairs = []*iavl.KVPair{}
+	for j := 0; j < len(key); j++ {
+		cs.Pairs = append(cs.Pairs, &iavl.KVPair{Key: key[j], Delete: true})
+	}
+
+	ncs := &proto.NamedChangeSet{
+		Name:      storeKey,
+		Changeset: *cs,
+	}
+
+	return db.ApplyChangeset(version, ncs)
+}