diff --git a/README.md b/README.md index 2cfa33e6f..242bccd28 100644 --- a/README.md +++ b/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: diff --git a/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md b/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md new file mode 100644 index 000000000..95f530579 --- /dev/null +++ b/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. diff --git a/docs/filesystem.md b/docs/filesystem.md index c1d5373a4..c1d707712 100644 --- a/docs/filesystem.md +++ b/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 diff --git a/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go b/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go new file mode 100644 index 000000000..0976f29a5 --- /dev/null +++ b/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) +} diff --git a/internal/filesystem/fuseadapter/stale_mount.go b/internal/filesystem/fuseadapter/stale_mount.go index 0b647adaa..981f55255 100644 --- a/internal/filesystem/fuseadapter/stale_mount.go +++ b/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) } diff --git a/internal/filesystem/fuseadapter/stale_mount_test.go b/internal/filesystem/fuseadapter/stale_mount_test.go index 8a11c8bf6..fe591b23a 100644 --- a/internal/filesystem/fuseadapter/stale_mount_test.go +++ b/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 diff --git a/main.go b/main.go index 5dcbe4f82..a1ef938c5 100644 --- a/main.go +++ b/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 { diff --git a/main_leader_balance.go b/main_leader_balance.go index b7fa14751..ce5cef050 100644 --- a/main_leader_balance.go +++ b/main_leader_balance.go @@ -86,14 +86,15 @@ type leaderBalanceDecision struct { reason string } -func startLeaderBalanceScheduler(ctx context.Context, eg *errgroup.Group, runtimes []*raftGroupRuntime, cfg leaderBalanceConfig) { +func startLeaderBalanceScheduler(ctx context.Context, eg *errgroup.Group, runtimes []*raftGroupRuntime, cfg leaderBalanceConfig, done chan<- struct{}) { if cfg.logger == nil { cfg.logger = slog.Default() } if cfg.metrics != nil { cfg.metrics.setEnabled(cfg.enabled) } - if !cfg.enabled || eg == nil { + if !cfg.enabled || eg == nil || ctx.Err() != nil { + close(done) return } cfg.normalize() @@ -105,6 +106,7 @@ func startLeaderBalanceScheduler(ctx context.Context, eg *errgroup.Group, runtim groupCooldownUntil: make(map[uint64]time.Time), } eg.Go(func() error { + defer close(done) defer func() { if err := loop.connCache.Close(); err != nil { cfg.logger.Warn("leader-balance: failed to close grpc cache", "err", err) @@ -158,6 +160,9 @@ func (c *leaderBalanceConfig) normalize() { } func (l *leaderBalanceLoop) run(ctx context.Context) { + if ctx.Err() != nil { + return + } if l.defaultRuntime == nil { l.cfg.logger.Warn("leader-balance: default raft group runtime not found", "group", l.cfg.defaultGroupID) return @@ -176,6 +181,9 @@ func (l *leaderBalanceLoop) run(ctx context.Context) { } func (l *leaderBalanceLoop) tick(ctx context.Context, now time.Time) { + if ctx.Err() != nil { + return + } defaultEngine := l.defaultRuntime.snapshotEngine() defaultLeader := defaultEngine != nil && defaultEngine.State() == raftengine.StateLeader if !defaultLeader { @@ -248,6 +256,9 @@ func (l *leaderBalanceLoop) killSwitchActive() bool { } func (l *leaderBalanceLoop) executeMove(ctx context.Context, move leaderBalanceMove) error { + if err := ctx.Err(); err != nil { + return errors.WithStack(err) + } if len(move.candidates) == 0 { return errors.New("leader-balance: move has no target candidates") } diff --git a/main_shutdown.go b/main_shutdown.go new file mode 100644 index 000000000..b0cffc77f --- /dev/null +++ b/main_shutdown.go @@ -0,0 +1,179 @@ +package main + +import ( + "context" + "log/slog" + "math" + "sync" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/cockroachdb/errors" + "golang.org/x/sync/errgroup" +) + +const ( + shutdownTransferTimeout = 5 * time.Second + shutdownTransferRetryDelay = 50 * time.Millisecond + shutdownVoterSuffrage = "voter" +) + +type shutdownLeadershipController interface { + raftengine.LeaderView + raftengine.ConfigReader + TransferLeadershipToServerIfEligible(context.Context, []raftengine.TransferTarget, uint64) error +} + +// shutdownHandoff keeps startup cancellation immediate, then switches the +// existing signal handler to request a handoff once startup has completed. +type shutdownHandoff struct { + mu sync.Mutex + request context.CancelFunc + cancel context.CancelFunc +} + +func newShutdownHandoff(cancel context.CancelFunc) *shutdownHandoff { + return &shutdownHandoff{request: cancel, cancel: cancel} +} + +func (h *shutdownHandoff) requestShutdown() { + h.mu.Lock() + request := h.request + h.mu.Unlock() + request() +} + +func (h *shutdownHandoff) arm(ctx context.Context, eg *errgroup.Group, runtimes []*raftGroupRuntime, quiesce func(context.Context) error) { + h.mu.Lock() + defer h.mu.Unlock() + if ctx.Err() != nil { + return + } + requests := make(chan struct{}) + h.request = sync.OnceFunc(func() { close(requests) }) + eg.Go(func() error { + shutdownAfterRequest(ctx, h.cancel, runtimes, requests, shutdownTransferTimeout, quiesce) + return nil + }) +} + +func shutdownAfterRequest(ctx context.Context, cancel context.CancelFunc, runtimes []*raftGroupRuntime, requests <-chan struct{}, timeout time.Duration, quiesce func(context.Context) error) { + select { + case <-ctx.Done(): + return + case <-requests: + } + defer cancel() + // Keep the server context alive until the transfers finish: canceling it + // first would close the Raft transport needed to elect the successors. + transferCtx, stop := context.WithTimeout(ctx, timeout) + defer stop() + done := make(chan struct{}) + go func() { + defer close(done) + if quiesce != nil { + if err := quiesce(transferCtx); err != nil { + slog.Warn("shutdown: leader balance did not quiesce before handoff", "err", err) + return + } + } + transferLeadershipOnShutdown(transferCtx, runtimes) + }() + // Logging may be backpressured. It must not hold root cancellation or + // the errgroup beyond the handoff deadline. + select { + case <-done: + case <-transferCtx.Done(): + } +} + +func stopLeaderBalanceForShutdown(ctx context.Context, cancel context.CancelFunc, done <-chan struct{}) error { + cancel() + select { + case <-done: + return nil + case <-ctx.Done(): + return errors.Wrap(ctx.Err(), "join leader balance before shutdown handoff") + } +} + +func transferLeadershipOnShutdown(ctx context.Context, runtimes []*raftGroupRuntime) { + ticker := time.NewTicker(shutdownTransferRetryDelay) + defer ticker.Stop() + for ctx.Err() == nil && transferShutdownLeaderRound(ctx, runtimes) { + // Revisit followers after each round: they may have acquired a new + // leadership while another group's handoff was still running. + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +func transferShutdownLeaderRound(ctx context.Context, runtimes []*raftGroupRuntime) bool { + var transfers sync.WaitGroup + foundLeader := false + for _, rt := range runtimes { + engine := rt.snapshotEngine() + admin, ok := engine.(shutdownLeadershipController) + if !ok || admin.State() != raftengine.StateLeader { + continue + } + foundLeader = true + transfers.Go(func() { + slog.Info("transferring leadership before shutdown", "group", rt.spec.id) + if err := transferGroupLeadershipOnShutdown(ctx, admin); err != nil { + slog.Warn("shutdown leadership transfer failed; continuing shutdown", "group", rt.spec.id, "error", err) + return + } + slog.Info("shutdown leadership transfer completed", "group", rt.spec.id) + }) + } + // All groups share one deadline, so an unavailable group does not delay + // healthy groups or multiply the shutdown budget by the shard count. + transfers.Wait() + return foundLeader +} + +func transferGroupLeadershipOnShutdown(ctx context.Context, admin shutdownLeadershipController) error { + ticker := time.NewTicker(shutdownTransferRetryDelay) + defer ticker.Stop() + for admin.State() == raftengine.StateLeader { + if err := ctx.Err(); err != nil { + return errors.WithStack(err) + } + cfg, err := admin.Configuration(ctx) + if err != nil { + return errors.Wrap(err, "read shutdown transfer membership") + } + candidates := shutdownTransferCandidates(cfg, admin.Leader().ID) + if len(candidates) == 0 { + return errors.WithStack(raftengine.ErrLeadershipTransferNoHealthyTarget) + } + // Eligibility excludes inactive replicas and learners. Let Raft catch + // up the selected voter during transfer rather than imposing a lag cap. + err = admin.TransferLeadershipToServerIfEligible(ctx, candidates, math.MaxUint64) + if err == nil || errors.Is(err, raftengine.ErrNotLeader) { + return nil + } + // A pending config change, another transfer, or a replica that is not + // active yet may settle while the shutdown deadline still has time. + select { + case <-ctx.Done(): + return errors.Wrap(errors.Join(err, ctx.Err()), "shutdown transfer deadline") + case <-ticker.C: + } + } + return nil +} + +func shutdownTransferCandidates(cfg raftengine.Configuration, localID string) []raftengine.TransferTarget { + candidates := make([]raftengine.TransferTarget, 0, len(cfg.Servers)) + for _, server := range cfg.Servers { + if server.ID != localID && server.Suffrage == shutdownVoterSuffrage { + candidates = append(candidates, raftengine.TransferTarget{ID: server.ID, Address: server.Address}) + } + } + return candidates +} diff --git a/main_shutdown_balance_test.go b/main_shutdown_balance_test.go new file mode 100644 index 000000000..a39beba7e --- /dev/null +++ b/main_shutdown_balance_test.go @@ -0,0 +1,101 @@ +package main + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +func TestShutdownJoinsBalancerBeforeHandoff(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + balanceCtx, stopBalance := context.WithCancel(ctx) + defer stopBalance() + stopping, balanceDone, release := make(chan struct{}), make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + t.Cleanup(unblock) + go func() { + <-balanceCtx.Done() + close(stopping) + <-release + close(balanceDone) + }() + var transfers atomic.Int32 + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + select { + case <-balanceDone: + default: + t.Error("handoff started before balancing stopped") + } + require.NoError(t, ctx.Err(), "quiescing must keep Raft transport alive") + transfers.Add(1) + return nil + }) + requests, finished := make(chan struct{}), make(chan struct{}) + close(requests) + go func() { + defer close(finished) + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, requests, time.Second, + func(ctx context.Context) error { return stopLeaderBalanceForShutdown(ctx, stopBalance, balanceDone) }) + }() + select { + case <-stopping: + case <-time.After(shutdownTestTimeout): + t.Fatal("balancing was not cancelled") + } + require.NoError(t, ctx.Err()) + require.Zero(t, transfers.Load()) + unblock() + select { + case <-finished: + case <-time.After(shutdownTestTimeout): + t.Fatal("handoff did not finish after balancing joined") + } + require.Equal(t, int32(1), transfers.Load()) + require.ErrorIs(t, ctx.Err(), context.Canceled) +} + +func TestShutdownBalancerJoinSharesHandoffDeadline(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + balanceCtx, stopBalance := context.WithCancel(ctx) + defer stopBalance() + var transfers atomic.Int32 + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + transfers.Add(1) + return nil + }) + requests, balanceDone := make(chan struct{}), make(chan struct{}) + close(requests) + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, requests, 20*time.Millisecond, + func(ctx context.Context) error { return stopLeaderBalanceForShutdown(ctx, stopBalance, balanceDone) }) + require.ErrorIs(t, balanceCtx.Err(), context.Canceled) + require.ErrorIs(t, ctx.Err(), context.Canceled) + require.Zero(t, transfers.Load(), "do not transfer concurrently with an unjoined balancing task") +} + +func TestShutdownBeforeBalancerStartsCompletesJoin(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(t.Context()) + cancel() + done := make(chan struct{}) + eg := &errgroup.Group{} + startLeaderBalanceScheduler(ctx, eg, nil, leaderBalanceConfig{enabled: true}, done) + select { + case <-done: + default: + t.Fatal("cancelled startup left balancing join pending") + } + require.NoError(t, eg.Wait()) + // A ready tick must not inspect engines or dispatch after cancellation. + (&leaderBalanceLoop{}).tick(ctx, time.Now()) + require.ErrorIs(t, (&leaderBalanceLoop{}).executeMove(ctx, leaderBalanceMove{}), context.Canceled) +} diff --git a/main_shutdown_e2e_test.go b/main_shutdown_e2e_test.go new file mode 100644 index 000000000..274fb59bd --- /dev/null +++ b/main_shutdown_e2e_test.go @@ -0,0 +1,147 @@ +package main + +import ( + "bytes" + "context" + "flag" + "os" + "os/exec" + "path/filepath" + "strings" + "syscall" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + pb "github.com/bootjp/elastickv/proto" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +const shutdownProcessTestEnv = "ELASTICKV_SHUTDOWN_PROCESS_TEST" + +// Run the production startup/shutdown path in a subprocess so SIGTERM cannot +// terminate the test runner or interfere with another test's signal handlers. +func TestShutdownProcess(t *testing.T) { + if os.Getenv(shutdownProcessTestEnv) != "1" { + return + } + require.NoError(t, flag.CommandLine.Parse(flag.Args())) + require.NoError(t, ignoreCanceledAfterSignal(run())) +} + +func TestShutdownSIGTERMTransfersEveryRaftGroupE2E(t *testing.T) { + const nodeCount = 3 + groupIDs := []uint64{1, 2} + baseDir := t.TempDir() + endpoints, listeners := allocateBootstrapE2EMultiGroupEndpoints(t, nodeCount, groupIDs) + t.Cleanup(func() { closeBootstrapE2EMultiGroupListeners(listeners) }) + groupPeers := bootstrapGroupPeersArg(endpoints, groupIDs) + nodes := make([]*bootstrapE2ENode, 0, nodeCount-1) + t.Cleanup(func() { closeBootstrapE2ENodes(t, nodes) }) + for i := range nodeCount - 1 { + node, err := startBootstrapE2EMultiGroupNode(baseDir, endpoints[i], listeners[i], true, groupIDs, groupPeers, raftEngineEtcd) + require.NoError(t, err) + nodes = append(nodes, node) + } + ep := endpoints[nodeCount-1] + closeBootstrapE2EMultiGroupListeners(listeners[nodeCount-1:]) + process, exited, logPath := startShutdownTestProcess(t, baseDir, ep, groupIDs, groupPeers) + + for _, groupID := range groupIDs { + moveShutdownTestLeadership(t, nodes, ep, groupID) + } + conn, err := grpc.NewClient(ep.raftAddrs[groupIDs[0]], grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + key, value := []byte("shutdown-key"), []byte("before-sigterm") + require.NoError(t, rawPutWithTimeout(pb.NewRawKVClient(conn), key, value)) + + require.NoError(t, process.Signal(syscall.SIGTERM)) + select { + case err := <-exited: + require.NoError(t, err, "the production server must exit cleanly on SIGTERM") + case <-time.After(bootstrapE2EWaitTimeout): + t.Fatal("the server did not exit after SIGTERM") + } + output, err := os.ReadFile(logPath) + require.NoError(t, err) + require.NotContains(t, string(output), "shutdown leadership transfer failed") + require.Equal(t, len(groupIDs), strings.Count(string(output), "shutdown leadership transfer completed")) + for _, groupID := range groupIDs { + waitForGroupLeader(t, nodes, groupID, bootstrapE2EWaitTimeout, bootstrapE2EWaitInterval) + } + connAfter, err := grpc.NewClient(endpoints[0].raftAddrs[groupIDs[0]], grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + t.Cleanup(func() { _ = connAfter.Close() }) + client := pb.NewRawKVClient(connAfter) + require.Eventually(t, func() bool { + resp, getErr := rawGetWithTimeout(client, key, bootstrapE2ERPCRequestTimeout) + return getErr == nil && resp.Exists && bytes.Equal(resp.Value, value) + }, bootstrapE2EWaitTimeout, bootstrapE2EWaitInterval) + require.Eventually(t, func() bool { + return rawPutWithTimeout(client, key, []byte("after-sigterm")) == nil + }, bootstrapE2EWaitTimeout, bootstrapE2EWaitInterval) +} + +func startShutdownTestProcess(t *testing.T, baseDir string, ep bootstrapE2EMultiGroupEndpoint, groupIDs []uint64, groupPeers string) (*os.Process, <-chan error, string) { + t.Helper() + logPath := filepath.Join(baseDir, "shutdown-process.log") + logFile, err := os.Create(logPath) + require.NoError(t, err) + t.Cleanup(func() { + _ = logFile.Close() + if t.Failed() { + output, readErr := os.ReadFile(logPath) + require.NoError(t, readErr) + t.Log(string(output)) + } + }) + //nolint:gosec // Re-executes this test binary with local test endpoints and a temporary data directory. + cmd := exec.CommandContext(t.Context(), os.Args[0], "-test.run=^TestShutdownProcess$", "--", + "-raftId="+ep.id, "-raftDataDir="+baseDir, + "-address="+ep.raftAddrs[groupIDs[0]], "-redisAddress="+ep.redisAddr, "-dynamoAddress="+ep.dynamoAddr, + "-raftGroups="+raftGroupsArg(ep, groupIDs), "-raftGroupPeers="+groupPeers, + "-metricsAddress=127.0.0.1:0", "-pprofAddress=") + cmd.Env = append(os.Environ(), shutdownProcessTestEnv+"=1") + cmd.Stdout, cmd.Stderr = logFile, logFile + require.NoError(t, cmd.Start()) + exited := make(chan error, 1) + go func() { + exited <- cmd.Wait() + close(exited) + }() + t.Cleanup(func() { + _ = cmd.Process.Kill() + <-exited + }) + return cmd.Process, exited, logPath +} + +func moveShutdownTestLeadership(t *testing.T, nodes []*bootstrapE2ENode, ep bootstrapE2EMultiGroupEndpoint, groupID uint64) { + t.Helper() + conn, err := grpc.NewClient(ep.raftAddrs[groupID], grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + defer func() { _ = conn.Close() }() + client := pb.NewRaftAdminClient(conn) + require.Eventually(t, func() bool { + ctx, cancel := context.WithTimeout(context.Background(), bootstrapE2ERPCRequestTimeout) + defer cancel() + status, err := client.Status(ctx, &pb.RaftAdminStatusRequest{}) + if err != nil { + return false + } + if status.State == pb.RaftAdminState_RAFT_ADMIN_STATE_LEADER { + return true + } + for _, node := range nodes { + engine := node.engineForGroup(groupID) + if engine.State() == raftengine.StateLeader { + admin, ok := engine.(raftengine.Admin) + return ok && admin.TransferLeadershipToServer(ctx, ep.id, ep.raftAddrs[groupID]) == nil + } + } + return false + }, bootstrapE2EWaitTimeout, bootstrapE2EWaitInterval) +} diff --git a/main_shutdown_handoff_test.go b/main_shutdown_handoff_test.go new file mode 100644 index 000000000..85a42a27b --- /dev/null +++ b/main_shutdown_handoff_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "context" + "log/slog" + "os" + "sync" + "syscall" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +func TestShutdownHandoffStartupSignalCancelsBeforeArming(t *testing.T) { + resetSignalShutdown(t) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + handoff := newShutdownHandoff(cancel) + stop := installShutdownSignalHandler(handoff.requestShutdown, syscall.SIGUSR2) + t.Cleanup(stop) + require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGUSR2)) + select { + case <-ctx.Done(): + case <-time.After(shutdownTestTimeout): + t.Fatal("a signal during startup must cancel immediately") + } + eg, runCtx := errgroup.WithContext(ctx) + handoff.arm(runCtx, eg, nil, nil) + require.NoError(t, eg.Wait(), "arming after startup cancellation must not wait for another signal") + require.True(t, signalShutdown.Load()) +} + +func TestShutdownHandoffSignalWaitsForTransferAndIgnoresRepeats(t *testing.T) { + resetSignalShutdown(t) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + started := make(chan struct{}) + released := make(chan struct{}) + release := sync.OnceFunc(func() { close(released) }) + t.Cleanup(release) + engine := shutdownTestEngine(func(transferCtx context.Context, _ []raftengine.TransferTarget, _ uint64) error { + close(started) + select { + case <-released: + return nil + case <-transferCtx.Done(): + return transferCtx.Err() + } + }) + handoff := newShutdownHandoff(cancel) + stop := installShutdownSignalHandler(handoff.requestShutdown, syscall.SIGUSR2) + t.Cleanup(stop) + eg, runCtx := errgroup.WithContext(ctx) + handoff.arm(runCtx, eg, []*raftGroupRuntime{{spec: groupSpec{id: 1}, engine: engine}}, nil) + require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGUSR2)) + select { + case <-started: + case <-time.After(shutdownTestTimeout): + t.Fatal("the signal did not start the leadership handoff") + } + require.NoError(t, ctx.Err(), "root cancellation must wait until the handoff finishes") + require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGUSR2)) + select { + case <-ctx.Done(): + t.Fatal("a repeated signal canceled the in-flight handoff") + case <-time.After(shutdownTransferRetryDelay): + } + release() + done := make(chan error, 1) + go func() { done <- eg.Wait() }() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(shutdownTestTimeout): + t.Fatal("shutdown did not finish after the handoff") + } + require.ErrorIs(t, ctx.Err(), context.Canceled) + require.True(t, signalShutdown.Load()) +} + +func TestShutdownHandoffDeadlineSurvivesBlockedLogging(t *testing.T) { + handler := &blockingLogHandler{release: make(chan struct{})} + release := sync.OnceFunc(func() { close(handler.release) }) + old := slog.Default() + slog.SetDefault(slog.New(handler)) + t.Cleanup(func() { + release() + slog.SetDefault(old) + }) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { return nil }) + requests := make(chan struct{}) + close(requests) + done := make(chan struct{}) + go func() { + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, requests, shutdownTransferRetryDelay, nil) + close(done) + }() + select { + case <-done: + case <-time.After(shutdownTestTimeout): + t.Fatal("a blocked handoff log held back shutdown past its deadline") + } + require.ErrorIs(t, ctx.Err(), context.Canceled) +} diff --git a/main_shutdown_test.go b/main_shutdown_test.go new file mode 100644 index 000000000..2255afffc --- /dev/null +++ b/main_shutdown_test.go @@ -0,0 +1,204 @@ +package main + +import ( + "context" + "errors" + "math" + "sync/atomic" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/stretchr/testify/require" +) + +type shutdownEngineStub struct { + raftengine.Engine + leader atomic.Bool + servers []raftengine.Server + transfer func(context.Context, []raftengine.TransferTarget, uint64) error +} + +func (e *shutdownEngineStub) State() raftengine.State { + if e.leader.Load() { + return raftengine.StateLeader + } + return raftengine.StateFollower +} +func (e *shutdownEngineStub) Leader() raftengine.LeaderInfo { + return raftengine.LeaderInfo{ID: "local"} +} +func (e *shutdownEngineStub) Configuration(context.Context) (raftengine.Configuration, error) { + return raftengine.Configuration{Servers: e.servers}, nil +} +func (e *shutdownEngineStub) TransferLeadershipToServerIfEligible(ctx context.Context, candidates []raftengine.TransferTarget, maxLag uint64) error { + err := e.transfer(ctx, candidates, maxLag) + if err == nil || errors.Is(err, raftengine.ErrNotLeader) { + e.leader.Store(false) + } + return err +} + +func shutdownTestEngine(transfer func(context.Context, []raftengine.TransferTarget, uint64) error) *shutdownEngineStub { + engine := &shutdownEngineStub{ + servers: []raftengine.Server{ + {ID: "local", Address: "local:1", Suffrage: "voter"}, + {ID: "learner", Address: "learner:1", Suffrage: "learner"}, + {ID: "voter", Address: "voter:1", Suffrage: "voter"}, + }, + transfer: transfer, + } + engine.leader.Store(true) + return engine +} + +func TestShutdownSignalTransfersAllLeadersBeforeCancellation(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var calls atomic.Int32 + transfer := func(transferCtx context.Context, candidates []raftengine.TransferTarget, maxLag uint64) error { + require.NoError(t, ctx.Err(), "the Raft transport must stay alive during transfer") + require.NoError(t, transferCtx.Err()) + require.Equal(t, []raftengine.TransferTarget{{ID: "voter", Address: "voter:1"}}, candidates) + require.Equal(t, uint64(math.MaxUint64), maxLag) + calls.Add(1) + return nil + } + follower := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + t.Error("a follower must not initiate transfer") + return nil + }) + follower.leader.Store(false) + runtimes := []*raftGroupRuntime{ + nil, + {}, + {spec: groupSpec{id: 1}, engine: shutdownTestEngine(transfer)}, + {spec: groupSpec{id: 2}, engine: shutdownTestEngine(transfer)}, + {spec: groupSpec{id: 3}, engine: follower}, + } + requests := make(chan struct{}) + close(requests) + shutdownAfterRequest(ctx, cancel, runtimes, requests, time.Second, nil) + require.ErrorIs(t, ctx.Err(), context.Canceled) + require.Equal(t, int32(2), calls.Load()) +} + +func TestShutdownSignalTransfersLeadershipAcquiredDuringHandoff(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var followerTransfers atomic.Int32 + follower := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + require.NoError(t, ctx.Err(), "newly acquired leadership must transfer before server cancellation") + followerTransfers.Add(1) + return nil + }) + follower.leader.Store(false) + leader := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + // The first runtime was already observed as a follower when this + // group's transfer causes it to acquire leadership. + follower.leader.Store(true) + return nil + }) + requests := make(chan struct{}) + close(requests) + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{ + {spec: groupSpec{id: 1}, engine: follower}, + {spec: groupSpec{id: 2}, engine: leader}, + }, requests, time.Second, nil) + require.Equal(t, int32(1), followerTransfers.Load()) + require.Equal(t, raftengine.StateFollower, follower.State()) + require.ErrorIs(t, ctx.Err(), context.Canceled) +} + +func TestShutdownSignalSharesDeadlineAcrossGroups(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + started := make(chan struct{}, 2) + transfer := func(transferCtx context.Context, _ []raftengine.TransferTarget, _ uint64) error { + started <- struct{}{} + <-transferCtx.Done() + return transferCtx.Err() + } + runtimes := []*raftGroupRuntime{ + {spec: groupSpec{id: 1}, engine: shutdownTestEngine(transfer)}, + {spec: groupSpec{id: 2}, engine: shutdownTestEngine(transfer)}, + } + requests := make(chan struct{}) + close(requests) + done := make(chan struct{}) + go func() { + shutdownAfterRequest(ctx, cancel, runtimes, requests, 200*time.Millisecond, nil) + close(done) + }() + for range runtimes { + select { + case <-started: + case <-done: + t.Fatal("all transfers must start before the shared deadline") + } + } + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("shutdown did not finish after the transfer deadline") + } + require.ErrorIs(t, ctx.Err(), context.Canceled) +} + +func TestShutdownTransferRetriesTransientErrors(t *testing.T) { + t.Parallel() + for _, transientErr := range []error{ + raftengine.ErrLeadershipTransferConfChangePending, + raftengine.ErrLeadershipTransferInProgress, + raftengine.ErrLeadershipTransferNoHealthyTarget, + } { + t.Run(transientErr.Error(), func(t *testing.T) { + t.Parallel() + calls := 0 + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + calls++ + if calls == 1 { + return transientErr + } + return nil + }) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + require.NoError(t, transferGroupLeadershipOnShutdown(ctx, engine)) + require.Equal(t, 2, calls) + }) + } +} + +func TestShutdownTransferWithoutAnotherVoter(t *testing.T) { + t.Parallel() + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + t.Error("a single voter must not initiate transfer") + return nil + }) + engine.servers = engine.servers[:2] // local voter and a learner + err := transferGroupLeadershipOnShutdown(t.Context(), engine) + require.ErrorIs(t, err, raftengine.ErrLeadershipTransferNoHealthyTarget) +} + +func TestShutdownTransferLosingLeadershipIsSuccess(t *testing.T) { + t.Parallel() + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + return errors.Join(errors.New("transfer raced an election"), raftengine.ErrNotLeader) + }) + require.NoError(t, transferGroupLeadershipOnShutdown(t.Context(), engine)) +} + +func TestShutdownAlreadyCanceledDoesNotTransfer(t *testing.T) { + t.Parallel() + engine := shutdownTestEngine(func(context.Context, []raftengine.TransferTarget, uint64) error { + t.Error("an already stopped server must not initiate transfer") + return nil + }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, make(chan struct{}), time.Second, nil) +}