main: transfer Raft leadership before shutdown

README.md

@@ -26,6 +26,23 @@ Elastickv is an experimental distributed key-value store backed by Raft. It expo
 ## Development Status
 Elastickv remains experimental. Protocol and POSIX compatibility are limited to the operations documented below. Contributions and feedback are welcome.
 
+## Graceful shutdown
+
+The production server (`go run .`) handles SIGTERM and SIGINT by transferring
+leadership of every locally led Raft group before stopping its servers and
+closing storage. Transfers run in parallel with a shared five-second deadline;
+groups are rechecked for leadership acquired during another handoff. The Raft
+transport stays available until they finish. Only active voting
+replicas can receive leadership; learners are excluded.
+
+If a group has no eligible successor or a transfer times out, the server logs a
+warning and continues shutdown. Followers shut down without initiating a
+transfer. gRPC requests then have up to ten seconds to drain before remaining
+streams are closed, using the common bounded shutdown path. Allow the five-second
+handoff window in addition to the listener and FUSE drain budgets and storage
+cleanup when setting the supervisor's termination grace period. During startup,
+signals cancel immediately; repeated signals are ignored throughout cleanup.
+
 ## Architecture
 
 Architecture diagrams are available in:

docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md

@@ -0,0 +1,104 @@
+# SIGTERM leadership transfer
+
+Status: proposed; implementation and regression tests accompany this change.
+Issue: https://github.com/bootjp/elastickv/issues/1284
+
+## Goal
+
+When the production node receives SIGTERM, hand off every locally led Raft
+group before stopping transport and closing storage. SIGINT uses the same path.
+A follower, a single-voter group, or a group without a healthy successor must
+still be able to shut down.
+
+## Shutdown order
+
+1. Reuse the single early `installShutdownSignalHandler` from PR #1273. Before
+   startup finishes, its callback cancels immediately. After all servers and the
+   filesystem start, `shutdownHandoff.arm` switches the callback to a handoff
+   request. Repeated signals remain ignored through storage cleanup; the existing
+   clean-exit handling and bounded signal-handler join are preserved.
+2. Cancel the local leader-balancer's separate context and join its loop/RPC
+   before handing off. This join shares the same five-second deadline as all
+   transfers; a stuck balancer skips handoff and still permits bounded shutdown.
+   Raft transport and ordinary server contexts remain alive during this step.
+   Snapshot each runtime's engine through `snapshotEngine`. For each local
+   leader, run a transfer concurrently under one five-second deadline. Re-scan
+   after each round until no local leaders remain or that deadline expires, so
+   followers promoted during another group's handoff are not missed.
+3. Read the group's committed membership, exclude the local voter and learners,
+   and call the existing `TransferLeadershipToServerIfEligible` API. The engine
+   checks recent activity; Raft catches up the target before sending TimeoutNow.
+   Retry failed attempts while the node remains leader and the deadline permits.
+   A concurrent loss of leadership ends the attempt successfully.
+4. Log transfer failures and continue shutdown. Cancel the server context only
+   after the attempts finish, so transfer messages can still use the transport.
+5. Reuse #1273's `stopGRPCServerBounded`: drain gRPC requests for up to ten
+   seconds, then force remaining streams closed. There is one shared drain policy. Finally, the existing cleanup stack closes services, engines, and
+   storage. The supervisor's grace period must include the handoff window, the
+   listener/FUSE drain budgets, and storage cleanup.
+
+The bounded gRPC drain is necessary because established Raft streams can keep
+`GracefulStop` blocked indefinitely. The initial three-node reproduction
+completed both group transfers but remained in `grpc.Server.stop` until killed.
+
+## Scope and safety
+
+The implementation is in `main_shutdown.go`, wired into `run` and
+`startRaftServers`. It changes no membership, wire format, persistence format,
+timestamp issuance, or Raft election rules. In-flight writes retain the existing
+Raft transfer and cancellation semantics. The five-second transfer budget is
+shared by all groups, including group 0 when configured.
+
+Successful handoff requires a reachable voting successor. Failure to hand off
+is a warning rather than a shutdown error, including on a single-voter node.
+An internal error or memory-pressure cancellation proceeds directly to cleanup
+without waiting for a handoff. The separate `cmd/server` demo is outside this
+production-node lifecycle change.
+
+## Verification
+
+- Unit tests cover multiple leaders, followers, nil runtimes, learner exclusion,
+  shared deadlines, transient transfer errors, concurrent leadership loss, and
+  pre-existing context cancellation.
+- An in-memory gRPC health watch covers draining a persistent stream.
+- A three-node, two-group E2E starts the production `run` path in a subprocess,
+  moves both leaderships onto it, commits a write, sends an actual SIGTERM, and
+  checks clean exit, both handoffs, preserved data, and a subsequent write.
+- Run the shutdown tests with the race detector, the available Go package suite,
+  lint, and both Jepsen unit suites. The pre-existing FUSE mount test uses a
+  Linux-only `Statfs_t.Namelen` field and currently prevents full-suite test and
+  lint typechecking on macOS.
+
+Initial validation on macOS with Go 1.26.6, before integrating PR #1273:
+
+- `go test . -run '^TestShutdown' -count=1`: passed, including actual SIGTERM.
+- The same command with `-race`: passed.
+- Lint across all packages except `internal/filesystem/fuseadapter`: zero issues.
+- Root Jepsen `lein -o test`: 204 tests, 566 assertions, zero failures/errors.
+- The Go suite excluding FUSE passed every package except `adapter`, which hit
+  the default ten-minute timeout in existing
+  `Test_consistency_satisfy_write_after_read_sequence` and `Test_grpc_transaction`.
+  This change does not modify the adapter package.
+- The upstream Redis submodule's unit suite could not load `jepsen.redis`;
+  its checked-in test references that absent namespace and contains a placeholder
+  failing assertion. The Redis workload tests in the root Jepsen suite passed.
+
+
+## Review integration with PR #1273
+
+This branch incorporates dependency commit `3eb6989351d360079d7b9e8416a52de8357a2fa0`
+from #1273; that PR should merge first. Its early signal handler and bounded
+gRPC/HTTP/Redis/FUSE cleanup are reused rather than registering a second signal
+handler or retaining a second gRPC drain implementation. The five-second handoff
+manager also bounds cancellation independently of backpressured log writes.
+
+The follower-acquisition regression failed before the re-scan fix. The actual
+SIGTERM E2E also failed when both original signal registrations were present.
+After integration, the shutdown, signal, clean-exit, gRPC, and filesystem lifecycle
+tests passed with the race detector, and full-package lint reported zero issues.
+The FUSE Linux-only typecheck issue listed in the initial validation is fixed by
+the dependency's platform tags; kernel-mount checks still require Linux FUSE.
+The inherited recording-helper fixture now uses shell redirection so it also
+runs on macOS without `/bin/touch`. The FUSE package passed its full race suite;
+targeted adapter/monitoring shutdown tests also passed with the race detector.
+The root Jepsen suite again passed all 204 tests and 566 assertions.

docs/filesystem.md

@@ -228,15 +228,21 @@ filesystem it was asked to expose.
 ## Stopping and restarting
 
 `SIGTERM` and `SIGINT` stop the server gracefully: the first signal is logged
-as `shutdown signal received, stopping`, the FUSE mount is unmounted, the
+as `shutdown signal received, stopping`. Once startup has completed, the server
+first transfers its Raft leaderships under one shared five-second deadline,
+rechecking groups that acquire leadership during another handoff. The transport
+stays available during that window. Startup signals cancel immediately so mount
+startup can unwind. Afterwards, the FUSE mount is unmounted, the
 listeners drain, and the WAL is synced before the process exits with status
 0. That status is for a clean shutdown, one whose only errors are the
 cancellation itself; a component that fails while stopping still makes the
 process exit with an error after the same cleanup. Repeated
 `SIGTERM`/`SIGINT` while that shutdown is running are logged and ignored so
 a supervisor's second stop attempt cannot cut the cleanup short.
-This is the same path the server takes when a component fails or the memory
-watchdog trips (which still exits with status 2). If the unmount fails because
+After the handoff, cleanup follows the same path as a component failure or a
+memory-watchdog stop (which still exits with status 2); those stops skip the
+handoff. Include its five-second window in the supervisor's grace period in
+addition to the drain budgets below and storage cleanup. If the unmount fails because
 a client still holds files on the mount open (`EBUSY`), the server detaches
 the mount lazily instead: new lookups see the plain directory at once, and
 the server stops serving, so a request still in flight, or one made through

internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go

@@ -0,0 +1,80 @@
+package fuseadapter
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDelayedMountHelperLookupKeepsFallbackSnapshot(t *testing.T) {
+	t.Setenv("PATH", t.TempDir())
+	oldDir := t.TempDir()
+	want := writeFakeMountHelper(t, oldDir, "fusermount3")
+	newDir := t.TempDir()
+	writeFakeMountHelper(t, newDir, "fusermount3")
+	previous := mountHelperFallbackDir
+	mountHelperFallbackDir = oldDir
+	t.Cleanup(func() { mountHelperFallbackDir = previous })
+
+	lookup := snapshotMountHelperLookup()
+	release := make(chan struct{})
+	type result struct {
+		path string
+		err  error
+	}
+	finished := make(chan result, 1)
+	go func() {
+		<-release
+		path, err := lookup("fusermount3")
+		finished <- result{path: path, err: err}
+	}()
+	// A caller can time out before its lookup runs. The next test/config
+	// owner must not change that lookup's already-admitted dependencies.
+	mountHelperFallbackDir = newDir
+	close(release)
+	got := <-finished
+	require.NoError(t, got.err)
+	require.Equal(t, want, got.path)
+}
+
+func TestCanceledMountHelperLookupDoesNotLaunchWorker(t *testing.T) {
+	previous := lookupMountHelperFn
+	called := make(chan struct{}, 1)
+	lookupMountHelperFn = func(string) (string, error) {
+		called <- struct{}{}
+		return "", nil
+	}
+	t.Cleanup(func() { lookupMountHelperFn = previous })
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	_, err := lookupMountHelperContext(ctx, "fusermount3")
+	require.ErrorIs(t, err, context.Canceled)
+	select {
+	case <-called:
+		t.Fatal("already-cancelled lookup launched background work")
+	case <-time.After(10 * time.Millisecond):
+	}
+}
+
+func TestMountHelperLookupSnapshotSurvivesConcurrentOwner(t *testing.T) {
+	t.Setenv("PATH", t.TempDir())
+	previous := mountHelperFallbackDir
+	mountHelperFallbackDir = t.TempDir()
+	nextDir := t.TempDir()
+	t.Cleanup(func() { mountHelperFallbackDir = previous })
+	lookup := snapshotMountHelperLookup()
+	started := make(chan struct{})
+	finished := make(chan error, 1)
+	go func() {
+		close(started)
+		_, err := lookup("fusermount3")
+		finished <- err
+	}()
+	<-started
+	// No release handshake orders this mutation before or after the worker's
+	// lookup. A lookup that still reads the global races under -race.
+	mountHelperFallbackDir = nextDir
+	require.Error(t, <-finished)
+}

internal/filesystem/fuseadapter/stale_mount.go

@@ -480,8 +480,19 @@ var mountHelperFallbackDir = "/bin"
 
 // lookupMountHelperFn locates a helper; a variable so tests can stand in a
 // lookup that blocks, as exec.LookPath does when a PATH directory sits on an
-// unresponsive filesystem.
-var lookupMountHelperFn = lookupMountHelper
+// unresponsive filesystem. Nil selects the default lookup with a snapshot of
+// its fallback directory, taken before launching the bounded worker.
+var lookupMountHelperFn func(string) (string, error)
+
+func snapshotMountHelperLookup() func(string) (string, error) {
+	if lookup := lookupMountHelperFn; lookup != nil {
+		return lookup
+	}
+	fallbackDir := mountHelperFallbackDir
+	return func(name string) (string, error) {
+		return lookupMountHelperIn(name, fallbackDir)
+	}
+}
 
 // lookupMountHelperContext runs the helper lookup under the same share of
 // ctx's deadline a helper invocation gets. exec.LookPath stats each PATH
@@ -490,17 +501,24 @@ var lookupMountHelperFn = lookupMountHelper
 // wait is bounded; a lookup that does not answer in time counts as the
 // helper not being found, which keeps the kernel fallback reachable.
 func lookupMountHelperContext(ctx context.Context, name string) (string, error) {
+	if err := ctx.Err(); err != nil {
+		return "", errors.Wrapf(err, "locate mount helper %s", name)
+	}
+	timeout := detachHelperTimeout(ctx)
+	if timeout <= 0 {
+		return "", errors.Errorf("locate mount helper %s: no answer within its share of the detach budget", name)
+	}
 	type lookupResult struct {
 		path string
 		err  error
 	}
 	results := make(chan lookupResult, 1)
-	lookup := lookupMountHelperFn
+	lookup := snapshotMountHelperLookup()
 	go func() {
 		path, err := lookup(name)
 		results <- lookupResult{path: path, err: err}
 	}()
-	timer := time.NewTimer(detachHelperTimeout(ctx))
+	timer := time.NewTimer(timeout)
 	defer timer.Stop()
 	select {
 	case res := <-results:
@@ -515,10 +533,14 @@ func lookupMountHelperContext(ctx context.Context, name string) (string, error)
 // lookupMountHelper finds a fusermount helper the way go-fuse does: PATH
 // first, then mountHelperFallbackDir.
 func lookupMountHelper(name string) (string, error) {
+	return lookupMountHelperIn(name, mountHelperFallbackDir)
+}
+
+func lookupMountHelperIn(name, fallbackDir string) (string, error) {
 	if path, err := exec.LookPath(name); err == nil {
 		return path, nil
 	}
-	path, err := exec.LookPath(filepath.Join(mountHelperFallbackDir, name))
+	path, err := exec.LookPath(filepath.Join(fallbackDir, name))
 	if err != nil {
 		return "", errors.Wrapf(err, "locate mount helper %s", name)
 	}

internal/filesystem/fuseadapter/stale_mount_test.go

@@ -747,7 +747,9 @@ func writeRecordingMountHelper(t *testing.T) string {
 	t.Helper()
 	helperDir := t.TempDir()
 	marker := filepath.Join(helperDir, "ran")
-	script := "#!/bin/sh\n: > " + marker + "\nexit 0\n"
+	t.Setenv("ELASTICKV_TEST_DETACH_MARKER", marker)
+	// Shell redirection keeps this helper portable across Linux and macOS.
+	script := "#!/bin/sh\n: > \"$ELASTICKV_TEST_DETACH_MARKER\"\nexit 0\n"
 	require.NoError(t, os.WriteFile(filepath.Join(helperDir, "fusermount3"), []byte(script), 0o700)) //nolint:gosec // the helper stand-in must be executable.
 	onlyMountHelpersIn(t, helperDir)
 	return marker

main.go

@@ -299,7 +299,7 @@ var shutdownSignals = []os.Signal{syscall.SIGTERM, syscall.SIGINT}
 const shutdownSignalHandlerJoinTimeout = time.Second
 
 // signalShutdown is set by the signal handler once a shutdown signal has
-// cancelled the root context, so main can tell a requested stop from a
+// requested a stop, so main can tell a requested stop from a
 // failure when the errgroup surfaces context.Canceled while unwinding.
 var signalShutdown atomic.Bool
 
@@ -542,7 +542,8 @@ func run() error {
 	// closers so it is still swallowing repeat signals while CleanupStack
 	// syncs and closes them.
 	ctx, cancel := context.WithCancel(context.Background())
-	cleanup.Add(installShutdownSignalHandler(cancel))
+	handoff := newShutdownHandoff(cancel)
+	cleanup.Add(installShutdownSignalHandler(handoff.requestShutdown))
 
 	// Stage 6D-6c: buildShardGroupsWithEncryptionWiring assembles the
 	// storage-envelope write-path wiring (cipher + deterministic nonce
@@ -719,11 +720,18 @@ func run() error {
 	}); err != nil {
 		return err
 	}
+	balanceCtx, stopBalance := context.WithCancel(runCtx)
+	defer stopBalance()
+	balanceDone := make(chan struct{})
+	handoff.arm(runCtx, eg, runtimes, func(ctx context.Context) error {
+		return stopLeaderBalanceForShutdown(ctx, stopBalance, balanceDone)
+	})
 	startLeaderBalanceScheduler(
-		runCtx,
+		balanceCtx,
 		eg,
 		runtimes,
 		leaderBalanceConfigFromFlags(*raftId, cfg.defaultGroup, cfg.sqsFifoPartitionMap, metricsRegistry.Registerer()),
+		balanceDone,
 	)
 
 	return waitRunGroup(eg)
@@ -3385,10 +3393,11 @@ func stopGRPCServerBounded(ctx context.Context, srv *grpc.Server, name string) {
 }
 
 // installShutdownSignalHandler routes SIGTERM/SIGINT (or the given signals)
-// into cancel so the process unwinds through the same path the memory
-// watchdog uses: the errgroup drains, the FUSE mount is unmounted, and the
-// CleanupStack syncs the WAL. The first signal cancels; later ones are
-// logged and otherwise ignored so a second SIGTERM from a supervisor cannot
+// into the supplied shutdown callback. Cleanup follows the memory watchdog's
+// path: the errgroup drains, the FUSE mount is unmounted, and the
+// CleanupStack syncs the WAL. Startup passes an immediate cancellation;
+// after startup, the callback requests a leadership handoff first. Later signals
+// are logged and ignored so a second SIGTERM from a supervisor cannot
 // cut the shutdown short (SIGKILL still can, which no handler prevents).
 // The returned function releases the registration and must be called once
 // the cleanup that needs protection has run.
@@ -3406,7 +3415,7 @@ func installShutdownSignalHandler(cancel context.CancelFunc, signals ...os.Signa
 			select {
 			case sig := <-sigCh:
 				if signalShutdown.CompareAndSwap(false, true) {
-					// Cancel before logging: a log write that blocks (a
+					// Request shutdown before logging: a log write that blocks (a
 					// backpressured stderr, a synchronous shipper) must not
 					// hold the shutdown back, and later signals are ignored.
 					cancel()
@@ -3438,9 +3447,10 @@ func installShutdownSignalHandler(cancel context.CancelFunc, signals ...os.Signa
 // watcher is off by default; it is enabled only when the operator sets
 // ELASTICKV_MEMORY_SHUTDOWN_THRESHOLD_MB. On threshold crossing the
 // callback flips the memoryPressureExit sentinel and cancels the root
-// context, routing through the exact same shutdown path SIGTERM would
-// use (errgroup unwinds, CleanupStack runs, WAL is synced). We do NOT
-// send a signal, call os.Exit, or touch the raft engine directly here.
+// context so errgroup unwinds, CleanupStack runs, and the WAL is synced.
+// It skips the signal handler's leadership handoff to relieve pressure
+// immediately. We do NOT send a signal, call os.Exit, or touch the raft
+// engine directly here.
 func startMemoryWatchdog(ctx context.Context, eg *errgroup.Group, cancel context.CancelFunc) {
 	cfg, enabled := memwatchConfigFromEnv()
 	if !enabled {