Shutdown gracefully, close DB connection

2022-11-21 12:14:06 -05:00 · 2022-11-21 12:14:06 -05:00 · cd41d9a419
parent 5f3f7afb90
commit cd41d9a419
7 changed files with 163 additions and 111 deletions
--- a/cmd/root.go
+++ b/cmd/root.go
@ -2,6 +2,7 @@ package cmd

 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"time"
@ -12,13 +13,15 @@ import (
 	"github.com/navidrome/navidrome/log"
 	"github.com/navidrome/navidrome/resources"
 	"github.com/navidrome/navidrome/scheduler"
-	"github.com/oklog/run"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"golang.org/x/sync/errgroup"

 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )

+var interrupted = errors.New("service was interrupted")
+
 var (
 	cfgFile  string
 	noBanner bool
@ -55,118 +58,79 @@ func preRun() {

 func runNavidrome() {
 	db.EnsureLatestVersion()
+	defer func() {
+		if err := db.Close(); err != nil {
+			log.Error("Error closing DB", err)
+		}
+		log.Info("Navidrome stopped gracefully. Bye.")
+	}()

-	var g run.Group
+	g, ctx := errgroup.WithContext(context.Background())
+	g.Go(startServer(ctx))
+	g.Go(startSignaler(ctx))
+	g.Go(startScheduler(ctx))
+	g.Go(schedulePeriodicScan(ctx))

-	g.Add(startServer())
-	g.Add(startSignaler())
-	g.Add(startScheduler())
-
-	schedule := conf.Server.ScanSchedule
-	if schedule != "" {
-		go schedulePeriodicScan(schedule)
-	} else {
-		log.Warn("Periodic scan is DISABLED")
-	}
-
-	if err := g.Run(); err != nil {
+	if err := g.Wait(); err != nil && !errors.Is(err, interrupted) {
 		log.Error("Fatal error in Navidrome. Aborting", err)
-		os.Exit(1)
 	}
 }

-func startServer() (func() error, func(err error)) {
+func startServer(ctx context.Context) func() error {
 	return func() error {
-			a := CreateServer(conf.Server.MusicFolder)
-			a.MountRouter("Native API", consts.URLPathNativeAPI, CreateNativeAPIRouter())
-			a.MountRouter("Subsonic API", consts.URLPathSubsonicAPI, CreateSubsonicAPIRouter())
-			if conf.Server.LastFM.Enabled {
-				a.MountRouter("LastFM Auth", consts.URLPathNativeAPI+"/lastfm", CreateLastFMRouter())
-			}
-			if conf.Server.ListenBrainz.Enabled {
-				a.MountRouter("ListenBrainz Auth", consts.URLPathNativeAPI+"/listenbrainz", CreateListenBrainzRouter())
-			}
-			if conf.Server.Prometheus.Enabled {
-				a.MountRouter("Prometheus metrics", conf.Server.Prometheus.MetricsPath, promhttp.Handler())
-			}
-			return a.Run(fmt.Sprintf("%s:%d", conf.Server.Address, conf.Server.Port))
-		}, func(err error) {
-			if err != nil {
-				log.Error("Shutting down Server due to error", err)
-			} else {
-				log.Info("Shutting down Server")
-			}
+		a := CreateServer(conf.Server.MusicFolder)
+		a.MountRouter("Native API", consts.URLPathNativeAPI, CreateNativeAPIRouter())
+		a.MountRouter("Subsonic API", consts.URLPathSubsonicAPI, CreateSubsonicAPIRouter())
+		if conf.Server.LastFM.Enabled {
+			a.MountRouter("LastFM Auth", consts.URLPathNativeAPI+"/lastfm", CreateLastFMRouter())
 		}
-}
-
-var sigChan = make(chan os.Signal, 1)
-
-func startSignaler() (func() error, func(err error)) {
-	scanner := GetScanner()
-	ctx, cancel := context.WithCancel(context.Background())
-
-	return func() error {
-			for {
-				select {
-				case sig := <-sigChan:
-					log.Info(ctx, "Received signal, triggering a new scan", "signal", sig)
-					start := time.Now()
-					err := scanner.RescanAll(ctx, false)
-					if err != nil {
-						log.Error(ctx, "Error scanning", err)
-					}
-					log.Info(ctx, "Triggered scan complete", "elapsed", time.Since(start).Round(100*time.Millisecond))
-				case <-ctx.Done():
-					return nil
-				}
-			}
-		}, func(err error) {
-			cancel()
-			if err != nil {
-				log.Error("Shutting down Signaler due to error", err)
-			} else {
-				log.Info("Shutting down Signaler")
-			}
+		if conf.Server.ListenBrainz.Enabled {
+			a.MountRouter("ListenBrainz Auth", consts.URLPathNativeAPI+"/listenbrainz", CreateListenBrainzRouter())
 		}
+		if conf.Server.Prometheus.Enabled {
+			a.MountRouter("Prometheus metrics", conf.Server.Prometheus.MetricsPath, promhttp.Handler())
+		}
+		return a.Run(ctx, fmt.Sprintf("%s:%d", conf.Server.Address, conf.Server.Port))
+	}
 }

-func schedulePeriodicScan(schedule string) {
-	scanner := GetScanner()
-	schedulerInstance := scheduler.GetInstance()
-
-	log.Info("Scheduling periodic scan", "schedule", schedule)
-	err := schedulerInstance.Add(schedule, func() {
-		_ = scanner.RescanAll(context.Background(), false)
-	})
-	if err != nil {
-		log.Error("Error scheduling periodic scan", err)
-	}
-
-	time.Sleep(2 * time.Second) // Wait 2 seconds before the initial scan
-	log.Debug("Executing initial scan")
-	if err := scanner.RescanAll(context.Background(), false); err != nil {
-		log.Error("Error executing initial scan", err)
-	}
-	log.Debug("Finished initial scan")
-}
-
-func startScheduler() (func() error, func(err error)) {
-	log.Info("Starting scheduler")
-	schedulerInstance := scheduler.GetInstance()
-	ctx, cancel := context.WithCancel(context.Background())
-
+func schedulePeriodicScan(ctx context.Context) func() error {
 	return func() error {
-			schedulerInstance.Run(ctx)
-
+		schedule := conf.Server.ScanSchedule
+		if schedule == "" {
+			log.Warn("Periodic scan is DISABLED")
 			return nil
-		}, func(err error) {
-			cancel()
-			if err != nil {
-				log.Error("Shutting down Scheduler due to error", err)
-			} else {
-				log.Info("Shutting down Scheduler")
-			}
 		}
+
+		scanner := GetScanner()
+		schedulerInstance := scheduler.GetInstance()
+
+		log.Info("Scheduling periodic scan", "schedule", schedule)
+		err := schedulerInstance.Add(schedule, func() {
+			_ = scanner.RescanAll(ctx, false)
+		})
+		if err != nil {
+			log.Error("Error scheduling periodic scan", err)
+		}
+
+		time.Sleep(2 * time.Second) // Wait 2 seconds before the initial scan
+		log.Debug("Executing initial scan")
+		if err := scanner.RescanAll(ctx, false); err != nil {
+			log.Error("Error executing initial scan", err)
+		}
+		log.Debug("Finished initial scan")
+		return nil
+	}
+}
+
+func startScheduler(ctx context.Context) func() error {
+	log.Info(ctx, "Starting scheduler")
+	schedulerInstance := scheduler.GetInstance()
+
+	return func() error {
+		schedulerInstance.Run(ctx)
+		return nil
+	}
 }

 // TODO: Implement some struct tags to map flags to viper
--- a/cmd/signaler_nonunix.go
+++ b/cmd/signaler_nonunix.go
@ -0,0 +1,26 @@
+//go:build !unix
+
+package cmd
+
+import (
+	"context"
+	"os"
+	"os/signal"
+
+	"github.com/navidrome/navidrome/log"
+)
+
+func startSignaler(ctx context.Context) func() error {
+	log.Info(ctx, "Starting signaler")
+
+	return func() error {
+		var sigChan = make(chan os.Signal, 1)
+		signal.Notify(sigChan, os.Interrupt)
+		select {
+		case <-sigChan:
+			return interrupted
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
--- a/cmd/signaler_unix.go
+++ b/cmd/signaler_unix.go
@ -1,16 +1,50 @@
-//go:build !windows && !plan9
+//go:build unix

 package cmd

 import (
+	"context"
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
+
+	"github.com/navidrome/navidrome/log"
 )

-func init() {
-	signals := []os.Signal{
-		syscall.SIGUSR1,
+const triggerScanSignal = syscall.SIGUSR1
+
+func startSignaler(ctx context.Context) func() error {
+	log.Info(ctx, "Starting signaler")
+	scanner := GetScanner()
+
+	return func() error {
+		var sigChan = make(chan os.Signal, 1)
+		signal.Notify(
+			sigChan,
+			os.Interrupt,
+			triggerScanSignal,
+			syscall.SIGHUP,
+			syscall.SIGTERM,
+			syscall.SIGABRT,
+		)
+
+		for {
+			select {
+			case sig := <-sigChan:
+				if sig != triggerScanSignal {
+					return interrupted
+				}
+				log.Info(ctx, "Received signal, triggering a new scan", "signal", sig)
+				start := time.Now()
+				err := scanner.RescanAll(ctx, false)
+				if err != nil {
+					log.Error(ctx, "Error scanning", err)
+				}
+				log.Info(ctx, "Triggered scan complete", "elapsed", time.Since(start).Round(100*time.Millisecond))
+			case <-ctx.Done():
+				return nil
+			}
+		}
 	}
-	signal.Notify(sigChan, signals...)
 }
--- a/db/db.go
+++ b/db/db.go
@ -34,6 +34,11 @@ func Db() *sql.DB {
 	})
 }

+func Close() error {
+	log.Info("Closing Database")
+	return Db().Close()
+}
+
 func EnsureLatestVersion() {
 	db := Db()

--- a/go.mod
+++ b/go.mod
@ -32,7 +32,6 @@ require (
 	github.com/mattn/go-zglob v0.0.3
 	github.com/microcosm-cc/bluemonday v1.0.20
 	github.com/mileusna/useragent v1.2.1
-	github.com/oklog/run v1.1.0
 	github.com/onsi/ginkgo/v2 v2.2.0
 	github.com/onsi/gomega v1.20.2
 	github.com/pressly/goose v2.7.0+incompatible
@ -45,6 +44,7 @@ require (
 	github.com/unrolled/secure v1.13.0
 	github.com/xrash/smetrics v0.0.0-20200730060457-89a2a8a1fb0b
 	golang.org/x/image v0.0.0-20191009234506-e7c1f5e7dbb8
+	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
 	golang.org/x/text v0.3.7
 	golang.org/x/tools v0.1.12
 )
@ -229,7 +229,6 @@ require (
 	golang.org/x/exp/typeparams v0.0.0-20220613132600-b0d781184e0d // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
 	golang.org/x/net v0.0.0-20220826154423-83b083e8dc8b // indirect
-	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4 // indirect
 	golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
--- a/go.sum
+++ b/go.sum
@ -503,8 +503,6 @@ github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
 github.com/ogier/pflag v0.0.1 h1:RW6JSWSu/RkSatfcLtogGfFgpim5p7ARQ10ECk5O750=
 github.com/ogier/pflag v0.0.1/go.mod h1:zkFki7tvTa0tafRvTBIZTvzYyAu6kQhPZFnshFFPE+g=
-github.com/oklog/run v1.1.0 h1:GEenZ1cK0+q0+wsJew9qUg/DyD8k3JzYsZAi5gYi2mA=
-github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DVU=
 github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
 github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
--- a/server/server.go
+++ b/server/server.go
@ -1,6 +1,8 @@
 package server

 import (
+	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"path"
@ -42,16 +44,40 @@ func (s *Server) MountRouter(description, urlPath string, subRouter http.Handler

 var startTime = time.Now()

-func (s *Server) Run(addr string) error {
+func (s *Server) Run(ctx context.Context, addr string) error {
 	s.MountRouter("WebUI", consts.URLPathUI, s.frontendAssetsHandler())
-	log.Info("Navidrome server is ready!", "address", addr, "startupTime", time.Since(startTime))
 	server := &http.Server{
 		Addr:              addr,
 		ReadHeaderTimeout: consts.ServerReadHeaderTimeout,
 		Handler:           s.router,
 	}

-	return server.ListenAndServe()
+	// Start HTTP server in its own goroutine, send a signal (errC) if failed to start
+	errC := make(chan error)
+	go func() {
+		if err := server.ListenAndServe(); err != http.ErrServerClosed {
+			log.Error(ctx, "Could not start server. Aborting", err)
+			errC <- err
+		}
+	}()
+
+	log.Info(ctx, "Navidrome server is ready!", "address", addr, "startupTime", time.Since(startTime))
+
+	// Wait for a signal to terminate (or an error during startup)
+	select {
+	case err := <-errC:
+		return err
+	case <-ctx.Done():
+	}
+
+	// Try to stop the HTTP server gracefully
+	log.Info(ctx, "Stopping HTTP server")
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	if err := server.Shutdown(ctx); err != nil && !errors.Is(err, context.DeadlineExceeded) {
+		log.Error(ctx, "Unexpected error in http.Shutdown()", err)
+	}
+	return nil
 }

 func (s *Server) initRoutes() {