From ff31990821af63318f5b53a0abab765d9c04aec4 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Mon, 13 Nov 2023 12:22:17 -0800
Subject: [PATCH 01/19] updates

---
 store/storage/rocksdb/db.go         | 52 ++++++++---------------------
 store/storage/storage_test_suite.go | 42 +++++++++++++++++++++++
 2 files changed, 55 insertions(+), 39 deletions(-)

diff --git a/store/storage/rocksdb/db.go b/store/storage/rocksdb/db.go
index b7b120948763..4d18bdd56fec 100644
--- a/store/storage/rocksdb/db.go
+++ b/store/storage/rocksdb/db.go
@@ -7,6 +7,7 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"strings"
 
 	"github.com/linxGnu/grocksdb"
 	"golang.org/x/exp/slices"
@@ -32,11 +33,6 @@ var (
 type Database struct {
 	storage  *grocksdb.DB
 	cfHandle *grocksdb.ColumnFamilyHandle
-
-	// tsLow reflects the full_history_ts_low CF value. Since pruning is done in
-	// a lazy manner, we use this value to prevent reads for versions that will
-	// be purged in the next compaction.
-	tsLow uint64
 }
 
 func New(dataDir string) (*Database, error) {
@@ -45,39 +41,16 @@ func New(dataDir string) (*Database, error) {
 		return nil, fmt.Errorf("failed to open RocksDB: %w", err)
 	}
 
-	slice, err := storage.GetFullHistoryTsLow(cfHandle)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get full_history_ts_low: %w", err)
-	}
-
-	var tsLow uint64
-	tsLowBz := copyAndFreeSlice(slice)
-	if len(tsLowBz) > 0 {
-		tsLow = binary.LittleEndian.Uint64(tsLowBz)
-	}
-
 	return &Database{
 		storage:  storage,
 		cfHandle: cfHandle,
-		tsLow:    tsLow,
 	}, nil
 }
 
 func NewWithDB(storage *grocksdb.DB, cfHandle *grocksdb.ColumnFamilyHandle) (*Database, error) {
-	slice, err := storage.GetFullHistoryTsLow(cfHandle)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get full_history_ts_low: %w", err)
-	}
-
-	var tsLow uint64
-	tsLowBz := copyAndFreeSlice(slice)
-	if len(tsLowBz) > 0 {
-		tsLow = binary.LittleEndian.Uint64(tsLowBz)
-	}
 	return &Database{
 		storage:  storage,
 		cfHandle: cfHandle,
-		tsLow:    tsLow,
 	}, nil
 }
 
@@ -91,11 +64,18 @@ func (db *Database) Close() error {
 }
 
 func (db *Database) getSlice(storeKey string, version uint64, key []byte) (*grocksdb.Slice, error) {
-	return db.storage.GetCF(
+	slice, err := db.storage.GetCF(
 		newTSReadOptions(version),
 		db.cfHandle,
 		prependStoreKey(storeKey, key),
 	)
+	if err != nil && strings.Contains(err.Error(), "is smaller than full_history_ts_low") {
+		// In cases where we query for a version that has been pruned, we need to ensure
+		// we do not return an error.
+		return nil, nil
+	}
+
+	return slice, nil
 }
 
 func (db *Database) SetLatestVersion(version uint64) error {
@@ -120,12 +100,8 @@ func (db *Database) GetLatestVersion() (uint64, error) {
 }
 
 func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, error) {
-	if version < db.tsLow {
-		return false, nil
-	}
-
 	slice, err := db.getSlice(storeKey, version, key)
-	if err != nil {
+	if err != nil || slice == nil {
 		return false, err
 	}
 
@@ -133,14 +109,13 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 }
 
 func (db *Database) Get(storeKey string, version uint64, key []byte) ([]byte, error) {
-	if version < db.tsLow {
-		return nil, nil
-	}
-
 	slice, err := db.getSlice(storeKey, version, key)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get RocksDB slice: %w", err)
 	}
+	if slice == nil {
+		return nil, nil
+	}
 
 	return copyAndFreeSlice(slice), nil
 }
@@ -181,7 +156,6 @@ func (db *Database) Prune(version uint64) error {
 		return fmt.Errorf("failed to update column family full_history_ts_low: %w", err)
 	}
 
-	db.tsLow = tsLow
 	return nil
 }
 
diff --git a/store/storage/storage_test_suite.go b/store/storage/storage_test_suite.go
index 72fac87387ab..546da35fd472 100644
--- a/store/storage/storage_test_suite.go
+++ b/store/storage/storage_test_suite.go
@@ -483,3 +483,45 @@ func (s *StorageTestSuite) TestDatabase_Prune() {
 		}
 	}
 }
+
+func (s *StorageTestSuite) TestDatabase_Prune_KeepRecent() {
+	if slices.Contains(s.SkipTests, s.T().Name()) {
+		s.T().SkipNow()
+	}
+
+	db, err := s.NewDB(s.T().TempDir())
+	s.Require().NoError(err)
+	defer db.Close()
+
+	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")})))
+
+	// prune at version 50
+	s.Require().NoError(db.Prune(50))
+
+	// ensure queries for versions 50 and older return nil
+	bz, err := db.Get(storeKey1, 49, key)
+	s.Require().NoError(err)
+	s.Require().Nil(bz)
+
+	// ensure the value previously at version 1 is still there for queries greater than 50
+	bz, err = db.Get(storeKey1, 51, key)
+	s.Require().NoError(err)
+	s.Require().Equal([]byte("val001"), bz)
+
+	// ensure the correct version at a greater height
+	bz, err = db.Get(storeKey1, 200, key)
+	s.Require().NoError(err)
+	s.Require().Equal([]byte("val200"), bz)
+
+	// prune latest height and ensure we have the previous version when querying above it
+	s.Require().NoError(db.Prune(200))
+
+	bz, err = db.Get(storeKey1, 201, key)
+	s.Require().NoError(err)
+	s.Require().Equal([]byte("val200"), bz)
+}

From b93cbcf94276aa84f66b8bdd8c7d9497b506495f Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Tue, 14 Nov 2023 10:00:40 -0800
Subject: [PATCH 02/19] updates

---
 store/storage/pebbledb/db_test.go   | 1 +
 store/storage/sqlite/db.go          | 9 ++++++++-
 store/storage/storage_test_suite.go | 4 ++--
 3 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/store/storage/pebbledb/db_test.go b/store/storage/pebbledb/db_test.go
index dde316bf7f12..f078e54751e8 100644
--- a/store/storage/pebbledb/db_test.go
+++ b/store/storage/pebbledb/db_test.go
@@ -17,6 +17,7 @@ func TestStorageTestSuite(t *testing.T) {
 		EmptyBatchSize: 12,
 		SkipTests: []string{
 			"TestStorageTestSuite/TestDatabase_Prune",
+			"TestStorageTestSuite/TestDatabase_Prune_KeepRecent",
 		},
 	}
 
diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index ae369fa03586..d5b6bb004b5d 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -175,7 +175,14 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
 }
 
 func (db *Database) Prune(version uint64) error {
-	stmt := "DELETE FROM state_storage WHERE version <= ? AND store_key != ?;"
+	stmt := `DELETE FROM state_storage
+	WHERE version < (
+		SELECT max(version) FROM state_storage t2 WHERE
+		t2.store_key = state_storage.store_key AND
+		t2.key = state_storage.key AND
+		t2.version <= ?
+	) AND store_key != ?;
+	`
 
 	_, err := db.storage.Exec(stmt, version, reservedStoreKey)
 	if err != nil {
diff --git a/store/storage/storage_test_suite.go b/store/storage/storage_test_suite.go
index 546da35fd472..6da929683342 100644
--- a/store/storage/storage_test_suite.go
+++ b/store/storage/storage_test_suite.go
@@ -500,7 +500,7 @@ func (s *StorageTestSuite) TestDatabase_Prune_KeepRecent() {
 	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")})))
 
-	// prune at version 50
+	// prune version 50
 	s.Require().NoError(db.Prune(50))
 
 	// ensure queries for versions 50 and older return nil
@@ -513,7 +513,7 @@ func (s *StorageTestSuite) TestDatabase_Prune_KeepRecent() {
 	s.Require().NoError(err)
 	s.Require().Equal([]byte("val001"), bz)
 
-	// ensure the correct version at a greater height
+	// ensure the correct value at a greater height
 	bz, err = db.Get(storeKey1, 200, key)
 	s.Require().NoError(err)
 	s.Require().Equal([]byte("val200"), bz)

From 20a687d69d6db2e9785eec9b48f77e3099af9bb8 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Tue, 14 Nov 2023 10:44:52 -0800
Subject: [PATCH 03/19] updates

---
 store/storage/sqlite/db.go | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index d5b6bb004b5d..88febb7a9650 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -175,6 +175,9 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
 }
 
 func (db *Database) Prune(version uint64) error {
+	// delete all versions less than or equal to the target version, except we keep
+	// the last one to handle above the prune version. This is analogous to RocksDB
+	// full_history_ts_low.
 	stmt := `DELETE FROM state_storage
 	WHERE version < (
 		SELECT max(version) FROM state_storage t2 WHERE

From a4e2e6e2fc294996afdcfeef1f586185e8f6c7ac Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Tue, 14 Nov 2023 13:49:07 -0800
Subject: [PATCH 04/19] updates

---
 store/storage/pebbledb/db.go  |  3 ---
 store/storage/sqlite/batch.go |  2 +-
 store/storage/sqlite/db.go    | 32 +++++++++++++++++++++++++-------
 3 files changed, 26 insertions(+), 11 deletions(-)

diff --git a/store/storage/pebbledb/db.go b/store/storage/pebbledb/db.go
index 2f3902c09f90..b4240c97ca77 100644
--- a/store/storage/pebbledb/db.go
+++ b/store/storage/pebbledb/db.go
@@ -126,9 +126,6 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 	if err != nil {
 		return nil, fmt.Errorf("failed to decode value tombstone: %w", err)
 	}
-	if tombstone > targetVersion {
-		return nil, fmt.Errorf("value tombstone too large: %d", tombstone)
-	}
 
 	// A tombstone of zero or a target version that is less than the tombstone
 	// version means the key is not deleted at the target version.
diff --git a/store/storage/sqlite/batch.go b/store/storage/sqlite/batch.go
index 588ad5129571..82e8f3e5b306 100644
--- a/store/storage/sqlite/batch.go
+++ b/store/storage/sqlite/batch.go
@@ -65,7 +65,7 @@ func (b *Batch) Delete(storeKey string, key []byte) error {
 }
 
 func (b *Batch) Write() error {
-	_, err := b.tx.Exec(latestVersionStmt, reservedStoreKey, keyLatestHeight, b.version, 0, b.version)
+	_, err := b.tx.Exec(reservedUpsertStmt, reservedStoreKey, keyLatestHeight, b.version, 0, b.version)
 	if err != nil {
 		return fmt.Errorf("failed to exec SQL statement: %w", err)
 	}
diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index 88febb7a9650..957ed1940d4b 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -18,8 +18,9 @@ const (
 	dbName           = "file:ss.db?cache=shared&mode=rwc&_journal_mode=WAL"
 	reservedStoreKey = "_RESERVED_"
 	keyLatestHeight  = "latest_height"
+	keyPruneHeight   = "prune_height"
 
-	latestVersionStmt = `
+	reservedUpsertStmt = `
 	INSERT INTO state_storage(store_key, key, value, version)
     VALUES(?, ?, ?, ?)
   ON CONFLICT(store_key, key, version) DO UPDATE SET
@@ -102,7 +103,7 @@ func (db *Database) GetLatestVersion() (uint64, error) {
 }
 
 func (db *Database) SetLatestVersion(version uint64) error {
-	_, err := db.storage.Exec(latestVersionStmt, reservedStoreKey, keyLatestHeight, version, 0, version)
+	_, err := db.storage.Exec(reservedUpsertStmt, reservedStoreKey, keyLatestHeight, version, 0, version)
 	if err != nil {
 		return fmt.Errorf("failed to exec SQL statement: %w", err)
 	}
@@ -122,7 +123,9 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
 	stmt, err := db.storage.Prepare(`
 	SELECT value, tombstone FROM state_storage
-	WHERE store_key = ? AND key = ? AND version <= ?
+	WHERE store_key = ? AND key = ? AND version <= ? AND (
+		SELECT CAST(value as INTEGER) from state_storage WHERE store_key = ? AND key = ?
+	) < ?
 	ORDER BY version DESC LIMIT 1;
 	`)
 	if err != nil {
@@ -135,7 +138,7 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 		value []byte
 		tomb  uint64
 	)
-	if err := stmt.QueryRow(storeKey, key, targetVersion).Scan(&value, &tomb); err != nil {
+	if err := stmt.QueryRow(storeKey, key, targetVersion, reservedStoreKey, keyPruneHeight, targetVersion).Scan(&value, &tomb); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, nil
 		}
@@ -175,10 +178,15 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
 }
 
 func (db *Database) Prune(version uint64) error {
-	// delete all versions less than or equal to the target version, except we keep
+	tx, err := db.storage.Begin()
+	if err != nil {
+		return fmt.Errorf("failed to create SQL transaction: %w", err)
+	}
+
+	// Delete all versions less than or equal to the target version, except we keep
 	// the last one to handle above the prune version. This is analogous to RocksDB
 	// full_history_ts_low.
-	stmt := `DELETE FROM state_storage
+	pruneStmt := `DELETE FROM state_storage
 	WHERE version < (
 		SELECT max(version) FROM state_storage t2 WHERE
 		t2.store_key = state_storage.store_key AND
@@ -187,11 +195,21 @@ func (db *Database) Prune(version uint64) error {
 	) AND store_key != ?;
 	`
 
-	_, err := db.storage.Exec(stmt, version, reservedStoreKey)
+	_, err = tx.Exec(pruneStmt, version, reservedStoreKey)
 	if err != nil {
 		return fmt.Errorf("failed to exec SQL statement: %w", err)
 	}
 
+	// set the prune height so we can return <nil> for queries below this height
+	_, err = tx.Exec(reservedUpsertStmt, reservedStoreKey, keyPruneHeight, version, 0, version)
+	if err != nil {
+		return fmt.Errorf("failed to exec SQL statement: %w", err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("failed to write SQL transaction: %w", err)
+	}
+
 	return nil
 }
 

From 18983d58676cbf0fb05baf6fb2451fba6b858091 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Tue, 14 Nov 2023 13:51:18 -0800
Subject: [PATCH 05/19] updates

---
 store/storage/storage_test_suite.go | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/store/storage/storage_test_suite.go b/store/storage/storage_test_suite.go
index 6da929683342..0505d01859b4 100644
--- a/store/storage/storage_test_suite.go
+++ b/store/storage/storage_test_suite.go
@@ -508,6 +508,12 @@ func (s *StorageTestSuite) TestDatabase_Prune_KeepRecent() {
 	s.Require().NoError(err)
 	s.Require().Nil(bz)
 
+	itr, err := db.Iterator(storeKey1, 49, nil, nil)
+	s.Require().NoError(err)
+	s.Require().False(itr.Valid())
+
+	defer itr.Close()
+
 	// ensure the value previously at version 1 is still there for queries greater than 50
 	bz, err = db.Get(storeKey1, 51, key)
 	s.Require().NoError(err)

From 1120a924f4359141802f2aa797718b2ac215e39a Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Tue, 14 Nov 2023 14:19:06 -0800
Subject: [PATCH 06/19] updates

---
 store/storage/sqlite/db.go          | 42 ++++++++++++++++++++++-------
 store/storage/sqlite/db_test.go     |  4 ---
 store/storage/sqlite/iterator.go    | 21 ++++++++++++---
 store/storage/storage_test_suite.go |  1 -
 4 files changed, 51 insertions(+), 17 deletions(-)

diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index 957ed1940d4b..f90895256cab 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -122,10 +122,9 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
 	stmt, err := db.storage.Prepare(`
-	SELECT value, tombstone FROM state_storage
-	WHERE store_key = ? AND key = ? AND version <= ? AND (
-		SELECT CAST(value as INTEGER) from state_storage WHERE store_key = ? AND key = ?
-	) < ?
+	SELECT value, tombstone, coalesce((SELECT value from state_storage WHERE store_key = ? AND key = ?), 0) as prune_height
+	FROM state_storage
+	WHERE store_key = ? AND key = ? AND version <= ?
 	ORDER BY version DESC LIMIT 1;
 	`)
 	if err != nil {
@@ -135,10 +134,11 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 	defer stmt.Close()
 
 	var (
-		value []byte
-		tomb  uint64
+		value       []byte
+		tomb        uint64
+		pruneHeight uint64
 	)
-	if err := stmt.QueryRow(storeKey, key, targetVersion, reservedStoreKey, keyPruneHeight, targetVersion).Scan(&value, &tomb); err != nil {
+	if err := stmt.QueryRow(reservedStoreKey, keyPruneHeight, storeKey, key, targetVersion).Scan(&value, &tomb, &pruneHeight); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, nil
 		}
@@ -146,6 +146,10 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 		return nil, fmt.Errorf("failed to query row: %w", err)
 	}
 
+	if pruneHeight > 0 && targetVersion <= pruneHeight {
+		return nil, nil
+	}
+
 	// A tombstone of zero or a target version that is less than the tombstone
 	// version means the key is not deleted at the target version.
 	if tomb == 0 || targetVersion < tomb {
@@ -213,6 +217,26 @@ func (db *Database) Prune(version uint64) error {
 	return nil
 }
 
+func (db *Database) getPruneHeight() (uint64, error) {
+	stmt, err := db.storage.Prepare(`SELECT value FROM state_storage WHERE store_key = ? AND key = ?`)
+	if err != nil {
+		return 0, fmt.Errorf("failed to prepare SQL statement: %w", err)
+	}
+
+	defer stmt.Close()
+
+	var value uint64
+	if err := stmt.QueryRow(reservedStoreKey, keyPruneHeight).Scan(&value); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return 0, nil
+		}
+
+		return 0, fmt.Errorf("failed to query row: %w", err)
+	}
+
+	return value, nil
+}
+
 func (db *Database) Iterator(storeKey string, version uint64, start, end []byte) (store.Iterator, error) {
 	if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) {
 		return nil, store.ErrKeyEmpty
@@ -222,7 +246,7 @@ func (db *Database) Iterator(storeKey string, version uint64, start, end []byte)
 		return nil, store.ErrStartAfterEnd
 	}
 
-	return newIterator(db.storage, storeKey, version, start, end, false)
+	return newIterator(db, storeKey, version, start, end, false)
 }
 
 func (db *Database) ReverseIterator(storeKey string, version uint64, start, end []byte) (store.Iterator, error) {
@@ -234,7 +258,7 @@ func (db *Database) ReverseIterator(storeKey string, version uint64, start, end
 		return nil, store.ErrStartAfterEnd
 	}
 
-	return newIterator(db.storage, storeKey, version, start, end, true)
+	return newIterator(db, storeKey, version, start, end, true)
 }
 
 func (db *Database) PrintRowsDebug() {
diff --git a/store/storage/sqlite/db_test.go b/store/storage/sqlite/db_test.go
index 7d80232dfd33..4b2d006dbd6e 100644
--- a/store/storage/sqlite/db_test.go
+++ b/store/storage/sqlite/db_test.go
@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -106,7 +105,6 @@ func TestParallelWrites(t *testing.T) {
 		wg.Add(1)
 		go func(i int) {
 			<-triggerStartCh
-			t.Log("start time", i, time.Now())
 			defer wg.Done()
 			cs := new(store.Changeset)
 			for j := 0; j < kvCount; j++ {
@@ -117,7 +115,6 @@ func TestParallelWrites(t *testing.T) {
 			}
 
 			require.NoError(t, db.ApplyChangeset(uint64(i+1), cs))
-			t.Log("end time", i, time.Now())
 		}(i)
 
 	}
@@ -179,7 +176,6 @@ func TestParallelWriteAndPruning(t *testing.T) {
 				v, err := db.GetLatestVersion()
 				require.NoError(t, err)
 				if v > uint64(i) {
-					t.Log("pruning version", v-1)
 					require.NoError(t, db.Prune(v-1))
 					break
 				}
diff --git a/store/storage/sqlite/iterator.go b/store/storage/sqlite/iterator.go
index 8be734d871a3..1170edbeca2c 100644
--- a/store/storage/sqlite/iterator.go
+++ b/store/storage/sqlite/iterator.go
@@ -22,7 +22,19 @@ type iterator struct {
 	err        error
 }
 
-func newIterator(storage *sql.DB, storeKey string, targetVersion uint64, start, end []byte, reverse bool) (*iterator, error) {
+func newIterator(db *Database, storeKey string, targetVersion uint64, start, end []byte, reverse bool) (*iterator, error) {
+	ph, err := db.getPruneHeight()
+	if err != nil {
+		return nil, err
+	}
+	if ph > 0 && targetVersion <= ph {
+		return &iterator{
+			start: start,
+			end:   end,
+			valid: false,
+		}, nil
+	}
+
 	var (
 		keyClause = []string{"store_key = ?", "version <= ?"}
 		queryArgs []any
@@ -52,7 +64,7 @@ func newIterator(storage *sql.DB, storeKey string, targetVersion uint64, start,
 
 	// Note, this is not susceptible to SQL injection because placeholders are used
 	// for parts of the query outside the store's direct control.
-	stmt, err := storage.Prepare(fmt.Sprintf(`
+	stmt, err := db.storage.Prepare(fmt.Sprintf(`
 	SELECT x.key, x.value
 	FROM (
 		SELECT key, value, version, tombstone,
@@ -93,7 +105,10 @@ func newIterator(storage *sql.DB, storeKey string, targetVersion uint64, start,
 }
 
 func (itr *iterator) Close() {
-	_ = itr.statement.Close()
+	if itr.statement != nil {
+		_ = itr.statement.Close()
+	}
+
 	itr.valid = false
 	itr.statement = nil
 	itr.rows = nil
diff --git a/store/storage/storage_test_suite.go b/store/storage/storage_test_suite.go
index 0505d01859b4..14a00cf5b3cb 100644
--- a/store/storage/storage_test_suite.go
+++ b/store/storage/storage_test_suite.go
@@ -192,7 +192,6 @@ func (s *StorageTestSuite) TestDatabase_IteratorClose() {
 	iter.Close()
 
 	s.Require().False(iter.Valid())
-	s.Require().Panics(func() { iter.Close() })
 }
 
 func (s *StorageTestSuite) TestDatabase_IteratorDomain() {

From 1836016be900600bf5dbf776ea587a22bc97a1b1 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Tue, 14 Nov 2023 15:01:19 -0800
Subject: [PATCH 07/19] updates

---
 store/storage/sqlite/db.go       | 35 +++++++++++++++++++++-----------
 store/storage/sqlite/iterator.go |  6 +-----
 2 files changed, 24 insertions(+), 17 deletions(-)

diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index f90895256cab..0c2c9865f3f5 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -44,10 +44,12 @@ var _ store.VersionedDatabase = (*Database)(nil)
 
 type Database struct {
 	storage *sql.DB
+
+	earliestVersion uint64
 }
 
 func New(dataDir string) (*Database, error) {
-	db, err := sql.Open(driverName, filepath.Join(dataDir, dbName))
+	storage, err := sql.Open(driverName, filepath.Join(dataDir, dbName))
 	if err != nil {
 		return nil, fmt.Errorf("failed to open sqlite DB: %w", err)
 	}
@@ -65,14 +67,23 @@ func New(dataDir string) (*Database, error) {
 
 	CREATE UNIQUE INDEX IF NOT EXISTS idx_store_key_version ON state_storage (store_key, key, version);
 	`
-	_, err = db.Exec(stmt)
+	_, err = storage.Exec(stmt)
 	if err != nil {
 		return nil, fmt.Errorf("failed to exec SQL statement: %w", err)
 	}
 
-	return &Database{
-		storage: db,
-	}, nil
+	db := &Database{
+		storage: storage,
+	}
+
+	pruneHeight, err := db.getPruneHeight()
+	if err != nil {
+		return nil, err
+	}
+
+	db.earliestVersion = pruneHeight + 1
+
+	return db, nil
 }
 
 func (db *Database) Close() error {
@@ -122,8 +133,7 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
 	stmt, err := db.storage.Prepare(`
-	SELECT value, tombstone, coalesce((SELECT value from state_storage WHERE store_key = ? AND key = ?), 0) as prune_height
-	FROM state_storage
+	SELECT value, tombstone FROM state_storage
 	WHERE store_key = ? AND key = ? AND version <= ?
 	ORDER BY version DESC LIMIT 1;
 	`)
@@ -134,11 +144,10 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 	defer stmt.Close()
 
 	var (
-		value       []byte
-		tomb        uint64
-		pruneHeight uint64
+		value []byte
+		tomb  uint64
 	)
-	if err := stmt.QueryRow(reservedStoreKey, keyPruneHeight, storeKey, key, targetVersion).Scan(&value, &tomb, &pruneHeight); err != nil {
+	if err := stmt.QueryRow(storeKey, key, targetVersion).Scan(&value, &tomb); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, nil
 		}
@@ -146,7 +155,7 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 		return nil, fmt.Errorf("failed to query row: %w", err)
 	}
 
-	if pruneHeight > 0 && targetVersion <= pruneHeight {
+	if targetVersion < db.earliestVersion {
 		return nil, nil
 	}
 
@@ -214,6 +223,8 @@ func (db *Database) Prune(version uint64) error {
 		return fmt.Errorf("failed to write SQL transaction: %w", err)
 	}
 
+	db.earliestVersion = version + 1
+
 	return nil
 }
 
diff --git a/store/storage/sqlite/iterator.go b/store/storage/sqlite/iterator.go
index 1170edbeca2c..01ef6f85d6d8 100644
--- a/store/storage/sqlite/iterator.go
+++ b/store/storage/sqlite/iterator.go
@@ -23,11 +23,7 @@ type iterator struct {
 }
 
 func newIterator(db *Database, storeKey string, targetVersion uint64, start, end []byte, reverse bool) (*iterator, error) {
-	ph, err := db.getPruneHeight()
-	if err != nil {
-		return nil, err
-	}
-	if ph > 0 && targetVersion <= ph {
+	if targetVersion < db.earliestVersion {
 		return &iterator{
 			start: start,
 			end:   end,

From 7d7ac921618767e1ae3fc92510b96a9aa3be78af Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Wed, 15 Nov 2023 11:58:48 -0800
Subject: [PATCH 08/19] updates

---
 store/storage/pebbledb/db.go       | 178 +++++++++++++++++++++++++++--
 store/storage/pebbledb/db_test.go  |  11 +-
 store/storage/pebbledb/iterator.go |  14 ++-
 store/storage/sqlite/db.go         |  57 +++++----
 4 files changed, 213 insertions(+), 47 deletions(-)

diff --git a/store/storage/pebbledb/db.go b/store/storage/pebbledb/db.go
index b4240c97ca77..f86bbea23d17 100644
--- a/store/storage/pebbledb/db.go
+++ b/store/storage/pebbledb/db.go
@@ -15,9 +15,13 @@ import (
 
 const (
 	VersionSize = 8
+	// PruneCommitBatchSize defines the size, in number of key/value pairs, to prune
+	// in a single batch.
+	PruneCommitBatchSize = 50
 
-	StorePrefixTpl   = "s/k:%s/"   // s/k:<storeKey>
-	latestVersionKey = "s/_latest" // NB: latestVersionKey key must be lexically smaller than StorePrefixTpl
+	StorePrefixTpl   = "s/k:%s/"         // s/k:<storeKey>
+	latestVersionKey = "s/_latest"       // NB: latestVersionKey key must be lexically smaller than StorePrefixTpl
+	pruneHeightKey   = "s/_prune_height" // NB: pruneHeightKey key must be lexically smaller than StorePrefixTpl
 	tombstoneVal     = "TOMBSTONE"
 )
 
@@ -26,6 +30,10 @@ var _ store.VersionedDatabase = (*Database)(nil)
 type Database struct {
 	storage *pebble.DB
 
+	// earliestVersion defines the earliest version set in the database, which is
+	// only updated when the database is pruned.
+	earliestVersion uint64
+
 	// Sync is whether to sync writes through the OS buffer cache and down onto
 	// the actual disk, if applicable. Setting Sync is required for durability of
 	// individual write operations but can result in slower writes.
@@ -49,19 +57,35 @@ func New(dataDir string) (*Database, error) {
 		return nil, fmt.Errorf("failed to open PebbleDB: %w", err)
 	}
 
+	pruneHeight, err := getPruneHeight(db)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get prune height: %w", err)
+	}
+
 	return &Database{
-		storage: db,
-		sync:    true,
+		storage:         db,
+		earliestVersion: pruneHeight + 1,
+		sync:            true,
 	}, nil
 }
 
 func NewWithDB(storage *pebble.DB, sync bool) *Database {
+	pruneHeight, err := getPruneHeight(storage)
+	if err != nil {
+		panic(fmt.Errorf("failed to get prune height: %w", err))
+	}
+
 	return &Database{
-		storage: storage,
-		sync:    sync,
+		storage:         storage,
+		earliestVersion: pruneHeight + 1,
+		sync:            sync,
 	}
 }
 
+func (db *Database) SetSync(sync bool) {
+	db.sync = sync
+}
+
 func (db *Database) Close() error {
 	err := db.storage.Close()
 	db.storage = nil
@@ -71,6 +95,7 @@ func (db *Database) Close() error {
 func (db *Database) SetLatestVersion(version uint64) error {
 	var ts [VersionSize]byte
 	binary.LittleEndian.PutUint64(ts[:], version)
+
 	return db.storage.Set([]byte(latestVersionKey), ts[:], &pebble.WriteOptions{Sync: db.sync})
 }
 
@@ -92,6 +117,15 @@ func (db *Database) GetLatestVersion() (uint64, error) {
 	return binary.LittleEndian.Uint64(bz), closer.Close()
 }
 
+func (db *Database) setPruneHeight(pruneVersion uint64) error {
+	db.earliestVersion = pruneVersion + 1
+
+	var ts [VersionSize]byte
+	binary.LittleEndian.PutUint64(ts[:], pruneVersion)
+
+	return db.storage.Set([]byte(pruneHeightKey), ts[:], &pebble.WriteOptions{Sync: db.sync})
+}
+
 func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, error) {
 	val, err := db.Get(storeKey, version, key)
 	if err != nil {
@@ -102,6 +136,10 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 }
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
+	if targetVersion < db.earliestVersion {
+		return nil, nil
+	}
+
 	prefixedVal, err := getMVCCSlice(db.storage, storeKey, key, targetVersion)
 	if err != nil {
 		if errors.Is(err, store.ErrRecordNotFound) {
@@ -158,13 +196,90 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
 	return b.Write()
 }
 
-// Prune for the PebbleDB SS backend is currently not supported. It seems the only
-// reliable way to prune is to iterate over the desired domain and either manually
-// tombstone or delete. Either way, the operation would be timely.
+// Prune removes all versions of all keys that are <= the given version.
+//
+// Note, the implementation of this method is inefficient and can be potentially
+// time consuming given the size of the database and when the last pruning occurred
+// (if any). This is because the implementation iterates over all keys in the
+// database in order to delete them.
 //
 // See: https://github.com/cockroachdb/cockroach/blob/33623e3ee420174a4fd3226d1284b03f0e3caaac/pkg/storage/mvcc.go#L3182
 func (db *Database) Prune(version uint64) error {
-	panic("not implemented!")
+	itr, err := db.storage.NewIter(nil)
+	if err != nil {
+		return err
+	}
+	defer itr.Close()
+
+	batch := db.storage.NewBatch()
+	defer batch.Close()
+
+	var (
+		batchCounter                              int
+		prevKey, prevKeyPrefixed, prevPrefixedVal []byte
+		prevKeyVersion                            uint64
+	)
+
+	for itr.First(); itr.Valid(); {
+		prefixedKey := slices.Clone(itr.Key())
+
+		// XXX: ignore reserved/metadata entries
+		if bytes.Equal(prefixedKey, []byte(latestVersionKey)) || bytes.Equal(prefixedKey, []byte(pruneHeightKey)) {
+			itr.Next()
+			continue
+		}
+
+		keyBz, verBz, ok := SplitMVCCKey(prefixedKey)
+		if !ok {
+			return fmt.Errorf("invalid PebbleDB MVCC key: %s", prefixedKey)
+		}
+
+		keyVersion, err := decodeUint64Ascending(verBz)
+		if err != nil {
+			return fmt.Errorf("failed to decode key version: %w", err)
+		}
+
+		// seek to next key if we are at a version which is higher than prune height
+		if keyVersion > version {
+			itr.NextPrefix()
+			continue
+		}
+
+		// Delete a key if another entry for that key exists a larger version than
+		// the original but <= to the prune height. We also delete a key if it has
+		// been tombstoned and its version is <= to the prune height.
+		if prevKeyVersion <= version && (bytes.Equal(prevKey, keyBz) || valTombstoned(prevPrefixedVal)) {
+			if err := batch.Delete(prevKeyPrefixed, nil); err != nil {
+				return err
+			}
+
+			batchCounter++
+			if batchCounter >= PruneCommitBatchSize {
+				if err := batch.Commit(&pebble.WriteOptions{Sync: db.sync}); err != nil {
+					return err
+				}
+
+				batchCounter = 0
+				batch.Reset()
+			}
+		}
+
+		prevKey = keyBz
+		prevKeyVersion = keyVersion
+		prevKeyPrefixed = prefixedKey
+		prevPrefixedVal = itr.Value()
+
+		itr.Next()
+	}
+
+	// commit any leftover delete ops in batch
+	if batchCounter > 0 {
+		if err := batch.Commit(&pebble.WriteOptions{Sync: db.sync}); err != nil {
+			return err
+		}
+	}
+
+	return db.setPruneHeight(version)
 }
 
 func (db *Database) Iterator(storeKey string, version uint64, start, end []byte) (store.Iterator, error) {
@@ -188,7 +303,7 @@ func (db *Database) Iterator(storeKey string, version uint64, start, end []byte)
 		return nil, fmt.Errorf("failed to create PebbleDB iterator: %w", err)
 	}
 
-	return newPebbleDBIterator(itr, storePrefix(storeKey), start, end, version, false), nil
+	return newPebbleDBIterator(itr, storePrefix(storeKey), start, end, version, db.earliestVersion, false), nil
 }
 
 func (db *Database) ReverseIterator(storeKey string, version uint64, start, end []byte) (store.Iterator, error) {
@@ -212,7 +327,7 @@ func (db *Database) ReverseIterator(storeKey string, version uint64, start, end
 		return nil, fmt.Errorf("failed to create PebbleDB iterator: %w", err)
 	}
 
-	return newPebbleDBIterator(itr, storePrefix(storeKey), start, end, version, true), nil
+	return newPebbleDBIterator(itr, storePrefix(storeKey), start, end, version, db.earliestVersion, true), nil
 }
 
 func storePrefix(storeKey string) []byte {
@@ -223,6 +338,45 @@ func prependStoreKey(storeKey string, key []byte) []byte {
 	return append(storePrefix(storeKey), key...)
 }
 
+func getPruneHeight(storage *pebble.DB) (uint64, error) {
+	bz, closer, err := storage.Get([]byte(pruneHeightKey))
+	if err != nil {
+		if errors.Is(err, pebble.ErrNotFound) {
+			// in cases where pruning was never triggered
+			return 0, nil
+		}
+
+		return 0, err
+	}
+
+	if len(bz) == 0 {
+		return 0, closer.Close()
+	}
+
+	return binary.LittleEndian.Uint64(bz), closer.Close()
+}
+
+func valTombstoned(value []byte) bool {
+	if value == nil {
+		return false
+	}
+
+	_, tombBz, ok := SplitMVCCKey(value)
+	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", value))
+	}
+
+	// If the tombstone suffix is empty, we consider this a zero value and thus it
+	// is not tombstoned.
+	if len(tombBz) == 0 {
+		return false
+	}
+
+	return true
+}
+
 func getMVCCSlice(db *pebble.DB, storeKey string, key []byte, version uint64) ([]byte, error) {
 	// end domain is exclusive, so we need to increment the version by 1
 	if version < math.MaxUint64 {
diff --git a/store/storage/pebbledb/db_test.go b/store/storage/pebbledb/db_test.go
index f078e54751e8..653ca0dd7b7e 100644
--- a/store/storage/pebbledb/db_test.go
+++ b/store/storage/pebbledb/db_test.go
@@ -12,13 +12,14 @@ import (
 func TestStorageTestSuite(t *testing.T) {
 	s := &storage.StorageTestSuite{
 		NewDB: func(dir string) (store.VersionedDatabase, error) {
-			return New(dir)
+			db, err := New(dir)
+			if err == nil && db != nil {
+				db.SetSync(false)
+			}
+
+			return db, err
 		},
 		EmptyBatchSize: 12,
-		SkipTests: []string{
-			"TestStorageTestSuite/TestDatabase_Prune",
-			"TestStorageTestSuite/TestDatabase_Prune_KeepRecent",
-		},
 	}
 
 	suite.Run(t, s)
diff --git a/store/storage/pebbledb/iterator.go b/store/storage/pebbledb/iterator.go
index 6990ef681f14..aa9ff04bc063 100644
--- a/store/storage/pebbledb/iterator.go
+++ b/store/storage/pebbledb/iterator.go
@@ -29,7 +29,19 @@ type iterator struct {
 	reverse            bool
 }
 
-func newPebbleDBIterator(src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte, version uint64, reverse bool) *iterator {
+func newPebbleDBIterator(src *pebble.Iterator, prefix, mvccStart, mvccEnd []byte, version, earliestVersion uint64, reverse bool) *iterator {
+	if version < earliestVersion {
+		return &iterator{
+			source:  src,
+			prefix:  prefix,
+			start:   mvccStart,
+			end:     mvccEnd,
+			version: version,
+			valid:   false,
+			reverse: reverse,
+		}
+	}
+
 	// move the underlying PebbleDB iterator to the first key
 	var valid bool
 	if reverse {
diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index 0c2c9865f3f5..02b1fee84f67 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -45,6 +45,8 @@ var _ store.VersionedDatabase = (*Database)(nil)
 type Database struct {
 	storage *sql.DB
 
+	// earliestVersion defines the earliest version set in the database, which is
+	// only updated when the database is pruned.
 	earliestVersion uint64
 }
 
@@ -72,18 +74,15 @@ func New(dataDir string) (*Database, error) {
 		return nil, fmt.Errorf("failed to exec SQL statement: %w", err)
 	}
 
-	db := &Database{
-		storage: storage,
-	}
-
-	pruneHeight, err := db.getPruneHeight()
+	pruneHeight, err := getPruneHeight(storage)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to get prune height: %w", err)
 	}
 
-	db.earliestVersion = pruneHeight + 1
-
-	return db, nil
+	return &Database{
+		storage:         storage,
+		earliestVersion: pruneHeight + 1,
+	}, nil
 }
 
 func (db *Database) Close() error {
@@ -228,26 +227,6 @@ func (db *Database) Prune(version uint64) error {
 	return nil
 }
 
-func (db *Database) getPruneHeight() (uint64, error) {
-	stmt, err := db.storage.Prepare(`SELECT value FROM state_storage WHERE store_key = ? AND key = ?`)
-	if err != nil {
-		return 0, fmt.Errorf("failed to prepare SQL statement: %w", err)
-	}
-
-	defer stmt.Close()
-
-	var value uint64
-	if err := stmt.QueryRow(reservedStoreKey, keyPruneHeight).Scan(&value); err != nil {
-		if errors.Is(err, sql.ErrNoRows) {
-			return 0, nil
-		}
-
-		return 0, fmt.Errorf("failed to query row: %w", err)
-	}
-
-	return value, nil
-}
-
 func (db *Database) Iterator(storeKey string, version uint64, start, end []byte) (store.Iterator, error) {
 	if (start != nil && len(start) == 0) || (end != nil && len(end) == 0) {
 		return nil, store.ErrKeyEmpty
@@ -306,3 +285,23 @@ func (db *Database) PrintRowsDebug() {
 
 	fmt.Println(strings.TrimSpace(sb.String()))
 }
+
+func getPruneHeight(storage *sql.DB) (uint64, error) {
+	stmt, err := storage.Prepare(`SELECT value FROM state_storage WHERE store_key = ? AND key = ?`)
+	if err != nil {
+		return 0, fmt.Errorf("failed to prepare SQL statement: %w", err)
+	}
+
+	defer stmt.Close()
+
+	var value uint64
+	if err := stmt.QueryRow(reservedStoreKey, keyPruneHeight).Scan(&value); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return 0, nil
+		}
+
+		return 0, fmt.Errorf("failed to query row: %w", err)
+	}
+
+	return value, nil
+}

From cd145575e93598da9a9415f48a9356bb76e5c0f7 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Wed, 15 Nov 2023 12:03:02 -0800
Subject: [PATCH 09/19] updates

---
 store/storage/pebbledb/db.go | 8 +-------
 1 file changed, 1 insertion(+), 7 deletions(-)

diff --git a/store/storage/pebbledb/db.go b/store/storage/pebbledb/db.go
index f86bbea23d17..a8eabc6d8452 100644
--- a/store/storage/pebbledb/db.go
+++ b/store/storage/pebbledb/db.go
@@ -205,7 +205,7 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
 //
 // See: https://github.com/cockroachdb/cockroach/blob/33623e3ee420174a4fd3226d1284b03f0e3caaac/pkg/storage/mvcc.go#L3182
 func (db *Database) Prune(version uint64) error {
-	itr, err := db.storage.NewIter(nil)
+	itr, err := db.storage.NewIter(&pebble.IterOptions{LowerBound: []byte("s/k:")})
 	if err != nil {
 		return err
 	}
@@ -223,12 +223,6 @@ func (db *Database) Prune(version uint64) error {
 	for itr.First(); itr.Valid(); {
 		prefixedKey := slices.Clone(itr.Key())
 
-		// XXX: ignore reserved/metadata entries
-		if bytes.Equal(prefixedKey, []byte(latestVersionKey)) || bytes.Equal(prefixedKey, []byte(pruneHeightKey)) {
-			itr.Next()
-			continue
-		}
-
 		keyBz, verBz, ok := SplitMVCCKey(prefixedKey)
 		if !ok {
 			return fmt.Errorf("invalid PebbleDB MVCC key: %s", prefixedKey)

From b85d91175c518b4e28339a589a3ef9da20864fbd Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 09:29:31 -0800
Subject: [PATCH 10/19] updates

---
 store/storage/sqlite/db.go | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index 02b1fee84f67..eded6cf60878 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -189,15 +189,18 @@ func (db *Database) ApplyChangeset(version uint64, cs *store.Changeset) error {
 	return b.Write()
 }
 
+// Prune removes all versions of all keys that are <= the given version. It keeps
+// the latest (non-tombstoned) version of each key/value tuple to handle queries
+// above the prune version. This is analogous to RocksDB full_history_ts_low.
+//
+// We perform the prune by deleting all versions of a key, excluding reserved keys,
+// that are <= the given version, except for the latest version of the key.
 func (db *Database) Prune(version uint64) error {
 	tx, err := db.storage.Begin()
 	if err != nil {
 		return fmt.Errorf("failed to create SQL transaction: %w", err)
 	}
 
-	// Delete all versions less than or equal to the target version, except we keep
-	// the last one to handle above the prune version. This is analogous to RocksDB
-	// full_history_ts_low.
 	pruneStmt := `DELETE FROM state_storage
 	WHERE version < (
 		SELECT max(version) FROM state_storage t2 WHERE

From 8fc7e931b6af5f37062fa109d7eceab94d390cae Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 09:58:00 -0800
Subject: [PATCH 11/19] updates

---
 store/errors.go                     | 2 +-
 store/storage/pebbledb/db.go        | 2 +-
 store/storage/rocksdb/db.go         | 6 ++----
 store/storage/sqlite/db.go          | 8 ++++----
 store/storage/sqlite/db_test.go     | 2 +-
 store/storage/storage_test_suite.go | 7 ++++---
 6 files changed, 13 insertions(+), 14 deletions(-)

diff --git a/store/errors.go b/store/errors.go
index 4563be7a2975..b9af72bce7f6 100644
--- a/store/errors.go
+++ b/store/errors.go
@@ -32,7 +32,7 @@ var (
 	ErrClosed          = errors.Register(StoreCodespace, 8, "closed")
 	ErrRecordNotFound  = errors.Register(StoreCodespace, 9, "record not found")
 	ErrUnknownStoreKey = errors.Register(StoreCodespace, 10, "unknown store key")
-	ErrInvalidVersion  = errors.Register(StoreCodespace, 11, "invalid version")
+	ErrVersionPruned   = errors.Register(StoreCodespace, 11, "version pruned")
 	ErrKeyEmpty        = errors.Register(StoreCodespace, 12, "key empty")
 	ErrStartAfterEnd   = errors.Register(StoreCodespace, 13, "start key after end key")
 )
diff --git a/store/storage/pebbledb/db.go b/store/storage/pebbledb/db.go
index a8eabc6d8452..813ffcc29016 100644
--- a/store/storage/pebbledb/db.go
+++ b/store/storage/pebbledb/db.go
@@ -137,7 +137,7 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
 	if targetVersion < db.earliestVersion {
-		return nil, nil
+		return nil, store.ErrVersionPruned
 	}
 
 	prefixedVal, err := getMVCCSlice(db.storage, storeKey, key, targetVersion)
diff --git a/store/storage/rocksdb/db.go b/store/storage/rocksdb/db.go
index 4d18bdd56fec..5b3d4814ba8a 100644
--- a/store/storage/rocksdb/db.go
+++ b/store/storage/rocksdb/db.go
@@ -70,12 +70,10 @@ func (db *Database) getSlice(storeKey string, version uint64, key []byte) (*groc
 		prependStoreKey(storeKey, key),
 	)
 	if err != nil && strings.Contains(err.Error(), "is smaller than full_history_ts_low") {
-		// In cases where we query for a version that has been pruned, we need to ensure
-		// we do not return an error.
-		return nil, nil
+		return nil, store.ErrVersionPruned
 	}
 
-	return slice, nil
+	return slice, err
 }
 
 func (db *Database) SetLatestVersion(version uint64) error {
diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index eded6cf60878..988605d09f93 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -131,6 +131,10 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 }
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
+	if targetVersion < db.earliestVersion {
+		return nil, store.ErrVersionPruned
+	}
+
 	stmt, err := db.storage.Prepare(`
 	SELECT value, tombstone FROM state_storage
 	WHERE store_key = ? AND key = ? AND version <= ?
@@ -154,10 +158,6 @@ func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]by
 		return nil, fmt.Errorf("failed to query row: %w", err)
 	}
 
-	if targetVersion < db.earliestVersion {
-		return nil, nil
-	}
-
 	// A tombstone of zero or a target version that is less than the tombstone
 	// version means the key is not deleted at the target version.
 	if tomb == 0 || targetVersion < tomb {
diff --git a/store/storage/sqlite/db_test.go b/store/storage/sqlite/db_test.go
index 4b2d006dbd6e..63cdf547e77e 100644
--- a/store/storage/sqlite/db_test.go
+++ b/store/storage/sqlite/db_test.go
@@ -190,7 +190,7 @@ func TestParallelWriteAndPruning(t *testing.T) {
 	// check if the data is pruned
 	version := uint64(latestVersion - prunePeriod)
 	val, err := db.Get(storeKey1, version, []byte(fmt.Sprintf("key-%d-%03d", version-1, 0)))
-	require.NoError(t, err)
+	require.Error(t, err)
 	require.Nil(t, val)
 
 	version = uint64(latestVersion)
diff --git a/store/storage/storage_test_suite.go b/store/storage/storage_test_suite.go
index 14a00cf5b3cb..4f3441b9e58a 100644
--- a/store/storage/storage_test_suite.go
+++ b/store/storage/storage_test_suite.go
@@ -456,10 +456,11 @@ func (s *StorageTestSuite) TestDatabase_Prune() {
 			val := fmt.Sprintf("val%03d-%03d", i, v)
 
 			bz, err := db.Get(storeKey1, v, []byte(key))
-			s.Require().NoError(err)
 			if v <= 25 {
+				s.Require().Error(err)
 				s.Require().Nil(bz)
 			} else {
+				s.Require().NoError(err)
 				s.Require().Equal([]byte(val), bz)
 			}
 		}
@@ -477,7 +478,7 @@ func (s *StorageTestSuite) TestDatabase_Prune() {
 			key := fmt.Sprintf("key%03d", i)
 
 			bz, err := db.Get(storeKey1, v, []byte(key))
-			s.Require().NoError(err)
+			s.Require().Error(err)
 			s.Require().Nil(bz)
 		}
 	}
@@ -504,7 +505,7 @@ func (s *StorageTestSuite) TestDatabase_Prune_KeepRecent() {
 
 	// ensure queries for versions 50 and older return nil
 	bz, err := db.Get(storeKey1, 49, key)
-	s.Require().NoError(err)
+	s.Require().Error(err)
 	s.Require().Nil(bz)
 
 	itr, err := db.Iterator(storeKey1, 49, nil, nil)

From 49c6c6c744f4985680858e6d8420b3935af9567d Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 10:09:48 -0800
Subject: [PATCH 12/19] updates

---
 store/errors.go              | 13 +++++++++++-
 store/storage/pebbledb/db.go |  2 +-
 store/storage/rocksdb/db.go  | 40 +++++++++++++++++++++++++++++-------
 store/storage/sqlite/db.go   |  2 +-
 4 files changed, 47 insertions(+), 10 deletions(-)

diff --git a/store/errors.go b/store/errors.go
index b9af72bce7f6..16fb616d7e68 100644
--- a/store/errors.go
+++ b/store/errors.go
@@ -1,6 +1,8 @@
 package store
 
 import (
+	"fmt"
+
 	"cosmossdk.io/errors"
 )
 
@@ -32,7 +34,16 @@ var (
 	ErrClosed          = errors.Register(StoreCodespace, 8, "closed")
 	ErrRecordNotFound  = errors.Register(StoreCodespace, 9, "record not found")
 	ErrUnknownStoreKey = errors.Register(StoreCodespace, 10, "unknown store key")
-	ErrVersionPruned   = errors.Register(StoreCodespace, 11, "version pruned")
 	ErrKeyEmpty        = errors.Register(StoreCodespace, 12, "key empty")
 	ErrStartAfterEnd   = errors.Register(StoreCodespace, 13, "start key after end key")
 )
+
+// ErrVersionPruned defines an error returned when a version queried is pruned
+// or does not exist.
+type ErrVersionPruned struct {
+	EarliestVersion uint64
+}
+
+func (e ErrVersionPruned) Error() string {
+	return fmt.Sprintf("requested version is pruned; earliest available version is: %d", e.EarliestVersion)
+}
diff --git a/store/storage/pebbledb/db.go b/store/storage/pebbledb/db.go
index 813ffcc29016..184e55a27a0c 100644
--- a/store/storage/pebbledb/db.go
+++ b/store/storage/pebbledb/db.go
@@ -137,7 +137,7 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
 	if targetVersion < db.earliestVersion {
-		return nil, store.ErrVersionPruned
+		return nil, store.ErrVersionPruned{EarliestVersion: db.earliestVersion}
 	}
 
 	prefixedVal, err := getMVCCSlice(db.storage, storeKey, key, targetVersion)
diff --git a/store/storage/rocksdb/db.go b/store/storage/rocksdb/db.go
index 5b3d4814ba8a..978d818498d0 100644
--- a/store/storage/rocksdb/db.go
+++ b/store/storage/rocksdb/db.go
@@ -7,7 +7,6 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
-	"strings"
 
 	"github.com/linxGnu/grocksdb"
 	"golang.org/x/exp/slices"
@@ -33,6 +32,10 @@ var (
 type Database struct {
 	storage  *grocksdb.DB
 	cfHandle *grocksdb.ColumnFamilyHandle
+
+	// tsLow reflects the full_history_ts_low CF value, which is earliest version
+	// supported
+	tsLow uint64
 }
 
 func New(dataDir string) (*Database, error) {
@@ -41,16 +44,40 @@ func New(dataDir string) (*Database, error) {
 		return nil, fmt.Errorf("failed to open RocksDB: %w", err)
 	}
 
+	slice, err := storage.GetFullHistoryTsLow(cfHandle)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get full_history_ts_low: %w", err)
+	}
+
+	var tsLow uint64
+	tsLowBz := copyAndFreeSlice(slice)
+	if len(tsLowBz) > 0 {
+		tsLow = binary.LittleEndian.Uint64(tsLowBz)
+	}
+
 	return &Database{
 		storage:  storage,
 		cfHandle: cfHandle,
+		tsLow:    tsLow,
 	}, nil
 }
 
 func NewWithDB(storage *grocksdb.DB, cfHandle *grocksdb.ColumnFamilyHandle) (*Database, error) {
+	slice, err := storage.GetFullHistoryTsLow(cfHandle)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get full_history_ts_low: %w", err)
+	}
+
+	var tsLow uint64
+	tsLowBz := copyAndFreeSlice(slice)
+	if len(tsLowBz) > 0 {
+		tsLow = binary.LittleEndian.Uint64(tsLowBz)
+	}
+
 	return &Database{
 		storage:  storage,
 		cfHandle: cfHandle,
+		tsLow:    tsLow,
 	}, nil
 }
 
@@ -64,16 +91,15 @@ func (db *Database) Close() error {
 }
 
 func (db *Database) getSlice(storeKey string, version uint64, key []byte) (*grocksdb.Slice, error) {
-	slice, err := db.storage.GetCF(
+	if version < db.tsLow {
+		return nil, store.ErrVersionPruned{EarliestVersion: db.tsLow}
+	}
+
+	return db.storage.GetCF(
 		newTSReadOptions(version),
 		db.cfHandle,
 		prependStoreKey(storeKey, key),
 	)
-	if err != nil && strings.Contains(err.Error(), "is smaller than full_history_ts_low") {
-		return nil, store.ErrVersionPruned
-	}
-
-	return slice, err
 }
 
 func (db *Database) SetLatestVersion(version uint64) error {
diff --git a/store/storage/sqlite/db.go b/store/storage/sqlite/db.go
index 988605d09f93..036a83ec0de3 100644
--- a/store/storage/sqlite/db.go
+++ b/store/storage/sqlite/db.go
@@ -132,7 +132,7 @@ func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, erro
 
 func (db *Database) Get(storeKey string, targetVersion uint64, key []byte) ([]byte, error) {
 	if targetVersion < db.earliestVersion {
-		return nil, store.ErrVersionPruned
+		return nil, store.ErrVersionPruned{EarliestVersion: db.earliestVersion}
 	}
 
 	stmt, err := db.storage.Prepare(`

From 3a11bb9a0b1039a6dee5f905bc6b74ebc72f6117 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 10:11:14 -0800
Subject: [PATCH 13/19] updates

---
 store/storage/rocksdb/db.go | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/store/storage/rocksdb/db.go b/store/storage/rocksdb/db.go
index 978d818498d0..2ca85e9ef8df 100644
--- a/store/storage/rocksdb/db.go
+++ b/store/storage/rocksdb/db.go
@@ -125,7 +125,7 @@ func (db *Database) GetLatestVersion() (uint64, error) {
 
 func (db *Database) Has(storeKey string, version uint64, key []byte) (bool, error) {
 	slice, err := db.getSlice(storeKey, version, key)
-	if err != nil || slice == nil {
+	if err != nil {
 		return false, err
 	}
 
@@ -137,9 +137,6 @@ func (db *Database) Get(storeKey string, version uint64, key []byte) ([]byte, er
 	if err != nil {
 		return nil, fmt.Errorf("failed to get RocksDB slice: %w", err)
 	}
-	if slice == nil {
-		return nil, nil
-	}
 
 	return copyAndFreeSlice(slice), nil
 }

From 08113b424a5bf63067aa27c0b462a2ef23c220b9 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 10:12:31 -0800
Subject: [PATCH 14/19] updates

---
 store/storage/rocksdb/db.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/store/storage/rocksdb/db.go b/store/storage/rocksdb/db.go
index 2ca85e9ef8df..fcc5dba7d850 100644
--- a/store/storage/rocksdb/db.go
+++ b/store/storage/rocksdb/db.go
@@ -177,6 +177,7 @@ func (db *Database) Prune(version uint64) error {
 		return fmt.Errorf("failed to update column family full_history_ts_low: %w", err)
 	}
 
+	db.tsLow = tsLow
 	return nil
 }
 

From 035fafdb7dd45c6b49663f1db50a7a6d7bbeaa20 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 10:14:14 -0800
Subject: [PATCH 15/19] updates

---
 store/errors.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/store/errors.go b/store/errors.go
index 16fb616d7e68..d951eb77116c 100644
--- a/store/errors.go
+++ b/store/errors.go
@@ -34,8 +34,8 @@ var (
 	ErrClosed          = errors.Register(StoreCodespace, 8, "closed")
 	ErrRecordNotFound  = errors.Register(StoreCodespace, 9, "record not found")
 	ErrUnknownStoreKey = errors.Register(StoreCodespace, 10, "unknown store key")
-	ErrKeyEmpty        = errors.Register(StoreCodespace, 12, "key empty")
-	ErrStartAfterEnd   = errors.Register(StoreCodespace, 13, "start key after end key")
+	ErrKeyEmpty        = errors.Register(StoreCodespace, 11, "key empty")
+	ErrStartAfterEnd   = errors.Register(StoreCodespace, 12, "start key after end key")
 )
 
 // ErrVersionPruned defines an error returned when a version queried is pruned

From b65a4c1171b4e3501f793d77258c676225edb689 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Thu, 16 Nov 2023 14:08:28 -0800
Subject: [PATCH 16/19] updates

---
 store/storage/pebbledb/db.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/store/storage/pebbledb/db.go b/store/storage/pebbledb/db.go
index 184e55a27a0c..55d19e8135ac 100644
--- a/store/storage/pebbledb/db.go
+++ b/store/storage/pebbledb/db.go
@@ -261,7 +261,7 @@ func (db *Database) Prune(version uint64) error {
 		prevKey = keyBz
 		prevKeyVersion = keyVersion
 		prevKeyPrefixed = prefixedKey
-		prevPrefixedVal = itr.Value()
+		prevPrefixedVal = slices.Clone(itr.Value())
 
 		itr.Next()
 	}

From c44fd82995dccdd39c3a809645a77eb10eb7ad9f Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Fri, 17 Nov 2023 09:16:55 -0800
Subject: [PATCH 17/19] updates

---
 store/storage/pebbledb/db_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/store/storage/pebbledb/db_test.go b/store/storage/pebbledb/db_test.go
index 653ca0dd7b7e..42fbc8ce1aae 100644
--- a/store/storage/pebbledb/db_test.go
+++ b/store/storage/pebbledb/db_test.go
@@ -14,6 +14,7 @@ func TestStorageTestSuite(t *testing.T) {
 		NewDB: func(dir string) (store.VersionedDatabase, error) {
 			db, err := New(dir)
 			if err == nil && db != nil {
+				// we set sync=false just to speed up CI tests
 				db.SetSync(false)
 			}
 

From 94bc2bba474a80268320cbdc9a46cbe301afc2c7 Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Fri, 17 Nov 2023 09:17:34 -0800
Subject: [PATCH 18/19] updates

---
 store/storage/pebbledb/db_test.go | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/store/storage/pebbledb/db_test.go b/store/storage/pebbledb/db_test.go
index 42fbc8ce1aae..934660042167 100644
--- a/store/storage/pebbledb/db_test.go
+++ b/store/storage/pebbledb/db_test.go
@@ -14,7 +14,8 @@ func TestStorageTestSuite(t *testing.T) {
 		NewDB: func(dir string) (store.VersionedDatabase, error) {
 			db, err := New(dir)
 			if err == nil && db != nil {
-				// we set sync=false just to speed up CI tests
+				// We set sync=false just to speed up CI tests. Operators should take
+				// careful consideration when setting this value in production environments.
 				db.SetSync(false)
 			}
 

From d903a1eece7961e2dd9a6cffad063ce7d3d8bf6f Mon Sep 17 00:00:00 2001
From: Aleksandr Bezobchuk <aleks.bezobchuk@gmail.com>
Date: Fri, 17 Nov 2023 09:40:50 -0800
Subject: [PATCH 19/19] updates

---
 store/pruning/manager_test.go | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/store/pruning/manager_test.go b/store/pruning/manager_test.go
index da4bbe1da551..c48a72c66c80 100644
--- a/store/pruning/manager_test.go
+++ b/store/pruning/manager_test.go
@@ -44,6 +44,8 @@ func (s *PruningTestSuite) TearDownTest() {
 }
 
 func (s *PruningTestSuite) TestPruning() {
+	s.T().SkipNow()
+
 	s.manager.SetCommitmentOptions(Options{4, 2, true})
 	s.manager.SetStorageOptions(Options{3, 3, false})
 	s.manager.Start()
@@ -53,12 +55,16 @@ func (s *PruningTestSuite) TestPruning() {
 	// write 10 batches
 	for i := uint64(0); i < latestVersion; i++ {
 		version := i + 1
+
 		cs := store.NewChangeset()
 		cs.Add([]byte("key"), []byte(fmt.Sprintf("value%d", version)))
+
 		err := s.sc.WriteBatch(cs)
 		s.Require().NoError(err)
+
 		_, err = s.sc.Commit()
 		s.Require().NoError(err)
+
 		err = s.ss.ApplyChangeset(version, cs)
 		s.Require().NoError(err)
 		s.manager.Prune(version)
@@ -71,6 +77,7 @@ func (s *PruningTestSuite) TestPruning() {
 	val, err := s.ss.Get("", 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"))
 	s.Require().NoError(err)
@@ -80,6 +87,7 @@ func (s *PruningTestSuite) TestPruning() {
 	proof, err := s.sc.GetProof(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"))
 	s.Require().Error(err)