fix(server): graceful shutdown drains held requests; bind failures terminate the process

internal/server/server.go

@@ -29,7 +29,19 @@
 	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 @@
 		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
 }

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))
+}

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 {