diff --git a/internal/server/server.go b/internal/server/server.go index 24185870..c64607c7 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -29,7 +29,19 @@ func setupRouter(ctx context.Context, logger *slog.Logger, serverConf config.Ser return r } -func Start(ctx context.Context, logger *slog.Logger, serverConf config.Server, tracingConf config.Tracing, s *api.ServeStrategy) { +// minDrainTimeout is the floor for the shutdown drain window, and drainGrace +// is added on top of the blocking strategy's default timeout so a request held +// right up to that timeout still has time to write its response. +const ( + minDrainTimeout = 15 * time.Second + drainGrace = 5 * time.Second +) + +// Start runs the HTTP server until ctx is cancelled, then drains in-flight +// requests before returning. It returns nil after a clean shutdown, or the +// fatal serve error (e.g. the port is already in use) so the caller can +// terminate the process instead of running on without a listener. +func Start(ctx context.Context, logger *slog.Logger, serverConf config.Server, tracingConf config.Tracing, s *api.ServeStrategy) error { start := time.Now() if logger.Enabled(ctx, slog.LevelDebug) { @@ -51,24 +63,44 @@ func Start(ctx context.Context, logger *slog.Logger, serverConf config.Server, t slog.String("mode", gin.Mode()), ) - go StartHttp(server, logger) + errC := make(chan error, 1) + go func() { + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errC <- err + } + }() + + select { + case err := <-errC: + // ListenAndServe failed outright (port in use, bad address, ...). + return fmt.Errorf("server: %w", err) + case <-ctx.Done(): + } - // Graceful web server shutdown. - <-ctx.Done() - logger.Info("server: shutting down") - err := server.Close() - if err != nil { - logger.Error("server: shutdown failed", slog.Any("error", err)) + // Drain in-flight requests instead of resetting them: the blocking + // strategy holds requests open up to its configured timeout by design, so + // the drain window must outlive it. The health endpoint already reports + // 503 once ctx is cancelled, taking the instance out of rotation while it + // drains. + drain := drainTimeout(s.StrategyConfig.Blocking.DefaultTimeout) + logger.Info("server: shutting down, draining in-flight requests", slog.Duration("drain_timeout", drain)) + shutdownCtx, cancel := context.WithTimeout(context.Background(), drain) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + logger.Error("server: drain did not complete in time, closing remaining connections", slog.Any("error", err)) + _ = server.Close() + return nil } + logger.Info("server: shutdown complete") + return nil } -// StartHttp starts the Web server in http mode. -func StartHttp(s *http.Server, logger *slog.Logger) { - if err := s.ListenAndServe(); err != nil { - if errors.Is(err, http.ErrServerClosed) { - logger.Info("server: shutdown complete") - } else { - logger.Error("server failed to start", slog.Any("error", err)) - } +// drainTimeout returns the shutdown drain window for the given blocking- +// strategy default timeout: the timeout plus a response grace, never below +// minDrainTimeout. +func drainTimeout(blockingTimeout time.Duration) time.Duration { + if d := blockingTimeout + drainGrace; d > minDrainTimeout { + return d } + return minDrainTimeout } diff --git a/internal/server/server_test.go b/internal/server/server_test.go new file mode 100644 index 00000000..8b1bd1cb --- /dev/null +++ b/internal/server/server_test.go @@ -0,0 +1,167 @@ +package server + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "testing" + "time" + + "github.com/neilotoole/slogt" + "github.com/sablierapp/sablier/internal/api" + "github.com/sablierapp/sablier/internal/api/apitest" + "github.com/sablierapp/sablier/pkg/config" + "github.com/sablierapp/sablier/pkg/metrics" + "github.com/sablierapp/sablier/pkg/sablier" + "github.com/sablierapp/sablier/pkg/theme" + "go.uber.org/mock/gomock" + "gotest.tools/v3/assert" +) + +func testStrategy(t *testing.T) *api.ServeStrategy { + t.Helper() + th, err := theme.New(slogt.New(t)) + assert.NilError(t, err) + return &api.ServeStrategy{ + Theme: th, + Metrics: metrics.Noop{}, + StrategyConfig: config.NewStrategyConfig(), + SessionsConfig: config.NewSessionsConfig(), + } +} + +func readySession() *sablier.SessionState { + return &sablier.SessionState{Instances: map[string]sablier.InstanceInfoWithError{ + "test": {Instance: sablier.InstanceInfo{Name: "test", Status: sablier.InstanceStatusReady, CurrentReplicas: 1, DesiredReplicas: 1}}, + }} +} + +// freePort reserves an ephemeral port and releases it for the server to bind. +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + assert.NilError(t, err) + port := l.Addr().(*net.TCPAddr).Port + assert.NilError(t, l.Close()) + return port +} + +func waitReachable(url string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + resp, err := http.Get(url) //nolint:gosec // local test URL + if err == nil { + _ = resp.Body.Close() + return nil + } + time.Sleep(25 * time.Millisecond) + } + return fmt.Errorf("%s not reachable within %s", url, timeout) +} + +// TestStartReturnsOnServeError pins the port-conflict behavior: when the +// listener cannot bind, Start must return the error instead of logging it and +// leaving the caller blocked forever with no HTTP server (a zombie process +// that looks healthy but serves nothing). +func TestStartReturnsOnServeError(t *testing.T) { + // Occupy the wildcard address, which is what the server binds. + l, err := net.Listen("tcp", ":0") + assert.NilError(t, err) + defer l.Close() //nolint:errcheck + port := l.Addr().(*net.TCPAddr).Port + + conf := config.NewServerConfig() + conf.Port = port // already taken by l + + done := make(chan error, 1) + go func() { + done <- Start(t.Context(), slogt.New(t), conf, config.NewTracingConfig(), testStrategy(t)) + }() + + select { + case err := <-done: + assert.ErrorContains(t, err, "server:") + case <-time.After(5 * time.Second): + t.Fatal("Start did not return after a bind failure") + } +} + +// TestStartDrainsInFlightRequestsOnShutdown pins graceful shutdown: a request +// being processed when ctx is cancelled must complete with a full response. +// The previous implementation called server.Close, which resets live +// connections; the blocking strategy holds requests open by design, so every +// restart cut them off mid-wait. +func TestStartDrainsInFlightRequestsOnShutdown(t *testing.T) { + port := freePort(t) + conf := config.NewServerConfig() + conf.Port = port + + strategy := testStrategy(t) + ctrl := gomock.NewController(t) + mock := apitest.NewMockSablier(ctrl) + strategy.Sablier = mock + + inFlight := make(chan struct{}) // closed when the handler enters the session call + release := make(chan struct{}) // closed by the test after shutdown has started + mock.EXPECT().RequestReadySession(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + DoAndReturn(func(context.Context, []string, time.Duration, time.Duration) (*sablier.SessionState, error) { + close(inFlight) + <-release + return readySession(), nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { + done <- Start(ctx, slogt.New(t), conf, config.NewTracingConfig(), strategy) + }() + + base := fmt.Sprintf("http://127.0.0.1:%d", port) + assert.NilError(t, waitReachable(base+"/health", 5*time.Second)) + + respC := make(chan error, 1) + go func() { + resp, err := http.Get(base + "/api/strategies/blocking?names=test") + if err != nil { + respC <- err + return + } + defer resp.Body.Close() //nolint:errcheck + if _, err := io.ReadAll(resp.Body); err != nil { + respC <- err + return + } + if resp.StatusCode != http.StatusOK { + respC <- fmt.Errorf("unexpected status %d", resp.StatusCode) + return + } + respC <- nil + }() + + <-inFlight // the request is provably being processed + cancel() // shutdown begins while the request is held + + // Give the shutdown path time to act on the connection: the old Close-based + // implementation resets it here, the drain-based one waits. + time.Sleep(100 * time.Millisecond) + close(release) // let the handler finish + + assert.NilError(t, <-respC, "in-flight request must complete during the drain window") + + select { + case err := <-done: + assert.NilError(t, err) + case <-time.After(30 * time.Second): + t.Fatal("Start did not return after shutdown") + } +} + +func TestDrainTimeout(t *testing.T) { + // The floor applies when the blocking timeout is short. + assert.Equal(t, minDrainTimeout, drainTimeout(time.Second)) + // Long blocking timeouts extend the window by the grace period. + assert.Equal(t, 5*time.Minute+drainGrace, drainTimeout(5*time.Minute)) +} diff --git a/pkg/sabliercmd/start.go b/pkg/sabliercmd/start.go index c473a295..c8d30930 100644 --- a/pkg/sabliercmd/start.go +++ b/pkg/sabliercmd/start.go @@ -183,22 +183,31 @@ func Start(ctx context.Context, conf config.Config) error { SessionsConfig: conf.Sessions, } - go server.Start(ctx, logger, conf.Server, conf.Tracing, strategy) - - // Listen for the interrupt signal. - <-ctx.Done() - - stop() - logger.InfoContext(ctx, "shutting down gracefully, press Ctrl+C again to force") + serverErr := make(chan error, 1) + go func() { + serverErr <- server.Start(ctx, logger, conf.Server, conf.Tracing, strategy) + }() - ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() + // Block until the process is signalled or the HTTP server dies. A serve + // failure (e.g. the port is already in use) must terminate the process: + // otherwise the watchers keep running with no listener and the instance + // looks healthy while serving nothing. + select { + case <-ctx.Done(): + stop() + logger.Info("shutting down gracefully, press Ctrl+C again to force") + // Wait for the server to finish draining in-flight requests before + // persisting state and exiting. + err = <-serverErr + case err = <-serverErr: + stop() + } save() - logger.InfoContext(ctx, "Server exiting") + logger.Info("Server exiting") - return nil + return err } func buildRecorder(enabled bool) metrics.Recorder {