Merge branch 'main' into nonces-requeuing

cmd/blobstream/base/config.go

@@ -85,6 +85,9 @@ const (
 
 	FlagLogLevel  = "log.level"
 	FlagLogFormat = "log.format"
+
+	FlagBackupRelayer         = "relayer.backup"
+	FlagBackupRelayerWaitTime = "relayer.wait-time"
 )
 
 func AddLogLevelFlag(cmd *cobra.Command) {
@@ -271,6 +274,40 @@ func GetEVMGasLimitFlag(cmd *cobra.Command) (uint64, bool, error) {
 	return val, changed, nil
 }
 
+func AddBackupRelayerFlag(cmd *cobra.Command) {
+	cmd.Flags().Bool(
+		FlagBackupRelayer,
+		false,
+		"Set the relayer to be a backup, i.e. not relay attestations until the `relayer.wait-time` is elapsed and no primary relayer has relayed any",
+	)
+}
+
+func GetBackupRelayerFlag(cmd *cobra.Command) (bool, bool, error) {
+	changed := cmd.Flags().Changed(FlagBackupRelayer)
+	val, err := cmd.Flags().GetBool(FlagBackupRelayer)
+	if err != nil {
+		return false, changed, err
+	}
+	return val, changed, nil
+}
+
+func AddBackupRelayerWaitTimeFlag(cmd *cobra.Command) {
+	cmd.Flags().Uint64(
+		FlagBackupRelayerWaitTime,
+		15,
+		"The wait time, in minutes, to wait for primary relayers to relay attestations before proceeding to relay them",
+	)
+}
+
+func GetBackupRelayerWaitTimeFlag(cmd *cobra.Command) (uint64, bool, error) {
+	changed := cmd.Flags().Changed(FlagBackupRelayerWaitTime)
+	val, err := cmd.Flags().GetUint64(FlagBackupRelayerWaitTime)
+	if err != nil {
+		return 0, changed, err
+	}
+	return val, changed, nil
+}
+
 func AddHomeFlag(cmd *cobra.Command, serviceName string, defaultHomeDir string) {
 	cmd.Flags().String(FlagHome, defaultHomeDir, fmt.Sprintf("The Blobstream %s home directory", serviceName))
 }

cmd/blobstream/orchestrator/cmd.go

@@ -75,22 +75,30 @@ func Start() *cobra.Command {
 			defer cancel()
 
 			stopFuncs := make([]func() error, 0)
+			defer func() {
+				for _, f := range stopFuncs {
+					err := f()
+					if err != nil {
+						logger.Error(err.Error())
+					}
+				}
+			}()
 
-			tmQuerier, appQuerier, stops, err := common.NewTmAndAppQuerier(logger, config.CoreRPC, config.CoreGRPC, config.GRPCInsecure)
-			stopFuncs = append(stopFuncs, stops...)
+			tmQuerier, appQuerier, storeStops, err := common.NewTmAndAppQuerier(logger, config.CoreRPC, config.CoreGRPC, config.GRPCInsecure)
+			stopFuncs = append(stopFuncs, storeStops...)
 			if err != nil {
 				return err
 			}
 
-			s, stops, err := common.OpenStore(logger, config.Home, store.OpenOptions{
+			s, storeStops, err := common.OpenStore(logger, config.Home, store.OpenOptions{
 				HasDataStore:      true,
 				BadgerOptions:     store.DefaultBadgerOptions(config.Home),
 				HasSignatureStore: false,
 				HasEVMKeyStore:    true,
 				HasP2PKeyStore:    true,
 			})
-			stopFuncs = append(stopFuncs, stops...)
 			if err != nil {
+				stopFuncs = append(stopFuncs, storeStops...)
 				return err
 			}
 
@@ -110,25 +118,14 @@ func Start() *cobra.Command {
 				return err
 			}
 			stopFuncs = append(stopFuncs, func() error { return dht.Close() })
+			stopFuncs = append(stopFuncs, storeStops...)
 
 			// creating the p2p querier
 			p2pQuerier := p2p.NewQuerier(dht, logger)
 			retrier := helpers.NewRetrier(logger, 5, 30*time.Second)
 
-			defer func() {
-				for _, f := range stopFuncs {
-					err := f()
-					if err != nil {
-						logger.Error(err.Error())
-					}
-				}
-			}()
-
 			// creating the broadcaster
 			broadcaster := orchestrator.NewBroadcaster(p2pQuerier.BlobstreamDHT)
-			if err != nil {
-				return err
-			}
 
 			// creating the orchestrator
 			orch := orchestrator.New(