Skip to content

feat: turn snapshot prune async #2809

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
blindchaser wants to merge 3 commits into from
Closed

feat: turn snapshot prune async #2809

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
057cec0
feat: turn snapshot prune async
blindchaser Feb 5, 2026
79afb3b
update to use cas
blindchaser Feb 5, 2026
10c7ff4
fix unit test
blindchaser Feb 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 22 additions & 8 deletions sei-db/sc/memiavl/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
"strconv"
"strings"
"sync"
"sync/atomic"
"time"

"github.com/alitto/pond"
Expand Down Expand Up @@ -69,8 +70,6 @@
// timestamp of the last successful snapshot creation
// Protected by db.mtx (only accessed in Commit call chain)
lastSnapshotTime time.Time
// make sure only one snapshot rewrite is running
pruneSnapshotLock sync.Mutex

// the changelog stream persists all the changesets
streamHandler types.Stream[proto.ChangelogEntry]
Expand All @@ -88,6 +87,9 @@
mtx sync.Mutex
// worker goroutine IdleTimeout = 5s
snapshotWriterPool *pond.WorkerPool

// pruningInProgress guards concurrent prune operations (CAS-based)
pruningInProgress atomic.Bool
}

const (
Expand Down Expand Up @@ -481,7 +483,7 @@
TotalMemNodeSize.Store(0)
TotalNumOfMemNode.Store(0)
db.logger.Info("switched to new memiavl snapshot", "version", db.MultiTree.Version())
db.pruneSnapshots()
go db.pruneSnapshots()

default:
}
Expand All @@ -491,9 +493,16 @@

// pruneSnapshot prune the old snapshots
func (db *DB) pruneSnapshots() {
// wait until last prune finish
db.pruneSnapshotLock.Lock()
defer db.pruneSnapshotLock.Unlock()
// CAS: only one prune can run at a time
if !db.pruningInProgress.CompareAndSwap(false, true) {
Copy link
Copy Markdown
Collaborator

@masih masih Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What if a prune takes just long enough to overlap with the next prune?
In that case the new prune cycle will be skipped and we end up with 2 cycles of unpruned state in the next prune

Copy link
Copy Markdown
Contributor Author

@blindchaser blindchaser Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yup, looking into the code. we will need to optimize some pruning logic and where/when pruning happens. creating a ticket for that.

Copy link
Copy Markdown
Contributor

@yzang2019 yzang2019 Feb 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Each prune should prune all snapshots that need to be pruned, not just delete one snapshot. We also should make the pruning logic to be triggered after each commit, instead of only after snapshot creation

return
}
defer db.pruningInProgress.Store(false)

startTime := time.Now()

Check warning

Code scanning / CodeQL

Calling the system time Warning

Calling the system time may be a possible source of non-determinism
defer func() {
db.logger.Info("pruneSnapshots completed", "duration", time.Since(startTime))
}()

currentVersion, err := currentVersion(db.dir)
if err != nil {
Expand Down Expand Up @@ -532,6 +541,9 @@
db.logger.Error("failed to find first snapshot", "err", err)
}

if db.streamHandler == nil {
return
Comment thread
blindchaser marked this conversation as resolved.
}
if err := db.streamHandler.TruncateBefore(utils.VersionToIndex(earliestVersion+1, db.initialVersion.Load())); err != nil {
Comment thread
blindchaser marked this conversation as resolved.
db.logger.Error("failed to truncate rlog", "err", err, "version", earliestVersion+1)
}
Expand Down Expand Up @@ -833,9 +845,11 @@
db.logger.Info("Closing memiavl db...")
db.mtx.Lock()
defer db.mtx.Unlock()
// Wait for any ongoing prune to finish, then block new prunes
for !db.pruningInProgress.CompareAndSwap(false, true) {
time.Sleep(time.Millisecond)
}
errs := []error{}
db.pruneSnapshotLock.Lock()
defer db.pruneSnapshotLock.Unlock()
// Close stream handler
db.logger.Info("Closing stream handler...")
if db.streamHandler != nil {
Expand Down
6 changes: 4 additions & 2 deletions sei-db/sc/memiavl/db_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,8 +153,10 @@ func TestRewriteSnapshotBackground(t *testing.T) {
close(stopCh)
wg.Wait()

db.pruneSnapshotLock.Lock()
defer db.pruneSnapshotLock.Unlock()
// Wait for any ongoing prune to finish
require.Eventually(t, func() bool {
return !db.pruningInProgress.Load()
}, time.Second, 10*time.Millisecond, "prune should complete")

entries, err := os.ReadDir(db.dir)
require.NoError(t, err)
Expand Down
Loading