raftengine/etcd: purge old .snap files after snapshot persistence

The etcd Snapshotter.SaveSnap() writes new .snap files but never
removes old ones. In upstream etcd this is handled by
fileutil.PurgeFile in the etcd server layer, which elastickv does not
use.

Add purgeOldSnapFiles() that retains the most recent 3 .snap files
and call it from both persistLocalSnapshotPayload and
persistCreatedSnapshot. This prevents unbounded disk growth on all
cluster nodes.

Author: bootjp

internal/raftengine/etcd/engine.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/binary"
 	"log/slog"
+	"path/filepath"
 	"sort"
 	"strconv"
 	"sync"
@@ -1317,6 +1318,11 @@ func (e *Engine) persistCreatedSnapshot(snap raftpb.Snapshot) error {
 	if err := e.persist.Release(snap); err != nil {
 		return errors.WithStack(err)
 	}
+
+	snapDir := filepath.Join(e.dataDir, snapDirName)
+	if purgeErr := purgeOldSnapFiles(snapDir, defaultMaxSnapFiles); purgeErr != nil {
+		return errors.Wrap(purgeErr, "purge old snap files")
+	}
 	return nil
 }
 
@@ -2262,16 +2268,18 @@ func (e *Engine) persistLocalSnapshotPayload(index uint64, payload []byte) error
 	_, err = persistLocalSnapshotPayload(e.storage, e.persist, index, payload)
 	switch {
 	case err == nil:
-		return nil
 	case errors.Is(err, etcdraft.ErrCompacted):
-		return nil
 	case errors.Is(err, etcdraft.ErrUnavailable):
-		return nil
 	case errors.Is(err, etcdraft.ErrSnapOutOfDate):
-		return nil
 	default:
 		return err
 	}
+
+	snapDir := filepath.Join(e.dataDir, snapDirName)
+	if purgeErr := purgeOldSnapFiles(snapDir, defaultMaxSnapFiles); purgeErr != nil {
+		return errors.Wrap(purgeErr, "purge old snap files")
+	}
+	return nil
 }
 
 func encodeReadContext(id uint64) []byte {

internal/raftengine/etcd/snapshot_spool_test.go

@@ -1,6 +1,7 @@
 package etcd
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
@@ -42,3 +43,66 @@ func TestCleanupStaleSnapshotSpoolsEmptyDir(t *testing.T) {
 	dir := t.TempDir()
 	require.NoError(t, cleanupStaleSnapshotSpools(dir))
 }
+
+// createSnapFile creates a fake .snap file with the etcd naming convention.
+func createSnapFile(t *testing.T, dir string, term, index uint64) string {
+	t.Helper()
+	name := fmt.Sprintf("%016x-%016x.snap", term, index)
+	path := filepath.Join(dir, name)
+	require.NoError(t, os.WriteFile(path, []byte("fake"), 0o644))
+	return path
+}
+
+func TestPurgeOldSnapFiles(t *testing.T) {
+	dir := t.TempDir()
+
+	// Create 6 snap files at increasing indices.
+	for i := uint64(1); i <= 6; i++ {
+		createSnapFile(t, dir, 1, i*10000)
+	}
+
+	// Create a non-snap file that must be preserved.
+	other := filepath.Join(dir, "db.tmp.12345")
+	require.NoError(t, os.WriteFile(other, []byte("x"), 0o644))
+
+	require.NoError(t, purgeOldSnapFiles(dir, defaultMaxSnapFiles))
+
+	entries, err := os.ReadDir(dir)
+	require.NoError(t, err)
+
+	var snaps []string
+	for _, e := range entries {
+		if filepath.Ext(e.Name()) == ".snap" {
+			snaps = append(snaps, e.Name())
+		}
+	}
+
+	// Only the newest 3 should remain.
+	require.Len(t, snaps, 3)
+	require.Equal(t, fmt.Sprintf("%016x-%016x.snap", 1, uint64(40000)), snaps[0])
+	require.Equal(t, fmt.Sprintf("%016x-%016x.snap", 1, uint64(50000)), snaps[1])
+	require.Equal(t, fmt.Sprintf("%016x-%016x.snap", 1, uint64(60000)), snaps[2])
+
+	// Non-snap file preserved.
+	_, err = os.Stat(other)
+	require.NoError(t, err)
+}
+
+func TestPurgeOldSnapFilesUnderLimit(t *testing.T) {
+	dir := t.TempDir()
+
+	// Only 2 files — under the limit of 3, nothing should be removed.
+	createSnapFile(t, dir, 1, 1000)
+	createSnapFile(t, dir, 1, 2000)
+
+	require.NoError(t, purgeOldSnapFiles(dir, defaultMaxSnapFiles))
+
+	entries, err := os.ReadDir(dir)
+	require.NoError(t, err)
+	require.Len(t, entries, 2)
+}
+
+func TestPurgeOldSnapFilesEmptyDir(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, purgeOldSnapFiles(dir, defaultMaxSnapFiles))
+}

internal/raftengine/etcd/wal_store.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"os"
 	"path/filepath"
+	"sort"
 
 	"github.com/cockroachdb/errors"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
@@ -340,6 +341,45 @@ func persistLocalSnapshotPayload(storage *etcdraft.MemoryStorage, persist etcdst
 	return snapshot, nil
 }
 
+// defaultMaxSnapFiles is the number of .snap files to retain in the snap
+// directory. etcd itself purges old snap files via fileutil.PurgeFile; the
+// elastickv etcd engine must do this explicitly.
+const defaultMaxSnapFiles = 3
+
+// purgeOldSnapFiles removes old .snap files from snapDir, keeping the most
+// recent maxSnap files. Snap file names encode term and index in hex and sort
+// lexicographically from oldest to newest, matching etcd's Snapshotter
+// convention.
+func purgeOldSnapFiles(snapDir string, maxSnap int) error {
+	entries, err := os.ReadDir(snapDir)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	var snaps []string
+	for _, e := range entries {
+		if !e.IsDir() && filepath.Ext(e.Name()) == ".snap" {
+			snaps = append(snaps, e.Name())
+		}
+	}
+
+	if len(snaps) <= maxSnap {
+		return nil
+	}
+
+	// Sort ascending (oldest first) — snap names are zero-padded hex so
+	// lexicographic order equals chronological order.
+	sort.Strings(snaps)
+
+	var combined error
+	for _, name := range snaps[:len(snaps)-maxSnap] {
+		if removeErr := os.Remove(filepath.Join(snapDir, name)); removeErr != nil && !os.IsNotExist(removeErr) {
+			combined = errors.CombineErrors(combined, errors.WithStack(removeErr))
+		}
+	}
+	return errors.WithStack(combined)
+}
+
 func buildLocalSnapshot(storage *etcdraft.MemoryStorage, applied uint64, payload []byte) (raftpb.Snapshot, error) {
 	_, confState, err := storage.InitialState()
 	if err != nil {