refactor: history storage

cmd/shisui/main.go

@@ -23,7 +23,6 @@ import (
 	"github.com/ethereum/go-ethereum/portalnetwork/beacon"
 	"github.com/ethereum/go-ethereum/portalnetwork/history"
 	"github.com/ethereum/go-ethereum/portalnetwork/storage"
-	"github.com/ethereum/go-ethereum/portalnetwork/storage/sqlite"
 	"github.com/ethereum/go-ethereum/rpc"
 	_ "github.com/mattn/go-sqlite3"
 	"github.com/protolambda/zrnt/eth2/configs"
@@ -181,7 +180,15 @@ func initDiscV5(config Config, conn discover.UDPConn) (*discover.UDPv5, *enode.L
 }
 
 func initHistory(config Config, server *rpc.Server, conn discover.UDPConn, localNode *enode.LocalNode, discV5 *discover.UDPv5) error {
-	contentStorage, err := sqlite.NewContentStorage(config.DataCapacity, localNode.ID(), config.DataDir)
+	db, err := history.NewDB(config.DataDir)
+	if err != nil {
+		return err
+	}
+	contentStorage, err := history.NewHistoryStorage(storage.PortalStorageConfig{
+		StorageCapacityMB: config.DataCapacity,
+		DB:                db,
+		NodeId:            localNode.ID(),
+	})
 	if err != nil {
 		return err
 	}

portalnetwork/beacon/types.go

@@ -68,7 +68,10 @@ func (flcb *ForkedLightClientBootstrap) Deserialize(spec *common.Spec, dr *codec
 }
 
 func (flcb *ForkedLightClientBootstrap) Serialize(spec *common.Spec, w *codec.EncodingWriter) error {
-	return w.FixedLenContainer(flcb.ForkDigest, spec.Wrap(flcb.Bootstrap))
+	if err := w.Write(flcb.ForkDigest[:]); err != nil {
+		return err
+	}
+	return flcb.Bootstrap.Serialize(spec, w)
 }
 
 func (flcb *ForkedLightClientBootstrap) FixedLength(_ *common.Spec) uint64 {
@@ -113,7 +116,10 @@ func (flcu *ForkedLightClientUpdate) Deserialize(spec *common.Spec, dr *codec.De
 }
 
 func (flcu *ForkedLightClientUpdate) Serialize(spec *common.Spec, w *codec.EncodingWriter) error {
-	return w.FixedLenContainer(flcu.ForkDigest, spec.Wrap(flcu.LightClientUpdate))
+	if err := w.Write(flcu.ForkDigest[:]); err != nil {
+		return err
+	}
+	return flcu.LightClientUpdate.Serialize(spec, w)
 }
 
 func (flcu *ForkedLightClientUpdate) FixedLength(_ *common.Spec) uint64 {
@@ -195,7 +201,10 @@ func (flcou *ForkedLightClientOptimisticUpdate) Deserialize(spec *common.Spec, d
 }
 
 func (flcou *ForkedLightClientOptimisticUpdate) Serialize(spec *common.Spec, w *codec.EncodingWriter) error {
-	return w.FixedLenContainer(flcou.ForkDigest, spec.Wrap(flcou.LightClientOptimisticUpdate))
+	if err := w.Write(flcou.ForkDigest[:]); err != nil {
+		return err
+	}
+	return flcou.LightClientOptimisticUpdate.Serialize(spec, w)
 }
 
 func (flcou *ForkedLightClientOptimisticUpdate) FixedLength(_ *common.Spec) uint64 {
@@ -240,7 +249,10 @@ func (flcfu *ForkedLightClientFinalityUpdate) Deserialize(spec *common.Spec, dr
 }
 
 func (flcfu *ForkedLightClientFinalityUpdate) Serialize(spec *common.Spec, w *codec.EncodingWriter) error {
-	return w.FixedLenContainer(flcfu.ForkDigest, spec.Wrap(flcfu.LightClientFinalityUpdate))
+	if err := w.Write(flcfu.ForkDigest[:]); err != nil {
+		return err
+	}
+	return flcfu.LightClientFinalityUpdate.Serialize(spec, w)
 }
 
 func (flcfu *ForkedLightClientFinalityUpdate) FixedLength(_ *common.Spec) uint64 {

portalnetwork/history/storage.go

@@ -0,0 +1,422 @@
+package history
+
+import (
+	"bytes"
+	"database/sql"
+	"errors"
+	"math/big"
+	"os"
+	"path"
+	"strings"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/portalnetwork/storage"
+	"github.com/holiman/uint256"
+	"github.com/mattn/go-sqlite3"
+)
+
+const (
+	sqliteName              = "history.sqlite"
+	contentDeletionFraction = 0.05 // 5% of the content will be deleted when the storage capacity is hit and radius gets adjusted.
+	// SQLite Statements
+	createSql = `CREATE TABLE IF NOT EXISTS kvstore (
+		key BLOB PRIMARY KEY,
+		value BLOB
+	);`
+	getSql                     = "SELECT value FROM kvstore WHERE key = (?1);"
+	putSql                     = "INSERT OR REPLACE INTO kvstore (key, value) VALUES (?1, ?2);"
+	deleteSql                  = "DELETE FROM kvstore WHERE key = (?1);"
+	containSql                 = "SELECT 1 FROM kvstore WHERE key = (?1);"
+	getAllOrderedByDistanceSql = "SELECT key, length(value), xor(key, (?1)) as distance FROM kvstore ORDER BY distance DESC;"
+	deleteOutOfRadiusStmt      = "DELETE FROM kvstore WHERE greater(xor(key, (?1)), (?2)) = 1"
+	XorFindFarthestQuery       = `SELECT
+		xor(key, (?1)) as distance
+		FROM kvstore
+		ORDER BY distance DESC`
+)
+
+var (
+	maxDistance = uint256.MustFromHex("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff")
+)
+
+var _ storage.ContentStorage = &ContentStorage{}
+var once sync.Once
+
+type ContentStorage struct {
+	nodeId                 enode.ID
+	storageCapacityInBytes uint64
+	radius                 *uint256.Int
+	sqliteDB               *sql.DB
+	getStmt                *sql.Stmt
+	putStmt                *sql.Stmt
+	delStmt                *sql.Stmt
+	containStmt            *sql.Stmt
+	log                    log.Logger
+}
+
+func xor(contentId, nodeId []byte) []byte {
+	// length of contentId maybe not 32bytes
+	padding := make([]byte, 32)
+	if len(contentId) != len(nodeId) {
+		copy(padding, contentId)
+	} else {
+		padding = contentId
+	}
+	res := make([]byte, len(padding))
+	for i := range padding {
+		res[i] = padding[i] ^ nodeId[i]
+	}
+	return res
+}
+
+// a > b return 1; a = b return 0; else return -1
+func greater(a, b []byte) int {
+	return bytes.Compare(a, b)
+}
+
+func NewDB(dataDir string) (*sql.DB, error) {
+	dbPath := path.Join(dataDir, "history")
+	err := os.MkdirAll(dbPath, 0755)
+	if err != nil {
+		return nil, err
+	}
+	// avoid repeated register in tests
+	once.Do(func() {
+		sql.Register("sqlite3_custom", &sqlite3.SQLiteDriver{
+			ConnectHook: func(conn *sqlite3.SQLiteConn) error {
+				if err := conn.RegisterFunc("xor", xor, false); err != nil {
+					return err
+				}
+				if err := conn.RegisterFunc("greater", greater, false); err != nil {
+					return err
+				}
+				return nil
+			},
+		})
+	})
+	sqlDb, err := sql.Open("sqlite3_custom", path.Join(dbPath, sqliteName))
+	return sqlDb, err
+}
+
+func NewHistoryStorage(config storage.PortalStorageConfig) (storage.ContentStorage, error) {
+	hs := &ContentStorage{
+		nodeId:                 config.NodeId,
+		sqliteDB:               config.DB,
+		storageCapacityInBytes: config.StorageCapacityMB * 1000000,
+		radius:                 maxDistance,
+		log:                    log.New("history_storage"),
+	}
+	err := hs.createTable()
+	if err != nil {
+		return nil, err
+	}
+
+	err = hs.initStmts()
+
+	// Check whether we already have data, and use it to set radius
+
+	return hs, err
+}
+
+// Get the content according to the contentId
+func (p *ContentStorage) Get(contentKey []byte, contentId []byte) ([]byte, error) {
+	var res []byte
+	err := p.getStmt.QueryRow(contentId).Scan(&res)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, storage.ErrContentNotFound
+	}
+	return res, err
+}
+
+type PutResult struct {
+	err    error
+	pruned bool
+	count  int
+}
+
+func (p *PutResult) Err() error {
+	return p.err
+}
+
+func (p *PutResult) Pruned() bool {
+	return p.pruned
+}
+
+func (p *PutResult) PrunedCount() int {
+	return p.count
+}
+
+func newPutResultWithErr(err error) PutResult {
+	return PutResult{
+		err: err,
+	}
+}
+
+func (p *ContentStorage) Put(contentKey []byte, contentId []byte, content []byte) error {
+	res := p.put(contentId, content)
+	return res.Err()
+}
+
+// Put saves the contentId and content
+func (p *ContentStorage) put(contentId []byte, content []byte) PutResult {
+	_, err := p.putStmt.Exec(contentId, content)
+	if err != nil {
+		return newPutResultWithErr(err)
+	}
+
+	dbSize, err := p.UsedSize()
+	if err != nil {
+		return newPutResultWithErr(err)
+	}
+	if dbSize > p.storageCapacityInBytes {
+		count, err := p.deleteContentFraction(contentDeletionFraction)
+		//
+		if err != nil {
+			log.Warn("failed to delete oversize item")
+			return newPutResultWithErr(err)
+		}
+		return PutResult{pruned: true, count: count}
+	}
+
+	return PutResult{}
+}
+
+func (p *ContentStorage) Close() error {
+	err := p.getStmt.Close()
+	if err != nil {
+		return err
+	}
+	err = p.putStmt.Close()
+	if err != nil {
+		return err
+	}
+	err = p.delStmt.Close()
+	if err != nil {
+		return err
+	}
+	err = p.containStmt.Close()
+	if err != nil {
+		return err
+	}
+	return p.sqliteDB.Close()
+}
+
+func (p *ContentStorage) createTable() error {
+	stat, err := p.sqliteDB.Prepare(createSql)
+	if err != nil {
+		return err
+	}
+	defer func(stat *sql.Stmt) {
+		err = stat.Close()
+		if err != nil {
+			p.log.Error("failed to close statement", "err", err)
+		}
+	}(stat)
+	_, err = stat.Exec()
+	return err
+}
+
+func (p *ContentStorage) initStmts() error {
+	var stat *sql.Stmt
+	var err error
+	if stat, err = p.sqliteDB.Prepare(getSql); err != nil {
+		return nil
+	}
+	p.getStmt = stat
+	if stat, err = p.sqliteDB.Prepare(putSql); err != nil {
+		return nil
+	}
+	p.putStmt = stat
+	if stat, err = p.sqliteDB.Prepare(deleteSql); err != nil {
+		return nil
+	}
+	p.delStmt = stat
+	if stat, err = p.sqliteDB.Prepare(containSql); err != nil {
+		return nil
+	}
+	p.containStmt = stat
+	return nil
+}
+
+// Size get database size, content size and similar
+func (p *ContentStorage) Size() (uint64, error) {
+	sql := "SELECT page_count * page_size as size FROM pragma_page_count(), pragma_page_size();"
+	stmt, err := p.sqliteDB.Prepare(sql)
+	if err != nil {
+		return 0, err
+	}
+	var res uint64
+	err = stmt.QueryRow().Scan(&res)
+	return res, err
+}
+
+func (p *ContentStorage) UnusedSize() (uint64, error) {
+	sql := "SELECT freelist_count * page_size as size FROM pragma_freelist_count(), pragma_page_size();"
+	return p.queryRowUint64(sql)
+}
+
+// UsedSize = Size - UnusedSize
+func (p *ContentStorage) UsedSize() (uint64, error) {
+	size, err := p.Size()
+	if err != nil {
+		return 0, err
+	}
+	unusedSize, err := p.UnusedSize()
+	if err != nil {
+		return 0, err
+	}
+	return size - unusedSize, err
+}
+
+// ContentCount return the total content count
+func (p *ContentStorage) ContentCount() (uint64, error) {
+	sql := "SELECT COUNT(key) FROM kvstore;"
+	return p.queryRowUint64(sql)
+}
+
+func (p *ContentStorage) ContentSize() (uint64, error) {
+	sql := "SELECT SUM(length(value)) FROM kvstore"
+	return p.queryRowUint64(sql)
+}
+
+func (p *ContentStorage) queryRowUint64(sql string) (uint64, error) {
+	// sql := "SELECT SUM(length(value)) FROM kvstore"
+	stmt, err := p.sqliteDB.Prepare(sql)
+	if err != nil {
+		return 0, err
+	}
+	var res uint64
+	err = stmt.QueryRow().Scan(&res)
+	return res, err
+}
+
+// GetLargestDistance find the largest distance
+func (p *ContentStorage) GetLargestDistance() (*uint256.Int, error) {
+	stmt, err := p.sqliteDB.Prepare(XorFindFarthestQuery)
+	if err != nil {
+		return nil, err
+	}
+	var distance []byte
+
+	err = stmt.QueryRow(p.nodeId[:]).Scan(&distance)
+	if err != nil {
+		return nil, err
+	}
+	res := uint256.NewInt(0)
+	err = res.UnmarshalSSZ(distance)
+
+	return res, err
+	// reverse the distance, because big.SetBytes is big-endian
+	// utils.ReverseBytesInPlace(distance)
+	// bigNum := new(big.Int).SetBytes(distance)
+	// res := uint256.MustFromBig(bigNum)
+	// return res, nil
+}
+
+// EstimateNewRadius calculates an estimated new radius based on the current radius, used size, and storage capacity.
+// The method takes the currentRadius as input and returns the estimated new radius and an error (if any).
+// It calculates the size ratio of usedSize to storageCapacityInBytes and adjusts the currentRadius accordingly.
+// If the size ratio is greater than 0, it performs the adjustment; otherwise, it returns the currentRadius unchanged.
+// The method returns an error if there is any issue in determining the used size.
+func (p *ContentStorage) EstimateNewRadius(currentRadius *uint256.Int) (*uint256.Int, error) {
+	currrentSize, err := p.UsedSize()
+	if err != nil {
+		return nil, err
+	}
+	sizeRatio := currrentSize / p.storageCapacityInBytes
+	if sizeRatio > 0 {
+		bigFormat := new(big.Int).SetUint64(sizeRatio)
+		return new(uint256.Int).Div(currentRadius, uint256.MustFromBig(bigFormat)), nil
+	}
+	return currentRadius, nil
+}
+
+func (p *ContentStorage) deleteContentFraction(fraction float64) (deleteCount int, err error) {
+	if fraction <= 0 || fraction >= 1 {
+		return deleteCount, errors.New("fraction should be between 0 and 1")
+	}
+	totalContentSize, err := p.ContentSize()
+	if err != nil {
+		return deleteCount, err
+	}
+	bytesToDelete := uint64(fraction * float64(totalContentSize))
+	// deleteElements := 0
+	deleteBytes := 0
+
+	rows, err := p.sqliteDB.Query(getAllOrderedByDistanceSql, p.nodeId[:])
+	if err != nil {
+		return deleteCount, err
+	}
+	defer func(rows *sql.Rows) {
+		err = rows.Close()
+		if err != nil {
+			p.log.Error("failed to close rows", "err", err)
+		}
+	}(rows)
+	idsToDelete := make([][]byte, 0)
+	for deleteBytes < int(bytesToDelete) && rows.Next() {
+		var contentId []byte
+		var payloadLen int
+		var distance []byte
+		err = rows.Scan(&contentId, &payloadLen, &distance)
+		if err != nil {
+			return deleteCount, err
+		}
+		idsToDelete = append(idsToDelete, contentId)
+		// err = p.del(contentId)
+		if err != nil {
+			return deleteCount, err
+		}
+		deleteBytes += payloadLen
+		deleteCount++
+	}
+	// row must close first, or database is locked
+	// rows.Close() can call multi times
+	err = rows.Close()
+	if err != nil {
+		return 0, err
+	}
+	err = p.batchDel(idsToDelete)
+	return
+}
+
+func (p *ContentStorage) del(contentId []byte) error {
+	_, err := p.delStmt.Exec(contentId)
+	return err
+}
+
+func (p *ContentStorage) batchDel(ids [][]byte) error {
+	query := "DELETE FROM kvstore WHERE key IN (?" + strings.Repeat(", ?", len(ids)-1) + ")"
+	args := make([]interface{}, len(ids))
+	for i, id := range ids {
+		args[i] = id
+	}
+
+	// delete items
+	_, err := p.sqliteDB.Exec(query, args...)
+	return err
+}
+
+// ReclaimSpace reclaims space in the ContentStorage's SQLite database by performing a VACUUM operation.
+// It returns an error if the VACUUM operation encounters any issues.
+func (p *ContentStorage) ReclaimSpace() error {
+	_, err := p.sqliteDB.Exec("VACUUM;")
+	return err
+}
+
+func (p *ContentStorage) deleteContentOutOfRadius(radius *uint256.Int) error {
+	res, err := p.sqliteDB.Exec(deleteOutOfRadiusStmt, p.nodeId[:], radius.Bytes())
+	if err != nil {
+		return err
+	}
+	count, _ := res.RowsAffected()
+	p.log.Trace("delete %d items", count)
+	return err
+}
+
+// ForcePrune delete the content which distance is further than the given radius
+func (p *ContentStorage) ForcePrune(radius *uint256.Int) error {
+	return p.deleteContentOutOfRadius(radius)
+}

portalnetwork/history/storage_test.go

@@ -0,0 +1,315 @@
+package history
+
+import (
+	"fmt"
+	"math"
+	"os"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/portalnetwork/storage"
+	contentStorage "github.com/ethereum/go-ethereum/portalnetwork/storage"
+	"github.com/holiman/uint256"
+	"github.com/stretchr/testify/assert"
+)
+
+const nodeDataDir = "./unit_test"
+
+func clearNodeData() {
+	_ = os.Remove(fmt.Sprintf("%s/%s/%s", nodeDataDir, "history", sqliteName))
+}
+
+func genBytes(length int) []byte {
+	res := make([]byte, length)
+	for i := 0; i < length; i++ {
+		res[i] = byte(i)
+	}
+	return res
+}
+
+func newContentStorage(storageCapacityInMB uint64, nodeId enode.ID, nodeDataDir string) (*ContentStorage, error) {
+	db, err := NewDB(nodeDataDir)
+	if err != nil {
+		return nil, err
+	}
+	hs, err := NewHistoryStorage(storage.PortalStorageConfig{
+		DB:                db,
+		StorageCapacityMB: storageCapacityInMB,
+		NodeId:            nodeId,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return hs.(*ContentStorage), nil
+}
+
+func TestBasicStorage(t *testing.T) {
+	zeroNodeId := uint256.NewInt(0).Bytes32()
+	storage, err := newContentStorage(math.MaxUint32, zeroNodeId, nodeDataDir)
+	assert.NoError(t, err)
+	defer clearNodeData()
+	defer storage.Close()
+
+	contentId := []byte("test")
+	content := []byte("value")
+
+	_, err = storage.Get(nil, contentId)
+	assert.Equal(t, contentStorage.ErrContentNotFound, err)
+
+	pt := storage.put(contentId, content)
+	assert.NoError(t, pt.Err())
+
+	val, err := storage.Get(nil, contentId)
+	assert.NoError(t, err)
+	assert.Equal(t, content, val)
+
+	count, err := storage.ContentCount()
+	assert.NoError(t, err)
+	assert.Equal(t, count, uint64(1))
+
+	size, err := storage.Size()
+	assert.NoError(t, err)
+	assert.True(t, size > 0)
+
+	unusedSize, err := storage.UnusedSize()
+	assert.NoError(t, err)
+
+	usedSize, err := storage.UsedSize()
+	assert.NoError(t, err)
+	assert.True(t, usedSize == size-unusedSize)
+}
+
+func TestDBSize(t *testing.T) {
+	zeroNodeId := uint256.NewInt(0).Bytes32()
+	storage, err := newContentStorage(math.MaxUint32, zeroNodeId, nodeDataDir)
+	assert.NoError(t, err)
+	defer clearNodeData()
+	defer storage.Close()
+
+	numBytes := 10000
+
+	size1, err := storage.Size()
+	assert.NoError(t, err)
+	putResult := storage.put(uint256.NewInt(1).Bytes(), genBytes(numBytes))
+	assert.Nil(t, putResult.Err())
+
+	size2, err := storage.Size()
+	assert.NoError(t, err)
+	putResult = storage.put(uint256.NewInt(2).Bytes(), genBytes(numBytes))
+	assert.NoError(t, putResult.Err())
+
+	size3, err := storage.Size()
+	assert.NoError(t, err)
+	putResult = storage.put(uint256.NewInt(2).Bytes(), genBytes(numBytes))
+	assert.NoError(t, putResult.Err())
+
+	size4, err := storage.Size()
+	assert.NoError(t, err)
+	usedSize, err := storage.UsedSize()
+	assert.NoError(t, err)
+
+	assert.True(t, size2 > size1)
+	assert.True(t, size3 > size2)
+	assert.True(t, size4 == size3)
+	assert.True(t, usedSize == size4)
+
+	err = storage.del(uint256.NewInt(2).Bytes())
+	assert.NoError(t, err)
+	err = storage.del(uint256.NewInt(1).Bytes())
+	assert.NoError(t, err)
+
+	usedSize1, err := storage.UsedSize()
+	assert.NoError(t, err)
+	size5, err := storage.Size()
+	assert.NoError(t, err)
+
+	assert.True(t, size4 == size5)
+	assert.True(t, usedSize1 < size5)
+
+	err = storage.ReclaimSpace()
+	assert.NoError(t, err)
+
+	usedSize2, err := storage.UsedSize()
+	assert.NoError(t, err)
+	size6, err := storage.Size()
+	assert.NoError(t, err)
+
+	assert.Equal(t, size1, size6)
+	assert.Equal(t, usedSize2, size6)
+}
+
+func TestDBPruning(t *testing.T) {
+	storageCapacity := uint64(1)
+
+	zeroNodeId := uint256.NewInt(0).Bytes32()
+	storage, err := newContentStorage(storageCapacity, zeroNodeId, nodeDataDir)
+	assert.NoError(t, err)
+	defer clearNodeData()
+	defer storage.Close()
+
+	furthestElement := uint256.NewInt(40)
+	secondFurthest := uint256.NewInt(30)
+	thirdFurthest := uint256.NewInt(20)
+
+	numBytes := 100_000
+	// test with private put method
+	pt1 := storage.put(uint256.NewInt(1).Bytes(), genBytes(numBytes))
+	assert.NoError(t, pt1.Err())
+	pt2 := storage.put(thirdFurthest.Bytes(), genBytes(numBytes))
+	assert.NoError(t, pt2.Err())
+	pt3 := storage.put(uint256.NewInt(3).Bytes(), genBytes(numBytes))
+	assert.NoError(t, pt3.Err())
+	pt4 := storage.put(uint256.NewInt(10).Bytes(), genBytes(numBytes))
+	assert.NoError(t, pt4.Err())
+	pt5 := storage.put(uint256.NewInt(5).Bytes(), genBytes(numBytes))
+	assert.NoError(t, pt5.Err())
+	pt6 := storage.put(uint256.NewInt(11).Bytes(), genBytes(numBytes))
+	assert.NoError(t, pt6.Err())
+	pt7 := storage.put(furthestElement.Bytes(), genBytes(40000))
+	assert.NoError(t, pt7.Err())
+	pt8 := storage.put(secondFurthest.Bytes(), genBytes(30000))
+	assert.NoError(t, pt8.Err())
+	pt9 := storage.put(uint256.NewInt(2).Bytes(), genBytes(numBytes*2))
+	assert.NoError(t, pt9.Err())
+
+	res, _ := storage.GetLargestDistance()
+
+	assert.Equal(t, res, uint256.NewInt(40))
+	pt10 := storage.put(uint256.NewInt(4).Bytes(), genBytes(132000))
+	assert.NoError(t, pt10.Err())
+
+	assert.False(t, pt1.Pruned())
+	assert.False(t, pt2.Pruned())
+	assert.False(t, pt3.Pruned())
+	assert.False(t, pt4.Pruned())
+	assert.False(t, pt5.Pruned())
+	assert.False(t, pt6.Pruned())
+	assert.False(t, pt7.Pruned())
+	assert.False(t, pt8.Pruned())
+	assert.False(t, pt9.Pruned())
+	assert.True(t, pt10.Pruned())
+
+	assert.Equal(t, pt10.PrunedCount(), 2)
+	usedSize, err := storage.UsedSize()
+	assert.NoError(t, err)
+	assert.True(t, usedSize < storage.storageCapacityInBytes)
+
+	_, err = storage.Get(nil, furthestElement.Bytes())
+	assert.Equal(t, contentStorage.ErrContentNotFound, err)
+
+	_, err = storage.Get(nil, secondFurthest.Bytes())
+	assert.Equal(t, contentStorage.ErrContentNotFound, err)
+
+	val, err := storage.Get(nil, thirdFurthest.Bytes())
+	assert.NoError(t, err)
+	assert.NotNil(t, val)
+}
+
+func TestGetLargestDistance(t *testing.T) {
+	storageCapacity := uint64(1)
+
+	zeroNodeId := uint256.NewInt(0).Bytes32()
+	storage, err := newContentStorage(storageCapacity, zeroNodeId, nodeDataDir)
+	assert.NoError(t, err)
+	defer clearNodeData()
+	defer storage.Close()
+
+	furthestElement := uint256.NewInt(40)
+	secondFurthest := uint256.NewInt(30)
+
+	pt7 := storage.put(furthestElement.Bytes(), genBytes(2000))
+	assert.NoError(t, pt7.Err())
+
+	val, err := storage.Get(nil, furthestElement.Bytes())
+	assert.NoError(t, err)
+	assert.NotNil(t, val)
+	pt8 := storage.put(secondFurthest.Bytes(), genBytes(2000))
+	assert.NoError(t, pt8.Err())
+	res, err := storage.GetLargestDistance()
+	assert.NoError(t, err)
+	assert.Equal(t, furthestElement, res)
+}
+
+func TestSimpleForcePruning(t *testing.T) {
+	storageCapacity := uint64(100_000)
+
+	zeroNodeId := uint256.NewInt(0).Bytes32()
+	storage, err := newContentStorage(storageCapacity, zeroNodeId, nodeDataDir)
+	assert.NoError(t, err)
+	defer clearNodeData()
+	defer storage.Close()
+
+	furthestElement := uint256.NewInt(40)
+	secondFurthest := uint256.NewInt(30)
+	third := uint256.NewInt(10)
+
+	pt1 := storage.put(furthestElement.Bytes(), genBytes(2000))
+	assert.NoError(t, pt1.Err())
+
+	pt2 := storage.put(secondFurthest.Bytes(), genBytes(2000))
+	assert.NoError(t, pt2.Err())
+
+	pt3 := storage.put(third.Bytes(), genBytes(2000))
+	assert.NoError(t, pt3.Err())
+	res, err := storage.GetLargestDistance()
+	assert.NoError(t, err)
+	assert.Equal(t, furthestElement, res)
+
+	err = storage.ForcePrune(uint256.NewInt(20))
+	assert.NoError(t, err)
+
+	_, err = storage.Get(nil, furthestElement.Bytes())
+	assert.Equal(t, contentStorage.ErrContentNotFound, err)
+
+	_, err = storage.Get(nil, secondFurthest.Bytes())
+	assert.Equal(t, contentStorage.ErrContentNotFound, err)
+
+	_, err = storage.Get(nil, third.Bytes())
+	assert.NoError(t, err)
+}
+
+func TestForcePruning(t *testing.T) {
+	const startCap = uint64(14_159_872)
+	const endCapacity = uint64(5000_000)
+	const amountOfItems = 10_000
+
+	maxUint256 := uint256.MustFromHex("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff")
+
+	nodeId := uint256.MustFromHex("0x30994892f3e4889d99deb5340050510d1842778acc7a7948adffa475fed51d6e").Bytes()
+	content := genBytes(1000)
+
+	storage, err := newContentStorage(startCap, enode.ID(nodeId), nodeDataDir)
+	assert.NoError(t, err)
+	defer clearNodeData()
+	defer storage.Close()
+
+	storage.storageCapacityInBytes = startCap
+
+	increment := uint256.NewInt(0).Div(maxUint256, uint256.NewInt(amountOfItems))
+	remainder := uint256.NewInt(0).Mod(maxUint256, uint256.NewInt(amountOfItems))
+
+	id := uint256.NewInt(0)
+	putCount := 0
+	// id < maxUint256 - remainder
+	for id.Cmp(uint256.NewInt(0).Sub(maxUint256, remainder)) == -1 {
+		res := storage.put(id.Bytes(), content)
+		assert.NoError(t, res.Err())
+		id = id.Add(id, increment)
+		putCount++
+	}
+
+	storage.storageCapacityInBytes = endCapacity
+
+	oldDistance, err := storage.GetLargestDistance()
+	assert.NoError(t, err)
+	newDistance, err := storage.EstimateNewRadius(oldDistance)
+	assert.NoError(t, err)
+	assert.NotEqual(t, oldDistance.Cmp(newDistance), -1)
+	err = storage.ForcePrune(newDistance)
+	assert.NoError(t, err)
+
+	var total int64
+	err = storage.sqliteDB.QueryRow("SELECT count(*) FROM kvstore where greater(xor(key, (?1)), (?2)) = 1", storage.nodeId[:], newDistance.Bytes()).Scan(&total)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), total)
+}