feat(store/v2): remove the pruning manager (#19411)
Co-authored-by: Aleksandr Bezobchuk <alexanderbez@users.noreply.github.com>
This commit is contained in:
cool-developer
co-authored by
Aleksandr Bezobchuk
Aleksandr Bezobchuk
parent
92eb6de6e3
commit
8fb9ca87b2
@@ -1,15 +0,0 @@
|
||||
# Pruning
|
||||
|
||||
## Overview
|
||||
|
||||
Pruning is the mechanism for deleting old versions data from both state storage and commitment. The pruning operation is triggered periodically.
|
||||
|
||||
## Pruning Options
|
||||
|
||||
Generally, there are three configurable parameters for pruning options:
|
||||
|
||||
- `pruning-keep-recent`: the number of recent versions to keep.
|
||||
- `pruning-interval`: the interval between two pruning operations.
|
||||
- `pruning-sync`: the flag to sync/async the pruning operation.
|
||||
|
||||
Different options will be applied to the state storage and commitment. The pruning option have an effect on the snapshot operation, but it will not manage the conflict resolution in SDK, it is the responsibility of the dedicated backend.
|
||||
@@ -1,166 +0,0 @@
|
||||
package pruning
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"cosmossdk.io/log"
|
||||
"cosmossdk.io/store/v2"
|
||||
)
|
||||
|
||||
// Manager is an abstraction to handle pruning of SS and SC backends.
|
||||
type Manager struct {
|
||||
mtx sync.Mutex
|
||||
isStarted bool
|
||||
|
||||
stateStorage store.VersionedDatabase
|
||||
stateCommitment store.Committer
|
||||
|
||||
logger log.Logger
|
||||
storageOpts Options
|
||||
commitmentOpts Options
|
||||
|
||||
chStorage chan struct{}
|
||||
chCommitment chan struct{}
|
||||
}
|
||||
|
||||
// NewManager creates a new Manager instance.
|
||||
func NewManager(
|
||||
logger log.Logger,
|
||||
ss store.VersionedDatabase,
|
||||
sc store.Committer,
|
||||
) *Manager {
|
||||
return &Manager{
|
||||
stateStorage: ss,
|
||||
stateCommitment: sc,
|
||||
logger: logger,
|
||||
storageOpts: DefaultOptions(),
|
||||
commitmentOpts: DefaultOptions(),
|
||||
}
|
||||
}
|
||||
|
||||
// SetStorageOptions sets the state storage options.
|
||||
func (m *Manager) SetStorageOptions(opts Options) {
|
||||
m.mtx.Lock()
|
||||
defer m.mtx.Unlock()
|
||||
|
||||
m.storageOpts = opts
|
||||
}
|
||||
|
||||
// SetCommitmentOptions sets the state commitment options.
|
||||
func (m *Manager) SetCommitmentOptions(opts Options) {
|
||||
m.mtx.Lock()
|
||||
defer m.mtx.Unlock()
|
||||
|
||||
m.commitmentOpts = opts
|
||||
}
|
||||
|
||||
// Start starts the manager.
|
||||
func (m *Manager) Start() {
|
||||
m.mtx.Lock()
|
||||
defer m.mtx.Unlock()
|
||||
|
||||
if m.isStarted {
|
||||
return
|
||||
}
|
||||
m.isStarted = true
|
||||
|
||||
if !m.storageOpts.Sync {
|
||||
m.chStorage = make(chan struct{}, 1)
|
||||
m.chStorage <- struct{}{}
|
||||
}
|
||||
if !m.commitmentOpts.Sync {
|
||||
m.chCommitment = make(chan struct{}, 1)
|
||||
m.chCommitment <- struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
// Stop stops the manager and waits for all goroutines to finish.
|
||||
func (m *Manager) Stop() {
|
||||
m.mtx.Lock()
|
||||
defer m.mtx.Unlock()
|
||||
|
||||
if !m.isStarted {
|
||||
return
|
||||
}
|
||||
m.isStarted = false
|
||||
|
||||
if !m.storageOpts.Sync {
|
||||
<-m.chStorage
|
||||
close(m.chStorage)
|
||||
}
|
||||
if !m.commitmentOpts.Sync {
|
||||
<-m.chCommitment
|
||||
close(m.chCommitment)
|
||||
}
|
||||
}
|
||||
|
||||
// Prune prunes the state storage and state commitment.
|
||||
// It will check the pruning conditions and prune if necessary.
|
||||
func (m *Manager) Prune(height uint64) {
|
||||
m.mtx.Lock()
|
||||
defer m.mtx.Unlock()
|
||||
|
||||
if !m.isStarted {
|
||||
return
|
||||
}
|
||||
|
||||
// storage pruning
|
||||
if m.storageOpts.Interval > 0 && height > m.storageOpts.KeepRecent && height%m.storageOpts.Interval == 0 {
|
||||
pruneHeight := height - m.storageOpts.KeepRecent - 1
|
||||
if m.storageOpts.Sync {
|
||||
m.pruneStorage(pruneHeight)
|
||||
} else {
|
||||
// it will not block if the previous pruning is still running
|
||||
select {
|
||||
case _, stillOpen := <-m.chStorage:
|
||||
if stillOpen {
|
||||
go func() {
|
||||
m.pruneStorage(pruneHeight)
|
||||
m.chStorage <- struct{}{}
|
||||
}()
|
||||
}
|
||||
|
||||
default:
|
||||
m.logger.Debug("storage pruning is still running; skipping", "version", pruneHeight)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// commitment pruning
|
||||
if m.commitmentOpts.Interval > 0 && height > m.commitmentOpts.KeepRecent && height%m.commitmentOpts.Interval == 0 {
|
||||
pruneHeight := height - m.commitmentOpts.KeepRecent - 1
|
||||
if m.commitmentOpts.Sync {
|
||||
m.pruneCommitment(pruneHeight)
|
||||
} else {
|
||||
// it will not block if the previous pruning is still running
|
||||
select {
|
||||
case _, stillOpen := <-m.chCommitment:
|
||||
if stillOpen {
|
||||
go func() {
|
||||
m.pruneCommitment(pruneHeight)
|
||||
m.chCommitment <- struct{}{}
|
||||
}()
|
||||
}
|
||||
|
||||
default:
|
||||
m.logger.Debug("commitment pruning is still running; skipping", "version", pruneHeight)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) pruneStorage(height uint64) {
|
||||
m.logger.Debug("pruning state storage", "height", height)
|
||||
|
||||
if err := m.stateStorage.Prune(height); err != nil {
|
||||
m.logger.Error("failed to prune state storage", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) pruneCommitment(height uint64) {
|
||||
m.logger.Debug("pruning state commitment", "height", height)
|
||||
|
||||
if err := m.stateCommitment.Prune(height); err != nil {
|
||||
m.logger.Error("failed to prune state commitment", "err", err)
|
||||
}
|
||||
}
|
||||
@@ -1,105 +0,0 @@
|
||||
package pruning
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/suite"
|
||||
|
||||
"cosmossdk.io/log"
|
||||
"cosmossdk.io/store/v2"
|
||||
"cosmossdk.io/store/v2/commitment"
|
||||
"cosmossdk.io/store/v2/commitment/iavl"
|
||||
dbm "cosmossdk.io/store/v2/db"
|
||||
"cosmossdk.io/store/v2/storage"
|
||||
"cosmossdk.io/store/v2/storage/sqlite"
|
||||
)
|
||||
|
||||
const defaultStoreKey = "default"
|
||||
|
||||
type PruningTestSuite struct {
|
||||
suite.Suite
|
||||
|
||||
manager *Manager
|
||||
ss store.VersionedDatabase
|
||||
sc store.Committer
|
||||
}
|
||||
|
||||
func TestPruningTestSuite(t *testing.T) {
|
||||
suite.Run(t, &PruningTestSuite{})
|
||||
}
|
||||
|
||||
func (s *PruningTestSuite) SetupTest() {
|
||||
logger := log.NewNopLogger()
|
||||
if testing.Verbose() {
|
||||
logger = log.NewTestLogger(s.T())
|
||||
}
|
||||
|
||||
sqliteDB, err := sqlite.New(s.T().TempDir())
|
||||
s.Require().NoError(err)
|
||||
ss := storage.NewStorageStore(sqliteDB)
|
||||
|
||||
tree := iavl.NewIavlTree(dbm.NewMemDB(), log.NewNopLogger(), iavl.DefaultConfig())
|
||||
sc, err := commitment.NewCommitStore(map[string]commitment.Tree{"default": tree}, dbm.NewMemDB(), logger)
|
||||
s.Require().NoError(err)
|
||||
|
||||
s.manager = NewManager(logger, ss, sc)
|
||||
s.ss = ss
|
||||
s.sc = sc
|
||||
}
|
||||
|
||||
func (s *PruningTestSuite) TearDownTest() {
|
||||
s.manager.Start()
|
||||
s.manager.Stop()
|
||||
}
|
||||
|
||||
func (s *PruningTestSuite) TestPruning() {
|
||||
s.manager.SetCommitmentOptions(Options{4, 2, true})
|
||||
s.manager.SetStorageOptions(Options{3, 3, true})
|
||||
s.manager.Start()
|
||||
|
||||
latestVersion := uint64(100)
|
||||
|
||||
// write batches
|
||||
for i := uint64(0); i < latestVersion; i++ {
|
||||
version := i + 1
|
||||
|
||||
cs := store.NewChangesetWithPairs(map[string]store.KVPairs{defaultStoreKey: {}})
|
||||
cs.AddKVPair(defaultStoreKey, store.KVPair{
|
||||
Key: []byte("key"),
|
||||
Value: []byte(fmt.Sprintf("value%d", version)),
|
||||
})
|
||||
err := s.sc.WriteBatch(cs)
|
||||
s.Require().NoError(err)
|
||||
|
||||
_, err = s.sc.Commit(version)
|
||||
s.Require().NoError(err)
|
||||
|
||||
err = s.ss.ApplyChangeset(version, cs)
|
||||
s.Require().NoError(err)
|
||||
s.manager.Prune(version)
|
||||
}
|
||||
|
||||
// wait for pruning to finish
|
||||
s.manager.Stop()
|
||||
|
||||
// check the store for the version 96
|
||||
val, err := s.ss.Get(defaultStoreKey, 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(defaultStoreKey, 50, []byte("key"))
|
||||
s.Require().Error(err)
|
||||
s.Require().Nil(val)
|
||||
|
||||
// check the commitment for the version 96
|
||||
proofOps, err := s.sc.GetProof(defaultStoreKey, latestVersion-4, []byte("key"))
|
||||
s.Require().NoError(err)
|
||||
s.Require().Len(proofOps, 2)
|
||||
|
||||
// check the commitment for the version 95
|
||||
proofOps, err = s.sc.GetProof(defaultStoreKey, latestVersion-5, []byte("key"))
|
||||
s.Require().Error(err)
|
||||
s.Require().Nil(proofOps)
|
||||
}
|
||||
@@ -1,25 +0,0 @@
|
||||
package pruning
|
||||
|
||||
// Options defines the pruning configuration.
|
||||
type Options struct {
|
||||
// KeepRecent sets the number of recent versions to keep.
|
||||
KeepRecent uint64
|
||||
|
||||
// Interval sets the number of how often to prune.
|
||||
// If set to 0, no pruning will be done.
|
||||
Interval uint64
|
||||
|
||||
// Sync when set to true ensure that pruning will be performed
|
||||
// synchronously, otherwise by default it will be done asynchronously.
|
||||
Sync bool
|
||||
}
|
||||
|
||||
// DefaultOptions returns the default pruning options.
|
||||
// Interval is set to 0, which means no pruning will be done.
|
||||
func DefaultOptions() Options {
|
||||
return Options{
|
||||
KeepRecent: 0,
|
||||
Interval: 0,
|
||||
Sync: false,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user