POS-86: Graceful shutdown of Heimdall deamon, REST and Bridge (#753)

* POS-50: Run Rest and Bridge service alongside the Heimdall daemon.

* Added graceful stop for rest server

* New function for starting Bridge with shutdown

* Heimdall, Bridge and REST can shutdown

* Removed loop

* Added graceful stop for rest server

* New function for starting Bridge with shutdown

* Heimdall, Bridge and REST can shutdown

* Change order of service startup and add shutdown to bridge while waiting for sync.

* Bridge process order of shutdown.

* Bridge process order of shutdown.

* Removed err group

* Fixed fork rebase

* Renamed passed ctx variable

* Renamed ctx variable

* Bridge flags fixed

* Bridge flags fixed

* Tidy comments.

Co-authored-by: Sasa Prsic <sasa@ethernal.tech>
Co-authored-by: Nemanja0x <nemanja@ethernal.tech>

bridge/bridge.go

@@ -5,10 +5,14 @@ import (
 	"os"
 
 	"github.com/maticnetwork/heimdall/bridge/cmd"
+	"github.com/maticnetwork/heimdall/helper"
+	"github.com/spf13/viper"
 )
 
 func main() {
-	rootCmd := cmd.BridgeCommands()
+	var logger = helper.Logger.With("module", "bridge/cmd/")
+	rootCmd := cmd.BridgeCommands(viper.GetViper(), logger, "bridge-main")
+
 	if err := rootCmd.Execute(); err != nil {
 		fmt.Println(err)
 		os.Exit(1)

bridge/cmd/root.go

@@ -1,12 +1,14 @@
 package cmd
 
 import (
+	"fmt"
 	"path/filepath"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
 	"github.com/maticnetwork/heimdall/helper"
+	logger "github.com/tendermint/tendermint/libs/log"
 )
 
 const (
@@ -26,58 +28,72 @@ var rootCmd = &cobra.Command{
 }
 
 // BridgeCommands returns command for bridge service
-func BridgeCommands() *cobra.Command {
+func BridgeCommands(v *viper.Viper, loggerInstance logger.Logger, caller string) *cobra.Command {
+	DecorateWithBridgeRootFlags(rootCmd, v, loggerInstance, caller)
 	return rootCmd
 }
 
-// initTendermintViperConfig sets global viper configuration needed to heimdall
-func initTendermintViperConfig(cmd *cobra.Command) {
-	tendermintNode, _ := cmd.Flags().GetString(helper.TendermintNodeFlag)
-	homeValue, _ := cmd.Flags().GetString(helper.HomeFlag)
-	withHeimdallConfigValue, _ := cmd.Flags().GetString(helper.WithHeimdallConfigFlag)
-	bridgeDBValue, _ := cmd.Flags().GetString(bridgeDBFlag)
-	borChainIDValue, _ := cmd.Flags().GetString(borChainIDFlag)
-
-	// bridge-db directory (default storage)
-	if bridgeDBValue == "" {
-		bridgeDBValue = filepath.Join(homeValue, "bridge", "storage")
+// function is called when bridge flags needs to be added to command
+func DecorateWithBridgeRootFlags(cmd *cobra.Command, v *viper.Viper, loggerInstance logger.Logger, caller string) {
+	cmd.PersistentFlags().StringP(helper.TendermintNodeFlag, "n", helper.DefaultTendermintNode, "Node to connect to")
+	if err := v.BindPFlag(helper.TendermintNodeFlag, cmd.PersistentFlags().Lookup(helper.TendermintNodeFlag)); err != nil {
+		loggerInstance.Error(fmt.Sprintf("%v | BindPFlag | %v", caller, helper.TendermintNodeFlag), "Error", err)
 	}
 
-	// set to viper
-	viper.Set(helper.TendermintNodeFlag, tendermintNode)
-	viper.Set(helper.HomeFlag, homeValue)
-	viper.Set(helper.WithHeimdallConfigFlag, withHeimdallConfigValue)
-	viper.Set(bridgeDBFlag, bridgeDBValue)
-	viper.Set(borChainIDFlag, borChainIDValue)
-
-	// start heimdall config
-	helper.InitHeimdallConfig("")
-}
+	cmd.PersistentFlags().String(helper.HomeFlag, helper.DefaultNodeHome, "directory for config and data")
+	if err := v.BindPFlag(helper.HomeFlag, cmd.PersistentFlags().Lookup(helper.HomeFlag)); err != nil {
+		loggerInstance.Error(fmt.Sprintf("%v | BindPFlag | %v", caller, helper.HomeFlag), "Error", err)
+	}
 
-func init() {
-	var logger = helper.Logger.With("module", "bridge/cmd/")
-	rootCmd.PersistentFlags().StringP(helper.TendermintNodeFlag, "n", helper.DefaultTendermintNode, "Node to connect to")
-	rootCmd.PersistentFlags().String(helper.HomeFlag, helper.DefaultNodeHome, "directory for config and data")
-	rootCmd.PersistentFlags().String(
+	cmd.PersistentFlags().String(
 		helper.WithHeimdallConfigFlag,
 		"",
 		"Heimdall config file path (default <home>/config/heimdall-config.json)",
 	)
+	if err := v.BindPFlag(helper.WithHeimdallConfigFlag, cmd.PersistentFlags().Lookup(helper.WithHeimdallConfigFlag)); err != nil {
+		loggerInstance.Error(fmt.Sprintf("%v | BindPFlag | %v", caller, helper.WithHeimdallConfigFlag), "Error", err)
+	}
+
 	// bridge storage db
-	rootCmd.PersistentFlags().String(
+	cmd.PersistentFlags().String(
 		bridgeDBFlag,
 		"",
 		"Bridge db path (default <home>/bridge/storage)",
 	)
+	if err := v.BindPFlag(bridgeDBFlag, cmd.PersistentFlags().Lookup(bridgeDBFlag)); err != nil {
+		loggerInstance.Error(fmt.Sprintf("%v | BindPFlag | %v", caller, bridgeDBFlag), "Error", err)
+	}
+
 	// bridge chain id
-	rootCmd.PersistentFlags().String(
+	cmd.PersistentFlags().String(
 		borChainIDFlag,
 		helper.DefaultBorChainID,
 		"Bor chain id",
 	)
+	if err := v.BindPFlag(borChainIDFlag, cmd.PersistentFlags().Lookup(borChainIDFlag)); err != nil {
+		loggerInstance.Error(fmt.Sprintf("%v | BindPFlag | %v", caller, borChainIDFlag), "Error", err)
+	}
+}
+
+// function is called to set appropriate bridge db path
+func AdjustBridgeDBValue(cmd *cobra.Command, v *viper.Viper) {
+	bridgeDBValue, _ := cmd.Flags().GetString(bridgeDBFlag)
+	homeValue, _ := cmd.Flags().GetString(helper.HomeFlag)
 
-	// bind all flags with viper
-	if err := viper.BindPFlags(rootCmd.Flags()); err != nil {
-		logger.Error("init | BindPFlag | rootCmd.Flags", "Error", err)
+	// bridge-db directory (default storage)
+	if bridgeDBValue == "" {
+		bridgeDBValue = filepath.Join(homeValue, "bridge", "storage")
 	}
+
+	v.Set(bridgeDBFlag, bridgeDBValue)
+}
+
+// initTendermintViperConfig sets global viper configuration needed to heimdall
+func initTendermintViperConfig(cmd *cobra.Command) {
+
+	// set appropriate bridge DB
+	AdjustBridgeDBValue(cmd, viper.GetViper())
+
+	// start heimdall config
+	helper.InitHeimdallConfig("")
 }

bridge/cmd/start.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"os/signal"
@@ -19,6 +20,7 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/tendermint/tendermint/libs/common"
 	httpClient "github.com/tendermint/tendermint/rpc/client"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/maticnetwork/heimdall/helper"
 	"github.com/spf13/viper"
@@ -28,6 +30,105 @@ const (
 	waitDuration = 1 * time.Minute
 )
 
+// StartBridgeWithCtx starts bridge service and is able to shutdow gracefully
+// returns service errors, if any
+func StartBridgeWithCtx(shutdownCtx context.Context) error {
+	var logger = helper.Logger.With("module", "bridge/cmd/")
+
+	// create codec
+	cdc := app.MakeCodec()
+	// queue connector & http client
+	_queueConnector := queue.NewQueueConnector(helper.GetConfig().AmqpURL)
+	_queueConnector.StartWorker()
+
+	_txBroadcaster := broadcaster.NewTxBroadcaster(cdc)
+	_httpClient := httpClient.NewHTTP(helper.GetConfig().TendermintRPCUrl, "/websocket")
+
+	// selected services to start
+	services := []common.Service{}
+	services = append(services,
+		listener.NewListenerService(cdc, _queueConnector, _httpClient),
+		processor.NewProcessorService(cdc, _queueConnector, _httpClient, _txBroadcaster),
+	)
+
+	// Start http client
+	err := _httpClient.Start()
+	if err != nil {
+		logger.Error("Error connecting to server: %v", err)
+		return err
+	}
+
+	// cli context
+	cliCtx := cliContext.NewCLIContext().WithCodec(cdc)
+	cliCtx.BroadcastMode = client.BroadcastAsync
+	cliCtx.TrustNode = true
+
+	// start bridge services only when node fully synced
+	loop := true
+	for loop {
+		select {
+		case <-shutdownCtx.Done():
+			return nil
+		case <-time.After(waitDuration):
+			if !util.IsCatchingUp(cliCtx) {
+				logger.Info("Node up to date, starting bridge services")
+				loop = false
+			} else {
+				logger.Info("Waiting for heimdall to be synced")
+			}
+		}
+	}
+
+	// start services
+	g := new(errgroup.Group)
+	for _, service := range services {
+		// loop variable must be captured
+		srv := service
+		g.Go(func() error {
+			if err := srv.Start(); err != nil {
+				logger.Error("GetStartCmd | serv.Start", "Error", err)
+				return err
+			}
+			<-srv.Quit()
+			return nil
+		})
+	}
+
+	// shutdown phase
+	g.Go(func() error {
+		// wait for interrupt and start shut down
+		<-shutdownCtx.Done()
+
+		logger.Info("Received stop signal - Stopping all services")
+		for _, service := range services {
+			srv := service
+			if srv.IsRunning() {
+				if err := srv.Stop(); err != nil {
+					logger.Error("GetStartCmd | service.Stop", "Error", err)
+					return err
+				}
+			}
+		}
+		// stop http client
+		if err := _httpClient.Stop(); err != nil {
+			logger.Error("GetStartCmd | _httpClient.Stop", "Error", err)
+			return err
+		}
+		// stop db instance
+		util.CloseBridgeDBInstance()
+
+		return nil
+	})
+
+	// wait for all routines to finish and log error
+	if err := g.Wait(); err != nil {
+		logger.Error("Bridge stopped", "err", err)
+		return err
+	}
+
+	return nil
+}
+
 // StartBridge starts bridge service, isStandAlone prevents os.Exit if the bridge started as side service
 func StartBridge(isStandAlone bool) {
 	var logger = helper.Logger.With("module", "bridge/cmd/")
@@ -102,7 +203,7 @@ func StartBridge(isStandAlone bool) {
 		time.Sleep(waitDuration)
 	}
 
-	// strt all processes
+	// start all processes
 	for _, service := range services {
 		go func(serv common.Service) {
 			defer wg.Done()