diff --git a/cmd/tapir/main.go b/cmd/tapir/main.go index b94fe5d..f09e6dc 100644 --- a/cmd/tapir/main.go +++ b/cmd/tapir/main.go @@ -17,9 +17,9 @@ import ( "errors" "fmt" "log/slog" + "net" "net/http" "os" - "os/signal" "sync" "time" @@ -41,7 +41,7 @@ func main() { } log := slog.New(slog.NewTextHandler(os.Stderr, nil)) - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + ctx, stop := shutdownContext(context.Background()) defer stop() var err error @@ -215,9 +215,11 @@ func cmdServe(ctx context.Context, log *slog.Logger) error { if err != nil { return err } + var workerDone chan struct{} // nil without uploads: nothing to wait for at shutdown wireUploads(app, cfg, intake) if intake != nil { - go worker.Run(ctx) + workerDone = make(chan struct{}) + go func() { worker.Run(ctx); close(workerDone) }() log.Info("audio upload enabled", "model", cfg.TranscribeModel, "fallback", cfg.TranscribeFallbackModel, "max_mb", cfg.UploadMaxBytes>>20, "max_minutes", int(cfg.UploadMaxDuration.Minutes())) log.Warn("audio upload assumes a SINGLE replica: the transcription queue is process-local (ADR-033, ADR-018)") @@ -350,20 +352,19 @@ func cmdServe(ctx context.Context, log *slog.Logger) error { }() } - // Graceful shutdown on signal: stop accepting, drain in-flight requests. - go func() { - <-ctx.Done() - shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _ = srv.Shutdown(shutdownCtx) - if metricsSrv != nil { - _ = metricsSrv.Shutdown(shutdownCtx) - } - }() - + // Graceful shutdown on signal: stop accepting, drain in-flight requests, and let the + // worker record the job it was running as interrupted (#50). + var extra []*http.Server + if metricsSrv != nil { + extra = append(extra, metricsSrv) + } + ln, err := net.Listen("tcp", srv.Addr) + if err != nil { + return fmt.Errorf("serve: listen: %w", err) + } log.Info("serving web ui", "addr", cfg.HTTPAddr, "user", cfg.UserID) - if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { - return fmt.Errorf("serve: %w", err) + if err := serveHTTP(ctx, srv, ln, extra, workerDone, shutdownGrace); err != nil { + return err } return nil } diff --git a/cmd/tapir/shutdown.go b/cmd/tapir/shutdown.go new file mode 100644 index 0000000..94ebd21 --- /dev/null +++ b/cmd/tapir/shutdown.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "os/signal" + "syscall" + "time" +) + +// shutdownGrace bounds each wait at shutdown (in-flight requests, then the worker). Two of +// them stay inside Kubernetes' default 30 s termination grace. +const shutdownGrace = 10 * time.Second + +// shutdownContext is cancelled by SIGINT (a terminal) or SIGTERM (what Kubernetes sends +// on every deploy). Handling only SIGINT meant a deploy killed the process before any +// shutdown code ran (#50). +func shutdownContext(parent context.Context) (context.Context, context.CancelFunc) { + return signal.NotifyContext(parent, os.Interrupt, syscall.SIGTERM) +} + +// serveHTTP serves srv on ln until it fails or ctx ends, then shuts every server down and +// waits for in-flight requests and for workerDone (nil: no worker) to finish, each for at +// most grace. It must not return earlier: Serve returns http.ErrServerClosed as soon as +// Shutdown STARTS, and returning from main kills the drain and the worker's last write. +func serveHTTP(ctx context.Context, srv *http.Server, ln net.Listener, extra []*http.Server, workerDone <-chan struct{}, grace time.Duration) error { + errc := make(chan error, 1) + go func() { errc <- srv.Serve(ln) }() + + select { + case err := <-errc: + if !errors.Is(err, http.ErrServerClosed) { + return fmt.Errorf("serve: %w", err) + } + return nil + case <-ctx.Done(): + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), grace) + defer cancel() + for _, s := range append([]*http.Server{srv}, extra...) { + _ = s.Shutdown(shutdownCtx) + } + if workerDone != nil { + select { + case <-workerDone: + case <-time.After(grace): + } + } + return nil +} diff --git a/cmd/tapir/shutdown_test.go b/cmd/tapir/shutdown_test.go index 458d266..610a782 100644 --- a/cmd/tapir/shutdown_test.go +++ b/cmd/tapir/shutdown_test.go @@ -81,7 +81,9 @@ func TestServeHTTP_GivesUpOnAWorkerThatNeverFinishesAfterTheGrace(t *testing.T) defer srv.Close() ctx, cancel := context.WithCancel(context.Background()) returned := make(chan error, 1) - go func() { returned <- serveHTTP(ctx, srv.Config, srv.Listener, nil, make(chan struct{}), 200*time.Millisecond) }() + go func() { + returned <- serveHTTP(ctx, srv.Config, srv.Listener, nil, make(chan struct{}), 200*time.Millisecond) + }() cancel()