fix(raft): bump snapshot spool cap to 16 GiB + env override

Receive-side snapshot spool was hardcoded to 1 GiB
(maxSnapshotPayloadBytes). Production FSM snapshots at 1.35 GiB
exceeded that ceiling: snapshotSpool.Write returned
errSnapshotPayloadTooLarge mid-stream, the gRPC SendSnapshot stream
broke, and etcd raft retried the snapshot indefinitely because
each retry hit the same wall.

Followers stuck at stale applied indices (the 213 follower in the
2026-05-08 incident never moved past applied=26,459,962 — over a
million entries behind), the leader sustained ~100 MB/s outbound for
hours sending the same 1.35 GiB snapshot over and over, and host
disks saturated at 73-99% util. Each receive cycle re-created an
elastickv-etcd-snapshot-* spool file with a fresh random suffix,
making the loop visible from the outside as continuously-changing
in-progress filenames.

Fix:
- Default cap raised to 16 GiB (~12x the production-observed FSM
  size) so it does not drift back into the runway as data grows.
- Cap is now resolved per spool creation via
  ELASTICKV_RAFT_MAX_SNAPSHOT_PAYLOAD_BYTES, so an operator can
  raise it without a binary rebuild if even the new default is
  ever insufficient.
- Each spool instance captures its own maxSize at construction
  rather than reading a package-level var on every Write, so a
  test or env flip cannot tear an in-flight receive.

The cap still exists -- defense against a misbehaving / compromised
peer streaming unbounded data into the spool dir is the original
intent, and that intent survives -- but the magnitude is now
realistic.

Self-review (5 lenses):
1. Data loss -- none. The cap was rejecting valid snapshots;
   raising it lets receivers actually accept FSM transfers they
   should already have been accepting. No persisted state changes.
2. Concurrency -- maxSize is captured at newSnapshotSpool time and
   read-only thereafter. No new locks. The env resolver is plain
   os.Getenv + ParseInt; no shared state.
3. Performance -- one Getenv + ParseInt per snapshot creation.
   Snapshots are infrequent (hours-scale on a stable cluster), so
   negligible. The 16 GiB default does NOT pre-allocate; the spool
   grows on disk only as bytes arrive.
4. Data consistency -- snapshot integrity unchanged. The fix only
   widens the reception envelope; the same chunk-validation,
   metadata, and final-flag handling apply.
5. Test coverage -- TestSnapshotSpool_DefaultCapAcceptsRealisticFSM
   pins the regression by writing 1.5 GiB through Write
   (skipped under -short to keep `make test` fast).
   TestSnapshotSpool_OverrideViaEnv exercises a lowered-cap value
   to confirm the env knob actually moves the cap and the
   errSnapshotPayloadTooLarge sentinel still surfaces past it.
   TestSnapshotSpool_OverrideInvalidFallsBack pins fail-soft on
   malformed env input so a typo doesn't zero the cap.

Test:
  go test -race -count=1 -short ./internal/raftengine/etcd
  -- 11.4s, all green.
  go test -race -count=1 \
    -run TestSnapshotSpool_DefaultCapAcceptsRealisticFSM \
    ./internal/raftengine/etcd
  -- 1.96s, green (1.5 GiB write succeeds).

Future work (separate PRs):
- snapshotSpool.Bytes() materializes the entire payload as []byte
  for RawNode.Step. With 16 GiB allowed, this is a real OOM risk
  on memory-constrained nodes. Streaming snapshot apply
  (RawNode.Step accepts an io.Reader, or the FSM-side path
  bypasses raftpb materialization) is the next step.
- Make the leader respect a follower-advertised receive cap so a
  cluster running mixed binaries can negotiate a safe value.

Author: bootjp

internal/raftengine/etcd/snapshot_spool.go

@@ -2,40 +2,77 @@ package etcd
 
 import (
 	"io"
+	"log/slog"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 
 	"github.com/cockroachdb/errors"
 )
 
-var (
-	// The current raftpb snapshot APIs still materialize payloads as []byte, so
-	// the prototype cannot stream snapshots end-to-end yet. Keep the payload on
-	// disk while assembling it and fail fast before unbounded growth.
-	maxSnapshotPayloadBytes int64 = 1 << 30 // 1 GiB
+// defaultMaxSnapshotPayloadBytes is the receive-side cap on a single snapshot
+// stream's spooled payload. Production hit a 1 GiB ceiling here that was
+// silently rejecting real-world FSM transfers (1.35 GiB+), so the receiver
+// returned errSnapshotPayloadTooLarge mid-stream, the gRPC stream broke,
+// and etcd raft retried — indefinitely, since each retry hit the same wall.
+// Followers stuck at stale applied indices, leader sustained ~100 MB/s
+// outbound, host disks saturated for hours.
+//
+// 16 GiB is sized as ~12× the production-observed FSM size so the limit
+// does not drift back into the runway as data grows. The cap still exists
+// so a misbehaving / compromised peer cannot stream unbounded data into
+// the spool dir; operators can raise it further via
+// ELASTICKV_RAFT_MAX_SNAPSHOT_PAYLOAD_BYTES if a real FSM ever exceeds
+// even this default.
+const defaultMaxSnapshotPayloadBytes int64 = 16 << 30 // 16 GiB
 
-	errSnapshotPayloadTooLarge = errors.New("etcd raft snapshot payload exceeds limit")
-)
+const maxSnapshotPayloadBytesEnvVar = "ELASTICKV_RAFT_MAX_SNAPSHOT_PAYLOAD_BYTES"
+
+// resolveMaxSnapshotPayloadBytes evaluates the env override once per spool
+// creation. Snapshots are infrequent enough that one Getenv + ParseInt per
+// spool is invisible in profiles, and resolving at construction means tests
+// that flip the env via t.Setenv don't have to mutate process-wide globals.
+func resolveMaxSnapshotPayloadBytes() int64 {
+	v := strings.TrimSpace(os.Getenv(maxSnapshotPayloadBytesEnvVar))
+	if v == "" {
+		return defaultMaxSnapshotPayloadBytes
+	}
+	n, err := strconv.ParseInt(v, 10, 64)
+	if err != nil || n <= 0 {
+		slog.Warn("invalid ELASTICKV_RAFT_MAX_SNAPSHOT_PAYLOAD_BYTES; using default",
+			"value", v, "default_bytes", defaultMaxSnapshotPayloadBytes)
+		return defaultMaxSnapshotPayloadBytes
+	}
+	return n
+}
+
+var errSnapshotPayloadTooLarge = errors.New("etcd raft snapshot payload exceeds limit")
 
 const snapshotSpoolPattern = "elastickv-etcd-snapshot-*"
 
 type snapshotSpool struct {
-	file *os.File
-	path string
-	size int64
+	file    *os.File
+	path    string
+	size    int64
+	maxSize int64
 }
 
 func newSnapshotSpool(dir string) (*snapshotSpool, error) {
 	file, err := os.CreateTemp(dir, snapshotSpoolPattern)
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
-	return &snapshotSpool{file: file, path: file.Name()}, nil
+	return &snapshotSpool{
+		file:    file,
+		path:    file.Name(),
+		maxSize: resolveMaxSnapshotPayloadBytes(),
+	}, nil
 }
 
 func (s *snapshotSpool) Write(p []byte) (int, error) {
-	if int64(len(p))+s.size > maxSnapshotPayloadBytes {
-		return 0, errors.Wrapf(errSnapshotPayloadTooLarge, "%d > %d", int64(len(p))+s.size, maxSnapshotPayloadBytes)
+	if int64(len(p))+s.size > s.maxSize {
+		return 0, errors.Wrapf(errSnapshotPayloadTooLarge, "%d > %d", int64(len(p))+s.size, s.maxSize)
 	}
 	n, err := s.file.Write(p)
 	s.size += int64(n)
@@ -50,7 +87,7 @@ func (s *snapshotSpool) Bytes() ([]byte, error) {
 		return nil, errors.WithStack(err)
 	}
 	// Read incrementally instead of sizing a buffer from s.size so malformed
-	// inputs stay bounded by maxSnapshotPayloadBytes and file-backed I/O.
+	// inputs stay bounded by s.maxSize and file-backed I/O.
 	data, err := io.ReadAll(s.file)
 	if err != nil {
 		return nil, errors.WithStack(err)

internal/raftengine/etcd/snapshot_spool_test.go

@@ -1,14 +1,87 @@
 package etcd
 
 import (
+	"bytes"
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
 	"testing"
 
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
+// TestSnapshotSpool_DefaultCapAcceptsRealisticFSM pins the regression behind
+// the 2026-05-08 incident: with the prior 1 GiB hardcoded cap, any real-world
+// FSM (production observed 1.35 GiB) failed mid-stream with
+// errSnapshotPayloadTooLarge, breaking the gRPC snapshot stream and locking
+// the leader/follower into a retransmit loop. The default cap must accept at
+// least 1.5 GiB without env override.
+func TestSnapshotSpool_DefaultCapAcceptsRealisticFSM(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping: writes 1.5 GiB to a temp file")
+	}
+	dir := t.TempDir()
+	spool, err := newSnapshotSpool(dir)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = spool.Close() })
+
+	// 1.5 GiB exceeds the legacy 1 GiB ceiling and matches realistic
+	// production FSM sizes within the same order of magnitude.
+	const target = int64(1536) << 20 // 1.5 GiB
+	const chunk = 8 << 20            // 8 MiB writes mirror the gRPC snapshot chunk size order
+	buf := bytes.Repeat([]byte{0xAB}, chunk)
+
+	var written int64
+	for written < target {
+		toWrite := chunk
+		if remaining := target - written; remaining < int64(chunk) {
+			toWrite = int(remaining)
+		}
+		n, err := spool.Write(buf[:toWrite])
+		require.NoError(t, err, "write at offset %d unexpectedly failed", written)
+		require.Equal(t, toWrite, n)
+		written += int64(n)
+	}
+	require.Equal(t, target, spool.size)
+}
+
+// TestSnapshotSpool_OverrideViaEnv confirms the env knob actually moves the
+// cap. Tests deliberately *lower* it (cheap to write past) instead of
+// raising — the upper-bound test above already proves a generous cap works.
+func TestSnapshotSpool_OverrideViaEnv(t *testing.T) {
+	const cap = int64(4096)
+	t.Setenv(maxSnapshotPayloadBytesEnvVar, strconv.FormatInt(cap, 10))
+
+	spool, err := newSnapshotSpool(t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = spool.Close() })
+
+	require.Equal(t, cap, spool.maxSize)
+
+	// Write up to the cap — succeeds.
+	_, err = spool.Write(bytes.Repeat([]byte{0x01}, int(cap)))
+	require.NoError(t, err)
+
+	// One byte past — fails with the documented sentinel so callers can
+	// errors.Is against errSnapshotPayloadTooLarge for telemetry.
+	_, err = spool.Write([]byte{0x02})
+	require.Error(t, err)
+	require.True(t, errors.Is(err, errSnapshotPayloadTooLarge), "got %v", err)
+}
+
+// TestSnapshotSpool_OverrideInvalidFallsBack pins the resolver's
+// fail-soft behaviour: a malformed env value must NOT zero the cap (which
+// would make every receive fail) — it falls back to the default.
+func TestSnapshotSpool_OverrideInvalidFallsBack(t *testing.T) {
+	t.Setenv(maxSnapshotPayloadBytesEnvVar, "not-a-number")
+	spool, err := newSnapshotSpool(t.TempDir())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = spool.Close() })
+	require.Equal(t, defaultMaxSnapshotPayloadBytes, spool.maxSize)
+}
+
 func TestCleanupStaleSnapshotSpools(t *testing.T) {
 	dir := t.TempDir()