Merge pull request #405 from shutter-network/fix/context-done-service

Graceful shutdown on service errors

Author: ulope

rolling-shutter/chainobserver/observer.go

@@ -88,12 +88,17 @@ func (chainobs *ChainObserver) Observe(ctx context.Context, eventTypes []*events
 	})
 	errorgroup.Go(func() error {
 		for {
-			eventSyncUpdate, err := syncer.Next(errorctx)
-			if err != nil {
-				return err
-			}
-			if err := chainobs.handleEventSyncUpdate(errorctx, eventSyncUpdate); err != nil {
-				return err
+			select {
+			case <-errorctx.Done():
+				return errorctx.Err()
+			default:
+				eventSyncUpdate, err := syncer.Next(errorctx)
+				if err != nil {
+					return err
+				}
+				if err := chainobs.handleEventSyncUpdate(errorctx, eventSyncUpdate); err != nil {
+					return err
+				}
 			}
 		}
 	})

rolling-shutter/keyper/keyper.go

@@ -119,7 +119,6 @@ func (kpr *keyper) Start(ctx context.Context, runner service.Runner) error {
 	if kpr.config.Metrics.Enabled {
 		epochkghandler.InitMetrics()
 		kpr.metricsServer = metricsserver.New(kpr.config.Metrics)
-		runner.Defer(kpr.metricsServer.Shutdown)
 	}
 
 	kpr.dbpool = dbpool

rolling-shutter/medley/metricsserver/metricsserver.go

@@ -22,7 +22,7 @@ func New(config *MetricsConfig) *MetricsServer {
 	return &MetricsServer{config: config, mux: http.NewServeMux()}
 }
 
-func (srv *MetricsServer) Start(_ context.Context, group service.Runner) error {
+func (srv *MetricsServer) Start(ctx context.Context, group service.Runner) error { //nolint:unparam
 	group.Go(func() error {
 		srv.mux.Handle("/metrics", promhttp.Handler())
 		addr := fmt.Sprintf("%s:%d", srv.config.Host, srv.config.Port)
@@ -39,6 +39,11 @@ func (srv *MetricsServer) Start(_ context.Context, group service.Runner) error {
 		}
 		return nil
 	})
+	group.Go(func() error {
+		<-ctx.Done()
+		srv.Shutdown()
+		return ctx.Err()
+	})
 	return nil
 }
 

rolling-shutter/medley/rootcmd/root.go

@@ -162,10 +162,9 @@ func Cmd() *cobra.Command {
 
 func Main(c *cobra.Command) {
 	status := 0
-
 	if err := c.Execute(); err != nil {
+		fmt.Fprintln(os.Stderr, err)
 		status = 1
 	}
-
 	os.Exit(status)
 }

rolling-shutter/snapshot/snapshot.go

@@ -86,14 +86,11 @@ func (snp *Snapshot) Start(ctx context.Context, runner service.Runner) error {
 			return err
 		}
 		snp.metricsServer = metricsserver.New(snp.Config.Metrics)
-		runner.Defer(snp.metricsServer.Shutdown)
 	}
 
 	hub := hubapi.New(snp.Config.SnapshotHubURL)
 	snp.hubapi = hub
 
-	runner.Defer(snp.jrpc.Shutdown)
-
 	snp.setupP2PHandler()
 	return runner.StartService(snp.getServices()...)
 }

rolling-shutter/snapshot/snpjrpc/server.go

@@ -145,7 +145,7 @@ func New(
 	return &jrpc
 }
 
-func (snpjrpc *SnpJRPC) Start(_ context.Context, group service.Runner) error {
+func (snpjrpc *SnpJRPC) Start(ctx context.Context, group service.Runner) error { //nolint:unparam
 	group.Go(func() error {
 		httpServer := snpjrpc.Server.Prepare()
 		log.Info().Str("address", snpjrpc.Server.Host).Msg("Running JSON-RPC server at")
@@ -154,6 +154,11 @@ func (snpjrpc *SnpJRPC) Start(_ context.Context, group service.Runner) error {
 		}
 		return nil
 	})
+	group.Go(func() error {
+		<-ctx.Done()
+		snpjrpc.Shutdown()
+		return ctx.Err()
+	})
 	return nil
 }
 

rolling-shutter/snapshotkeyper/snapshotkeyper.go

@@ -90,7 +90,6 @@ func (snkpr *snapshotkeyper) Start(ctx context.Context, runner service.Runner) e
 	if snkpr.config.Metrics.Enabled {
 		epochkghandler.InitMetrics()
 		snkpr.metricsServer = metricsserver.New(snkpr.config.Metrics)
-		runner.Defer(snkpr.metricsServer.Shutdown)
 	}
 
 	snkpr.dbpool = dbpool