fix(serve): handle SIGTERM and wait for in-flight requests and the worker before exiting (#50)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
+17
-16
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user