fix(raft): remove orphan .fsm on apply failure (PR #747 r4)

Round-3 review on commit 944ab86 from chatgpt-codex (P2):

> The streaming branch finalizes and renames the spool into
> fsmSnapDir before the message is handed to SendSnapshot's
> t.handle path, but FinalizeAsFSMFile clears the spool path so
> deferred spool.Close() cannot clean it up afterward. If
> t.handle returns an error (for example during transient
> engine/raft failures), SendSnapshot returns failure to the
> sender while leaving the newly written .fsm file behind with
> no corresponding applied snapshot, which can accumulate
> orphaned large files across retries with different snapshot
> indexes.

Confirmed:
- After receiveSnapshotStream succeeds, msg.Snapshot.Data is a
  17-byte EKVT token and the .fsm file lives at
  fsmSnapPath(fsmSnapDir, index).
- SendSnapshot then calls t.handle(ctx, msg). The engine's
  applySnapshot is synchronous to t.handle, so a non-nil return
  guarantees applied_index was NOT advanced — the .fsm file is
  unreferenced.
- Same-index retries are safe (os.Rename atomically replaces),
  but the leader can take a fresh snapshot at a higher index
  before the apply finally succeeds, and each failed attempt at
  a different index leaves an orphan .fsm.
- cleanupStaleFSMSnaps only runs at startup (prepareDataDirs), so
  during a long-lived process, orphans accumulate to disk-size
  pressure.

Fix: SendSnapshot calls a new removeOrphanedFSMSnapshot helper
on the apply-failure branch. The helper:

1. Decodes the EKVT token from msg.Snapshot.Data; bails if the
   message used the legacy inline path (no .fsm file to clean).
2. Reads fsmSnapDir under t.mu; bails if it's unset (legacy
   receivers also use the inline path so there's nothing on
   disk).
3. os.Remove(fsmSnapPath(...)) — best-effort, IsNotExist is
   tolerated, all other errors are slog.Warn'd. The original
   apply error is the actionable signal returned to the sender;
   a failed Remove is a secondary concern that startup cleanup
   still picks up.

Test added (TestSendSnapshot_ApplyFailureRemovesFinalizedFSMFile):
- Wires SetFSMSnapDir + SetHandler-that-fails.
- Drives a real testStateMachine-framed payload through
  SendSnapshot.
- Asserts SendSnapshot surfaces the apply error.
- Asserts fsmSnapPath(fsmSnapDir, index) does NOT exist after
  the call — orphan cleanup fired.

Caller audit (semantic change requires it):
- SendSnapshot (this function): only gRPC service handler, no
  in-tree non-test callers. Modified.
- removeOrphanedFSMSnapshot: only called from SendSnapshot.
- t.handle (other callers): line 438 in Send (regular RPC for
  non-snapshot messages, no .fsm involvement) is untouched.
- client.SendSnapshot (lines 267, 455, 481): leader-side client
  invocations are different function, unaffected.

The cleanup is gated by isSnapshotToken(msg.Snapshot.Data) so
the legacy in-memory fallback path (when fsmSnapDir is unset)
sees no behavior change — there's no .fsm file to remove on
that path because Bytes() materializes inline.

Test:
  go test -race -count=1 -short ./internal/raftengine/etcd
  -- 11.3s, all green.
  golangci-lint run --enable-only cyclop ./internal/raftengine/etcd/...
  -- 0 issues.

Author: bootjp

internal/raftengine/etcd/grpc_transport.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"io"
 	"log/slog"
+	"os"
 	"sync"
 	"time"
 
@@ -378,11 +379,54 @@ func (t *GRPCTransport) SendSnapshot(stream pb.EtcdRaft_SendSnapshotServer) erro
 		return err
 	}
 	if err := t.handle(stream.Context(), msg); err != nil {
+		// If receive finalized the snapshot as a .fsm file (token in
+		// Snapshot.Data), the engine refused to apply it — likely a
+		// transient context cancel or raft error. Remove the on-disk
+		// file so retries at later indexes don't leak orphan .fsm
+		// payloads into fsmSnapDir until the next startup runs
+		// cleanupStaleFSMSnaps. Same-index retries are already safe
+		// because os.Rename atomically replaces the prior file.
+		t.removeOrphanedFSMSnapshot(msg)
 		return err
 	}
 	return errors.WithStack(stream.SendAndClose(&pb.EtcdRaftAck{}))
 }
 
+// removeOrphanedFSMSnapshot deletes the .fsm file that
+// receiveSnapshotStream finalized for `msg`, if any. Used by
+// SendSnapshot when the engine apply (`t.handle`) fails after the
+// receive succeeded — the engine has NOT applied the snapshot (apply is
+// synchronous to t.handle, so a non-nil return means applied_index was
+// not advanced), so the file is unreferenced and safe to remove.
+//
+// Best-effort: a cleanup failure here is logged but not returned because
+// the original apply error is the actionable signal; orphans get swept
+// by cleanupStaleFSMSnaps at the next engine restart even if Remove
+// races with another process.
+func (t *GRPCTransport) removeOrphanedFSMSnapshot(msg raftpb.Message) {
+	if msg.Snapshot == nil || !isSnapshotToken(msg.Snapshot.Data) {
+		return
+	}
+	tok, err := decodeSnapshotToken(msg.Snapshot.Data)
+	if err != nil {
+		return
+	}
+	t.mu.RLock()
+	fsmSnapDir := t.fsmSnapDir
+	t.mu.RUnlock()
+	if fsmSnapDir == "" {
+		return
+	}
+	path := fsmSnapPath(fsmSnapDir, tok.Index)
+	if rmErr := os.Remove(path); rmErr != nil && !os.IsNotExist(rmErr) {
+		slog.Warn("failed to remove orphaned fsm snapshot file after apply failure",
+			"path", path,
+			"index", tok.Index,
+			"err", rmErr,
+		)
+	}
+}
+
 func (t *GRPCTransport) Send(ctx context.Context, req *pb.EtcdRaftMessage) (*pb.EtcdRaftAck, error) {
 	if req == nil {
 		return &pb.EtcdRaftAck{}, nil

internal/raftengine/etcd/grpc_transport_test.go

@@ -282,6 +282,75 @@ func TestReceiveSnapshotStream_SpoolPlacedInFSMSnapDir(t *testing.T) {
 	require.NoError(t, statErr, "renamed .fsm file should exist at canonical path")
 }
 
+// TestSendSnapshot_ApplyFailureRemovesFinalizedFSMFile pins the
+// orphan-cleanup behaviour from PR #747 round-4 (Codex P2): when the
+// receive path successfully finalizes the snapshot as
+// fsmSnapDir/<index>.fsm but the engine's apply (t.handle) then fails
+// — transient context cancel, raft error, etc. — the finalized .fsm
+// file MUST be removed. Otherwise retries at later snapshot indexes
+// accumulate orphan .fsm payloads in fsmSnapDir until startup runs
+// cleanupStaleFSMSnaps. Same-index retries are already safe via
+// os.Rename's atomic-replace, so this test exercises the cross-index
+// case where the orphan would actually persist.
+func TestSendSnapshot_ApplyFailureRemovesFinalizedFSMFile(t *testing.T) {
+	const index = uint64(77)
+
+	// Build a real testStateMachine payload + framed bytes so receive
+	// finalizes a syntactically valid .fsm file.
+	senderFSM := &testStateMachine{}
+	senderFSM.Apply([]byte("entry-for-orphan-cleanup-test"))
+	snap, err := senderFSM.Snapshot()
+	require.NoError(t, err)
+	var buf bytes.Buffer
+	_, err = snap.WriteTo(&buf)
+	require.NoError(t, err)
+	require.NoError(t, snap.Close())
+	payload := buf.Bytes()
+
+	metadata := raftpb.Message{
+		Type: raftpb.MsgSnap,
+		From: 1,
+		To:   2,
+		Snapshot: &raftpb.Snapshot{
+			Metadata: raftpb.SnapshotMetadata{Index: index, Term: 1},
+		},
+	}
+	raw, err := metadata.Marshal()
+	require.NoError(t, err)
+
+	fsmSnapDir := t.TempDir()
+
+	transport := NewGRPCTransport(nil)
+	transport.SetSpoolDir(t.TempDir())
+	transport.SetFSMSnapDir(fsmSnapDir)
+
+	// Wire a handler that always fails so SendSnapshot exercises the
+	// orphan-cleanup branch.
+	applyErr := errors.New("simulated apply failure")
+	transport.SetHandler(func(_ context.Context, _ raftpb.Message) error {
+		return applyErr
+	})
+
+	stream := &testSendSnapshotServer{
+		chunks: []*pb.EtcdRaftSnapshotChunk{
+			{Metadata: raw},
+			{Chunk: payload, Final: true},
+		},
+	}
+
+	err = transport.SendSnapshot(stream)
+	require.Error(t, err)
+	require.ErrorIs(t, err, applyErr, "SendSnapshot must surface the apply failure")
+
+	// THE point: the .fsm file at the canonical path MUST have been
+	// removed. Without the cleanup, leader retries at later indexes
+	// would accumulate one .fsm per failed apply until next startup.
+	finalPath := fsmSnapPath(fsmSnapDir, index)
+	_, statErr := os.Stat(finalPath)
+	require.True(t, os.IsNotExist(statErr),
+		"orphan .fsm file at %s must be removed after apply failure (got stat err: %v)", finalPath, statErr)
+}
+
 // TestReceiveSnapshotStream_LegacyFallbackWhenNoFSMSnapDir pins the
 // behaviour when fsmSnapDir is unset: receive still works, just via the
 // pre-PR materialization path. Tests that don't wire an fsmSnapDir (most