From 34c6fd200beaea95fe2c3cfbd5132a543df7fb65 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 23:19:44 +0900 Subject: [PATCH 01/10] docs: propose graceful shutdown leadership transfer --- ...22_proposed_sigterm_leadership_transfer.md | 74 +++++++++++++++++++ 1 file changed, 74 insertions(+) create mode 100644 docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md 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..c89e5a506 --- /dev/null +++ b/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md @@ -0,0 +1,74 @@ +# 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. Register the signal handler after constructing the group runtimes, before + starting the servers. Stop the signal subscription during cleanup. +2. Snapshot each runtime's engine through `snapshotEngine`. For each local + leader, run a transfer concurrently under one five-second deadline. +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. Drain gRPC requests for up to five more seconds, then force remaining streams + closed. Finally, the existing cleanup stack closes services, engines, and + storage. The supervisor's grace period must also allow 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. + +Local validation on macOS with Go 1.26.6: + +- `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. From f133f118a41c969bfc0b916621a587900a57e16a Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 22 Oct 2026 23:20:00 +0900 Subject: [PATCH 02/10] main: transfer Raft leadership on SIGTERM --- README.md | 14 +++ main.go | 10 +- main_shutdown.go | 127 +++++++++++++++++++++++ main_shutdown_e2e_test.go | 147 +++++++++++++++++++++++++++ main_shutdown_test.go | 207 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 501 insertions(+), 4 deletions(-) create mode 100644 main_shutdown.go create mode 100644 main_shutdown_e2e_test.go create mode 100644 main_shutdown_test.go diff --git a/README.md b/README.md index 5541e5e23..fb4b397ae 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,20 @@ 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; +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 have another five seconds to drain before remaining +streams are closed. Set the supervisor's termination grace period longer than +ten seconds to leave time for storage cleanup. + ## Architecture Architecture diagrams are available in: diff --git a/main.go b/main.go index 6efe3f584..3dc619eac 100644 --- a/main.go +++ b/main.go @@ -568,6 +568,7 @@ func run() error { sqsAdvertisesHTFIFO(), slog.Default()) cleanup.Add(leadershipRefusalDeregister) eg, runCtx := errgroup.WithContext(ctx) + cleanup.Add(startSignalShutdown(runCtx, eg, cancel, runtimes)) startTSOModeReloader(runCtx, eg, tsoWiring) startRaftEngineLifecycleWatchers(runCtx, eg, runtimes) // setupDistributionCatalog + the Stage 7a process-start registration @@ -3091,9 +3092,10 @@ func loadS3BlobPeerTokenFile(path string) (string, error) { // 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 { @@ -3413,7 +3415,7 @@ func startRaftServers( go func() { select { case <-ctx.Done(): - srv.GracefulStop() + stopGRPCServer(srv, shutdownGRPCGracePeriod) _ = lis.Close() closeService() case <-stop: diff --git a/main_shutdown.go b/main_shutdown.go new file mode 100644 index 000000000..bd6ab9233 --- /dev/null +++ b/main_shutdown.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "log/slog" + "math" + "os" + "os/signal" + "sync" + "syscall" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/cockroachdb/errors" + "golang.org/x/sync/errgroup" + "google.golang.org/grpc" +) + +const ( + shutdownTransferTimeout = 5 * time.Second + shutdownTransferRetryDelay = 50 * time.Millisecond + shutdownGRPCGracePeriod = 5 * time.Second + shutdownVoterSuffrage = "voter" +) + +type shutdownLeadershipController interface { + raftengine.LeaderView + raftengine.ConfigReader + TransferLeadershipToServerIfEligible(context.Context, []raftengine.TransferTarget, uint64) error +} + +func stopGRPCServer(server *grpc.Server, gracePeriod time.Duration) { + // Raft and catalog streams can remain open indefinitely. Give in-flight + // requests time to finish, then release streams so storage cleanup can run. + forceStop := time.AfterFunc(gracePeriod, server.Stop) + defer forceStop.Stop() + server.GracefulStop() +} + +func startSignalShutdown(ctx context.Context, eg *errgroup.Group, cancel context.CancelFunc, runtimes []*raftGroupRuntime) func() { + signals := make(chan os.Signal, 1) + signal.Notify(signals, os.Interrupt, syscall.SIGTERM) + eg.Go(func() error { + shutdownOnSignal(ctx, cancel, runtimes, signals, shutdownTransferTimeout) + return nil + }) + return func() { signal.Stop(signals) } +} + +func shutdownOnSignal(ctx context.Context, cancel context.CancelFunc, runtimes []*raftGroupRuntime, signals <-chan os.Signal, timeout time.Duration) { + select { + case <-ctx.Done(): + return + case sig := <-signals: + slog.Info("shutdown signal received", "signal", sig) + } + 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() + transferLeadershipOnShutdown(transferCtx, runtimes) +} + +func transferLeadershipOnShutdown(ctx context.Context, runtimes []*raftGroupRuntime) { + var transfers sync.WaitGroup + for _, rt := range runtimes { + engine := rt.snapshotEngine() + admin, ok := engine.(shutdownLeadershipController) + if !ok || admin.State() != raftengine.StateLeader { + continue + } + 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() +} + +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_e2e_test.go b/main_shutdown_e2e_test.go new file mode 100644 index 000000000..e2b62cab1 --- /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, 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_test.go b/main_shutdown_test.go new file mode 100644 index 000000000..d609d1f52 --- /dev/null +++ b/main_shutdown_test.go @@ -0,0 +1,207 @@ +package main + +import ( + "context" + "errors" + "math" + "net" + "os" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/raftengine" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/health" + healthpb "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/test/bufconn" +) + +type shutdownEngineStub struct { + raftengine.Engine + state raftengine.State + servers []raftengine.Server + transfer func(context.Context, []raftengine.TransferTarget, uint64) error +} + +func (e *shutdownEngineStub) State() raftengine.State { return e.state } +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 { + return e.transfer(ctx, candidates, maxLag) +} + +func shutdownTestEngine(transfer func(context.Context, []raftengine.TransferTarget, uint64) error) *shutdownEngineStub { + return &shutdownEngineStub{ + state: raftengine.StateLeader, + 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, + } +} + +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.state = raftengine.StateFollower + runtimes := []*raftGroupRuntime{ + nil, + {}, + {spec: groupSpec{id: 1}, engine: shutdownTestEngine(transfer)}, + {spec: groupSpec{id: 2}, engine: shutdownTestEngine(transfer)}, + {spec: groupSpec{id: 3}, engine: follower}, + } + signals := make(chan os.Signal, 1) + signals <- syscall.SIGTERM + shutdownOnSignal(ctx, cancel, runtimes, signals, time.Second) + require.ErrorIs(t, ctx.Err(), context.Canceled) + require.Equal(t, int32(2), calls.Load()) +} + +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)}, + } + signals := make(chan os.Signal, 1) + signals <- syscall.SIGTERM + done := make(chan struct{}) + go func() { + shutdownOnSignal(ctx, cancel, runtimes, signals, 200*time.Millisecond) + 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() + shutdownOnSignal(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, make(chan os.Signal), time.Second) +} + +func TestShutdownGRPCServerClosesPersistentStreams(t *testing.T) { + t.Parallel() + const bufferSize = 1 << 20 + listener := bufconn.Listen(bufferSize) + server := grpc.NewServer() + healthpb.RegisterHealthServer(server, health.NewServer()) + t.Cleanup(server.Stop) + go func() { _ = server.Serve(listener) }() + conn, err := grpc.NewClient("passthrough:///shutdown-test", + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { + return listener.DialContext(ctx) + })) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + stream, err := healthpb.NewHealthClient(conn).Watch(t.Context(), &healthpb.HealthCheckRequest{}) + require.NoError(t, err) + _, err = stream.Recv() + require.NoError(t, err, "establish an active stream before stopping the server") + done := make(chan struct{}) + go func() { + stopGRPCServer(server, shutdownTransferRetryDelay) + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("a persistent stream blocked gRPC shutdown") + } + _, err = stream.Recv() + require.Error(t, err) +} From a7278357839879ba5c5eb8fe4102e04bd5f452b2 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 01:16:42 +0900 Subject: [PATCH 03/10] main: revisit leadership acquired during shutdown --- main_shutdown.go | 17 +++++++++++++++ main_shutdown_test.go | 50 +++++++++++++++++++++++++++++++++++++------ 2 files changed, 61 insertions(+), 6 deletions(-) diff --git a/main_shutdown.go b/main_shutdown.go index bd6ab9233..86e1a65ec 100644 --- a/main_shutdown.go +++ b/main_shutdown.go @@ -63,13 +63,29 @@ func shutdownOnSignal(ctx context.Context, cancel context.CancelFunc, runtimes [ } 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 { @@ -82,6 +98,7 @@ func transferLeadershipOnShutdown(ctx context.Context, runtimes []*raftGroupRunt // 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 { diff --git a/main_shutdown_test.go b/main_shutdown_test.go index d609d1f52..bcd94d905 100644 --- a/main_shutdown_test.go +++ b/main_shutdown_test.go @@ -22,12 +22,17 @@ import ( type shutdownEngineStub struct { raftengine.Engine - state raftengine.State + leader atomic.Bool servers []raftengine.Server transfer func(context.Context, []raftengine.TransferTarget, uint64) error } -func (e *shutdownEngineStub) State() raftengine.State { return e.state } +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"} } @@ -35,12 +40,15 @@ func (e *shutdownEngineStub) Configuration(context.Context) (raftengine.Configur return raftengine.Configuration{Servers: e.servers}, nil } func (e *shutdownEngineStub) TransferLeadershipToServerIfEligible(ctx context.Context, candidates []raftengine.TransferTarget, maxLag uint64) error { - return e.transfer(ctx, candidates, maxLag) + 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 { - return &shutdownEngineStub{ - state: raftengine.StateLeader, + engine := &shutdownEngineStub{ servers: []raftengine.Server{ {ID: "local", Address: "local:1", Suffrage: "voter"}, {ID: "learner", Address: "learner:1", Suffrage: "learner"}, @@ -48,6 +56,8 @@ func shutdownTestEngine(transfer func(context.Context, []raftengine.TransferTarg }, transfer: transfer, } + engine.leader.Store(true) + return engine } func TestShutdownSignalTransfersAllLeadersBeforeCancellation(t *testing.T) { @@ -67,7 +77,7 @@ func TestShutdownSignalTransfersAllLeadersBeforeCancellation(t *testing.T) { t.Error("a follower must not initiate transfer") return nil }) - follower.state = raftengine.StateFollower + follower.leader.Store(false) runtimes := []*raftGroupRuntime{ nil, {}, @@ -82,6 +92,34 @@ func TestShutdownSignalTransfersAllLeadersBeforeCancellation(t *testing.T) { 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 + }) + signals := make(chan os.Signal, 1) + signals <- syscall.SIGTERM + shutdownOnSignal(ctx, cancel, []*raftGroupRuntime{ + {spec: groupSpec{id: 1}, engine: follower}, + {spec: groupSpec{id: 2}, engine: leader}, + }, signals, time.Second) + 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()) From c62bbf4dd5963e3d194cc98b1a65bdc5e9af158f Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 02:41:23 +0900 Subject: [PATCH 04/10] main: join shutdown handlers and reap unmount helpers --- .../fuseadapter/helper_process_test.go | 36 +++++++ .../fuseadapter/helper_process_unix.go | 26 ++++++ .../filesystem/fuseadapter/stale_mount.go | 13 ++- .../fuseadapter/stale_mount_test.go | 2 +- .../filesystem/fuseadapter/unmount_context.go | 49 ++++++++++ main.go | 35 +++++-- main_admin.go | 5 +- main_admin_shutdown.go | 71 ++++++++++++++ main_admin_shutdown_test.go | 93 +++++++++++++++++++ main_filesystem_test.go | 11 +++ 10 files changed, 332 insertions(+), 9 deletions(-) create mode 100644 internal/filesystem/fuseadapter/helper_process_test.go create mode 100644 internal/filesystem/fuseadapter/helper_process_unix.go create mode 100644 internal/filesystem/fuseadapter/unmount_context.go create mode 100644 main_admin_shutdown.go create mode 100644 main_admin_shutdown_test.go diff --git a/internal/filesystem/fuseadapter/helper_process_test.go b/internal/filesystem/fuseadapter/helper_process_test.go new file mode 100644 index 000000000..68a4701b7 --- /dev/null +++ b/internal/filesystem/fuseadapter/helper_process_test.go @@ -0,0 +1,36 @@ +//go:build linux || darwin || freebsd + +package fuseadapter + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestUnmountHelperCancellationKillsLateDescendant(t *testing.T) { + dir := t.TempDir() + helper, ready, late := filepath.Join(dir, "helper"), filepath.Join(dir, "ready"), filepath.Join(dir, "late") + script := fmt.Sprintf("#!/bin/sh\n( /bin/sleep 1; printf late > '%s' ) &\nprintf ready > '%s'\nwait\n", late, ready) + require.NoError(t, os.WriteFile(helper, []byte(script), 0o700)) //nolint:gosec // Executable helper fixture, isolated in the test's temporary directory. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- runUnmountHelper(ctx, helper, dir, false) }() + require.Eventually(t, func() bool { _, err := os.Stat(ready); return err == nil }, time.Second, time.Millisecond) + cancel() + select { + case err := <-done: + require.Error(t, err) + case <-time.After(time.Second): + t.Fatal("unmount helper was not reaped after cancellation") + } + time.Sleep(1100 * time.Millisecond) + _, err := os.Stat(late) + require.ErrorIs(t, err, os.ErrNotExist, "an old helper descendant must not act after a replacement mount can appear") +} diff --git a/internal/filesystem/fuseadapter/helper_process_unix.go b/internal/filesystem/fuseadapter/helper_process_unix.go new file mode 100644 index 000000000..536563996 --- /dev/null +++ b/internal/filesystem/fuseadapter/helper_process_unix.go @@ -0,0 +1,26 @@ +//go:build linux || darwin || freebsd + +package fuseadapter + +import ( + "os/exec" + "syscall" + + "github.com/cockroachdb/errors" +) + +func isolateMountHelper(cmd *exec.Cmd) func() { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + kill := func() error { + if cmd.Process == nil { + return nil + } + err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + if errors.Is(err, syscall.ESRCH) { + return nil + } + return errors.WithStack(err) + } + cmd.Cancel = kill + return func() { _ = kill() } +} diff --git a/internal/filesystem/fuseadapter/stale_mount.go b/internal/filesystem/fuseadapter/stale_mount.go index 1358552d1..5ad24f6c4 100644 --- a/internal/filesystem/fuseadapter/stale_mount.go +++ b/internal/filesystem/fuseadapter/stale_mount.go @@ -614,10 +614,21 @@ func detachWithHelpers(ctx context.Context, mountPoint, fsType string) (string, // runDetachHelper runs one helper binary with `-u -z mountPoint` under its // share of ctx's deadline, reporting its output alongside a failure. func runDetachHelper(ctx context.Context, bin, mountPoint string) error { + return runUnmountHelper(ctx, bin, mountPoint, true) +} + +func runUnmountHelper(ctx context.Context, bin, mountPoint string, lazy bool) error { cmdCtx, cancel := context.WithTimeout(ctx, detachHelperTimeout(ctx)) defer cancel() + args := []string{"-u"} + if lazy { + args = append(args, "-z") + } + args = append(args, mountPoint) //nolint:gosec // bin comes from LookPath over two fixed helper names and mountPoint is the operator's configured path. - cmd := exec.CommandContext(cmdCtx, bin, "-u", "-z", mountPoint) + cmd := exec.CommandContext(cmdCtx, bin, args...) + finish := isolateMountHelper(cmd) + defer finish() // After the context kills the helper, do not wait on its output pipes // for longer than this: a child the helper left behind could otherwise // hold them open past the budget. diff --git a/internal/filesystem/fuseadapter/stale_mount_test.go b/internal/filesystem/fuseadapter/stale_mount_test.go index 20b77b9e9..f97f70f2d 100644 --- a/internal/filesystem/fuseadapter/stale_mount_test.go +++ b/internal/filesystem/fuseadapter/stale_mount_test.go @@ -543,7 +543,7 @@ func TestDetachFUSEMountReachesKernelFallbackWhenHelpersHoldPipes(t *testing.T) require.NotErrorIs(t, err, context.DeadlineExceeded, "the helpers' pipe waits must not pre-empt the kernel fallback") require.ErrorContains(t, err, lazyUnmountMethod) elapsed := time.Since(start) - require.GreaterOrEqual(t, elapsed, budget/2-50*time.Millisecond, "the first helper must have been given its share, pipe wait included") + require.GreaterOrEqual(t, elapsed, budget/2-detachHelperWaitDelay-50*time.Millisecond, "the first helper must have been given its execution share") require.Less(t, elapsed, budget, "the fallback must run inside the caller's budget") } diff --git a/internal/filesystem/fuseadapter/unmount_context.go b/internal/filesystem/fuseadapter/unmount_context.go new file mode 100644 index 000000000..de32e60ef --- /dev/null +++ b/internal/filesystem/fuseadapter/unmount_context.go @@ -0,0 +1,49 @@ +package fuseadapter + +import ( + "context" + + "github.com/cockroachdb/errors" +) + +// UnmountContext bounds and reaps the path-based helper before returning. +// Unlike go-fuse's Unmount, cancellation never leaves a helper that could +// later unmount a new filesystem at the same path. +func (s *Server) UnmountContext(ctx context.Context) error { + if s == nil { + return nil + } + if s.adapter != nil { + s.adapter.Close() + } + if s.inner == nil || s.isReleased() || s.mountPoint == "" { + return nil + } + return unmountFUSEMount(ctx, s.mountPoint, s.fsType) +} + +func unmountFUSEMount(ctx context.Context, mountPoint, fsType string) error { + if err := requireFUSEMount(ctx, mountPoint, fsType); err != nil { + if errors.Is(err, errNoFUSEMount) { + return nil + } + return errors.WithStack(err) + } + var helperErr error + for _, helper := range detachMountHelpers { + bin, err := lookupMountHelperContext(ctx, helper) + if err != nil { + helperErr = err + continue + } + if err := requireFUSEMount(ctx, mountPoint, fsType); err != nil { + return errors.WithStack(err) + } + if err := runUnmountHelper(ctx, bin, mountPoint, false); err != nil { + helperErr = err + continue + } + return nil + } + return errors.Wrap(helperErr, "ordinary FUSE unmount helper failed") +} diff --git a/main.go b/main.go index 32dda9263..5ce87dd8f 100644 --- a/main.go +++ b/main.go @@ -1029,6 +1029,7 @@ func mountFilesystemService( // lifecycle drives; tests substitute a fake to exercise the failure paths. type filesystemMountServer interface { Unmount() error + UnmountContext(context.Context) error UnmountLazy(context.Context) error } @@ -1048,7 +1049,9 @@ func installFilesystemServerLifecycle( serveDone <-chan struct{}, mountPoint string, ) { - unmounter := newFilesystemUnmounter(server.Unmount) + unmountCtx, cancelUnmount := context.WithCancel(context.Background()) + unmounter := newFilesystemUnmounter(func() error { return server.UnmountContext(unmountCtx) }) + unmounter.cancel = cancelUnmount cleanup.Add(func() { // Normally the lifecycle goroutine has already run the shutdown by // the time CleanupStack runs (eg.Wait precedes it); then there is @@ -1064,6 +1067,7 @@ func installFilesystemServerLifecycle( slog.Warn("filesystem FUSE unmount failed", "mount_point", mountPoint, "err", err) } case <-time.After(filesystemServeStopTimeout): + cancelUnmount() slog.Error("filesystem FUSE unmount did not return in time during cleanup, continuing", "mount_point", mountPoint, "timeout", filesystemServeStopTimeout, @@ -1090,10 +1094,9 @@ func installFilesystemServerLifecycle( // stopFilesystemServer unmounts during a shutdown within one bounded budget // (filesystemServeStopTimeout). The plain unmount runs in its own goroutine -// because go-fuse's Unmount waits for its event loops after the kernel -// unmount, and a loop stuck in a backend request (leader unavailable) never -// returns; a plain unmount also fails with EBUSY while a client holds files -// on the mount open. In either case the mount is detached lazily, which +// because a plain unmount can block or fail with EBUSY while a client holds +// files open. Its helper is context-bound and must be killed and reaped +// before the lazy detach can start. The mount is then detached lazily, which // also cancels the requests still in flight so a handler blocked in the // backend returns before the stores close, and the wait for the serve loop // is bounded, so the rest of the shutdown (WAL sync, store close) still runs @@ -1128,6 +1131,12 @@ func stopFilesystemServer( // Fall through: the reserved remainder of the budget still waits for // the serve loop, so a request that finishes in time is not cut off // by the store close that follows. + // Killing a context-aware unmount must also reap its helper before + // another path-based unmount can target a replacement filesystem. + if !unmounter.cancelAndJoin(stopCtx) { + slog.ErrorContext(ctx, "filesystem unmount helper did not exit; refusing concurrent lazy detach", "mount_point", mountPoint) + return + } lazyDetachFilesystem(ctx, stopCtx, server, mountPoint) } select { @@ -1140,13 +1149,14 @@ func stopFilesystemServer( } } -// filesystemUnmounter runs Server.Unmount at most once, in the background, +// filesystemUnmounter runs the context-bound unmount at most once, in the background, // and lets any number of waiters observe its completion without a later // caller ever blocking on an earlier call that is stuck (go-fuse's Unmount // waits for an event loop that may be held by a hung backend request; // sync.OnceValue would make the second caller wait for the first). type filesystemUnmounter struct { run func() error + cancel context.CancelFunc started atomic.Bool done chan struct{} err error @@ -1156,6 +1166,19 @@ type filesystemUnmounter struct { handled atomic.Bool } +func (u *filesystemUnmounter) cancelAndJoin(ctx context.Context) bool { + if u.cancel == nil { + return true + } + u.cancel() + select { + case <-u.done: + return true + case <-ctx.Done(): + return false + } +} + func newFilesystemUnmounter(run func() error) *filesystemUnmounter { return &filesystemUnmounter{run: run, done: make(chan struct{})} } diff --git a/main_admin.go b/main_admin.go index 089dff0d6..a10f6a827 100644 --- a/main_admin.go +++ b/main_admin.go @@ -1219,10 +1219,13 @@ func registerAdminLifecycle( version string, ) { addr := adminCfg.Listen + handlers := trackAdminHandlers(httpSrv) eg.Go(func() error { <-ctx.Done() slog.Info("shutting down admin listener", "address", addr, "reason", ctx.Err()) - return monitoring.ShutdownHTTPServer(ctx, httpSrv, "admin", adminShutdownTimeout) + err := monitoring.ShutdownHTTPServer(ctx, httpSrv, "admin", adminShutdownTimeout) + handlers.join(adminHandlerJoinTimeout) + return errors.WithStack(err) }) eg.Go(func() error { slog.Info("starting admin listener", "address", addr, "tls", tlsEnabled, "version", version) diff --git a/main_admin_shutdown.go b/main_admin_shutdown.go new file mode 100644 index 000000000..5bb0da391 --- /dev/null +++ b/main_admin_shutdown.go @@ -0,0 +1,71 @@ +package main + +import ( + "log/slog" + "net/http" + "sync" + "time" +) + +var adminHandlerJoinTimeout = 2 * time.Second + +type adminHandlerDrain struct { + mu sync.Mutex + active int + stopping bool + done chan struct{} +} + +func trackAdminHandlers(server *http.Server) *adminHandlerDrain { + d := &adminHandlerDrain{done: make(chan struct{})} + handler := server.Handler + if handler == nil { + handler = http.DefaultServeMux + } + server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !d.enter() { + http.Error(w, "server is stopping", http.StatusServiceUnavailable) + return + } + defer d.exit() + handler.ServeHTTP(w, r) + }) + return d +} + +func (d *adminHandlerDrain) enter() bool { + d.mu.Lock() + defer d.mu.Unlock() + if d.stopping { + return false + } + d.active++ + return true +} + +func (d *adminHandlerDrain) exit() { + d.mu.Lock() + defer d.mu.Unlock() + d.active-- + if d.stopping && d.active == 0 { + close(d.done) + } +} + +func (d *adminHandlerDrain) join(timeout time.Duration) { + d.mu.Lock() + if !d.stopping { + d.stopping = true + if d.active == 0 { + close(d.done) + } + } + d.mu.Unlock() + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-d.done: + case <-timer.C: + slog.Warn("admin handlers still running after forced connection close", "timeout", timeout) + } +} diff --git a/main_admin_shutdown_test.go b/main_admin_shutdown_test.go new file mode 100644 index 000000000..e72c0d588 --- /dev/null +++ b/main_admin_shutdown_test.go @@ -0,0 +1,93 @@ +package main + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/bootjp/elastickv/internal/admin" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +func TestAdminHandlerDrainJoinsHandlerAfterForcedClose(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + srv := &http.Server{ReadHeaderTimeout: time.Second, Handler: http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + close(entered) + <-release + })} + drain := trackAdminHandlers(srv) + go srv.Handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) + <-entered + done := make(chan struct{}) + go func() { drain.join(time.Second); close(done) }() + select { + case <-done: + t.Fatal("join returned while the admitted handler was running") + case <-time.After(10 * time.Millisecond): + } + close(release) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("handler did not join") + } + response := httptest.NewRecorder() + srv.Handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/", nil)) + require.Equal(t, http.StatusServiceUnavailable, response.Code) +} + +func TestAdminLifecycleWaitsForForceClosedHandler(t *testing.T) { + old := adminShutdownTimeout + adminShutdownTimeout = 10 * time.Millisecond + t.Cleanup(func() { adminShutdownTimeout = old }) + entered, canceled, release := make(chan struct{}), make(chan struct{}), make(chan struct{}) + srv := &http.Server{ReadHeaderTimeout: time.Second, Handler: http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + close(entered) + <-r.Context().Done() + close(canceled) + <-release + })} + listenConfig := &net.ListenConfig{} + listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + group := &errgroup.Group{} + registerAdminLifecycle(ctx, group, srv, listener, &admin.Config{Listen: listener.Addr().String()}, false, "test") + clientDone := make(chan struct{}) + request, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://"+listener.Addr().String(), nil) + require.NoError(t, err) + go func() { + defer close(clientDone) + client := &http.Client{Timeout: time.Second} + response, requestErr := client.Do(request) + if requestErr == nil { + _ = response.Body.Close() + } + }() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("admin handler did not start") + } + cancel() + select { + case <-canceled: + case <-time.After(time.Second): + t.Fatal("admin connection was not forced closed") + } + done := make(chan error, 1) + go func() { done <- group.Wait() }() + select { + case <-done: + t.Fatal("admin lifecycle returned before its handler unwound") + case <-time.After(10 * time.Millisecond): + } + close(release) + require.NoError(t, <-done) + <-clientDone +} diff --git a/main_filesystem_test.go b/main_filesystem_test.go index 256e39894..89bf536eb 100644 --- a/main_filesystem_test.go +++ b/main_filesystem_test.go @@ -294,6 +294,17 @@ func (f *fakeFilesystemMountServer) Unmount() error { return f.unmountErr } +func (f *fakeFilesystemMountServer) UnmountContext(ctx context.Context) error { + if f.unmountBlock != nil { + select { + case <-f.unmountBlock: + case <-ctx.Done(): + return ctx.Err() + } + } + return f.Unmount() +} + func (f *fakeFilesystemMountServer) UnmountLazy(context.Context) error { f.lazyCalls++ if f.onLazy != nil { From 5d2c7ed536bae6c44a43cbff0543182c01abd3ff Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 02:48:38 +0900 Subject: [PATCH 05/10] test: supply contexts to admin HTTP request fixtures --- main_admin_shutdown_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/main_admin_shutdown_test.go b/main_admin_shutdown_test.go index e72c0d588..330a4811e 100644 --- a/main_admin_shutdown_test.go +++ b/main_admin_shutdown_test.go @@ -20,7 +20,7 @@ func TestAdminHandlerDrainJoinsHandlerAfterForcedClose(t *testing.T) { <-release })} drain := trackAdminHandlers(srv) - go srv.Handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/", nil)) + go srv.Handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil)) <-entered done := make(chan struct{}) go func() { drain.join(time.Second); close(done) }() @@ -36,7 +36,7 @@ func TestAdminHandlerDrainJoinsHandlerAfterForcedClose(t *testing.T) { t.Fatal("handler did not join") } response := httptest.NewRecorder() - srv.Handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/", nil)) + srv.Handler.ServeHTTP(response, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil)) require.Equal(t, http.StatusServiceUnavailable, response.Code) } From 50f862525be10ac853670935b3e62b93f1c07bbe Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 03:07:02 +0900 Subject: [PATCH 06/10] fuseadapter: reap startup unmount helpers before detach --- .../fuseadapter/release_deadline_test.go | 14 ++-- internal/filesystem/fuseadapter/server.go | 66 +++++++------------ .../filesystem/fuseadapter/stale_mount.go | 8 ++- .../fuseadapter/stale_mount_test.go | 13 ++-- .../filesystem/fuseadapter/unmount_context.go | 23 +++++-- .../fuseadapter/unmount_context_test.go | 52 +++++++++++++++ 6 files changed, 110 insertions(+), 66 deletions(-) create mode 100644 internal/filesystem/fuseadapter/unmount_context_test.go diff --git a/internal/filesystem/fuseadapter/release_deadline_test.go b/internal/filesystem/fuseadapter/release_deadline_test.go index 794db8708..ddd542976 100644 --- a/internal/filesystem/fuseadapter/release_deadline_test.go +++ b/internal/filesystem/fuseadapter/release_deadline_test.go @@ -1,6 +1,7 @@ package fuseadapter import ( + "context" "os" "path/filepath" "testing" @@ -29,13 +30,10 @@ func TestReleaseAfterReadinessSharesTheUnmountDeadline(t *testing.T) { marker := writeRecordingMountHelper(t) ready := make(chan error, 1) time.AfterFunc(1600*time.Millisecond, func() { ready <- nil }) - hung := make(chan struct{}) - t.Cleanup(func() { close(hung) }) - start := time.Now() - releaseAfterReadiness(ready, func() error { - <-hung - return nil + releaseAfterReadiness(ready, func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() }, mountPoint, fuseTypeFor(filesystemName)) elapsed := time.Since(start) require.FileExists(t, marker, "the hung unmount must be followed by the lazy detach") @@ -56,7 +54,7 @@ func TestReleaseAfterReadinessBoundsTheLazyDetach(t *testing.T) { unmounted := make(chan struct{}, 1) start := time.Now() - releaseAfterReadiness(make(chan error), func() error { + releaseAfterReadiness(make(chan error), func(context.Context) error { unmounted <- struct{}{} return nil }, mountPoint, fuseTypeFor(filesystemName)) @@ -76,7 +74,7 @@ func TestReleaseAbandonedSkipsUnmountPastItsDeadline(t *testing.T) { marker := writeRecordingMountHelper(t) unmounted := make(chan struct{}, 1) - releaseAbandonedBy(time.Now().Add(-time.Millisecond), func() error { + releaseAbandonedBy(time.Now().Add(-time.Millisecond), func(context.Context) error { unmounted <- struct{}{} return nil }, mountPoint, fuseTypeFor(filesystemName)) diff --git a/internal/filesystem/fuseadapter/server.go b/internal/filesystem/fuseadapter/server.go index 3f257de87..99a5c2943 100644 --- a/internal/filesystem/fuseadapter/server.go +++ b/internal/filesystem/fuseadapter/server.go @@ -629,9 +629,8 @@ func newServerContext(ctx context.Context, fs *RawFileSystem, mountPoint string, } // abandonedUnmountTimeout and abandonedDetachTimeout split the one window a -// mount whose caller gave up on it is released in. The plain unmount (go-fuse -// runs the unmount helper without a bound and then waits for its event -// loops, either of which can hang), together with the wait for a cancelled +// mount whose caller gave up on it is released in. The context-bound plain +// unmount, together with the wait for a cancelled // readiness check before it, must finish within abandonedUnmountTimeout of // the start of the release; the lazy detach that may follow must finish // within abandonedDetachTimeout after that deadline. A release therefore @@ -647,10 +646,8 @@ var ( // serves, now and later (fs is its request bridge), so a client request that // reached the backend before the release returns instead of running on // while the caller's unwinding closes the stores. go-fuse's NewServer -// registers the event loop with its wait group and Unmount waits for that -// group, but only Serve runs the loop that releases it, so Serve must be -// running for Unmount to return; the loop exits by itself once the kernel -// unmount closes the connection. +// registers an event loop that only Serve can drain, so Serve must run +// while the context-bound unmount releases the kernel connection. func releaseAbandonedMount(results <-chan newServerResult, fs *RawFileSystem, mountPoint, fsType string) { res := <-results if res.server == nil { @@ -658,12 +655,12 @@ func releaseAbandonedMount(results <-chan newServerResult, fs *RawFileSystem, mo } fs.cancelInFlight() go res.server.Serve() - releaseAbandoned(res.server.Unmount, mountPoint, fsType) + releaseAbandoned(func(ctx context.Context) error { return unmountFUSEMount(ctx, mountPoint, fsType) }, mountPoint, fsType) } // releaseAbandoned releases a mount in a release window that starts now // (see releaseAbandonedBy). -func releaseAbandoned(unmount func() error, mountPoint, fsType string) { +func releaseAbandoned(unmount func(context.Context) error, mountPoint, fsType string) { releaseAbandonedBy(time.Now().Add(abandonedUnmountTimeout), unmount, mountPoint, fsType) } @@ -673,48 +670,30 @@ func releaseAbandoned(unmount func() error, mountPoint, fsType string) { // behind; the lazy detach itself refuses a path that no longer holds a // mount of fsType, so an unmount that did succeed late is not followed by a // detach of whatever sits underneath. When unmountBy has already passed, -// unmount is not started at all: go-fuse's unmount runs without a bound, -// and one left running beside the lazy detach could act, by path, on a -// mount the detach exposed. -func releaseAbandonedBy(unmountBy time.Time, unmount func() error, mountPoint, fsType string) { - if wait := time.Until(unmountBy); wait <= 0 { +// unmount is not started at all. The callback must reap its context-bound +// helper before returning so nothing can later act on a replacement mount. +func releaseAbandonedBy(unmountBy time.Time, unmount func(context.Context) error, mountPoint, fsType string) { + ctx, cancel := context.WithDeadline(context.Background(), unmountBy) + if ctx.Err() != nil { slog.Warn("filesystem FUSE unmount of an abandoned mount has no time left, detaching lazily", "mount_point", mountPoint, "timeout", abandonedUnmountTimeout, ) - } else if !unmountWithin(wait, unmount, mountPoint) { + } else if err := unmount(ctx); err == nil { + cancel() return + } else { + slog.Warn("filesystem FUSE unmount of an abandoned mount failed, detaching lazily", "mount_point", mountPoint, "err", err) } - ctx, cancel := context.WithDeadline(context.Background(), unmountBy.Add(abandonedDetachTimeout)) + // The context-bound callback reaps its helper before returning. + cancel() + ctx, cancel = context.WithDeadline(context.Background(), unmountBy.Add(abandonedDetachTimeout)) defer cancel() if _, err := detachFUSEMount(ctx, mountPoint, fsType); err != nil && !errors.Is(err, errNoFUSEMount) { slog.Warn("filesystem FUSE lazy detach of an abandoned mount failed", "mount_point", mountPoint, "err", err) } } -// unmountWithin runs unmount for up to wait and reports whether the mount -// still needs the lazy detach: true when unmount failed or did not return -// in time, false when it unmounted cleanly. -func unmountWithin(wait time.Duration, unmount func() error, mountPoint string) bool { - done := make(chan error, 1) - go func() { done <- unmount() }() - timer := time.NewTimer(wait) - defer timer.Stop() - select { - case err := <-done: - if err == nil { - return false - } - slog.Warn("filesystem FUSE unmount of an abandoned mount failed, detaching lazily", "mount_point", mountPoint, "err", err) - case <-timer.C: - slog.Warn("filesystem FUSE unmount of an abandoned mount did not return in time, detaching lazily", - "mount_point", mountPoint, - "timeout", abandonedUnmountTimeout, - ) - } - return true -} - func recoverIntentsBeforeMount(ctx context.Context, recoverer intentRecoverer) error { for { stats, err := recoverer.RecoverIntents(ctx, startupRecoveryBatch) @@ -814,7 +793,7 @@ func (s *Server) WaitMount() error { func (s *Server) release(ready <-chan error) { s.releaseOnce.Do(func() { s.fs.cancelInFlight() - releaseAfterReadiness(ready, s.inner.Unmount, s.mountPoint, s.fsType) + releaseAfterReadiness(ready, s.UnmountContext, s.mountPoint, s.fsType) if s.released != nil { close(s.released) } @@ -827,15 +806,14 @@ func (s *Server) release(ready <-chan error) { // deadline, so a check that returns late leaves the unmount only what is // left of it, and the lazy detach runs by its end. A check still running at // the unmount deadline means a backend request is ignoring its context, and -// go-fuse's Unmount (which must not run alongside the check, and would wait -// on the stuck request) is skipped for the lazy detach alone. -func releaseAfterReadiness(ready <-chan error, unmount func() error, mountPoint, fsType string) { +// the ordinary unmount is skipped for the lazy detach alone. +func releaseAfterReadiness(ready <-chan error, unmount func(context.Context) error, mountPoint, fsType string) { unmountBy := time.Now().Add(abandonedUnmountTimeout) timer := time.NewTimer(abandonedUnmountTimeout) select { case <-ready: case <-timer.C: - unmount = func() error { return errReadinessStuck } + unmount = func(context.Context) error { return errReadinessStuck } } timer.Stop() releaseAbandonedBy(unmountBy, unmount, mountPoint, fsType) diff --git a/internal/filesystem/fuseadapter/stale_mount.go b/internal/filesystem/fuseadapter/stale_mount.go index 5ad24f6c4..0b647adaa 100644 --- a/internal/filesystem/fuseadapter/stale_mount.go +++ b/internal/filesystem/fuseadapter/stale_mount.go @@ -618,13 +618,17 @@ func runDetachHelper(ctx context.Context, bin, mountPoint string) error { } func runUnmountHelper(ctx context.Context, bin, mountPoint string, lazy bool) error { - cmdCtx, cancel := context.WithTimeout(ctx, detachHelperTimeout(ctx)) - defer cancel() args := []string{"-u"} if lazy { args = append(args, "-z") } args = append(args, mountPoint) + return runMountHelper(ctx, bin, args) +} + +func runMountHelper(ctx context.Context, bin string, args []string) error { + cmdCtx, cancel := context.WithTimeout(ctx, detachHelperTimeout(ctx)) + defer cancel() //nolint:gosec // bin comes from LookPath over two fixed helper names and mountPoint is the operator's configured path. cmd := exec.CommandContext(cmdCtx, bin, args...) finish := isolateMountHelper(cmd) diff --git a/internal/filesystem/fuseadapter/stale_mount_test.go b/internal/filesystem/fuseadapter/stale_mount_test.go index 4a02cf088..fe591b23a 100644 --- a/internal/filesystem/fuseadapter/stale_mount_test.go +++ b/internal/filesystem/fuseadapter/stale_mount_test.go @@ -766,13 +766,10 @@ func TestReleaseAbandonedDetachesWhenUnmountHangs(t *testing.T) { mountPoint := t.TempDir() useFakeDetachMountTable(t, fuseTableLine(mountPoint)) marker := writeRecordingMountHelper(t) - release := make(chan struct{}) - t.Cleanup(func() { close(release) }) - start := time.Now() - releaseAbandoned(func() error { - <-release - return nil + releaseAbandoned(func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() }, mountPoint, fuseTypeFor(filesystemName)) require.Less(t, time.Since(start), 5*time.Second, "the release must not wait on the hung unmount") require.FileExists(t, marker, "the lazy detach must run once the unmount overran its bound") @@ -786,9 +783,9 @@ func TestReleaseAbandonedFallsBackOnlyWhenUnmountFails(t *testing.T) { useFakeDetachMountTable(t, fuseTableLine(mountPoint)) marker := writeRecordingMountHelper(t) - releaseAbandoned(func() error { return nil }, mountPoint, fuseTypeFor(filesystemName)) + releaseAbandoned(func(context.Context) error { return nil }, mountPoint, fuseTypeFor(filesystemName)) require.NoFileExists(t, marker, "a clean unmount needs no detach") - releaseAbandoned(func() error { return errors.New("device or resource busy") }, mountPoint, fuseTypeFor(filesystemName)) + releaseAbandoned(func(context.Context) error { return errors.New("device or resource busy") }, mountPoint, fuseTypeFor(filesystemName)) require.FileExists(t, marker, "a failed unmount falls back to the lazy detach") } diff --git a/internal/filesystem/fuseadapter/unmount_context.go b/internal/filesystem/fuseadapter/unmount_context.go index de32e60ef..dd8edcb2d 100644 --- a/internal/filesystem/fuseadapter/unmount_context.go +++ b/internal/filesystem/fuseadapter/unmount_context.go @@ -23,23 +23,27 @@ func (s *Server) UnmountContext(ctx context.Context) error { } func unmountFUSEMount(ctx context.Context, mountPoint, fsType string) error { - if err := requireFUSEMount(ctx, mountPoint, fsType); err != nil { + if err := verifyOrdinaryUnmount(ctx, mountPoint, fsType); err != nil { if errors.Is(err, errNoFUSEMount) { return nil } return errors.WithStack(err) } var helperErr error - for _, helper := range detachMountHelpers { + for _, helper := range []string{"fusermount3", "fusermount", "umount"} { bin, err := lookupMountHelperContext(ctx, helper) if err != nil { helperErr = err continue } - if err := requireFUSEMount(ctx, mountPoint, fsType); err != nil { + if err := verifyOrdinaryUnmount(ctx, mountPoint, fsType); err != nil { return errors.WithStack(err) } - if err := runUnmountHelper(ctx, bin, mountPoint, false); err != nil { + args := []string{mountPoint} + if helper != "umount" { + args = append([]string{"-u"}, args...) + } + if err := runMountHelper(ctx, bin, args); err != nil { helperErr = err continue } @@ -47,3 +51,14 @@ func unmountFUSEMount(ctx context.Context, mountPoint, fsType string) error { } return errors.Wrap(helperErr, "ordinary FUSE unmount helper failed") } + +func verifyOrdinaryUnmount(ctx context.Context, mountPoint, fsType string) error { + err := requireFUSEMount(ctx, mountPoint, fsType) + var unavailable *mountTableUnavailableError + if errors.As(err, &unavailable) { + // Startup permits this platform/container configuration. Only ordinary + // unmount is allowed without the table; lazy detach stays type-guarded. + return nil + } + return errors.WithStack(err) +} diff --git a/internal/filesystem/fuseadapter/unmount_context_test.go b/internal/filesystem/fuseadapter/unmount_context_test.go new file mode 100644 index 000000000..5a0939030 --- /dev/null +++ b/internal/filesystem/fuseadapter/unmount_context_test.go @@ -0,0 +1,52 @@ +package fuseadapter + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestOrdinaryUnmountWithoutMountTableUsesBoundedHelper(t *testing.T) { + helperDir := t.TempDir() + marker := filepath.Join(helperDir, "called") + t.Setenv("ELASTICKV_ORDINARY_MARKER", marker) + script := "#!/bin/sh\nprintf '%s' \"$*\" > \"$ELASTICKV_ORDINARY_MARKER\"\n" + //nolint:gosec // Executable helper fixture, isolated in the test's temporary directory. + require.NoError(t, os.WriteFile(filepath.Join(helperDir, "umount"), []byte(script), 0o700)) + onlyMountHelpersIn(t, helperDir) + table := useFakeDetachMountTable(t, "") + require.NoError(t, os.Remove(table)) + mountPoint := t.TempDir() + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + require.NoError(t, unmountFUSEMount(ctx, mountPoint, fuseTypeFor(filesystemName))) + data, err := os.ReadFile(marker) + require.NoError(t, err) + require.Equal(t, mountPoint, string(data), "ordinary umount receives no force/lazy flags") + _, err = detachFUSEMount(ctx, mountPoint, fuseTypeFor(filesystemName)) + require.Error(t, err, "lazy detach must still refuse an unverifiable mount") +} + +func TestAbandonedReleaseReapsOrdinaryHelperBeforeDetach(t *testing.T) { + shortenAbandonedRelease(t, 2*time.Second, 3*time.Second) + helperDir := t.TempDir() + detached, late := filepath.Join(helperDir, "detached"), filepath.Join(helperDir, "late") + t.Setenv("ELASTICKV_DETACHED_MARKER", detached) + t.Setenv("ELASTICKV_LATE_MARKER", late) + script := "#!/bin/sh\nif [ \"$2\" = '-z' ]; then : > \"$ELASTICKV_DETACHED_MARKER\"; exit 0; fi\n( /bin/sleep 1; : > \"$ELASTICKV_LATE_MARKER\" ) &\nexec /bin/sleep 5\n" + //nolint:gosec // Executable helper fixture, isolated in the test's temporary directory. + require.NoError(t, os.WriteFile(filepath.Join(helperDir, "fusermount3"), []byte(script), 0o700)) + onlyMountHelpersIn(t, helperDir) + mountPoint := t.TempDir() + useFakeDetachMountTable(t, fuseTableLine(mountPoint)) + releaseAbandoned(func(ctx context.Context) error { + return unmountFUSEMount(ctx, mountPoint, fuseTypeFor(filesystemName)) + }, mountPoint, fuseTypeFor(filesystemName)) + require.FileExists(t, detached) + time.Sleep(1100 * time.Millisecond) + require.NoFileExists(t, late, "ordinary helper descendants must not survive the lazy detach") +} From 2eabcc7090c7123e17fcd99921090d2dd6d02bec Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 03:59:20 +0900 Subject: [PATCH 07/10] Arm shutdown handoff immediately after server startup --- main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main.go b/main.go index 95dc08399..f0d023ee5 100644 --- a/main.go +++ b/main.go @@ -720,6 +720,7 @@ func run() error { }); err != nil { return err } + handoff.arm(runCtx, eg, runtimes) startLeaderBalanceScheduler( runCtx, eg, @@ -727,7 +728,6 @@ func run() error { leaderBalanceConfigFromFlags(*raftId, cfg.defaultGroup, cfg.sqsFifoPartitionMap, metricsRegistry.Registerer()), ) - handoff.arm(runCtx, eg, runtimes) return waitRunGroup(eg) } From c7a14b3230c44d934455ce8b11aafd9813cf11d8 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 04:24:47 +0900 Subject: [PATCH 08/10] Quiesce leader balancing before shutdown leadership handoff --- ...22_proposed_sigterm_leadership_transfer.md | 6 +- main.go | 10 +- main_leader_balance.go | 15 ++- main_shutdown.go | 24 ++++- main_shutdown_balance_test.go | 101 ++++++++++++++++++ main_shutdown_handoff_test.go | 6 +- main_shutdown_test.go | 8 +- 7 files changed, 154 insertions(+), 16 deletions(-) create mode 100644 main_shutdown_balance_test.go diff --git a/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md b/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md index 09d7f3446..95f530579 100644 --- a/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md +++ b/docs/design/2026_10_22_proposed_sigterm_leadership_transfer.md @@ -17,7 +17,11 @@ still be able to shut down. 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. Snapshot each runtime's engine through `snapshotEngine`. For each local +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. diff --git a/main.go b/main.go index f0d023ee5..5fa706168 100644 --- a/main.go +++ b/main.go @@ -720,12 +720,18 @@ func run() error { }); err != nil { return err } - handoff.arm(runCtx, eg, runtimes) + 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) 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 index 30d6048fc..b0cffc77f 100644 --- a/main_shutdown.go +++ b/main_shutdown.go @@ -43,7 +43,7 @@ func (h *shutdownHandoff) requestShutdown() { request() } -func (h *shutdownHandoff) arm(ctx context.Context, eg *errgroup.Group, runtimes []*raftGroupRuntime) { +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 { @@ -52,12 +52,12 @@ func (h *shutdownHandoff) arm(ctx context.Context, eg *errgroup.Group, runtimes requests := make(chan struct{}) h.request = sync.OnceFunc(func() { close(requests) }) eg.Go(func() error { - shutdownAfterRequest(ctx, h.cancel, runtimes, requests, shutdownTransferTimeout) + 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) { +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 @@ -70,8 +70,14 @@ func shutdownAfterRequest(ctx context.Context, cancel context.CancelFunc, runtim 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) - close(done) }() // Logging may be backpressured. It must not hold root cancellation or // the errgroup beyond the handoff deadline. @@ -81,6 +87,16 @@ func shutdownAfterRequest(ctx context.Context, cancel context.CancelFunc, runtim } } +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() 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_handoff_test.go b/main_shutdown_handoff_test.go index d2a3ac186..85a42a27b 100644 --- a/main_shutdown_handoff_test.go +++ b/main_shutdown_handoff_test.go @@ -28,7 +28,7 @@ func TestShutdownHandoffStartupSignalCancelsBeforeArming(t *testing.T) { t.Fatal("a signal during startup must cancel immediately") } eg, runCtx := errgroup.WithContext(ctx) - handoff.arm(runCtx, eg, nil) + 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()) } @@ -54,7 +54,7 @@ func TestShutdownHandoffSignalWaitsForTransferAndIgnoresRepeats(t *testing.T) { 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}}) + handoff.arm(runCtx, eg, []*raftGroupRuntime{{spec: groupSpec{id: 1}, engine: engine}}, nil) require.NoError(t, syscall.Kill(os.Getpid(), syscall.SIGUSR2)) select { case <-started: @@ -97,7 +97,7 @@ func TestShutdownHandoffDeadlineSurvivesBlockedLogging(t *testing.T) { close(requests) done := make(chan struct{}) go func() { - shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, requests, shutdownTransferRetryDelay) + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, requests, shutdownTransferRetryDelay, nil) close(done) }() select { diff --git a/main_shutdown_test.go b/main_shutdown_test.go index 15c6e9532..2255afffc 100644 --- a/main_shutdown_test.go +++ b/main_shutdown_test.go @@ -79,7 +79,7 @@ func TestShutdownSignalTransfersAllLeadersBeforeCancellation(t *testing.T) { } requests := make(chan struct{}) close(requests) - shutdownAfterRequest(ctx, cancel, runtimes, requests, time.Second) + shutdownAfterRequest(ctx, cancel, runtimes, requests, time.Second, nil) require.ErrorIs(t, ctx.Err(), context.Canceled) require.Equal(t, int32(2), calls.Load()) } @@ -106,7 +106,7 @@ func TestShutdownSignalTransfersLeadershipAcquiredDuringHandoff(t *testing.T) { shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{ {spec: groupSpec{id: 1}, engine: follower}, {spec: groupSpec{id: 2}, engine: leader}, - }, requests, time.Second) + }, 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) @@ -130,7 +130,7 @@ func TestShutdownSignalSharesDeadlineAcrossGroups(t *testing.T) { close(requests) done := make(chan struct{}) go func() { - shutdownAfterRequest(ctx, cancel, runtimes, requests, 200*time.Millisecond) + shutdownAfterRequest(ctx, cancel, runtimes, requests, 200*time.Millisecond, nil) close(done) }() for range runtimes { @@ -200,5 +200,5 @@ func TestShutdownAlreadyCanceledDoesNotTransfer(t *testing.T) { }) ctx, cancel := context.WithCancel(context.Background()) cancel() - shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, make(chan struct{}), time.Second) + shutdownAfterRequest(ctx, cancel, []*raftGroupRuntime{{engine: engine}}, make(chan struct{}), time.Second, nil) } From e784b44fe693d75ca1dca7067a7a9744fe0fa6fb Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 05:02:47 +0900 Subject: [PATCH 09/10] Snapshot FUSE helper fallback before asynchronous lookup --- .../helper_lookup_snapshot_test.go | 59 +++++++++++++++++++ .../filesystem/fuseadapter/stale_mount.go | 32 ++++++++-- 2 files changed, 86 insertions(+), 5 deletions(-) create mode 100644 internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go 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..9339cb16c --- /dev/null +++ b/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go @@ -0,0 +1,59 @@ +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): + } +} 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) } From aec5bf8c06bb294e5a4553879a2a9edffe0a680f Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 23 Oct 2026 05:08:59 +0900 Subject: [PATCH 10/10] Test FUSE lookup snapshots against concurrent fixture ownership --- .../helper_lookup_snapshot_test.go | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go b/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go index 9339cb16c..0976f29a5 100644 --- a/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go +++ b/internal/filesystem/fuseadapter/helper_lookup_snapshot_test.go @@ -57,3 +57,24 @@ func TestCanceledMountHelperLookupDoesNotLaunchWorker(t *testing.T) { 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) +}