raftengine/etcd: clean up orphaned snapshot spool files on startup

Snapshot spool files (elastickv-etcd-snapshot-*) were left behind when
the engine crashed before snapshotSpool.Close() could run. Since the
spool directory was moved from os temp to cfg.DataDir, these files
accumulated without bound, eventually filling the disk.

Add cleanupStaleSnapshotSpools() and call it from openDiskState() so
orphaned spools are removed on every engine restart.

Author: bootjp

internal/raftengine/etcd/snapshot_spool.go

@@ -3,6 +3,7 @@ package etcd
 import (
 	"io"
 	"os"
+	"path/filepath"
 
 	"github.com/cockroachdb/errors"
 )
@@ -64,6 +65,24 @@ func (s *snapshotSpool) Reader() (io.Reader, error) {
 	return s.file, nil
 }
 
+// cleanupStaleSnapshotSpools removes orphaned snapshot spool files left behind
+// by a previous engine instance that crashed before Close could run.
+func cleanupStaleSnapshotSpools(dir string) error {
+	matches, err := filepath.Glob(filepath.Join(dir, snapshotSpoolPattern))
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	var combined error
+	for _, match := range matches {
+		removeErr := os.Remove(match)
+		if removeErr == nil || os.IsNotExist(removeErr) {
+			continue
+		}
+		combined = errors.CombineErrors(combined, errors.WithStack(removeErr))
+	}
+	return errors.WithStack(combined)
+}
+
 func (s *snapshotSpool) Close() error {
 	if s == nil {
 		return nil

internal/raftengine/etcd/snapshot_spool_test.go

@@ -0,0 +1,44 @@
+package etcd
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCleanupStaleSnapshotSpools(t *testing.T) {
+	dir := t.TempDir()
+
+	// Create several orphaned spool files matching the pattern.
+	for i := 0; i < 5; i++ {
+		f, err := os.CreateTemp(dir, snapshotSpoolPattern)
+		require.NoError(t, err)
+		require.NoError(t, f.Close())
+	}
+
+	// Create an unrelated file that must not be removed.
+	unrelated := filepath.Join(dir, "keep-me.txt")
+	require.NoError(t, os.WriteFile(unrelated, []byte("data"), 0o644))
+
+	matches, err := filepath.Glob(filepath.Join(dir, snapshotSpoolPattern))
+	require.NoError(t, err)
+	require.Len(t, matches, 5)
+
+	require.NoError(t, cleanupStaleSnapshotSpools(dir))
+
+	// All spool files should be gone.
+	matches, err = filepath.Glob(filepath.Join(dir, snapshotSpoolPattern))
+	require.NoError(t, err)
+	require.Empty(t, matches)
+
+	// Unrelated file should still exist.
+	_, err = os.Stat(unrelated)
+	require.NoError(t, err)
+}
+
+func TestCleanupStaleSnapshotSpoolsEmptyDir(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, cleanupStaleSnapshotSpools(dir))
+}

internal/raftengine/etcd/wal_store.go

@@ -38,6 +38,10 @@ func openDiskState(cfg OpenConfig, peers []Peer) (*diskState, error) {
 		return nil, errors.WithStack(err)
 	}
 
+	if err := cleanupStaleSnapshotSpools(cfg.DataDir); err != nil {
+		return nil, errors.Wrap(err, "cleanup stale snapshot spools")
+	}
+
 	if wal.Exist(walDir) {
 		return loadWalState(logger, walDir, snapDir, cfg.StateMachine)
 	}